Refactor requesting new build tasks (#207)

* Refactor requesting new build tasks

* Fix review comments

Author: anfimovdm

build_node/build_node_builder.py

@@ -11,6 +11,7 @@
 import random
 import typing
 import urllib.parse
+from queue import Queue
 
 import requests
 import requests.adapters
@@ -41,7 +42,7 @@ def __init__(
         thread_num,
         terminated_event,
         graceful_terminated_event,
-        task_queue,
+        task_queue: Queue,
     ):
         """
         Build thread initialization.
@@ -56,6 +57,8 @@ def __init__(
             Shows, if process got "kill -15" signal.
         graceful_terminated_event : threading.Event
             Shows, if process got "kill -10" signal.
+        task_queue: queue.Queue
+            Shared queue with build tasks
         """
         super().__init__(
             thread_num=thread_num,
@@ -243,6 +246,7 @@ def run(self):
                     #       without root permissions
                     rm_sudo(task_dir)
                 self.__builder = None
+                self.__task_queue.task_done()
 
     @measure_stage("cas_notarize_artifacts")
     def __cas_notarize_artifacts(

build_node/build_node_supervisor.py

@@ -1,6 +1,7 @@
 import logging
 import traceback
 import urllib.parse
+from queue import Queue
 
 import requests
 import requests.adapters
@@ -14,7 +15,13 @@
 
 class BuilderSupervisor(BaseSupervisor):
 
-    def __init__(self, config, builders, terminated_event, task_queue):
+    def __init__(
+        self,
+        config,
+        builders,
+        terminated_event,
+        task_queue: Queue,
+    ):
         self.__session = None
         self.__task_queue = task_queue
         self.__cached_config = TTLCache(
@@ -44,42 +51,47 @@ def __generate_request_session(self):
         self.__session.mount('https://', adapter)
 
     def __request_build_task(self):
-        if not self.__task_queue.full():
-            supported_arches = [self.config.base_arch]
-            excluded_packages = self.get_excluded_packages()
-            if self.config.base_arch == 'x86_64':
-                supported_arches.append('i686')
-            if self.config.build_src:
-                supported_arches.append('src')
-            full_url = urllib.parse.urljoin(
-                self.config.master_url, 'build_node/get_task'
+        if self.__task_queue.unfinished_tasks >= self.config.threads_count:
+            return {}
+        supported_arches = [self.config.base_arch]
+        excluded_packages = self.get_excluded_packages()
+        if self.config.base_arch == 'x86_64':
+            supported_arches.append('i686')
+        if self.config.build_src:
+            supported_arches.append('src')
+        full_url = urllib.parse.urljoin(
+            self.config.master_url, 'build_node/get_task'
+        )
+        data = {
+            'supported_arches': supported_arches,
+            'excluded_packages': excluded_packages,
+        }
+        try:
+            response = self.__session.post(
+                full_url,
+                json=data,
+                timeout=self.config.request_timeout,
             )
-            data = {
-                'supported_arches': supported_arches,
-                'excluded_packages': excluded_packages,
-            }
-            try:
-                response = self.__session.post(
-                    full_url, json=data, timeout=self.config.request_timeout
-                )
-                response.raise_for_status()
-                return response.json()
-            except Exception:
-                logging.error(
-                    "Can't report active task to master:\n%s",
-                    traceback.format_exc(),
-                )
+            response.raise_for_status()
+            return response.json()
+        except Exception:
+            logging.exception(
+                "Failed to request build task from master:",
+            )
+            return {}
 
     def __report_active_tasks(self):
         active_tasks = self.get_active_tasks()
-        logging.debug('Sending active tasks: {}'.format(active_tasks))
+        logging.debug('Sending active tasks: %s', active_tasks)
         full_url = urllib.parse.urljoin(
             self.config.master_url, 'build_node/ping'
         )
         data = {'active_tasks': [int(item) for item in active_tasks]}
         try:
             self.__session.post(
-                full_url, json=data, timeout=self.config.request_timeout
+                full_url,
+                json=data,
+                timeout=self.config.request_timeout,
             )
         except Exception:
             logging.error(
@@ -110,7 +122,6 @@ def get_excluded_packages(self):
         logging.debug('Excluded packages in this node: %s', excluded_packages)
         return excluded_packages
 
-
     def run(self):
         self.__generate_request_session()
         while not self.terminated_event.is_set():