[tor-commits] [sbws/master] chg: scanner: Add function to finish the results

[juga at torproject.org]

commit dd9cb438f3c271d1b4fff0004c3a8c49512712cf
Author: juga0 <juga at riseup.net>
Date:   Mon Mar 4 18:53:29 2019 +0000

    chg: scanner: Add function to finish the results
---
 sbws/core/scanner.py | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py
index 484a440..2e88b53 100644
--- a/sbws/core/scanner.py
+++ b/sbws/core/scanner.py
@@ -7,6 +7,8 @@ import threading
 import traceback
 import uuid
 
+from multiprocessing.context import TimeoutError
+
 from ..lib.circuitbuilder import GapsCircuitBuilder as CB
 from ..lib.resultdump import ResultDump
 from ..lib.resultdump import ResultSuccess, ResultErrorCircuit
@@ -508,6 +510,34 @@ def wait_for_results(time_to_sleep, pending_results):
         dumpstacks()
 
 
+def force_get_results(pending_results):
+    """Try to get either the result or an exception, which is logged.
+
+    Get call by ``wait_for_results`` when the time waiting was already long.
+    To get either the Result or an exception, call `get` with timeout.
+    Timeout is low since we already waited.
+    `get` is not call before, because it blocks and the callbacks are not call.
+    """
+    for r in pending_results:
+        try:
+            result = r.get(timeout=0.1)
+            log.warning("Result %s was not stored, it took too long.",
+                        result)
+        # TimeoutError is raised when the result is not ready, ie. has not
+        # been processed yet
+        except TimeoutError:
+            log.warning("A result was not stored, it was not ready.")
+        # If the result raised an exception, get returns it,
+        # so, log any exception so that it can be fixed.
+        # This should not happen, since callback_err would have been call first
+        except Exception as e:
+            log.critical(FILLUP_TICKET_MSG)
+            log.exception("exception %s", e)
+            # if the exception happened in the threads:
+            log.warning("traceback %s",
+                        traceback.print_exception(type(e), e, e.__traceback__))
+
+
 def run_speedtest(args, conf):
     """Initializes all the data and threads needed to measure the relays.