February 2019 - tor-commits - lists.torproject.org

[sbws/master] settings: add class to store end event singleton
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit ea50a75040cbed3db363147788d6088347e8969e Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 14:46:52 2019 +0000 settings: add class to store end event singleton --- sbws/__init__.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/sbws/__init__.py b/sbws/__init__.py index 4256404..3871edb 100644 --- a/sbws/__init__.py +++ b/sbws/__init__.py @@ -1 +1,25 @@ __version__ = '1.0.3-dev0' + + +import threading # noqa + + +class Settings: + """Singleton settings for all the packages. + This way change settings can be seen by all the packages that import it. + + It lives in ``__init__.py`` to leave open the possibility of having a + ``settings.py`` module for user settings. + + .. note:: After refactoring, globals should only have constants. + Any other variable that needs to be modified when initializing + should be initialized here. + + """ + def __init__(self): + self.end_event = threading.Event() + + def set_end_event(self): + self.end_event.set() + +settings = Settings() # noqa

1 0

[sbws/master] scanner: stop raising exception in worker thread
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit 1ccca8abc3c8e32c32b76e3457782549bce9f63c Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:44:28 2019 +0000 scanner: stop raising exception in worker thread --- sbws/core/scanner.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index f16b148..34bfc5a 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -270,11 +270,7 @@ def measure_relay(args, conf, destinations, cb, rl, relay): def dispatch_worker_thread(*a, **kw): - try: - return measure_relay(*a, **kw) - except Exception as err: - log.exception('Unhandled exception in worker thread') - raise err + return measure_relay(*a, **kw) def _should_keep_result(did_request_maximum, result_time, download_times):

1 0

[sbws/master] scanner: catch exception when stopping scanner
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit eb74aecbe305c995ca77d532c4be723f2083c149 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:42:24 2019 +0000 scanner: catch exception when stopping scanner --- sbws/core/scanner.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 7246f12..f16b148 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -39,9 +39,11 @@ def timed_recv_from_server(session, dest, byte_range): try: requests_utils.get( session, dest.url, headers=headers, verify=dest.verify) - except requests.exceptions.ConnectionError as e: - return False, e - except requests.exceptions.ReadTimeout as e: + # NewConnectionError will be raised when shutting down. + except (requests.exceptions.ConnectionError, + requests.exceptions.ReadTimeout, + requests.exceptions.NewConnectionError) as e: + log.debug(e) return False, e end_time = time.time() return True, end_time - start_time

1 0

[sbws/master] scanner: lower logging level on uncatch exceptions
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit 0331ea584e2b630567c15f178bad739d925da933 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:46:10 2019 +0000 scanner: lower logging level on uncatch exceptions when worker thread finish. --- sbws/core/scanner.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 34bfc5a..ed58fe7 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -322,9 +322,10 @@ def result_putter_error(target): ''' Create a function that takes a single argument -- an error from a measurement -- and return that function so it can be used by someone else ''' - def closure(err): - log.error('Unhandled exception caught while measuring %s: %s %s', - target.nickname, type(err), err) + def closure(object): + # The only object that can be here if there is not any uncatched + # exception is stem.SocketClosed when stopping sbws + log.debug(type(object)) return closure

1 0

[sbws/master] scanner: use end event singleton
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit 27fb6cbacfdccb5b301d89989c35ba8d4ab986f2 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:50:04 2019 +0000 scanner: use end event singleton --- sbws/core/scanner.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index ed58fe7..bf6d660 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -14,16 +14,15 @@ import sbws.util.stem as stem_utils import sbws.util.requests as requests_utils from argparse import ArgumentDefaultsHelpFormatter from multiprocessing.dummy import Pool -from threading import Event import time import os import logging import requests import random +from .. import settings rng = random.SystemRandom() -end_event = Event() log = logging.getLogger(__name__) @@ -119,7 +118,7 @@ def measure_bandwidth_to_server(session, conf, dest, content_length): 'target': conf.getfloat('scanner', 'download_target'), 'max': conf.getfloat('scanner', 'download_max'), } - while len(results) < num_downloads: + while len(results) < num_downloads and not settings.end_event.is_set(): assert expected_amount >= min_dl assert expected_amount <= max_dl random_range = get_random_range_string(content_length, expected_amount) @@ -348,7 +347,7 @@ def run_speedtest(args, conf): time.sleep(15) rl = RelayList(args, conf, controller) cb = CB(args, conf, controller, rl) - rd = ResultDump(args, conf, end_event) + rd = ResultDump(args, conf) rp = RelayPrioritizer(args, conf, rl, rd) destinations, error_msg = DestinationList.from_config( conf, cb, rl, controller) @@ -407,9 +406,4 @@ def main(args, conf): state = State(conf.getpath('paths', 'state_fname')) state['scanner_started'] = now_isodt_str() - try: - run_speedtest(args, conf) - except KeyboardInterrupt as e: - raise e - finally: - end_event.set() + run_speedtest(args, conf)

1 0

[sbws/master] scanner: create global variables launching threads
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit 66ee7406733e5656528b2ea426403facba116b18 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:53:59 2019 +0000 scanner: create global variables launching threads so that the objects that manage the threads can be stop at any time. --- sbws/core/scanner.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index bf6d660..b926149 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -24,6 +24,11 @@ from .. import settings rng = random.SystemRandom() log = logging.getLogger(__name__) +# Declare the objects that manage the threads global so that sbws can exit +# gracefully at any time. +pool = None +rd = None +controller = None def timed_recv_from_server(session, dest, byte_range): @@ -329,6 +334,7 @@ def result_putter_error(target): def run_speedtest(args, conf): + global rd, pool, controller controller, _ = stem_utils.init_controller( path=conf.getpath('tor', 'control_socket')) if not controller:

1 0

[sbws/master] scanner: add documentation to function that run scanner
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit c24b2354d96269e4f126762256ffe1535bd4426d Author: juga0 <juga(a)riseup.net> Date: Wed Jan 23 09:02:30 2019 +0000 scanner: add documentation to function that run scanner --- sbws/core/scanner.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 08fb35a..bb7a2ca 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -439,6 +439,22 @@ def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, def run_speedtest(args, conf): + """Initializes all the data and threads needed to measure the relays. + + It launches or connect to Tor in a thread. + It initializes the list of relays seen in the Tor network. + It starts a thread to read the previous measurements and wait for new + measurements to write them to the disk. + It initializes a class that will be used to order the relays depending + on their measurements age. + It initializes the list of destinations that will be used for the + measurements. + It initializes the thread pool that will launch the measurement threads. + The pool starts 3 other threads that are not the measurement (worker) + threads. + Finally, it calls the function that will manage the measurement threads. + + """ global rd, pool, controller controller, _ = stem_utils.init_controller( path=conf.getpath('tor', 'control_socket'))

1 0

[sbws/master] scanner: add function to stop threads on SIGTERM
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit b85fb114278d81d74b0768ab1ba9f88cd1e929d4 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:57:46 2019 +0000 scanner: add function to stop threads on SIGTERM --- sbws/core/scanner.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index b926149..7c2a937 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -1,5 +1,8 @@ ''' Measure the relays. ''' +import signal +import sys + from ..lib.circuitbuilder import GapsCircuitBuilder as CB from ..lib.resultdump import ResultDump from ..lib.resultdump import ResultSuccess, ResultErrorCircuit @@ -31,6 +34,24 @@ rd = None controller = None +def stop_threads(signal, frame): + global rd, pool + log.debug('Stopping sbws.') + # Avoid new threads to start. + settings.set_end_event() + # Stop Pool threads + pool.close() + pool.join() + # Stop ResultDump thread + rd.thread.join() + # Stop Tor thread + controller.close() + sys.exit(0) + + +signal.signal(signal.SIGTERM, stop_threads) + + 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

1 0

[sbws/master] scanner: move log to function that starts measuring
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit c9ace7de89ed9fbad94129361cc7f0e32e8acab4 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 22 18:55:34 2019 +0000 scanner: move log to function that starts measuring since the relay might start being measured some time after apply_async is called if there're not available threads. --- sbws/core/scanner.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 166ce5d..2dc8277 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -202,6 +202,7 @@ def _pick_ideal_second_hop(relay, dest, rl, cont, is_exit): def measure_relay(args, conf, destinations, cb, rl, relay): + log.debug('Measuring %s %s', relay.nickname, relay.fingerprint) s = requests_utils.make_session( cb.controller, conf.getfloat('general', 'http_timeout')) # Pick a destionation @@ -409,8 +410,6 @@ def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, if settings.end_event.is_set(): break num_relays += 1 - log.debug('Measuring %s %s', target.nickname, - target.fingerprint[0:8]) # callback and callback_err must be non-blocking callback = result_putter(result_dump) callback_err = result_putter_error(target)

1 0

[sbws/master] scanner: catch SIGINT in the main loop
by juga＠torproject.org 04 Feb '19

04 Feb '19

commit fe16d6d8a82e53cd62b9bdda577e79b2f6e666d9 Author: juga0 <juga(a)riseup.net> Date: Tue Jan 8 15:59:00 2019 +0000 scanner: catch SIGINT in the main loop also split main function into an extra main_loop function to be able to stop the threads after they have started. Also check end event in the mean loop and before starting to measure a new relay. Fixes bug #28869. Bugfix v0.1.0. --- sbws/core/scanner.py | 115 ++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 90 insertions(+), 25 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 7c2a937..166ce5d 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -354,6 +354,88 @@ def result_putter_error(target): return closure +def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, + relay_prioritizer, destinations, max_pending_results, pool): + """Starts and reuse the threads that measure the relays forever. + + It starts a loop that will be run while there is not and event signaling + that sbws is stopping (because of SIGTERM or SIGINT). + + Then, it starts a second loop with an ordered list (generator) of relays + to measure that might a subset of all the current relays in the Network. + + For every relay, it starts a new thread which runs ``measure_relay`` to + measure the relay until there are ``max_pending_results`` threads. + After that, it will reuse a thread that has finished for every relay to + measure. + It is the the pool method ``apply_async`` which starts or reuse a thread. + This method returns an ``ApplyResult`` immediately, which has a ``ready`` + methods that tells whether the thread has finished or not. + + When the thread finish, ie. ``ApplyResult`` is ``ready``, it triggers + ``result_putter`` callback, which put the ``Result`` in ``ResultDump`` + queue and complete immediately. + + ``ResultDump`` thread (started before and out of this function) will get + the ``Result`` from the queue and write it to disk, so this doesn't block + the measurement threads. + + If there was an exception not catched by ``measure_relay``, it will call + instead ``result_putter_error``, which logs the error and complete + immediately. + + Before iterating over the next relay, it waits (non blocking, since it + happens in the main thread) until one of the ``max_pending_results`` + threads has finished. + + This is not needed, since otherwise async_result will queue the relays to + measure in order and won't start reusing a thread to measure a relay until + other thread has finished. But it makes the logic a bit more sequential. + + Before the outer loop iterates, it also waits (again non blocking) that all + the ``Results`` are ready. + This avoid to start measuring the same relay which might still being + measured. + + """ + pending_results = [] + # Do not start a new loop if sbws is stopping. + while not settings.end_event.is_set(): + log.debug("Starting a new measurement loop.") + num_relays = 0 + loop_tstart = time.time() + for target in relay_prioritizer.best_priority(): + # Don't start measuring a relay if sbws is stopping. + if settings.end_event.is_set(): + break + num_relays += 1 + log.debug('Measuring %s %s', target.nickname, + target.fingerprint[0:8]) + # callback and callback_err must be non-blocking + callback = result_putter(result_dump) + callback_err = result_putter_error(target) + async_result = pool.apply_async( + dispatch_worker_thread, + [args, conf, destinations, circuit_builder, relay_list, + target], {}, callback, callback_err) + pending_results.append(async_result) + # Instead of letting apply_async to queue the relays in order until + # a thread has finished, wait here until a thread has finished. + while len(pending_results) >= max_pending_results: + # sleep is non-blocking sine happens in the main process + time.sleep(5) + pending_results = [r for r in pending_results if not r.ready()] + while len(pending_results) > 0: + log.debug("There are %s pending measurements.", + len(pending_results)) + # sleep is non-blocking sine happens in the main process + time.sleep(5) + pending_results = [r for r in pending_results if not r.ready()] + loop_tstop = time.time() + loop_tdelta = (loop_tstop - loop_tstart) / 60 + log.debug("Measured %s relays in %s minutes", num_relays, loop_tdelta) + + def run_speedtest(args, conf): global rd, pool, controller controller, _ = stem_utils.init_controller( @@ -382,31 +464,14 @@ def run_speedtest(args, conf): fail_hard(error_msg) max_pending_results = conf.getint('scanner', 'measurement_threads') pool = Pool(max_pending_results) - pending_results = [] - while True: - num_relays = 0 - loop_tstart = time.time() - log.info("Starting a new loop to measure relays.") - for target in rp.best_priority(): - num_relays += 1 - log.debug('Measuring %s %s', target.nickname, - target.fingerprint[0:8]) - callback = result_putter(rd) - callback_err = result_putter_error(target) - async_result = pool.apply_async( - dispatch_worker_thread, - [args, conf, destinations, cb, rl, target], - {}, callback, callback_err) - pending_results.append(async_result) - 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.sleep(5) - pending_results = [r for r in pending_results if not r.ready()] - loop_tstop = time.time() - loop_tdelta = (loop_tstop - loop_tstart) / 60 - log.info("Measured %s relays in %s minutes", num_relays, loop_tdelta) + + try: + main_loop(args, conf, controller, rl, cb, rd, rp, destinations, + max_pending_results, pool) + except KeyboardInterrupt: + log.info("Interrupted by the user.") + finally: + stop_threads(signal.SIGINT, None) def gen_parser(sub):

1 0