Repository: MatterMiners/lapis

Issue:
Simulation doesn't await job execution after last scheduling cycle
I am under the impression that after the last update the simulation finishes after when the job queue is empty and all jobs are scheduled to drones but does not await the execution of all jobs in the drones job queue.

You can reproduce this by running lapis with the attached input files.

The simulation stops with the last scheduling cycle at 180 instead of processing the second job with a walltime of 1000.

I would expect the simulation to finish running all jobs.

Cluster input

    TotalSlotCPUs TotalSlotDisk TotalSlotMemory Count sitename
    1 44624348.0 8000 1 site1

Job Input

    [
    {
    "QDate": 1567155456,
    "RequestCpus": 1,
    "RequestWalltime": 60,
    "RequestMemory": 2000,
    "RequestDisk": 6000000,
    "RemoteWallClockTime": 100.0,
    "'Number of Allocated Processors'": 1,
    "MemoryUsage": 2867,
    "DiskUsage_RAW": 41898,
    "RemoteSysCpu": 10.0,
    "RemoteUserCpu": 40.0,
    "Inputfiles": {
    "a.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "b.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "c.root": {
    "filesize": 25,
    "usedsize": 20
    }
    }
    },
    {
    "QDate": 1567155456,
    "RequestCpus": 1,
    "RequestWalltime": 60,
    "RequestMemory": 2000,
    "RequestDisk": 6000000,
    "RemoteWallClockTime": 77.0,
    "'Number of Allocated Processors'": 1,
    "MemoryUsage": 1207,
    "DiskUsage_RAW": 45562,
    "RemoteSysCpu": 7.0,
    "RemoteUserCpu": 50.0,
    "Inputfiles": {
    "a.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "b.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "c.root": {
    "filesize": 25,
    "usedsize": 20
    }
    }
    },
    {
    "QDate": 1567155456,
    "RequestCpus": 1,
    "RequestWalltime": 60,
    "RequestMemory": 2000,
    "RequestDisk": 6000000,
    "RemoteWallClockTime": 63.0,
    "'Number of Allocated Processors'": 1,
    "MemoryUsage": 1207,
    "DiskUsage_RAW": 45562,
    "RemoteSysCpu": 3.0,
    "RemoteUserCpu": 40.0,
    "Inputfiles": {
    "a.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "b.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "c.root": {
    "filesize": 25,
    "usedsize": 20
    }
    }
    },
    {
    "QDate": 1567155456,
    "RequestCpus": 1,
    "RequestWalltime": 60,
    "RequestMemory": 2000,
    "RequestDisk": 6000000,
    "RemoteWallClockTime": 1000.0,
    "'Number of Allocated Processors'": 1,
    "MemoryUsage": 1207,
    "DiskUsage_RAW": 45562,
    "RemoteSysCpu": 7.0,
    "RemoteUserCpu": 40.3,
    "Inputfiles": {
    "a.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "b.root": {
    "filesize": 25,
    "usedsize": 20
    },
    "c.root": {
    "filesize": 25,
    "usedsize": 20
    }
    }
    }
    ]


Discussion:

Patch:
diff --git a/lapis/drone.py b/lapis/drone.py
index 9826c78..7126242 100644
--- a/lapis/drone.py
+++ b/lapis/drone.py
@@ -154,9 +154,8 @@ 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)
+            job.drone = None
+            await self.scheduler.job_finished(job)
             self._utilisation = self._allocation = None
             self.scheduler.update_drone(self)
             await sampling_required.put(self)
diff --git a/lapis/scheduler.py b/lapis/scheduler.py
index 7e9a42d..dc564b9 100644
--- a/lapis/scheduler.py
+++ b/lapis/scheduler.py
@@ -1,5 +1,5 @@
 from typing import Dict
-from usim import Scope, interval
+from usim import Scope, interval, Resources
 
 from lapis.drone import Drone
 from lapis.monitor import sampling_required
@@ -32,6 +32,7 @@ class CondorJobScheduler(object):
         self.interval = 60
         self.job_queue = JobQueue()
         self._collecting = True
+        self._processing = Resources(jobs=0)
 
     @property
     def drone_list(self):
@@ -100,19 +101,27 @@ class CondorJobScheduler(object):
                             for key, value in left_resources.items()
                         }
                         self._add_drone(best_match, left_resources)
-                if not self._collecting and not self.job_queue:
+                if (
+                    not self._collecting
+                    and not self.job_queue
+                    and self._processing.levels.jobs == 0
+                ):
                     break
                 await sampling_required.put(self)
 
     async def _collect_jobs(self):
         async for job in self._stream_queue:
             self.job_queue.append(job)
+            await self._processing.increase(jobs=1)
             # TODO: logging happens with each job
             await sampling_required.put(self.job_queue)
         self._collecting = False
 
-    async def retry_job(self, job):
-        await self._stream_queue.put(job)
+    async def job_finished(self, job):
+        if job.successful:
+            await self._processing.decrease(jobs=1)
+        else:
+            await self._stream_queue.put(job)
 
     def _schedule_job(self, job) -> Drone:
         priorities = {}
diff --git a/lapis/simulator.py b/lapis/simulator.py
index 9689ae2..2920202 100644
--- a/lapis/simulator.py
+++ b/lapis/simulator.py
@@ -32,6 +32,7 @@ class Simulator(object):
         self.cost = 0
         self._job_generators = []
         self.monitoring = None
+        self.duration = None
         self.enable_monitoring()
 
     def enable_monitoring(self):
@@ -77,7 +78,8 @@ class Simulator(object):
             for controller in self.controllers:
                 while_running.do(controller.run(), volatile=True)
             while_running.do(self.monitoring.run(), volatile=True)
-        print(f"Finished simulation at {time.now}")
+        self.duration = time.now
+        print(f"Finished simulation at {self.duration}")
 
     async def _queue_jobs(self, job_input, job_reader):
         await job_to_queue_scheduler(


Test patch:
diff --git a/lapis_tests/test_simulator.py b/lapis_tests/test_simulator.py
new file mode 100644
index 0000000..4875d6b
--- /dev/null
+++ b/lapis_tests/test_simulator.py
@@ -0,0 +1,40 @@
+from tempfile import NamedTemporaryFile
+
+from lapis.job_io.htcondor import htcondor_job_reader
+from lapis.pool import StaticPool
+from lapis.pool_io.htcondor import htcondor_pool_reader
+from lapis.scheduler import CondorJobScheduler
+from lapis.simulator import Simulator
+
+
+class TestSimulator(object):
+    def test_simulation_exit(self):
+        simulator = Simulator()
+        with NamedTemporaryFile(suffix=".csv") as machine_config, NamedTemporaryFile(
+            suffix=".csv"
+        ) as job_config:
+            with open(machine_config.name, "w") as write_stream:
+                write_stream.write(
+                    "TotalSlotCPUs TotalSlotDisk TotalSlotMemory Count\n"
+                    "1 44624348.0 8000 1"
+                )
+            with open(job_config.name, "w") as write_stream:
+                write_stream.write(
+                    "QDate RequestCpus RequestWalltime RequestMemory RequestDisk "
+                    "RemoteWallClockTime MemoryUsage DiskUsage_RAW RemoteSysCpu "
+                    "RemoteUserCpu\n"
+                    "1567155456 1 60 2000 6000000 100.0 2867 41898 10.0 40.0"
+                )
+            job_input = open(job_config.name, "r+")
+            machine_input = open(machine_config.name, "r+")
+            simulator.create_job_generator(
+                job_input=job_input, job_reader=htcondor_job_reader
+            )
+            simulator.create_scheduler(scheduler_type=CondorJobScheduler)
+            simulator.create_pools(
+                pool_input=machine_input,
+                pool_reader=htcondor_pool_reader,
+                pool_type=StaticPool,
+            )
+            simulator.run()
+            assert 180 == simulator.duration