Repository: virtool/virtool
Issue: Jobs not releasing resources reliably

**Describe the bug**
A clear and concise description of what the bug is.

**To Reproduce**
Steps to reproduce the behavior:
1. Run an instance for an extended period of time.
2. Start and complete a large number of jobs.
3. Eventually all jobs are waiting.

**Expected behavior**
Jobs should not wait endlessly when all resources are free.

**Server**
 - Distro: Debian 9.4
 - Version: v2.2.3

**Client**
 - OS: Linux Mint 18.3
 - Browser: Chrome 66


Changed files:
virtool/job_manager.py, tests/test_app.py, virtool/app.py, tests/test_job_manager.py

diff --git a/tests/test_app.py b/tests/test_app.py
index 1b4b5b571..9fffb2a08 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -112,6 +112,7 @@ async def test_init_job_manager(mocker, loop):
     app = web.Application(loop=loop)
 
     app["db"] = None
+    app["process_executor"] = mocker.MagicMock()
     app["settings"] = None
     app["dispatcher"] = mocker.MagicMock()
 
@@ -120,6 +121,7 @@ async def test_init_job_manager(mocker, loop):
     assert isinstance(app["job_manager"], virtool.job_manager.Manager)
 
     assert app["job_manager"].loop == loop
+    assert app["job_manager"].executor == app["process_executor"]
     assert app["job_manager"].db is None
     assert app["job_manager"].settings is None
     assert app["job_manager"].dispatch == app["dispatcher"].dispatch
diff --git a/tests/test_job_manager.py b/tests/test_job_manager.py
deleted file mode 100644
index ddfd16919..000000000
--- a/tests/test_job_manager.py
+++ /dev/null
@@ -1,362 +0,0 @@
-import pytest
-import asyncio
-import virtool.job
-import virtool.job_manager
-
-
-@pytest.fixture
-def test_job_manager(loop, mocker, tmpdir, test_motor, test_dispatch):
-    settings = {
-        "proc": 6,
-        "mem": 24,
-        "dummy_proc": 2,
-        "dummy_mem": 4,
-        "data_path": str(tmpdir)
-    }
-
-    tmpdir.mkdir("logs").mkdir("jobs")
-
-    manager = virtool.job_manager.Manager(loop, test_motor, settings, test_dispatch, mocker.stub("capture_exception"))
-
-    manager.start()
-
-    yield manager
-
-
-class TestStarted:
-
-    async def test(self, test_job_manager):
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 0,
-                "mem": 0
-            },
-            "available": {
-                "proc": 6,
-                "mem": 24
-            }
-        }
-
-        assert test_job_manager.started
-
-
-class TestNew:
-
-    @pytest.mark.parametrize("use_executor,result", [
-        (True, "I ran in an executor. My message is hello world"),
-        (False, "I didn't run in an executor. My message is hello world")
-    ])
-    async def test(self, loop, use_executor, result, test_job_manager, test_random_alphanumeric, static_time):
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 0,
-                "mem": 0
-            },
-            "available": {
-                "proc": 6,
-                "mem": 24
-            }
-        }
-
-        target = list()
-
-        await test_job_manager.new("dummy", {
-            "message": "hello world",
-            "target": target,
-            "use_executor": use_executor
-        }, "test")
-
-        await asyncio.sleep(5, loop=loop)
-
-        assert await test_job_manager.db.jobs.find_one() == {
-            "_id": test_random_alphanumeric.last_choice,
-            "args": {
-                "message": "hello world",
-                "target": [],
-                "use_executor": use_executor
-            },
-            "mem": 4,
-            "proc": 2,
-            "status": [
-                {
-                    "error": None,
-                    "progress": 0,
-                    "stage": None,
-                    "state": "waiting",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 0.25,
-                    "stage": "prepare",
-                    "state": "running",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 0.50,
-                    "stage": "say_message",
-                    "state": "running",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 0.75,
-                    "stage": "try_read_stdout",
-                    "state": "running",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 1,
-                    "stage": "try_read_stdout",
-                    "state": "complete",
-                    "timestamp": static_time
-                }
-            ],
-            "task": "dummy",
-            "user": {
-                "id": "test"
-            }
-        }
-
-        assert target == [result]
-
-    @pytest.mark.parametrize("use_executor", [True, False])
-    async def test_exception(self, loop, use_executor, test_job_manager, test_random_alphanumeric, static_time):
-        """
-        Test that an exception in a stage_method leads to an error log in the job document.
-
-        """
-        await test_job_manager.new("dummy", {
-            "message": 1337,
-            "use_executor": use_executor
-        }, "test")
-
-        count = 0
-
-        while True:
-            document = await test_job_manager.db.jobs.find_one()
-
-            last_status = document["status"].pop(-1)
-
-            if last_status["error"]:
-                break
-
-            if count == 15:
-                raise TimeoutError("Timed out waiting for job to error")
-
-            count += 1
-
-            await asyncio.sleep(0.3, loop=loop)
-
-        assert document == {
-            "_id": test_random_alphanumeric.last_choice,
-            "args": {
-                "message": 1337,
-                "use_executor": use_executor
-            },
-            "mem": 4,
-            "proc": 2,
-            "status": [
-                {
-                    "error": None,
-                    "progress": 0,
-                    "stage": None,
-                    "state": "waiting",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 0.25,
-                    "stage": "prepare",
-                    "state": "running",
-                    "timestamp": static_time
-                },
-                {
-                    "error": None,
-                    "progress": 0.5,
-                    "stage": "say_message",
-                    "state": "running",
-                    "timestamp": static_time
-                }
-            ],
-            "task": "dummy",
-            "user": {
-                "id": "test"
-            }
-        }
-
-        assert last_status["error"]["details"] == ["must be str, not int"]
-        assert last_status["error"]["type"] == "TypeError"
-
-
-class TestGetResources:
-
-    async def test(self, test_job_manager):
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 0,
-                "mem": 0
-            },
-            "available": {
-                "proc": 6,
-                "mem": 24
-            }
-        }
-
-
-class TestReserveResources:
-
-    async def test(self, test_job_manager):
-        mock_job = virtool.job.Job(
-            test_job_manager.loop,
-            test_job_manager.executor,
-            test_job_manager.db,
-            test_job_manager.settings,
-            test_job_manager.dispatch,
-            test_job_manager.capture_exception,
-            "foobar",
-            "dummy",
-            {},
-            2,
-            4
-        )
-
-        test_job_manager.reserve_resources(mock_job)
-
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 2,
-                "mem": 4
-            },
-            "available": {
-                "proc": 4,
-                "mem": 20
-            }
-        }
-
-    async def test_multiple(self, test_job_manager):
-        mock_job_1 = virtool.job.Job(
-            test_job_manager.loop,
-            test_job_manager.executor,
-            test_job_manager.db,
-            test_job_manager.settings,
-            test_job_manager.dispatch,
-            test_job_manager.capture_exception,
-            "foobar",
-            "dummy",
-            {},
-            2,
-            8
-        )
-
-        mock_job_2 = virtool.job.Job(
-            test_job_manager.loop,
-            test_job_manager.executor,
-            test_job_manager.db,
-            test_job_manager.settings,
-            test_job_manager.dispatch,
-            test_job_manager.capture_exception,
-            "foobar",
-            "dummy",
-            {},
-            1,
-            2
-        )
-
-        test_job_manager.reserve_resources(mock_job_1)
-        test_job_manager.reserve_resources(mock_job_2)
-
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 3,
-                "mem": 10
-            },
-            "available": {
-                "proc": 3,
-                "mem": 14
-            }
-        }
-
-
-class TestReleaseResources:
-
-    async def test(self, test_job_manager):
-        mock_job = virtool.job.Job(
-            test_job_manager.loop,
-            test_job_manager.executor,
-            test_job_manager.db,
-            test_job_manager.settings,
-            test_job_manager.dispatch,
-            test_job_manager.capture_exception,
-            "foobar",
-            "dummy",
-            {},
-            2,
-            4
-        )
-
-        test_job_manager.reserve_resources(mock_job)
-
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 2,
-                "mem": 4
-            },
-            "available": {
-                "proc": 4,
-                "mem": 20
-            }
-        }
-
-        test_job_manager.release_resources(mock_job)
-
-        assert test_job_manager.get_resources() == {
-            "limit": {
-                "proc": 6,
-                "mem": 24
-            },
-            "used": {
-                "proc": 0,
-                "mem": 0
-            },
-            "available": {
-                "proc": 6,
-                "mem": 24
-            }
-        }
-
-
-class TestClose:
-
-    async def test_stop(self, test_job_manager):
-        assert not test_job_manager._stop
-
-        await test_job_manager.close()
-
-        assert test_job_manager._stop
diff --git a/virtool/app.py b/virtool/app.py
index 56e6f24fe..b534cd325 100644
--- a/virtool/app.py
+++ b/virtool/app.py
@@ -44,7 +44,7 @@ def init_executors(app):
     :type app: :class:`aiohttp.web.Application`
 
     """
-    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+    executor = concurrent.futures.ThreadPoolExecutor()
     app.loop.set_default_executor(executor)
     app["executor"] = executor
 
@@ -218,6 +218,7 @@ async def init_job_manager(app):
 
     app["job_manager"] = virtool.job_manager.Manager(
         app.loop,
+        app["process_executor"],
         app["db"],
         app["settings"],
         app["dispatcher"].dispatch,
@@ -295,6 +296,8 @@ async def on_shutdown(app):
     if file_manager is not None:
         await file_manager.close()
 
+    app["process_executor"].shutdown(wait=True)
+
 
 def create_app(loop, db_name=None, disable_job_manager=False, disable_file_manager=False, force_version=None,
                ignore_settings=False, no_sentry=False, skip_db_checks=False, skip_setup=False):
@@ -335,8 +338,6 @@ def create_app(loop, db_name=None, disable_job_manager=False, disable_file_manag
         else:
             app["settings"] = dict()
 
-
-
         app.on_startup.append(init_executors)
         app.on_startup.append(init_dispatcher)
         app.on_startup.append(init_db)
diff --git a/virtool/job_manager.py b/virtool/job_manager.py
index 0d9246880..a9c9936f2 100644
--- a/virtool/job_manager.py
+++ b/virtool/job_manager.py
@@ -1,5 +1,4 @@
 import asyncio
-from concurrent.futures import ProcessPoolExecutor
 
 import virtool.job
 import virtool.utils
@@ -8,7 +7,7 @@ import virtool.job_classes
 
 class Manager:
 
-    def __init__(self, loop, db, settings, dispatch, capture_exception):
+    def __init__(self, loop, executor, db, settings, dispatch, capture_exception):
         #: The application IO loop
         self.loop = loop
 
@@ -24,12 +23,7 @@ class Manager:
         #: A reference to Sentry client's `captureException` method.
         self.capture_exception = capture_exception
 
-        self._used = {
-            "proc": 0,
-            "mem": 0
-        }
-
-        self.executor = ProcessPoolExecutor()
+        self.executor = executor
 
         #: A dict to store all the tracked job objects in.
         self._jobs_dict = dict()
@@ -48,43 +42,28 @@ class Manager:
         self._run_task = asyncio.ensure_future(self.run(), loop=self.loop)
 
     async def run(self):
-        iteration = 0
-
         while True:
             to_delete = list()
 
-            for job in self._jobs_dict.values():
-                if not self._stop and not job.started:
-                    try:
-                        self.reserve_resources(job)
-                        job.start()
-                    except InsufficientResourceError:
-                        pass
-
-                if job.finished:
-                    self.release_resources(job)
-                    to_delete.append(job.id)
+            if len(self._jobs_dict):
+                available = get_available_resources(self.settings, self._jobs_dict)
 
-            for job_id in to_delete:
-                del self._jobs_dict[job_id]
+                for job_id, job in self._jobs_dict.items():
+                    if not self._stop and not job.started:
+                        if job.proc <= available["proc"] and job.mem <= available["mem"]:
+                            job.start()
+                            break
 
-            iteration += 1
+                    if job.finished:
+                        to_delete.append(job_id)
 
-            if iteration == 100:
-                iteration = 0
-
-                ids = await virtool.job.get_waiting_and_running_ids(self.db)
-
-                await self.db.jobs.delete_many({
-                    "_id": {
-                        "$in": [job_id for job_id in ids if job_id not in self._jobs_dict]
-                    }
-                })
+                for job_id in to_delete:
+                    del self._jobs_dict[job_id]
 
             await asyncio.sleep(0.1, loop=self.loop)
 
             if self._stop and len(self._jobs_dict) == 0:
-                break
+                return
 
     async def new(self, task_name, task_args, user_id, job_id=None):
 
@@ -142,38 +121,21 @@ class Manager:
         for job in self._jobs_dict.values():
             await job.cancel()
 
-        self.executor.shutdown(wait=True)
-
-    def reserve_resources(self, job):
-        """
-        Reserve resources for the given job. Throws an :class:`InsufficientResourceError` if the required resources are
-        not available.
+        await self._run_task
 
-        :param job: the job object to reserve resources for
-        :type job: :class:`virtool.job.Job`
 
-        :raises: :class:`InsufficientResourceError`
+def get_available_resources(settings, jobs):
+    used = get_used_resources(jobs)
+    return {key: settings[key] - used[key] for key in ["proc", "mem"]}
 
-        """
-        available = self.get_resources()["available"]
 
-        if job.proc <= available["proc"] and job.mem <= available["mem"]:
-            self._used["proc"] += job.proc
-            self._used["mem"] += job.mem
-        else:
-            raise InsufficientResourceError
+def get_used_resources(jobs):
+    running_jobs = [j for j in jobs.values() if j.started]
 
-    def release_resources(self, job):
-        self._used["proc"] -= job.proc
-        self._used["mem"] -= job.mem
+    return {
+        "proc": sum(j.proc for j in running_jobs),
+        "mem": sum(j.mem for j in running_jobs)
+    }
 
-    def get_resources(self):
-        return {
-            "used": dict(self._used),
-            "available": {key: self.settings.get(key) - self._used[key] for key in self._used},
-            "limit": {key: self.settings.get(key) for key in self._used}
-        }
 
 
-class InsufficientResourceError(Exception):
-    pass