Merge branch 'bug28932'

sbws/core/scanner.py

 ''' Measure the relays. '''
 
+import sys
+import threading
+
 from ..lib.circuitbuilder import GapsCircuitBuilder as CB
 from ..lib.resultdump import ResultDump
 from ..lib.resultdump import ResultSuccess, ResultErrorCircuit
@@ -9,7 +12,7 @@ from ..lib.relayprioritizer import RelayPrioritizer
 from ..lib.destination import DestinationList
 from ..util.timestamp import now_isodt_str
 from ..util.state import State
-from sbws.globals import fail_hard
+from sbws.globals import fail_hard, TIMEOUT_MEASUREMENTS
 import sbws.util.stem as stem_utils
 import sbws.util.requests as requests_utils
 from argparse import ArgumentDefaultsHelpFormatter
@@ -27,6 +30,27 @@ end_event = Event()
 log = logging.getLogger(__name__)
 
 
+def dumpstacks():
+    import traceback
+    log.critical("sbws stop measuring relays, probably because of a bug."
+                 "Please, open a ticket in trac.torproject.org with this"
+                 "backtrace.")
+    thread_id2name = dict([(t.ident, t.name) for t in threading.enumerate()])
+    for thread_id, stack in sys._current_frames().items():
+        log.critical("Thread: %s(%d)",
+                     thread_id2name.get(thread_id, ""), thread_id)
+        log.critical(traceback.print_stack(stack))
+    # If logging level is less than DEBUG (more verbose), start pdb so that
+    # developers can debug the issue.
+    if log.getEffectiveLevel() < logging.DEBUG:
+        import pdb
+        pdb.set_trace()
+    # Otherwise exit.
+    else:
+        # Change to stop threads when #28869 is merged
+        sys.exit(1)
+
+
 def timed_recv_from_server(session, dest, byte_range):
     ''' Request the **byte_range** from the URL at **dest**. If successful,
     return True and the time it took to download. Otherwise return False and an
@@ -377,9 +401,16 @@ def run_speedtest(args, conf):
             while len(pending_results) >= max_pending_results:
                 time.sleep(5)
                 pending_results = [r for r in pending_results if not r.ready()]
-        while len(pending_results) > 0:
+        time_waiting = 0
+        while (len(pending_results) > 0
+               and time_waiting <= TIMEOUT_MEASUREMENTS):
+            log.debug("Number of pending measurement threads %s after "
+                      "a prioritization loop.", len(pending_results))
             time.sleep(5)
+            time_waiting += 5
             pending_results = [r for r in pending_results if not r.ready()]
+        if time_waiting > TIMEOUT_MEASUREMENTS:
+            dumpstacks()
         loop_tstop = time.time()
         loop_tdelta = (loop_tstop - loop_tstart) / 60
         log.info("Measured %s relays in %s minutes", num_relays, loop_tdelta)

sbws/globals.py

@@ -34,6 +34,7 @@ SUPERVISED_USER_CONFIG_PATH = "/etc/sbws/sbws.ini"
 SUPERVISED_RUN_DPATH = "/run/sbws/tor"
 
 SOCKET_TIMEOUT = 60  # seconds
+TIMEOUT_MEASUREMENTS = 60 * 3  # 3 minutes
 
 SBWS_SCALE_CONSTANT = 7500
 TORFLOW_SCALING = 1