Revert "Remove while loop from check_service"

This reverts commit 46d2863.

anitya/check_service.py

@@ -22,7 +22,7 @@
 """
 
 import logging
-from concurrent.futures import ThreadPoolExecutor, wait
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
 from threading import Lock
 from time import sleep
@@ -189,6 +189,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
@@ -199,26 +201,32 @@ def run(self):
         futures = {}
         pool_size = config.get("CRON_POOL")
         timeout = config.get("CHECK_TIMEOUT")
-        for i in range(0, len(queue), pool_size):
-            with ThreadPoolExecutor(pool_size) as pool:
-                # Wait till every project in chunk is checked
-                for project in queue[i : i + pool_size]:
+        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
-                (done, not_done) = wait(futures, timeout=timeout)
-                for future in done:
-                    # log any exception
-                    if future.exception():
-                        try:
-                            future.result()
-                        except Exception as e:
-                            _log.exception(e)
-                for future in not_done:
-                    future.cancel()
-                    # Don't actually kill the threads, they will be killed by finishing
-                    # the with statement
+                try:
+                    for future in as_completed(futures, timeout=timeout):
+                        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
+                except TimeoutError:
+                    projects_left -= 1
                     _log.info("Thread was killed because the execution took too long.")
                     with self.error_counter_lock:
                         self.error_counter += 1

anitya/tests/test_check_service.py

@@ -24,7 +24,6 @@
 """
 
 import unittest
-from concurrent.futures import Future
 from datetime import timedelta
 from unittest import mock
 
@@ -383,8 +382,8 @@ def test_run_small_pool_size(self, mock_check_project_release):
         self.assertEqual(len(run_objects), 1)
         self.assertEqual(run_objects[0].total_count, 2)
 
-    @mock.patch("anitya.check_service.wait", return_value=([], [Future(), Future()]))
-    def test_run_timeout(self, mock_wait):
+    @mock.patch("anitya.check_service.as_completed", side_effect=TimeoutError())
+    def test_run_timeout(self, mock_as_completed):
         """
         Assert that TimeoutError is thrown when TIMEOUT is reached.
         """