- tor-commits - lists.torproject.org

[torflow/master] Split bwauthority.py into parent and child
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit 7dc70f2efadf641a2f036a1d366715b622dc3a5d Author: aagbsn <aagbsn(a)extc.org> Date: Thu Jun 16 14:07:48 2011 -0700 Split bwauthority.py into parent and child This approach mitigates the memory leaks that plagued long-running bwauths by launching each slice as a separate process --- NetworkScanners/BwAuthority/bwauthority-parent.py | 48 +++++++++++++++ NetworkScanners/BwAuthority/bwauthority.py | 65 ++++++++++----------- 2 files changed, 80 insertions(+), 33 deletions(-) diff --git a/NetworkScanners/BwAuthority/bwauthority-parent.py b/NetworkScanners/BwAuthority/bwauthority-parent.py new file mode 100644 index 0000000..aadcd0e --- /dev/null +++ b/NetworkScanners/BwAuthority/bwauthority-parent.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python +import sys +import subprocess +import ConfigParser +import os +import traceback +sys.path.append("../../") +from TorCtl import TorUtil +from TorCtl.TorUtil import plog +import bwauthority + +p = None + +def main(argv): + TorUtil.read_config(argv[1]) + (start_pct,stop_pct,nodes_per_slice,save_every,circs_per_node,out_dir, + max_fetch_time,tor_dir,sleep_start,sleep_stop, + min_streams,pid_file_name) = bwauthority.read_config(argv[1]) + + if pid_file_name: + pidfd = file(pid_file_name, 'w') + pidfd.write('%d\n' % os.getpid()) + pidfd.close() + + slice_num = 0 + while True: + plog('INFO', 'Beginning time loop') + + p = subprocess.Popen("python bwauthority.py %s %s" % + (argv[1], str(slice_num)),shell=True) + p.wait() + if (p.returncode == 0): + slice_num += 1 + elif (p.returncode == bwauthority.STOP_PCT_REACHED): + slice_num = 0 + else: + plog('WARN', 'Child process returned %s' % p.returncode) + +if __name__ == '__main__': + try: + main(sys.argv) + except KeyboardInterrupt: + p.terminate() + plog('INFO', "Ctrl + C was pressed. Exiting ... ") + traceback.print_exc() + except Exception, e: + plog('ERROR', "An unexpected error occured.") + traceback.print_exc() diff --git a/NetworkScanners/BwAuthority/bwauthority.py b/NetworkScanners/BwAuthority/bwauthority.py index 70a8344..0eace2e 100755 --- a/NetworkScanners/BwAuthority/bwauthority.py +++ b/NetworkScanners/BwAuthority/bwauthority.py @@ -73,6 +73,9 @@ __selmgr = PathSupport.SelectionManager( use_guards=False, exit_ports=[443]) +# exit code to indicate scan completion +STOP_PCT_REACHED = -9 + def read_config(filename): config = ConfigParser.SafeConfigParser() config.read(filename) @@ -273,46 +276,42 @@ def main(argv): pidfd.write('%d\n' % os.getpid()) pidfd.close() - try: - (c,hdlr) = setup_handler(out_dir, tor_dir+"/control_auth_cookie") - except Exception, e: - traceback.print_exc() - plog("WARN", "Can't connect to Tor: "+str(e)) - - sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' - hdlr.attach_sql_listener('sqlite:///'+sql_file) + slice_num = int(argv[2]) - # set SOCKS proxy - socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, TorUtil.tor_host, TorUtil.tor_port) - socket.socket = socks.socksocket - plog("INFO", "Set socks proxy to "+TorUtil.tor_host+":"+str(TorUtil.tor_port)) + try: + (c,hdlr) = setup_handler(out_dir, tor_dir+"/control_auth_cookie") + except Exception, e: + traceback.print_exc() + plog("WARN", "Can't connect to Tor: "+str(e)) + + sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' + hdlr.attach_sql_listener('sqlite:///'+sql_file) + + # set SOCKS proxy + socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, TorUtil.tor_host, TorUtil.tor_port) + socket.socket = socks.socksocket + plog("INFO", "Set socks proxy to "+TorUtil.tor_host+":"+str(TorUtil.tor_port)) - while True: - pct = start_pct - plog('INFO', 'Beginning time loop') - slice_num = 0 - while pct < stop_pct: - hdlr.wait_for_consensus() - pct_step = hdlr.rank_to_percent(nodes_per_slice) - hdlr.reset_stats() - hdlr.commit() - plog('DEBUG', 'Reset stats') + hdlr.wait_for_consensus() + pct_step = hdlr.rank_to_percent(nodes_per_slice) - speedrace(hdlr, pct, pct+pct_step, circs_per_node, save_every, out_dir, - max_fetch_time, sleep_start, sleep_stop, slice_num, - min_streams, sql_file) + # check to see if we are done + if (slice_num * pct_step > stop_pct): + sys.exit(STOP_PCT_REACHED) - # For debugging memory leak.. - #TorUtil.dump_class_ref_counts(referrer_depth=1) + speedrace(hdlr, slice_num*pct_step + start_pct, (slice_num + 1)*pct_step + start_pct, circs_per_node, save_every, out_dir, + max_fetch_time, sleep_start, sleep_stop, slice_num, + min_streams, sql_file) - # TODO: Change pathlen to 3 and kill exit+ConserveExit restrictions - # And record circ failure rates.. + # For debugging memory leak.. + #TorUtil.dump_class_ref_counts(referrer_depth=1) - #circ_measure(hdlr, pct, pct+pct_step, circs_per_node, save_every, - # out_dir, max_fetch_time, sleep_start, sleep_stop, slice_num, sql_file) + # TODO: Change pathlen to 3 and kill exit+ConserveExit restrictions + # And record circ failure rates.. - pct += pct_step - slice_num += 1 + #circ_measure(hdlr, pct, pct+pct_step, circs_per_node, save_every, + # out_dir, max_fetch_time, sleep_start, sleep_stop, slice_num, sql_file) + sys.exit(0) def cleanup(c, f): plog("INFO", "Resetting __LeaveStreamsUnattached=0 and FetchUselessDescriptors="+f)

1 0

[torflow/master] Add support for other database backends
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit 2b799efc7095a31b54d6a18d2e9d12eed28c3600 Author: aagbsn <aagbsn(a)extc.org> Date: Thu Jun 16 19:26:52 2011 -0700 Add support for other database backends db_url can now be set in the configuration file if it is unset, BwAuthority will default to sqlite --- NetworkScanners/BwAuthority/bwauthority_child.py | 13 +++++++++---- 1 files changed, 9 insertions(+), 4 deletions(-) diff --git a/NetworkScanners/BwAuthority/bwauthority_child.py b/NetworkScanners/BwAuthority/bwauthority_child.py index 0eace2e..60f7a09 100755 --- a/NetworkScanners/BwAuthority/bwauthority_child.py +++ b/NetworkScanners/BwAuthority/bwauthority_child.py @@ -98,10 +98,11 @@ def read_config(filename): sleep_stop = tuple(map(int, sleep_stop.split(":"))) pid_file = config.get('BwAuthority', 'pid_file') + db_url = config.get('BwAuthority', 'db_url') return (start_pct,stop_pct,nodes_per_slice,save_every, circs_per_node,out_dir,max_fetch_time,tor_dir, - sleep_start,sleep_stop,min_streams,pid_file) + sleep_start,sleep_stop,min_streams,pid_file,db_url) def choose_url(percentile): for (pct, url) in urls: @@ -269,7 +270,7 @@ def main(argv): TorUtil.read_config(argv[1]) (start_pct,stop_pct,nodes_per_slice,save_every,circs_per_node,out_dir, max_fetch_time,tor_dir,sleep_start,sleep_stop, - min_streams,pid_file_name) = read_config(argv[1]) + min_streams,pid_file_name,db_url) = read_config(argv[1]) if pid_file_name: pidfd = file(pid_file_name, 'w') @@ -284,8 +285,12 @@ def main(argv): traceback.print_exc() plog("WARN", "Can't connect to Tor: "+str(e)) - sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' - hdlr.attach_sql_listener('sqlite:///'+sql_file) + if db_url: + hdlr.attach_sql_listener(db_url) + else: + plog("INFO", "db_url not found in config. Defaulting to sqlite") + sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' + hdlr.attach_sql_listener('sqlite:///'+sql_file) # set SOCKS proxy socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, TorUtil.tor_host, TorUtil.tor_port)

1 0

[torflow/master] renamed files for consistency
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit 9e9127b82b65b5ed5f397b8dd267cd101702b012 Author: aagbsn <aagbsn(a)extc.org> Date: Thu Jun 16 18:38:00 2011 -0700 renamed files for consistency --- NetworkScanners/BwAuthority/bwauthority-parent.py | 48 --- NetworkScanners/BwAuthority/bwauthority.py | 360 ++------------------- NetworkScanners/BwAuthority/bwauthority_child.py | 366 +++++++++++++++++++++ 3 files changed, 386 insertions(+), 388 deletions(-) diff --git a/NetworkScanners/BwAuthority/bwauthority-parent.py b/NetworkScanners/BwAuthority/bwauthority-parent.py deleted file mode 100644 index aadcd0e..0000000 --- a/NetworkScanners/BwAuthority/bwauthority-parent.py +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env python -import sys -import subprocess -import ConfigParser -import os -import traceback -sys.path.append("../../") -from TorCtl import TorUtil -from TorCtl.TorUtil import plog -import bwauthority - -p = None - -def main(argv): - TorUtil.read_config(argv[1]) - (start_pct,stop_pct,nodes_per_slice,save_every,circs_per_node,out_dir, - max_fetch_time,tor_dir,sleep_start,sleep_stop, - min_streams,pid_file_name) = bwauthority.read_config(argv[1]) - - if pid_file_name: - pidfd = file(pid_file_name, 'w') - pidfd.write('%d\n' % os.getpid()) - pidfd.close() - - slice_num = 0 - while True: - plog('INFO', 'Beginning time loop') - - p = subprocess.Popen("python bwauthority.py %s %s" % - (argv[1], str(slice_num)),shell=True) - p.wait() - if (p.returncode == 0): - slice_num += 1 - elif (p.returncode == bwauthority.STOP_PCT_REACHED): - slice_num = 0 - else: - plog('WARN', 'Child process returned %s' % p.returncode) - -if __name__ == '__main__': - try: - main(sys.argv) - except KeyboardInterrupt: - p.terminate() - plog('INFO', "Ctrl + C was pressed. Exiting ... ") - traceback.print_exc() - except Exception, e: - plog('ERROR', "An unexpected error occured.") - traceback.print_exc() diff --git a/NetworkScanners/BwAuthority/bwauthority.py b/NetworkScanners/BwAuthority/bwauthority.py old mode 100755 new mode 100644 index 0eace2e..0d57dfe --- a/NetworkScanners/BwAuthority/bwauthority.py +++ b/NetworkScanners/BwAuthority/bwauthority.py @@ -1,364 +1,44 @@ -#!/usr/bin/python -# -# 2009 Mike Perry, Karsten Loesing - -""" -Speedracer - -Speedracer continuously requests the Tor design paper over the Tor network -and measures how long circuit building and downloading takes. -""" - -import atexit -import socket -import time +#!/usr/bin/env python import sys -import urllib2 +import subprocess +import ConfigParser import os import traceback -import copy -import shutil -import threading -import ConfigParser -import sqlalchemy -import sets - sys.path.append("../../") - +from TorCtl import TorUtil from TorCtl.TorUtil import plog +import bwauthority_child -# WAAAYYYYYY too noisy. -#import gc -#gc.set_debug(gc.DEBUG_COLLECTABLE|gc.DEBUG_UNCOLLECTABLE|gc.DEBUG_INSTANCES|gc.DEBUG_OBJECTS) - -from TorCtl import ScanSupport,PathSupport,SQLSupport,TorCtl,TorUtil - -sys.path.append("../libs") -# Make our SocksiPy use our socket -__origsocket = socket.socket -socket.socket = PathSupport.SmartSocket -from SocksiPy import socks -socket.socket = __origsocket - -user_agent = "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; .NET CLR 1.0.3705; .NET CLR 1.1.4322)" - -# Note these urls should be https due to caching considerations. -# If you really must make them http, be sure to change exit_ports to [80] -# below, or else the scan will not finish. -# TODO: As the network balances, these can become more uniform in size -# TODO: We'll also want to try to prefer pairing unmeasured nodes -# together then, and use a different url set for them. -# cutoff percent URL -urls = [(5, "https://38.229.70.2/16M"), # fbw 1499k..500k - (10, "https://38.229.70.2/8M"), # fbw 500k..350k - (20, "https://38.229.70.2/4M"), # fbw 350k..200k - (30, "https://38.229.70.2/2M"), # fbw 200k..128k - (50, "https://38.229.70.2/512k"), # fbw 128k..50k - (80, "https://38.229.70.2/256k"), # fbw 50k..26k - (100, "https://38.229.70.2/128k")] # fbw 26k..10k - - -# Do NOT modify this object directly after it is handed to PathBuilder -# Use PathBuilder.schedule_selmgr instead. -# (Modifying the arguments here is OK) -__selmgr = PathSupport.SelectionManager( - pathlen=2, - order_exits=False, - percent_fast=100, - percent_skip=0, - min_bw=1024, - use_all_exits=False, - uniform=True, - use_exit=None, - use_guards=False, - exit_ports=[443]) - -# exit code to indicate scan completion -STOP_PCT_REACHED = -9 - -def read_config(filename): - config = ConfigParser.SafeConfigParser() - config.read(filename) - - start_pct = config.getint('BwAuthority', 'start_pct') - stop_pct = config.getint('BwAuthority', 'stop_pct') - - nodes_per_slice = config.getint('BwAuthority', 'nodes_per_slice') - save_every = config.getint('BwAuthority', 'save_every') - circs_per_node = config.getint('BwAuthority', 'circs_per_node') - min_streams = config.getint('BwAuthority', 'min_streams') - out_dir = config.get('BwAuthority', 'out_dir') - tor_dir = config.get('BwAuthority', 'tor_dir') - max_fetch_time = config.getint('BwAuthority', 'max_fetch_time') - - sleep_start = config.get('BwAuthority', 'sleep_start') - sleep_stop = config.get('BwAuthority', 'sleep_stop') - - sleep_start = tuple(map(int, sleep_start.split(":"))) - sleep_stop = tuple(map(int, sleep_stop.split(":"))) - - pid_file = config.get('BwAuthority', 'pid_file') - - return (start_pct,stop_pct,nodes_per_slice,save_every, - circs_per_node,out_dir,max_fetch_time,tor_dir, - sleep_start,sleep_stop,min_streams,pid_file) - -def choose_url(percentile): - for (pct, url) in urls: - if percentile < pct: - return url - #return "https://86.59.21.36/torbrowser/dist/tor-im-browser-1.2.0_ru_split/tor-im-br…" - raise PathSupport.NoNodesRemain("No nodes left for url choice!") - -def http_request(address): - ''' perform an http GET-request and return 1 for success or 0 for failure ''' - - request = urllib2.Request(address) - request.add_header('User-Agent', user_agent) - - try: - reply = urllib2.urlopen(request) - decl_length = reply.info().get("Content-Length") - read_len = len(reply.read()) - plog("DEBUG", "Read: "+str(read_len)+" of declared "+str(decl_length)) - return 1 - except (ValueError, urllib2.URLError): - plog('ERROR', 'The http-request address ' + address + ' is malformed') - return 0 - except (IndexError, TypeError): - plog('ERROR', 'An error occured while negotiating socks5 with Tor') - return 0 - except KeyboardInterrupt: - raise KeyboardInterrupt - except socks.Socks5Error, e: - if e.value[0] == 6: - plog("NOTICE", "Tor timed out our SOCKS stream request.") - else: - plog('ERROR', 'An unknown HTTP error occured') - traceback.print_exc() - return 0 - except: - plog('ERROR', 'An unknown HTTP error occured') - traceback.print_exc() - return 0 - -class BwScanHandler(ScanSupport.SQLScanHandler): - def is_count_met(self, count, num_streams, position=0): - cond = threading.Condition() - cond._finished = True # lol python haxx. Could make subclass, but why?? :) - def notlambda(this): - cond.acquire() - # TODO: Using the entry_gen router list is somewhat ghetto.. - if this.selmgr.bad_restrictions: - plog("NOTICE", - "Bad restrictions on last attempt. Declaring this slice finished") - elif (this.selmgr.path_selector.entry_gen.rstr_routers and \ - this.selmgr.path_selector.exit_gen.rstr_routers): - for r in this.selmgr.path_selector.entry_gen.rstr_routers: - if r._generated[position] < count: - cond._finished = False - plog("DEBUG", "Entry router "+r.idhex+"="+r.nickname+" not done: "+str(r._generated[position])+", down: "+str(r.down)+", OK: "+str(this.selmgr.path_selector.entry_gen.rstr_list.r_is_ok(r))+", sorted_r: "+str(r in this.sorted_r)) - # XXX: - #break - for r in this.selmgr.path_selector.exit_gen.rstr_routers: - if r._generated[position] < count: - cond._finished = False - plog("DEBUG", "Exit router "+r.idhex+"="+r.nickname+" not done: "+str(r._generated[position])+", down: "+str(r.down)+", OK: "+str(this.selmgr.path_selector.exit_gen.rstr_list.r_is_ok(r))+", sorted_r: "+str(r in this.sorted_r)) - # XXX: - #break - # Also run for at least 2*circs_per_node*nodes/3 successful fetches to - # ensure we don't skip slices in the case of temporary network failure - if cond._finished: - num_routers = len( - sets.Set(this.selmgr.path_selector.entry_gen.rstr_routers - + this.selmgr.path_selector.exit_gen.rstr_routers)) - # If more than 35% of the 2-hop paths failed, keep going to get - # more measurements - if num_streams < 0.65*((num_routers*count)/2.0): - plog("WARN", "Not enough streams yet. "+str(num_streams)+" < "+ - str(0.65*(num_routers*count/2.0))) - cond._finished = False - cond.notify() - cond.release() - plog("DEBUG", "Checking if scan count is met...") - cond.acquire() - self.schedule_low_prio(notlambda) - cond.wait() - cond.release() - plog("DEBUG", "Scan count met: "+str(cond._finished)) - return cond._finished - -def speedrace(hdlr, start_pct, stop_pct, circs_per_node, save_every, out_dir, - max_fetch_time, sleep_start_tp, sleep_stop_tp, slice_num, - min_streams, sql_file): - hdlr.set_pct_rstr(start_pct, stop_pct) - - attempt = 0 - successful = 0 - while True: - if hdlr.is_count_met(circs_per_node, successful): break - hdlr.wait_for_consensus() - - # Check local time. Do not scan between 01:30 and 05:30 local time - lt = time.localtime() - sleep_start = time.mktime(lt[0:3]+sleep_start_tp+(0,0,0)+(lt[-1],)) - sleep_stop = time.mktime(lt[0:3]+sleep_stop_tp+(0,0,0)+(lt[-1],)) - t0 = time.time() - if sleep_start <= t0 and t0 <= sleep_stop: - plog("NOTICE", "It's bedtime. Sleeping for "+str(round((sleep_stop-t0)/3600.0,1))+"h") - time.sleep(sleep_stop - t0) - t0 = time.time() - - hdlr.new_exit() - attempt += 1 - - # FIXME: This noise is due to a difficult to find Tor bug that - # causes some exits to hang forever on streams :( - timer = threading.Timer(max_fetch_time, lambda: hdlr.close_streams(7)) - timer.start() - url = choose_url(start_pct) - plog("DEBUG", "Launching stream request for url "+url+" in "+str(start_pct)+'-'+str(stop_pct) + '%') - ret = http_request(url) - timer.cancel() - PathSupport.SmartSocket.clear_port_table() - - delta_build = time.time() - t0 - if delta_build >= max_fetch_time: - plog('WARN', 'Timer exceeded limit: ' + str(delta_build) + '\n') - - build_exit = hdlr.get_exit_node() - if ret == 1 and build_exit: - successful += 1 - plog('DEBUG', str(start_pct) + '-' + str(stop_pct) + '% circuit build+fetch took ' + str(delta_build) + ' for ' + str(build_exit)) - else: - plog('DEBUG', str(start_pct)+'-'+str(stop_pct)+'% circuit build+fetch failed for ' + str(build_exit)) - - if save_every and ret and successful and (successful % save_every) == 0: - race_time = time.strftime("20%y-%m-%d-%H:%M:%S") - hdlr.close_circuits() - hdlr.commit() - lo = str(round(start_pct,1)) - hi = str(round(stop_pct,1)) - # Warning, don't remove the sql stats without changing the recompute - # param in write_strm_bws to True - hdlr.write_sql_stats(os.getcwd()+'/'+out_dir+'/sql-'+lo+':'+hi+"-"+str(successful)+"-"+race_time, sqlalchemy.or_(SQLSupport.RouterStats.circ_try_from > 0, SQLSupport.RouterStats.circ_try_to > 0)) - hdlr.write_strm_bws(os.getcwd()+'/'+out_dir+'/bws-'+lo+':'+hi+"-"+str(successful)+"-"+race_time, stats_filter=SQLSupport.RouterStats.strm_closed >= 1) - - plog('INFO', str(start_pct) + '-' + str(stop_pct) + '% ' + str(successful) + ' fetches took ' + str(attempt) + ' tries.') - - hdlr.close_circuits() - hdlr.commit() - - lo = str(round(start_pct,1)) - hi = str(round(stop_pct,1)) - - # There may be a consensus change between the point of speed - # racing and the writing of stats causing a discrepency - # between the immediate, current consensus result used to determine - # termination and this average-based result. - # So instead of using percentiles to filter here, we filter based on - # circuit chosen. - hdlr.write_sql_stats(os.getcwd()+'/'+out_dir+'/sql-'+lo+':'+hi+"-done-"+time.strftime("20%y-%m-%d-%H:%M:%S"), stats_filter=sqlalchemy.or_(SQLSupport.RouterStats.circ_try_from > 0, SQLSupport.RouterStats.circ_try_to > 0)) - # Warning, don't remove the sql stats call without changing the recompute - # param in write_strm_bws to True - hdlr.write_strm_bws(os.getcwd()+'/'+out_dir+'/bws-'+lo+':'+hi+"-done-"+time.strftime("20%y-%m-%d-%H:%M:%S"), slice_num, stats_filter=sqlalchemy.and_(SQLSupport.RouterStats.strm_closed >= min_streams, SQLSupport.RouterStats.filt_sbw >= 0, SQLSupport.RouterStats.sbw >=0 )) - plog('DEBUG', 'Wrote stats') - #hdlr.save_sql_file(sql_file, os.getcwd()+"/"+out_dir+"/bw-db-"+str(lo)+":"+str(hi)+"-"+time.strftime("20%y-%m-%d-%H:%M:%S")+".sqlite") def main(argv): TorUtil.read_config(argv[1]) (start_pct,stop_pct,nodes_per_slice,save_every,circs_per_node,out_dir, max_fetch_time,tor_dir,sleep_start,sleep_stop, - min_streams,pid_file_name) = read_config(argv[1]) + min_streams,pid_file_name) = bwauthority_child.read_config(argv[1]) if pid_file_name: pidfd = file(pid_file_name, 'w') pidfd.write('%d\n' % os.getpid()) pidfd.close() - slice_num = int(argv[2]) - - try: - (c,hdlr) = setup_handler(out_dir, tor_dir+"/control_auth_cookie") - except Exception, e: - traceback.print_exc() - plog("WARN", "Can't connect to Tor: "+str(e)) - - sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' - hdlr.attach_sql_listener('sqlite:///'+sql_file) - - # set SOCKS proxy - socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, TorUtil.tor_host, TorUtil.tor_port) - socket.socket = socks.socksocket - plog("INFO", "Set socks proxy to "+TorUtil.tor_host+":"+str(TorUtil.tor_port)) - - hdlr.wait_for_consensus() - pct_step = hdlr.rank_to_percent(nodes_per_slice) - - # check to see if we are done - if (slice_num * pct_step > stop_pct): - sys.exit(STOP_PCT_REACHED) - - speedrace(hdlr, slice_num*pct_step + start_pct, (slice_num + 1)*pct_step + start_pct, circs_per_node, save_every, out_dir, - max_fetch_time, sleep_start, sleep_stop, slice_num, - min_streams, sql_file) - - # For debugging memory leak.. - #TorUtil.dump_class_ref_counts(referrer_depth=1) - - # TODO: Change pathlen to 3 and kill exit+ConserveExit restrictions - # And record circ failure rates.. - - #circ_measure(hdlr, pct, pct+pct_step, circs_per_node, save_every, - # out_dir, max_fetch_time, sleep_start, sleep_stop, slice_num, sql_file) - sys.exit(0) - -def cleanup(c, f): - plog("INFO", "Resetting __LeaveStreamsUnattached=0 and FetchUselessDescriptors="+f) - try: - # XXX: Remember __LeaveStreamsUnattached and use saved value! - c.set_option("__LeaveStreamsUnattached", "0") - c.set_option("FetchUselessDescriptors", f) - except TorCtl.TorCtlClosed: - pass - -def setup_handler(out_dir, cookie_file): - plog('INFO', 'Connecting to Tor at '+TorUtil.control_host+":"+str(TorUtil.control_port)) - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((TorUtil.control_host,TorUtil.control_port)) - c = PathSupport.Connection(s) - #c.debug(file(out_dir+"/control.log", "w", buffering=0)) - c.authenticate_cookie(file(cookie_file, "r")) - #f = c.get_option("__LeaveStreamsUnattached")[0] - h = BwScanHandler(c, __selmgr, - strm_selector=PathSupport.SmartSocket.StreamSelector) - - c.set_event_handler(h) - #c.set_periodic_timer(2.0, "PULSE") - - c.set_events([TorCtl.EVENT_TYPE.STREAM, - TorCtl.EVENT_TYPE.BW, - TorCtl.EVENT_TYPE.NEWCONSENSUS, - TorCtl.EVENT_TYPE.NEWDESC, - TorCtl.EVENT_TYPE.CIRC, - TorCtl.EVENT_TYPE.STREAM_BW], True) - - c.set_option("__LeaveStreamsUnattached", "1") - f = c.get_option("FetchUselessDescriptors")[0][1] - c.set_option("FetchUselessDescriptors", "1") - atexit.register(cleanup, *(c, f)) - return (c,h) - -def usage(argv): - print "Usage: "+argv[0]+" <configfile>" - return + slice_num = 0 + while True: + plog('INFO', 'Beginning time loop') + global p + p = subprocess.Popen(["python", "bwauthority_child.py", argv[1], str(slice_num)]) + p.wait() + if (p.returncode == 0): + slice_num += 1 + elif (p.returncode == bwauthority_child.STOP_PCT_REACHED): + slice_num = 0 + else: + plog('WARN', 'Child process returned %s' % p.returncode) -# initiate the program if __name__ == '__main__': try: - if len(sys.argv) < 2: usage(sys.argv) - else: main(sys.argv) + main(sys.argv) except KeyboardInterrupt: + p.kill() plog('INFO', "Ctrl + C was pressed. Exiting ... ") traceback.print_exc() except Exception, e: diff --git a/NetworkScanners/BwAuthority/bwauthority_child.py b/NetworkScanners/BwAuthority/bwauthority_child.py new file mode 100755 index 0000000..0eace2e --- /dev/null +++ b/NetworkScanners/BwAuthority/bwauthority_child.py @@ -0,0 +1,366 @@ +#!/usr/bin/python +# +# 2009 Mike Perry, Karsten Loesing + +""" +Speedracer + +Speedracer continuously requests the Tor design paper over the Tor network +and measures how long circuit building and downloading takes. +""" + +import atexit +import socket +import time +import sys +import urllib2 +import os +import traceback +import copy +import shutil +import threading +import ConfigParser +import sqlalchemy +import sets + +sys.path.append("../../") + +from TorCtl.TorUtil import plog + +# WAAAYYYYYY too noisy. +#import gc +#gc.set_debug(gc.DEBUG_COLLECTABLE|gc.DEBUG_UNCOLLECTABLE|gc.DEBUG_INSTANCES|gc.DEBUG_OBJECTS) + +from TorCtl import ScanSupport,PathSupport,SQLSupport,TorCtl,TorUtil + +sys.path.append("../libs") +# Make our SocksiPy use our socket +__origsocket = socket.socket +socket.socket = PathSupport.SmartSocket +from SocksiPy import socks +socket.socket = __origsocket + +user_agent = "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; .NET CLR 1.0.3705; .NET CLR 1.1.4322)" + +# Note these urls should be https due to caching considerations. +# If you really must make them http, be sure to change exit_ports to [80] +# below, or else the scan will not finish. +# TODO: As the network balances, these can become more uniform in size +# TODO: We'll also want to try to prefer pairing unmeasured nodes +# together then, and use a different url set for them. +# cutoff percent URL +urls = [(5, "https://38.229.70.2/16M"), # fbw 1499k..500k + (10, "https://38.229.70.2/8M"), # fbw 500k..350k + (20, "https://38.229.70.2/4M"), # fbw 350k..200k + (30, "https://38.229.70.2/2M"), # fbw 200k..128k + (50, "https://38.229.70.2/512k"), # fbw 128k..50k + (80, "https://38.229.70.2/256k"), # fbw 50k..26k + (100, "https://38.229.70.2/128k")] # fbw 26k..10k + + +# Do NOT modify this object directly after it is handed to PathBuilder +# Use PathBuilder.schedule_selmgr instead. +# (Modifying the arguments here is OK) +__selmgr = PathSupport.SelectionManager( + pathlen=2, + order_exits=False, + percent_fast=100, + percent_skip=0, + min_bw=1024, + use_all_exits=False, + uniform=True, + use_exit=None, + use_guards=False, + exit_ports=[443]) + +# exit code to indicate scan completion +STOP_PCT_REACHED = -9 + +def read_config(filename): + config = ConfigParser.SafeConfigParser() + config.read(filename) + + start_pct = config.getint('BwAuthority', 'start_pct') + stop_pct = config.getint('BwAuthority', 'stop_pct') + + nodes_per_slice = config.getint('BwAuthority', 'nodes_per_slice') + save_every = config.getint('BwAuthority', 'save_every') + circs_per_node = config.getint('BwAuthority', 'circs_per_node') + min_streams = config.getint('BwAuthority', 'min_streams') + out_dir = config.get('BwAuthority', 'out_dir') + tor_dir = config.get('BwAuthority', 'tor_dir') + max_fetch_time = config.getint('BwAuthority', 'max_fetch_time') + + sleep_start = config.get('BwAuthority', 'sleep_start') + sleep_stop = config.get('BwAuthority', 'sleep_stop') + + sleep_start = tuple(map(int, sleep_start.split(":"))) + sleep_stop = tuple(map(int, sleep_stop.split(":"))) + + pid_file = config.get('BwAuthority', 'pid_file') + + return (start_pct,stop_pct,nodes_per_slice,save_every, + circs_per_node,out_dir,max_fetch_time,tor_dir, + sleep_start,sleep_stop,min_streams,pid_file) + +def choose_url(percentile): + for (pct, url) in urls: + if percentile < pct: + return url + #return "https://86.59.21.36/torbrowser/dist/tor-im-browser-1.2.0_ru_split/tor-im-br…" + raise PathSupport.NoNodesRemain("No nodes left for url choice!") + +def http_request(address): + ''' perform an http GET-request and return 1 for success or 0 for failure ''' + + request = urllib2.Request(address) + request.add_header('User-Agent', user_agent) + + try: + reply = urllib2.urlopen(request) + decl_length = reply.info().get("Content-Length") + read_len = len(reply.read()) + plog("DEBUG", "Read: "+str(read_len)+" of declared "+str(decl_length)) + return 1 + except (ValueError, urllib2.URLError): + plog('ERROR', 'The http-request address ' + address + ' is malformed') + return 0 + except (IndexError, TypeError): + plog('ERROR', 'An error occured while negotiating socks5 with Tor') + return 0 + except KeyboardInterrupt: + raise KeyboardInterrupt + except socks.Socks5Error, e: + if e.value[0] == 6: + plog("NOTICE", "Tor timed out our SOCKS stream request.") + else: + plog('ERROR', 'An unknown HTTP error occured') + traceback.print_exc() + return 0 + except: + plog('ERROR', 'An unknown HTTP error occured') + traceback.print_exc() + return 0 + +class BwScanHandler(ScanSupport.SQLScanHandler): + def is_count_met(self, count, num_streams, position=0): + cond = threading.Condition() + cond._finished = True # lol python haxx. Could make subclass, but why?? :) + def notlambda(this): + cond.acquire() + # TODO: Using the entry_gen router list is somewhat ghetto.. + if this.selmgr.bad_restrictions: + plog("NOTICE", + "Bad restrictions on last attempt. Declaring this slice finished") + elif (this.selmgr.path_selector.entry_gen.rstr_routers and \ + this.selmgr.path_selector.exit_gen.rstr_routers): + for r in this.selmgr.path_selector.entry_gen.rstr_routers: + if r._generated[position] < count: + cond._finished = False + plog("DEBUG", "Entry router "+r.idhex+"="+r.nickname+" not done: "+str(r._generated[position])+", down: "+str(r.down)+", OK: "+str(this.selmgr.path_selector.entry_gen.rstr_list.r_is_ok(r))+", sorted_r: "+str(r in this.sorted_r)) + # XXX: + #break + for r in this.selmgr.path_selector.exit_gen.rstr_routers: + if r._generated[position] < count: + cond._finished = False + plog("DEBUG", "Exit router "+r.idhex+"="+r.nickname+" not done: "+str(r._generated[position])+", down: "+str(r.down)+", OK: "+str(this.selmgr.path_selector.exit_gen.rstr_list.r_is_ok(r))+", sorted_r: "+str(r in this.sorted_r)) + # XXX: + #break + # Also run for at least 2*circs_per_node*nodes/3 successful fetches to + # ensure we don't skip slices in the case of temporary network failure + if cond._finished: + num_routers = len( + sets.Set(this.selmgr.path_selector.entry_gen.rstr_routers + + this.selmgr.path_selector.exit_gen.rstr_routers)) + # If more than 35% of the 2-hop paths failed, keep going to get + # more measurements + if num_streams < 0.65*((num_routers*count)/2.0): + plog("WARN", "Not enough streams yet. "+str(num_streams)+" < "+ + str(0.65*(num_routers*count/2.0))) + cond._finished = False + cond.notify() + cond.release() + plog("DEBUG", "Checking if scan count is met...") + cond.acquire() + self.schedule_low_prio(notlambda) + cond.wait() + cond.release() + plog("DEBUG", "Scan count met: "+str(cond._finished)) + return cond._finished + +def speedrace(hdlr, start_pct, stop_pct, circs_per_node, save_every, out_dir, + max_fetch_time, sleep_start_tp, sleep_stop_tp, slice_num, + min_streams, sql_file): + hdlr.set_pct_rstr(start_pct, stop_pct) + + attempt = 0 + successful = 0 + while True: + if hdlr.is_count_met(circs_per_node, successful): break + hdlr.wait_for_consensus() + + # Check local time. Do not scan between 01:30 and 05:30 local time + lt = time.localtime() + sleep_start = time.mktime(lt[0:3]+sleep_start_tp+(0,0,0)+(lt[-1],)) + sleep_stop = time.mktime(lt[0:3]+sleep_stop_tp+(0,0,0)+(lt[-1],)) + t0 = time.time() + if sleep_start <= t0 and t0 <= sleep_stop: + plog("NOTICE", "It's bedtime. Sleeping for "+str(round((sleep_stop-t0)/3600.0,1))+"h") + time.sleep(sleep_stop - t0) + t0 = time.time() + + hdlr.new_exit() + attempt += 1 + + # FIXME: This noise is due to a difficult to find Tor bug that + # causes some exits to hang forever on streams :( + timer = threading.Timer(max_fetch_time, lambda: hdlr.close_streams(7)) + timer.start() + url = choose_url(start_pct) + plog("DEBUG", "Launching stream request for url "+url+" in "+str(start_pct)+'-'+str(stop_pct) + '%') + ret = http_request(url) + timer.cancel() + PathSupport.SmartSocket.clear_port_table() + + delta_build = time.time() - t0 + if delta_build >= max_fetch_time: + plog('WARN', 'Timer exceeded limit: ' + str(delta_build) + '\n') + + build_exit = hdlr.get_exit_node() + if ret == 1 and build_exit: + successful += 1 + plog('DEBUG', str(start_pct) + '-' + str(stop_pct) + '% circuit build+fetch took ' + str(delta_build) + ' for ' + str(build_exit)) + else: + plog('DEBUG', str(start_pct)+'-'+str(stop_pct)+'% circuit build+fetch failed for ' + str(build_exit)) + + if save_every and ret and successful and (successful % save_every) == 0: + race_time = time.strftime("20%y-%m-%d-%H:%M:%S") + hdlr.close_circuits() + hdlr.commit() + lo = str(round(start_pct,1)) + hi = str(round(stop_pct,1)) + # Warning, don't remove the sql stats without changing the recompute + # param in write_strm_bws to True + hdlr.write_sql_stats(os.getcwd()+'/'+out_dir+'/sql-'+lo+':'+hi+"-"+str(successful)+"-"+race_time, sqlalchemy.or_(SQLSupport.RouterStats.circ_try_from > 0, SQLSupport.RouterStats.circ_try_to > 0)) + hdlr.write_strm_bws(os.getcwd()+'/'+out_dir+'/bws-'+lo+':'+hi+"-"+str(successful)+"-"+race_time, stats_filter=SQLSupport.RouterStats.strm_closed >= 1) + + plog('INFO', str(start_pct) + '-' + str(stop_pct) + '% ' + str(successful) + ' fetches took ' + str(attempt) + ' tries.') + + hdlr.close_circuits() + hdlr.commit() + + lo = str(round(start_pct,1)) + hi = str(round(stop_pct,1)) + + # There may be a consensus change between the point of speed + # racing and the writing of stats causing a discrepency + # between the immediate, current consensus result used to determine + # termination and this average-based result. + # So instead of using percentiles to filter here, we filter based on + # circuit chosen. + hdlr.write_sql_stats(os.getcwd()+'/'+out_dir+'/sql-'+lo+':'+hi+"-done-"+time.strftime("20%y-%m-%d-%H:%M:%S"), stats_filter=sqlalchemy.or_(SQLSupport.RouterStats.circ_try_from > 0, SQLSupport.RouterStats.circ_try_to > 0)) + # Warning, don't remove the sql stats call without changing the recompute + # param in write_strm_bws to True + hdlr.write_strm_bws(os.getcwd()+'/'+out_dir+'/bws-'+lo+':'+hi+"-done-"+time.strftime("20%y-%m-%d-%H:%M:%S"), slice_num, stats_filter=sqlalchemy.and_(SQLSupport.RouterStats.strm_closed >= min_streams, SQLSupport.RouterStats.filt_sbw >= 0, SQLSupport.RouterStats.sbw >=0 )) + plog('DEBUG', 'Wrote stats') + #hdlr.save_sql_file(sql_file, os.getcwd()+"/"+out_dir+"/bw-db-"+str(lo)+":"+str(hi)+"-"+time.strftime("20%y-%m-%d-%H:%M:%S")+".sqlite") + +def main(argv): + TorUtil.read_config(argv[1]) + (start_pct,stop_pct,nodes_per_slice,save_every,circs_per_node,out_dir, + max_fetch_time,tor_dir,sleep_start,sleep_stop, + min_streams,pid_file_name) = read_config(argv[1]) + + if pid_file_name: + pidfd = file(pid_file_name, 'w') + pidfd.write('%d\n' % os.getpid()) + pidfd.close() + + slice_num = int(argv[2]) + + try: + (c,hdlr) = setup_handler(out_dir, tor_dir+"/control_auth_cookie") + except Exception, e: + traceback.print_exc() + plog("WARN", "Can't connect to Tor: "+str(e)) + + sql_file = os.getcwd()+'/'+out_dir+'/bwauthority.sqlite' + hdlr.attach_sql_listener('sqlite:///'+sql_file) + + # set SOCKS proxy + socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, TorUtil.tor_host, TorUtil.tor_port) + socket.socket = socks.socksocket + plog("INFO", "Set socks proxy to "+TorUtil.tor_host+":"+str(TorUtil.tor_port)) + + hdlr.wait_for_consensus() + pct_step = hdlr.rank_to_percent(nodes_per_slice) + + # check to see if we are done + if (slice_num * pct_step > stop_pct): + sys.exit(STOP_PCT_REACHED) + + speedrace(hdlr, slice_num*pct_step + start_pct, (slice_num + 1)*pct_step + start_pct, circs_per_node, save_every, out_dir, + max_fetch_time, sleep_start, sleep_stop, slice_num, + min_streams, sql_file) + + # For debugging memory leak.. + #TorUtil.dump_class_ref_counts(referrer_depth=1) + + # TODO: Change pathlen to 3 and kill exit+ConserveExit restrictions + # And record circ failure rates.. + + #circ_measure(hdlr, pct, pct+pct_step, circs_per_node, save_every, + # out_dir, max_fetch_time, sleep_start, sleep_stop, slice_num, sql_file) + sys.exit(0) + +def cleanup(c, f): + plog("INFO", "Resetting __LeaveStreamsUnattached=0 and FetchUselessDescriptors="+f) + try: + # XXX: Remember __LeaveStreamsUnattached and use saved value! + c.set_option("__LeaveStreamsUnattached", "0") + c.set_option("FetchUselessDescriptors", f) + except TorCtl.TorCtlClosed: + pass + +def setup_handler(out_dir, cookie_file): + plog('INFO', 'Connecting to Tor at '+TorUtil.control_host+":"+str(TorUtil.control_port)) + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.connect((TorUtil.control_host,TorUtil.control_port)) + c = PathSupport.Connection(s) + #c.debug(file(out_dir+"/control.log", "w", buffering=0)) + c.authenticate_cookie(file(cookie_file, "r")) + #f = c.get_option("__LeaveStreamsUnattached")[0] + h = BwScanHandler(c, __selmgr, + strm_selector=PathSupport.SmartSocket.StreamSelector) + + c.set_event_handler(h) + #c.set_periodic_timer(2.0, "PULSE") + + c.set_events([TorCtl.EVENT_TYPE.STREAM, + TorCtl.EVENT_TYPE.BW, + TorCtl.EVENT_TYPE.NEWCONSENSUS, + TorCtl.EVENT_TYPE.NEWDESC, + TorCtl.EVENT_TYPE.CIRC, + TorCtl.EVENT_TYPE.STREAM_BW], True) + + c.set_option("__LeaveStreamsUnattached", "1") + f = c.get_option("FetchUselessDescriptors")[0][1] + c.set_option("FetchUselessDescriptors", "1") + atexit.register(cleanup, *(c, f)) + return (c,h) + +def usage(argv): + print "Usage: "+argv[0]+" <configfile>" + return + +# initiate the program +if __name__ == '__main__': + try: + if len(sys.argv) < 2: usage(sys.argv) + else: main(sys.argv) + except KeyboardInterrupt: + plog('INFO', "Ctrl + C was pressed. Exiting ... ") + traceback.print_exc() + except Exception, e: + plog('ERROR', "An unexpected error occured.") + traceback.print_exc()

1 0

[pytorctl/master] Add a little more rambling to reset_all().
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit dc1cb9ce39fd5c9c5ea50f542642130a44e71fee Author: Mike Perry <mikeperry-git(a)fscked.org> Date: Fri Jun 24 12:33:02 2011 -0700 Add a little more rambling to reset_all(). --- SQLSupport.py | 13 +++++++++---- 1 files changed, 9 insertions(+), 4 deletions(-) diff --git a/SQLSupport.py b/SQLSupport.py index cf2c753..1e2971b 100644 --- a/SQLSupport.py +++ b/SQLSupport.py @@ -658,8 +658,12 @@ class RouterStats(Entity): #################### Model Support ################ def reset_all(): plog("WARN", "SQLSupport.reset_all() called. See SQLSupport.py for details") - # Need to keep routers around.. - # WARNING! + # XXX: We still have a memory leak somewhere in here + # Current suspects are sqlite, python-sqlite, or sqlalchemy misuse... + # http://stackoverflow.com/questions/5552932/sqlalchemy-misuse-causing-memory… + # The bandwidth scanners switched to a parent/child model because of this. + + # XXX: WARNING! # Must keep the routers around because circ_status_event may # reference old Routers that are no longer in consensus # and will raise an ObjectDeletedError. See function circ_status_event in @@ -677,12 +681,13 @@ def reset_all(): tc_session.commit() tc_session.expunge_all() - # WARNING! + # XXX: WARNING! # May not clear relation all tables! (SQLAlchemy or Elixir bug) # Try: # tc_session.execute('delete from router_streams__stream;') - # WARNING! + + # XXX: WARNING! # This will cause Postgres databases to hang # on DROP TABLE. Possibly an issue with cascade. # Sqlite works though.

1 0

[pytorctl/master] Merge remote branch 'aagbsn/testing'
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit d0f593f2ae245e951323ff7fc3a65fa6a1f16b47 Merge: 3c438e0 eb736ba Author: Mike Perry <mikeperry-git(a)fscked.org> Date: Fri Jun 24 11:57:47 2011 -0700 Merge remote branch 'aagbsn/testing' SQLSupport.py | 101 ++++++++++++++++++++++++++++++++++++++++++++------------ ScanSupport.py | 18 +++++++++- 2 files changed, 96 insertions(+), 23 deletions(-)

1 0

[pytorctl/master] added reset_stats() to Scansupport.py
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit 13307d9cc0ed54bf5602c2c9b5e8db86113f3a45 Author: aagbsn <aagbsn(a)extc.org> Date: Mon Apr 18 10:38:50 2011 -0700 added reset_stats() to Scansupport.py calls parent and SQLSupport reset functions Tests show that SQLSupport.reset_all() may clear too much because if BwAuthority calls Scansupport.reset_stats() after each speedrace() run only the first slice is properly recorded; the rest are empty. See: https://trac.torproject.org/projects/tor/ticket/2947 --- ScanSupport.py | 13 ++++++++++++- 1 files changed, 12 insertions(+), 1 deletions(-) diff --git a/ScanSupport.py b/ScanSupport.py index f2a39d9..c425fd2 100644 --- a/ScanSupport.py +++ b/ScanSupport.py @@ -260,4 +260,15 @@ class SQLScanHandler(ScanHandler): cond.release() plog("INFO", "Consensus OK") - + def reset_stats(self): + cond = threading.Condition() + def notlambda(this): + cond.acquire() + ScanHandler.reset_stats(self) + SQLSupport.reset_all() + cond.notify() + cond.release() + cond.acquire() + self.schedule_low_prio(notlambda) + cond.wait() + cond.release()

1 0

[pytorctl/master] update consensus after resetting stats within the same job
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit e1266cb4d716a8a066ea3d92fb3d19305939a058 Author: aagbsn <aagbsn(a)extc.org> Date: Thu May 19 14:17:12 2011 -0700 update consensus after resetting stats within the same job --- ScanSupport.py | 2 ++ 1 files changed, 2 insertions(+), 0 deletions(-) diff --git a/ScanSupport.py b/ScanSupport.py index c425fd2..97c43a8 100644 --- a/ScanSupport.py +++ b/ScanSupport.py @@ -266,6 +266,8 @@ class SQLScanHandler(ScanHandler): cond.acquire() ScanHandler.reset_stats(self) SQLSupport.reset_all() + this.sql_consensus_listener.update_consensus() + this.sql_consensus_listener._update_rank_history(this.sql_consensus_listener.consensus.ns_map.iterkeys()) cond.notify() cond.release() cond.acquire()

1 0

[pytorctl/master] Added refresh_all() and warnings to reset_all()
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit eb736ba5e24d1e4259db921bda52ab918c043553 Author: aagbsn <aagbsn(a)extc.org> Date: Thu Jun 23 15:40:05 2011 -0700 Added refresh_all() and warnings to reset_all() SQLSupport.refresh_all() is required to keep Elixir and tc_session in sync. Otherwise it is possible for routers added by the consensus update to not show up in queries using the Elixir model i.e. Router.query.all() Also, warnings have been added to SQLSupport.reset_all() because this does not work properly -- in some cases relation tables were not being reset properly (this resulted in old bw measurements being repeated in future output!). Finally, even when reset_all() works properly, bwauthority memory usage continues to grow. --- SQLSupport.py | 27 +++++++++++++++++++++++++++ ScanSupport.py | 1 + 2 files changed, 28 insertions(+), 0 deletions(-) diff --git a/SQLSupport.py b/SQLSupport.py index 8eefa20..cf2c753 100644 --- a/SQLSupport.py +++ b/SQLSupport.py @@ -657,7 +657,13 @@ class RouterStats(Entity): #################### Model Support ################ def reset_all(): + plog("WARN", "SQLSupport.reset_all() called. See SQLSupport.py for details") # Need to keep routers around.. + # WARNING! + # Must keep the routers around because circ_status_event may + # reference old Routers that are no longer in consensus + # and will raise an ObjectDeletedError. See function circ_status_event in + # class CircuitListener in SQLSupport.py for r in Router.query.all(): # This appears to be needed. the relation tables do not get dropped # automatically. @@ -671,6 +677,16 @@ def reset_all(): tc_session.commit() tc_session.expunge_all() + # WARNING! + # May not clear relation all tables! (SQLAlchemy or Elixir bug) + # Try: + # tc_session.execute('delete from router_streams__stream;') + + # WARNING! + # This will cause Postgres databases to hang + # on DROP TABLE. Possibly an issue with cascade. + # Sqlite works though. + BwHistory.table.drop() # Will drop subclasses Extension.table.drop() Stream.table.drop() @@ -691,6 +707,17 @@ def reset_all(): plog("NOTICE", "Reset all SQL stats") +def refresh_all(): + # necessary to keep all sessions synchronized + # This is probably a bug. See reset_all() above. + # Call this after update_consensus(), _update_rank_history() + # See: ScanSupport.reset_stats() + # Could be a cascade problem too, see: + # http://stackoverflow.com/questions/3481976/sqlalchemy-objectdeletederror-in… + # Also see: + # http://groups.google.com/group/sqlalchemy/browse_thread/thread/c9099eaaffd7… + [tc_session.refresh(r) for r in Router.query.all()] + ##################### End Model Support #################### class ConsensusTrackerListener(TorCtl.DualEventListener): diff --git a/ScanSupport.py b/ScanSupport.py index 97c43a8..8985fd7 100644 --- a/ScanSupport.py +++ b/ScanSupport.py @@ -268,6 +268,7 @@ class SQLScanHandler(ScanHandler): SQLSupport.reset_all() this.sql_consensus_listener.update_consensus() this.sql_consensus_listener._update_rank_history(this.sql_consensus_listener.consensus.ns_map.iterkeys()) + SQLSupport.refresh_all() cond.notify() cond.release() cond.acquire()

1 0

[pytorctl/master] backward compatibility with SQLAlchemy 0.5.x
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit c8fee8b7b5ea6c584bb5a75647ee44737e60b66d Author: aagbsn <aagbsn(a)extc.org> Date: Sun Apr 10 23:59:09 2011 -0700 backward compatibility with SQLAlchemy 0.5.x --- SQLSupport.py | 6 +++++- 1 files changed, 5 insertions(+), 1 deletions(-) diff --git a/SQLSupport.py b/SQLSupport.py index 2bd0a8f..9e1a4cb 100644 --- a/SQLSupport.py +++ b/SQLSupport.py @@ -63,6 +63,10 @@ def setup_db(db_uri, echo=False, drop=False): # wouldn't kill you, you know. tc_session.add = tc_session.save_or_update + if sqlalchemy.__version__ < "0.6.0": + # clear() replaced with expunge_all + tc_session.clear = tc_session.expunge_all + class Router(Entity): using_options(shortnames=True, order_by='-published', session=tc_session, metadata=tc_metadata) using_mapper_options(save_on_init=False) @@ -374,7 +378,7 @@ class RouterStats(Entity): _compute_stats = Callable(_compute_stats) def _compute_ranks(): - tc_session.clear() + tc_session.expunge_all() min_r = select([func.min(BwHistory.rank)], BwHistory.table.c.router_idhex == RouterStats.table.c.router_idhex).as_scalar()

1 0

[pytorctl/master] rewrite query for mysql compatibility attempt 1
by mikeperry＠torproject.org 24 Jun '11

24 Jun '11

commit a416402b0a4c6b5406448a5f87538081f26d9368 Author: aagbsn <aagbsn(a)extc.org> Date: Thu Apr 21 18:28:17 2011 -0700 rewrite query for mysql compatibility attempt 1 --- SQLSupport.py | 11 ++++++++++- 1 files changed, 10 insertions(+), 1 deletions(-) diff --git a/SQLSupport.py b/SQLSupport.py index 9e1a4cb..8ef8b68 100644 --- a/SQLSupport.py +++ b/SQLSupport.py @@ -405,9 +405,18 @@ class RouterStats(Entity): #min_avg_rank = select([func.min(RouterStats.avg_rank)]).as_scalar() max_avg_rank = select([func.max(RouterStats.avg_rank)]).as_scalar() + # this query breaks MySQL! + #RouterStats.table.update(values= + # {RouterStats.table.c.percentile: + # (100.0*RouterStats.table.c.avg_rank)/max_avg_rank}).execute() + + # I think the problem is the reference to RouterStats.table.c.avg_rank -- let's just replace it with the query used above. I think the nested query will be OK + #query('UPDATE routerstats SET percentile=((%s * routerstats.avg_rank) / (SELECT max(routerstats.avg_rank) AS max_1 \nFROM routerstats))' (100.0,)) RouterStats.table.update(values= {RouterStats.table.c.percentile: - (100.0*RouterStats.table.c.avg_rank)/max_avg_rank}).execute() + (100.0*avg_r)/max_avg_rank}).execute() + + tc_session.commit() _compute_ranks = Callable(_compute_ranks)

1 0