fix: Handle exceptions from ThreadPoolExecutors

Co-authored-by: Andre Sailer <[email protected]>

Author: sfayer

integration_tests.py

@@ -9,7 +9,7 @@
 import sys
 import tempfile
 import time
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
 from typing import Optional
 
@@ -486,10 +486,15 @@ def logs(pattern: str = "*", lines: int = 10, follow: bool = True):
     if follow:
         base_cmd += ["-f"]
     with ThreadPoolExecutor(len(services)) as pool:
+        futures = []
         for service in fnmatch.filter(services, pattern):
             cmd = base_cmd + [f"{runit_dir}/{service}/log/current"]
             p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=None, text=True)
-            pool.submit(_log_popen_stdout, p)
+            futures.append(pool.submit(_log_popen_stdout, p))
+        for res in as_completed(futures):
+            err = res.exception()
+            if err:
+                raise err
 
 
 class TestExit(typer.Exit):

src/DIRAC/Core/DISET/private/MessageBroker.py

@@ -181,7 +181,14 @@ def __receiveMsgDataAndQueue(self, trid):
                 "from %s : %s" % (self.__trPool.get(trid).getFormattedCredentials(), result["Message"]),
             )
             return self.removeTransport(trid)
-        self.__threadPool.submit(self.__processIncomingData, trid, result)
+
+        def err_handler(res):
+            err = res.exception()
+            if err:
+                self.__log.exception("Exception in receiveMsgDataAndQueue thread", lException=err)
+
+        future = self.__threadPool.submit(self.__processIncomingData, trid, result)
+        future.add_done_callback(err_handler)
         return S_OK()
 
     def __processIncomingData(self, trid, receivedResult):

src/DIRAC/Core/DISET/private/Service.py

@@ -338,7 +338,14 @@ def handleConnection(self, clientTransport):
         if not self.activityMonitoring:
             self._stats["connections"] += 1
             self._monitor.setComponentExtraParam("queries", self._stats["connections"])
-        self._threadPool.submit(self._processInThread, clientTransport)
+
+        def err_handler(result):
+            err = result.exception()
+            if err:
+                gLogger.exception("Exception in handleConnection thread", lException=err)
+
+        future = self._threadPool.submit(self._processInThread, clientTransport)
+        future.add_done_callback(err_handler)
 
     @property
     def wantsThrottle(self):

src/DIRAC/FrameworkSystem/Agent/MyProxyRenewalAgent.py

@@ -72,4 +72,8 @@ def execute(self):
                 userDN = record[0]
                 userGroup = record[1]
                 futures.append(executor.submit(self.__renewProxyForCredentials, userDN, userGroup))
+            for res in concurrent.futures.as_completed(futures):
+                err = res.exception()
+                if err:
+                    raise err
         return S_OK()

src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py

@@ -23,7 +23,7 @@
 import random
 import socket
 from collections import defaultdict
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 import DIRAC
 from DIRAC import S_OK, gConfig
@@ -1079,8 +1079,13 @@ def updatePilotStatus(self):
         # Threads aim at overcoming such issues and thus 1 thread per queue is created to
         # update the status of pilots in transient states
         with ThreadPoolExecutor(max_workers=len(self.queueDict)) as executor:
+            futures = []
             for queue in self.queueDict:
-                executor.submit(self._updatePilotStatusPerQueue, queue, proxy)
+                futures.append(executor.submit(self._updatePilotStatusPerQueue, queue, proxy))
+            for res in as_completed(futures):
+                err = res.exception()
+                if err:
+                    self.log.exception("Update pilot status thread failed", lException=err)
 
         # The pilot can be in Done state set by the job agent check if the output is retrieved
         for queue in self.queueDict: