Repository: MatterMiners/lapis

Issue:
Job transfers seem to interrupt scheduling cycle
@tfesenbecker reported there seems to be an issue with the scheduling cycle when files are transferred from within jobs. As jobs should be absolutely independent from the scheduler itself, this should not happen at all. So we should ensure that file transfer can last longer than the scheduling cycle without any negative influence on the environment including scheduler, pools, or drones.

Discussion:

Patch:
diff --git a/lapis/drone.py b/lapis/drone.py
index b6d9cab..9826c78 100644
--- a/lapis/drone.py
+++ b/lapis/drone.py
@@ -1,5 +1,5 @@
 from cobald import interfaces
-from usim import time, Scope, instant, Capacities, ResourcesUnavailable
+from usim import time, Scope, instant, Capacities, ResourcesUnavailable, Queue
 
 from lapis.job import Job
 
@@ -44,6 +44,7 @@ class Drone(interfaces.Pool):
         self.jobs = 0
         self._allocation = None
         self._utilisation = None
+        self._job_queue = Queue()
 
     @property
     def theoretical_available_resources(self):
@@ -60,6 +61,9 @@ class Drone(interfaces.Pool):
         self._supply = 1
         self.scheduler.register_drone(self)
         await sampling_required.put(self)
+        async with Scope() as scope:
+            async for job, kill in self._job_queue:
+                scope.do(self._run_job(job=job, kill=kill))
 
     @property
     def supply(self) -> float:
@@ -103,7 +107,10 @@ class Drone(interfaces.Pool):
         await sampling_required.put(self)  # TODO: introduce state of drone
         await (time + 1)
 
-    async def start_job(self, job: Job, kill: bool = False):
+    async def schedule_job(self, job: Job, kill: bool = False):
+        await self._job_queue.put((job, kill))
+
+    async def _run_job(self, job: Job, kill: bool):
         """
         Method manages to start a job in the context of the given drone.
         The job is started independent of available resources. If resources of
@@ -147,6 +154,9 @@ class Drone(interfaces.Pool):
                 await instant
                 job_execution.cancel()
             self.jobs -= 1
+            if not job.successful:
+                job.drone = None
+                await self.scheduler.retry_job(job)
             self._utilisation = self._allocation = None
             self.scheduler.update_drone(self)
             await sampling_required.put(self)
diff --git a/lapis/pool.py b/lapis/pool.py
index 928b710..2a3e465 100644
--- a/lapis/pool.py
+++ b/lapis/pool.py
@@ -29,20 +29,21 @@ class Pool(interfaces.Pool):
         assert init <= capacity
         self.make_drone = make_drone
         self._drones = []
-        self.init_pool(init=init)
         self._demand = 1
         self._level = init
         self._capacity = capacity
         self._name = name
 
-    def init_pool(self, init: int = 0):
+    async def init_pool(self, scope: Scope, init: int = 0):
         """
         Initialisation of existing drones at creation time of pool.
 
         :param init: Number of drones to create.
         """
         for _ in range(init):
-            self._drones.append(self.make_drone(0))
+            drone = self.make_drone(0)
+            scope.do(drone.run())
+            self._drones.append(drone)
 
     # TODO: the run method currently needs to be called manually
     async def run(self):
@@ -52,6 +53,7 @@ class Pool(interfaces.Pool):
         initialising new drones. Otherwise drones get removed.
         """
         async with Scope() as scope:
+            await self.init_pool(scope=scope, init=self._level)
             async for _ in interval(1):
                 drones_required = min(self._demand, self._capacity) - self._level
                 while drones_required > 0:
@@ -145,5 +147,6 @@ class StaticPool(Pool):
         """
         Pool runs forever and does not check if number of drones needs to be adapted.
         """
-        while True:
+        async with Scope() as scope:
+            await self.init_pool(scope=scope, init=self._level)
             await eternity
diff --git a/lapis/scheduler.py b/lapis/scheduler.py
index c3ad63e..7e9a42d 100644
--- a/lapis/scheduler.py
+++ b/lapis/scheduler.py
@@ -90,7 +90,7 @@ class CondorJobScheduler(object):
                 for job in self.job_queue:
                     best_match = self._schedule_job(job)
                     if best_match:
-                        scope.do(best_match.start_job(job))
+                        await best_match.schedule_job(job)
                         self.job_queue.remove(job)
                         await sampling_required.put(self.job_queue)
                         self.unregister_drone(best_match)
@@ -111,6 +111,9 @@ class CondorJobScheduler(object):
             await sampling_required.put(self.job_queue)
         self._collecting = False
 
+    async def retry_job(self, job):
+        await self._stream_queue.put(job)
+
     def _schedule_job(self, job) -> Drone:
         priorities = {}
         for cluster in self.drone_cluster:


Test patch:
diff --git a/lapis_tests/test_job.py b/lapis_tests/test_job.py
index 13b4090..3c75916 100644
--- a/lapis_tests/test_job.py
+++ b/lapis_tests/test_job.py
@@ -47,9 +47,10 @@ class TestJob(object):
             pool_resources={"cores": 1, "memory": 1},
             scheduling_duration=0,
         )
+        await drone.run()
         async with Scope() as scope:
-            scope.do(drone.start_job(job=job))
-        assert 10 == time
+            scope.do(drone.schedule_job(job=job))
+        assert 10 == time.now
         assert 0 == job.waiting_time
         assert job.successful
 
@@ -65,8 +66,9 @@ class TestJob(object):
             pool_resources={"cores": 1, "memory": 1},
             scheduling_duration=0,
         )
+        await drone.run()
         async with Scope() as scope:
-            scope.do(drone.start_job(job=job))
+            scope.do(drone.schedule_job(job=job))
         assert 0 == time
         assert not job.successful
         assert 0 == job.waiting_time
@@ -87,9 +89,10 @@ class TestJob(object):
             pool_resources={"cores": 1, "memory": 1},
             scheduling_duration=0,
         )
+        await drone.run()
         async with Scope() as scope:
-            scope.do(drone.start_job(job=job_one))
-            scope.do(drone.start_job(job=job_two))
+            scope.do(drone.schedule_job(job=job_one))
+            scope.do(drone.schedule_job(job=job_two))
         assert 10 == time
         assert job_one.successful
         assert not job_two.successful
@@ -112,9 +115,10 @@ class TestJob(object):
             pool_resources={"cores": 2, "memory": 2},
             scheduling_duration=0,
         )
+        await drone.run()
         async with Scope() as scope:
-            scope.do(drone.start_job(job=job_one))
-            scope.do(drone.start_job(job=job_two))
+            scope.do(drone.schedule_job(job=job_one))
+            scope.do(drone.schedule_job(job=job_two))
         assert 10 == time
         assert job_one.successful
         assert job_two.successful