Merge pull request #69 from MatterMiners/fix/runningjobs-55

Drone as a requirement to run a job

lapis/drone.py

@@ -127,7 +127,7 @@ async def _run_job(self, job: Job, kill: bool):
 
             self._utilisation = self._allocation = None
 
-            job_execution = scope.do(job.run())
+            job_execution = scope.do(job.run(self))
             self.jobs += 1
             try:
                 async with self.resources.claim(
@@ -154,7 +154,6 @@ async def _run_job(self, job: Job, kill: bool):
                 await instant
                 job_execution.cancel()
             self.jobs -= 1
-            job.drone = None
             await self.scheduler.job_finished(job)
             self._utilisation = self._allocation = None
             self.scheduler.update_drone(self)

lapis/job.py

@@ -91,18 +91,23 @@ def waiting_time(self) -> float:
             return self.in_queue_until - self.in_queue_since
         return float("Inf")
 
-    async def run(self):
+    async def run(self, drone: "Drone"):
+        assert drone, "Jobs cannot run without a drone being assigned"
+        self.drone = drone
         self.in_queue_until = time.now
         self._success = None
         await sampling_required.put(self)
         try:
             await (time + self.walltime)
         except CancelTask:
+            self.drone = None
             self._success = False
         except BaseException:
+            self.drone = None
             self._success = False
             raise
         else:
+            self.drone = None
             self._success = True
         await sampling_required.put(self)
 

lapis_tests/__init__.py

@@ -43,3 +43,7 @@ def unregister_drone(drone: Drone):
     @staticmethod
     def update_drone(drone: Drone):
         pass
+
+
+class DummyDrone:
+    pass

lapis_tests/test_job.py

@@ -3,7 +3,7 @@
 
 from lapis.drone import Drone
 from lapis.job import Job
-from lapis_tests import via_usim, DummyScheduler
+from lapis_tests import via_usim, DummyScheduler, DummyDrone
 
 
 class TestJob(object):
@@ -27,10 +27,11 @@ def test_name(self):
 
     @via_usim
     async def test_run_job(self):
+        drone = DummyDrone()
         job = Job(resources={"walltime": 50}, used_resources={"walltime": 10})
         assert float("inf") == job.waiting_time
         async with Scope() as scope:
-            scope.do(job.run())
+            scope.do(job.run(drone))
         assert 10 == time
         assert 0 == job.waiting_time
         assert job.successful