Handle results separately, don't cumullate them

anitya/check_service.py

@@ -26,7 +26,7 @@
 from typing import List
 from datetime import datetime
 from time import sleep
-from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 import sqlalchemy as sa
 import arrow
@@ -157,6 +157,8 @@ def run(self):
         self.clear_counters()
         queue = self.construct_queue(time)
         total_count = len(queue)
+        projects_left = len(queue)
+        projects_iter = iter(queue)
 
         if not queue:
             return
@@ -165,18 +167,32 @@ def run(self):
         _log.info(
             "Starting check on {} for total of {} projects".format(time, total_count)
         )
+
+        futures = {}
         pool_size = config.get("CRON_POOL", 10)
-        pool = ThreadPoolExecutor(pool_size)
-        futures = [pool.submit(self.update_project, project) for project in queue]
-
-        # Wait till every project is checked
-        wait(futures, return_when=ALL_COMPLETED)
-        for future in futures:
-            if future.exception():
-                try:
-                    future.result()
-                except Exception as e:
-                    _log.exception(e)
+        with ThreadPoolExecutor(pool_size) as pool:
+            # Wait till every project in queue is checked
+            while projects_left:
+                for project in projects_iter:
+                    future = pool.submit(self.update_project, project)
+                    futures[future] = project
+                    if len(futures) > pool_size:
+                        break  # limit job submissions
+
+                # Wait for jobs that aren't completed yet
+                for future in as_completed(futures):
+                    projects_left -= 1  # one project down
+
+                    # log any exception
+                    if future.exception():
+                        try:
+                            future.result()
+                        except Exception as e:
+                            _log.exception(e)
+
+                    del futures[future]
+
+                    break  # give a chance to add more jobs
 
         # 3. Finalize
         _log.info(

anitya/tests/test_check_service.py

@@ -230,6 +230,36 @@ def test_run_nothing_to_check(self, mock_check_project_release):
         self.assertEqual(len(run_objects), 0)
         mock_check_project_release.assert_not_called()
 
+    @mock.patch("anitya.lib.utilities.check_project_release")
+    @mock.patch.dict("anitya.config.config", {"CRON_POOL": 1})
+    def test_run_small_pool_size(self, mock_check_project_release):
+        """
+        Assert that small pool size is handled correctly.
+        """
+        project = models.Project(
+            name="Foobar",
+            backend="GitHub",
+            homepage="www.fakeproject.com",
+            next_check=arrow.utcnow().datetime,
+        )
+        self.session.add(project)
+
+        project = models.Project(
+            name="Fake",
+            backend="GitHub",
+            homepage="www.fakeproject1.com",
+            next_check=arrow.utcnow().datetime,
+        )
+        self.session.add(project)
+        self.session.commit()
+
+        self.checker.run()
+
+        run_objects = models.Run.query.all()
+
+        self.assertEqual(len(run_objects), 1)
+        self.assertEqual(run_objects[0].total_count, 2)
+
     def test_clear_counters(self):
         """
         Assert that counters are cleared.

news/PR842.bug

@@ -0,0 +1 @@
+Fix OOM issue with check_service