[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[or-cvs] r11002: Moved CircuitHandler and StreamHandler classes to PathSuppor (in torflow/trunk: . TorCtl)
Author: renner
Date: 2007-07-31 01:58:18 -0400 (Tue, 31 Jul 2007)
New Revision: 11002
Modified:
torflow/trunk/TorCtl/GeoIPSupport.py
torflow/trunk/TorCtl/PathSupport.py
torflow/trunk/op-addon.py
Log:
Moved CircuitHandler and StreamHandler classes to PathSupport.py and introduced
saving of a network-model to a binary-file for being able to load it on startup.
Modified: torflow/trunk/TorCtl/GeoIPSupport.py
===================================================================
--- torflow/trunk/TorCtl/GeoIPSupport.py 2007-07-31 00:26:54 UTC (rev 11001)
+++ torflow/trunk/TorCtl/GeoIPSupport.py 2007-07-31 05:58:18 UTC (rev 11002)
@@ -56,39 +56,42 @@
# List of continents
continents = [africa, asia, europe, north_america, oceania, south_america]
-# Perform country -- continent mapping
def get_continent(country_code):
+ """ Perform country -- continent mapping """
for c in continents:
if c.contains(country_code):
return c.code
plog("INFO", country_code + " is not on any continent")
return None
-# Get the country code out of a GeoLiteCity record (not used)
+def get_country(ip):
+ """ Get the country via the library """
+ return geoip.country_code_by_addr(ip)
+
def get_country_from_record(ip):
+ """ Get the country code out of a GeoLiteCity record (not used) """
record = geoip.record_by_addr(ip)
if record != None:
return record['country_code']
-# Router class extended to GeoIP
class GeoIPRouter(TorCtl.Router):
+ """ Router class extended to GeoIP """
def __init__(self, router):
self.__dict__ = router.__dict__
- # Select method to get the country_code here
- self.country_code = geoip.country_code_by_addr(self.get_ip_dotted())
- #self.country_code = get_country_from_record(self.get_ip_dotted())
+ self.country_code = get_country(self.get_ip_dotted())
if self.country_code == None:
plog("INFO", self.nickname + ": Country code not found")
self.continent = None
else: self.continent = get_continent(self.country_code)
- # Convert long int back to dotted quad string
def get_ip_dotted(self):
+ """ Convert long int back to dotted quad string """
return socket.inet_ntoa(struct.pack('>I', self.ip))
class GeoIPConfig:
""" Class to configure GeoIP-based path building """
- def __init__(self, unique_countries, max_crossings, entry_country, exit_country, excludes):
+ def __init__(self, unique_countries, max_crossings, entry_country,
+ exit_country, excludes):
# TODO: Somehow ensure validity of the configuration
# Do not use a country twice in a route
Modified: torflow/trunk/TorCtl/PathSupport.py
===================================================================
--- torflow/trunk/TorCtl/PathSupport.py 2007-07-31 00:26:54 UTC (rev 11001)
+++ torflow/trunk/TorCtl/PathSupport.py 2007-07-31 05:58:18 UTC (rev 11002)
@@ -17,10 +17,11 @@
"AtLeastNNodeRestriction", "NotNodeRestriction", "Subnet16Restriction",
"UniqueRestriction", "UniformGenerator", "OrderedExitGenerator",
"BwWeightedGenerator", "PathSelector", "Connection", "NickRestriction",
-"IdHexRestriction", "PathBuilder", "SelectionManager",
-"CountryCodeRestriction", "CountryRestriction", "UniqueCountryRestriction",
-"SingleCountryRestriction", "ContinentRestriction",
-"ContinentJumperRestriction", "UniqueContinentRestriction"]
+"IdHexRestriction", "PathBuilder", "CircuitHandler", "StreamHandler",
+"SelectionManager", "CountryCodeRestriction", "CountryRestriction",
+"UniqueCountryRestriction", "SingleCountryRestriction",
+"ContinentRestriction", "ContinentJumperRestriction",
+"UniqueContinentRestriction"]
#################### Path Support Interfaces #####################
@@ -661,9 +662,12 @@
self.exit = None
self.built = False
self.dirty = False
+ self.closed = False
self.detached_cnt = 0
self.last_extended_at = time.time()
- self.pending_streams = [] # Which stream IDs are pending us
+ self.extend_times = [] # List of all extend-durations
+ self.setup_duration = None # Sum of extend-times
+ self.pending_streams = [] # Which stream IDs are pending us
def id_path(self): return map(lambda r: r.idhex, self.path)
@@ -979,9 +983,278 @@
def bandwidth_event(self, b): pass # For heartbeat only..
+################### CircuitHandler #############################
+
+class CircuitHandler(PathBuilder):
+ """ CircuitHandler that extends from PathBuilder """
+ def __init__(self, c, selmgr, num_circuits, RouterClass):
+ PathBuilder.__init__(self, c, selmgr, RouterClass)
+ self.num_circuits = num_circuits # Size of the circuit pool
+ self.check_circuit_pool() # Bring up the pool of circs
+
+ def check_circuit_pool(self):
+ """ Init or check the status of the circuit-pool """
+ # Get current number of circuits
+ n = len(self.circuits.values())
+ i = self.num_circuits-n
+ if i > 0:
+ plog("INFO", "Checked pool of circuits: we need to build " +
+ str(i) + " circuits")
+ # Schedule (num_circs-n) circuit-buildups
+ while (n < self.num_circuits):
+ self.build_circuit("255.255.255.255", 80)
+ plog("DEBUG", "Scheduled circuit No. " + str(n+1))
+ n += 1
+
+ def build_circuit(self, host, port):
+ """ Build a circuit """
+ circ = None
+ while circ == None:
+ try:
+ self.selmgr.set_target(host, port)
+ circ = self.c.build_circuit(self.selmgr.pathlen,
+ self.selmgr.path_selector)
+ self.circuits[circ.circ_id] = circ
+ except TorCtl.ErrorReply, e:
+ # FIXME: How come some routers are non-existant? Shouldn't
+ # we have gotten an NS event to notify us they disappeared?
+ plog("NOTICE", "Error building circuit: " + str(e.args))
+ return circ
+
+ def close_circuit(self, id):
+ """ Close a circuit with given id """
+ # TODO: Pass streams to another circ before closing?
+ self.circuits[id].closed = True
+ try: self.c.close_circuit(id)
+ except TorCtl.ErrorReply, e:
+ plog("ERROR", "Failed closing circuit " + str(id) + ": " + str(e))
+
+ def circ_status_event(self, c):
+ """ Handle circuit status events """
+ output = [c.event_name, str(c.circ_id), c.status]
+ if c.path: output.append(",".join(c.path))
+ if c.reason: output.append("REASON=" + c.reason)
+ if c.remote_reason: output.append("REMOTE_REASON=" + c.remote_reason)
+ plog("DEBUG", " ".join(output))
+
+ # Circuits we don't control get built by Tor
+ if c.circ_id not in self.circuits:
+ plog("DEBUG", "Ignoring circuit " + str(c.circ_id) +
+ " (controlled by Tor)")
+ return
+
+ # EXTENDED
+ if c.status == "EXTENDED":
+ # Compute elapsed time
+ extend_time = c.arrived_at-self.circuits[c.circ_id].last_extended_at
+ self.circuits[c.circ_id].extend_times.append(extend_time)
+ plog("INFO", "Circuit " + str(c.circ_id) + " extended in " +
+ str(extend_time) + " sec")
+ self.circuits[c.circ_id].last_extended_at = c.arrived_at
+
+ # FAILED & CLOSED
+ elif c.status == "FAILED" or c.status == "CLOSED":
+ # XXX: Can still get a STREAM FAILED for this circ after this
+ circ = self.circuits[c.circ_id]
+ # Actual removal of the circ
+ del self.circuits[c.circ_id]
+ # Give away pending streams
+ for stream in circ.pending_streams:
+ plog("DEBUG", "Finding new circ for " + str(stream.strm_id))
+ self.attach_stream_any(stream, stream.detached_from)
+ # Check if there are enough circs
+ self.check_circuit_pool()
+ return
+
+ # BUILT
+ elif c.status == "BUILT":
+ circ = self.circuits[c.circ_id]
+ circ.built = True
+ for stream in circ.pending_streams:
+ try:
+ self.c.attach_stream(stream.strm_id, c.circ_id)
+ except TorCtl.ErrorReply, e:
+ # No need to retry here. We should get the failed
+ # event for either the circ or stream next
+ plog("WARN", "Error attaching stream: " + str(e.args))
+ # Compute duration by summing up extend_times
+ duration = reduce(lambda x, y: x+y, circ.extend_times, 0.0)
+ plog("INFO", "Circuit " + str(c.circ_id) + " needed " +
+ str(duration) + " seconds to be built")
+ # Save the duration to the circuit for later use
+ circ.setup_duration = duration
+
+ # OTHER?
+ else:
+ # If this was e.g. a LAUNCHED
+ pass
+
+################### StreamHandler ##############################
+
+class StreamHandler(CircuitHandler):
+ """ StreamHandler that extends from the CircuitHandler """
+ def __init__(self, c, selmgr, num_circs, RouterClass):
+ CircuitHandler.__init__(self, c, selmgr, num_circs, RouterClass)
+ self.sorted_circs = None # optional sorted list
+
+ def clear_dns_cache(self):
+ """ Send signal CLEARDNSCACHE """
+ lines = self.c.sendAndRecv("SIGNAL CLEARDNSCACHE\r\n")
+ for _, msg, more in lines:
+ plog("DEBUG", "CLEARDNSCACHE: " + msg)
+
+ def close_stream(self, id, reason):
+ """ Close a stream with given id and reason """
+ self.c.close_stream(id, reason)
+
+ def create_and_attach(self, stream, unattached_streams):
+ """ Create a new circuit and attach (stream + unattached_streams) """
+ circ = self.build_circuit(stream.host, stream.port)
+ for u in unattached_streams:
+ plog("DEBUG", "Attaching " + str(u.strm_id) +
+ " pending build of circuit " + str(circ.circ_id))
+ u.pending_circ = circ
+ circ.pending_streams.extend(unattached_streams)
+ self.circuits[circ.circ_id] = circ
+ self.last_exit = circ.exit
+
+ def attach_stream_any(self, stream, badcircs):
+ """ Attach a regular user stream """
+ unattached_streams = [stream]
+ if self.new_nym:
+ self.new_nym = False
+ plog("DEBUG", "Obeying new nym")
+ for key in self.circuits.keys():
+ if (not self.circuits[key].dirty
+ and len(self.circuits[key].pending_streams)):
+ plog("WARN", "New nym called, destroying circuit "+str(key)
+ +" with "+str(len(self.circuits[key].pending_streams))
+ +" pending streams")
+ unattached_streams.extend(self.circuits[key].pending_streams)
+ del self.circuits[key].pending_streams[:]
+ # FIXME: Consider actually closing circs if no streams
+ self.circuits[key].dirty = True
+
+ # Check if there is a sorted list of circs
+ if self.sorted_circs: list = self.sorted_circs
+ else: list = self.circuits.values()
+ for circ in list:
+ # Check each circuit
+ if circ.built and not circ.closed and circ.circ_id not in badcircs and not circ.dirty:
+ if circ.exit.will_exit_to(stream.host, stream.port):
+ try:
+ self.c.attach_stream(stream.strm_id, circ.circ_id)
+ stream.pending_circ = circ # Only one possible here
+ circ.pending_streams.append(stream)
+ self.last_exit = circ.exit
+ except TorCtl.ErrorReply, e:
+ # No need to retry here. We should get the failed
+ # event for either the circ or stream next
+ plog("WARN", "Error attaching stream: " + str(e.args))
+ return
+ break
+ else:
+ plog("DEBUG", "Circuit " + str(circ.circ_id) + " won't exit")
+ else:
+ self.create_and_attach(stream, unattached_streams)
+
+ def stream_status_event(self, s):
+ """ Catch user stream events """
+ # Construct debugging output
+ output = [s.event_name, str(s.strm_id), s.status, str(s.circ_id), s.target_host, str(s.target_port)]
+ if s.reason: output.append("REASON=" + s.reason)
+ if s.remote_reason: output.append("REMOTE_REASON=" + s.remote_reason)
+ plog("DEBUG", " ".join(output))
+
+ # If target_host is not an IP-address
+ if not re.match(r"\d+.\d+.\d+.\d+", s.target_host):
+ s.target_host = "255.255.255.255" # ignore DNS for exit policy check
+
+ # NEW or NEWRESOLVE
+ if s.status == "NEW" or s.status == "NEWRESOLVE":
+ if s.status == "NEWRESOLVE" and not s.target_port:
+ s.target_port = self.resolve_port
+ # Set up the new stream
+ stream = Stream(s.strm_id, s.target_host, s.target_port, s.status)
+ self.streams[s.strm_id] = stream
+ self.attach_stream_any(self.streams[s.strm_id], self.streams[s.strm_id].detached_from)
+
+ # DETACHED
+ elif s.status == "DETACHED":
+ # Stream not found
+ if s.strm_id not in self.streams:
+ plog("WARN", "Detached stream " + str(s.strm_id) + " not found")
+ self.streams[s.strm_id] = Stream(s.strm_id, s.target_host, s.target_port, "NEW")
+ # Circuit not found
+ if not s.circ_id:
+ plog("WARN", "Stream " + str(s.strm_id) + " detached from no circuit!")
+ else:
+ self.streams[s.strm_id].detached_from.append(s.circ_id)
+ # Detect timeouts on user streams
+ if s.reason == "TIMEOUT":
+ # TODO: Count timeouts on streams?
+ #self.streams[s.strm_id].timeout_counter += 1
+ plog("DEBUG", "User stream timed out on circuit " + str(s.circ_id))
+ # Stream was pending
+ if self.streams[s.strm_id] in self.streams[s.strm_id].pending_circ.pending_streams:
+ self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
+ # Attach to another circ
+ self.streams[s.strm_id].pending_circ = None
+ self.attach_stream_any(self.streams[s.strm_id], self.streams[s.strm_id].detached_from)
+
+ # SUCCEEDED
+ if s.status == "SUCCEEDED":
+ if s.strm_id not in self.streams:
+ plog("NOTICE", "Succeeded stream " + str(s.strm_id) + " not found")
+ return
+ if s.circ_id and self.streams[s.strm_id].pending_circ.circ_id != s.circ_id:
+ # Hrmm.. this can happen on a new-nym.. Very rare, putting warn
+ # in because I'm still not sure this is correct
+ plog("WARN", "Mismatch of pending: "
+ + str(self.streams[s.strm_id].pending_circ.circ_id) + " vs "
+ + str(s.circ_id))
+ self.streams[s.strm_id].circ = self.circuits[s.circ_id]
+ else:
+ self.streams[s.strm_id].circ = self.streams[s.strm_id].pending_circ
+ self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
+ self.streams[s.strm_id].pending_circ = None
+ self.streams[s.strm_id].attached_at = s.arrived_at
+
+ # FAILED or CLOSED
+ elif s.status == "FAILED" or s.status == "CLOSED":
+ if s.strm_id not in self.streams:
+ plog("NOTICE", "Failed stream " + str(s.strm_id) + " not found")
+ return
+ # if not s.circ_id:
+ # plog("WARN", "Stream " + str(s.strm_id) + " closed/failed from no circuit")
+ # We get failed and closed for each stream, let CLOSED do the cleanup
+ if s.status == "FAILED":
+ # Avoid busted circuits that will not resolve or carry traffic
+ self.streams[s.strm_id].failed = True
+ if s.circ_id in self.circuits: self.circuits[s.circ_id].dirty = True
+ elif self.streams[s.strm_id].attached_at != 0:
+ plog("WARN", "Failed stream on unknown circuit " + str(s.circ_id))
+ return
+ # CLOSED
+ if self.streams[s.strm_id].pending_circ:
+ self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
+ # Actual removal of the stream
+ del self.streams[s.strm_id]
+
+ # REMAP
+ elif s.status == "REMAP":
+ if s.strm_id not in self.streams:
+ plog("WARN", "Remap id "+str(s.strm_id)+" not found")
+ else:
+ if not re.match(r"\d+.\d+.\d+.\d+", s.target_host):
+ s.target_host = "255.255.255.255"
+ plog("NOTICE", "Non-IP remap for "+str(s.strm_id) +
+ " to " + s.target_host)
+ self.streams[s.strm_id].host = s.target_host
+ self.streams[s.strm_id].port = s.target_port
+
########################## Unit tests ##########################
-
def do_unit(rst, r_list, plamb):
print "\n"
print "-----------------------------------"
Modified: torflow/trunk/op-addon.py
===================================================================
--- torflow/trunk/op-addon.py 2007-07-31 00:26:54 UTC (rev 11001)
+++ torflow/trunk/op-addon.py 2007-07-31 05:58:18 UTC (rev 11002)
@@ -52,7 +52,6 @@
GEOIP = "GeoIP"
RTT = "RTT"
-# Set global variables here
# Measure the circuits
measure_circs = config.getboolean(RTT, "measure_circs")
if measure_circs:
@@ -128,6 +127,9 @@
use_guards = config.getboolean(NODE_SELECTION, "use_guards"),
geoip_config = get_geoip_config())
+# Path to directory where to store files
+data_dir = "data/op-addon/"
+
## Connection #################################################################
class Connection(TorCtl.Connection):
@@ -200,7 +202,6 @@
## FileHandler ################################################################
-# TODO: Move this to TorCtl.TorUtil?
class FileHandler:
""" FileHandler class for writing/appending collected data to a file """
def __init__(self, filename):
@@ -236,11 +237,7 @@
self.timeout_counter = 0 # timeout limit
self.slowness_counter = 0 # slowness limit
self.rtt_created = False # if this was created from the model
- # TODO: Move these to PathSupport.py?
- self.closed = False # mark circuit closed
- self.extend_times = [] # list of all extend-times
- self.setup_duration = None # sum of extend-times
-
+
def add_rtt(self, rtt):
""" Add a new value and refresh the stats """
# Set current
@@ -276,7 +273,6 @@
## CircuitBuildingStats #######################################################
-# TODO: Move to TorCtl.TorUtil?
class CircuitBuildingStats(Stats):
""" Create an instance of this and gather overall circuit stats """
def __init__(self):
@@ -329,27 +325,43 @@
""" Create a string for printing out information """
s = ""
for l in self.links:
- # Get the single objects
- s += l.src.nickname + "--" + l.dest.nickname +\
- " (" + str(l.current_rtt) + ") " + ", "
- return "Route proposal: " + s + "--> " + str(self.rtt) + " sec"
+ s += str(l.src) + "--" + l.dest + " (" + str(l.current_rtt) + ") " + ", "
+ return s + "--> " + str(self.rtt) + " sec"
class NetworkModel:
""" This class is used to record measured RTTs for single links in a model
of the 'currently explored subnet' (undirected graph) """
- def __init__(self, rooter):
+ def __init__(self, logfile=None):
""" Constructor: pass the root of all our circuits """
- # Use XDiGraph() (= directed)?
- self.graph = networkx.XGraph(name="Explored Tor Subnet",
- selfloops=False, multiedges=False)
- # Initially add THIS proxy to the model
- self.root = rooter
- self.graph.add_node(self.root)
- self.proposals = []
- plog("DEBUG", "NetworkModel initiated: added " + self.root.nickname)
+ self.pickle_path = "data/op-addon/network-model.pickle"
+ self.logfile = logfile # Optional logfile
+ self.proposals = [] # Current list of circ-proposals
+ self.prefixes = {} # Prefixes for DFS
+ try:
+ self.graph = self.load_graph()
+ self.find_all_proposals()
+ self.print_info()
+ except:
+ plog("INFO", "Could not load a model, creating a new one ..")
+ self.graph = networkx.XGraph(name="Explored Tor Subnet")
+ self.graph.add_node(None)
+ plog("INFO", "NetworkModel initiated")
+ def save_graph(self):
+ """ Write the graph to a binary file """
+ start = time.time()
+ networkx.write_gpickle(self.graph, self.pickle_path)
+ plog("INFO", "Saved graph to '" + self.pickle_path +
+ "' in " + str(time.time()-start) + " sec")
+
+ def load_graph(self):
+ """ Load a graph from a binary file and return it """
+ graph = networkx.read_gpickle(self.pickle_path)
+ plog("INFO", "Loaded graph from '" + self.pickle_path + "'")
+ return graph
+
def add_link(self, src, dest, rtt):
- """ Add a link to the graph given src, dest & rtt """
+ """ Add link to the graph given src, dest (routers) & rtt (LinkInfo) """
self.graph.add_edge(src, dest, LinkInfo(src, dest, rtt))
def add_circuit(self, c):
@@ -363,14 +375,14 @@
# First hop --> add Link from Root to 1
if i == 1:
link_rtt = c.part_rtts[i]
- self.add_link(self.root, c.path[i-1], link_rtt)
+ self.add_link(None, c.path[i-1].idhex, link_rtt)
# Handle i -- (i+1)
if i+1 in c.part_rtts:
link_rtt = c.part_rtts[i+1] - c.part_rtts[i]
if link_rtt > 0:
plog("INFO", "Computed link-RTT " + str(i) + ": " + str(link_rtt))
# Save to NetworkModel
- self.add_link(c.path[i-1], c.path[i], link_rtt)
+ self.add_link(c.path[i-1].idhex, c.path[i].idhex, link_rtt)
else:
plog("WARN", "Negative link-RTT " + str(i) + ": " + str(link_rtt))
# Handle (n-1) -- n
@@ -380,7 +392,7 @@
if link_rtt > 0:
plog("INFO", "Computed link-RTT " + str(i) + ": " + str(link_rtt))
# Save to NetworkModel
- self.add_link(c.path[i-1], c.path[i], link_rtt)
+ self.add_link(c.path[i-1].idhex, c.path[i].idhex, link_rtt)
else:
plog("WARN", "Negative link-RTT " + str(i) + ": " + str(link_rtt))
@@ -392,36 +404,22 @@
links.append(self.graph.get_edge(path[i], path[i+1]))
return links
- def find_circuits(self):
+ def find_all_proposals(self):
+ """ Call visit on the root-node """
# Reset list of proposals and prefixes for DFS
self.proposals = []
- self.prefixes = {}
- # Measure for info
+ self.prefixes.clear()
start = time.time()
# Start the search
- self.visit(self.root, [])
+ self.visit(None, [])
# Sort proposals for their RTTs
sort_list(self.proposals, lambda x: x.rtt)
- # Some logging
- plog("DEBUG", "Finding " + str(len(self.proposals)) +
+ plog("INFO", "Finding " + str(len(self.proposals)) +
" proposals and sorting them took us " +
str(time.time()-start) + " seconds")
- # Print all of them for debugging/info
- for p in self.proposals:
- print(p.to_string())
- def get_proposals(self, n):
- """ Return all proposals with rtt <= n seconds """
- ret = []
- for p in self.proposals:
- if p.rtt <= n:
- ret.append(p)
- plog("DEBUG", "Found " + str(len(ret)) +
- " path proposals having RTT <= " + str(n) + " sec")
- return ret
-
def visit(self, node, path, i=1):
- """ Recursive Depth-First-Search: Maybe use some existing method? """
+ """ Recursive Depth-First-Search: Maybe use some existing methods """
if node not in path:
path.append(node)
# Root -- Exit
@@ -435,340 +433,57 @@
if n not in self.prefixes[i]:
self.visit(n, copy.copy(self.prefixes[i]), i+1)
- def print_graph(self):
- """ Print current info about the graph """
- print(self.graph.info())
+ def get_proposals(self, n):
+ """ Return all proposals having rtt <= n seconds """
+ ret = []
+ for p in self.proposals:
+ if p.rtt <= n:
+ ret.append(p)
+ plog("INFO", "Found " + str(len(ret)) +
+ " path proposals having RTT <= " + str(n) + " sec")
+ return ret
-## EventHandlers ##############################################################
-
-# TODO: Move to PathSupport
-class CircuitHandler(PathSupport.PathBuilder):
- """ CircuitHandler that extends from PathBuilder """
- def __init__(self, c, selmgr, num_circuits):
- # Init the PathBuilder
- PathSupport.PathBuilder.__init__(self, c, selmgr, GeoIPSupport.GeoIPRouter)
- self.num_circuits = num_circuits # size of the circuit pool
- self.check_circuit_pool() # bring up the pool of circs
- self.circ_stats = CircuitBuildingStats() # record setup-times
- # Filehandlers for saving stats about circuit building
- self.stats_logger = FileHandler("data/op-addon/circ-setup-stats")
- self.setup_logger = FileHandler("data/op-addon/circ-setup-durations")
-
- def check_circuit_pool(self):
- """ Init or check the status of our pool of circuits """
- # Get current number of circuits
- n = len(self.circuits.values())
- i = self.num_circuits - n
- if i > 0:
- plog("INFO", "Checked pool of circuits: we need to build " +
- str(i) + " circuits")
- # Schedule (num_circs - n) circuit-buildups
- while (n < self.num_circuits):
- self.build_idle_circuit()
- plog("DEBUG", "Scheduled circuit No. " + str(n+1))
- n += 1
-
- def close_circuit(self, id):
- """ Try to close a circuit with given id """
- # TODO: Pass streams to another circ before closing?
- self.circuits[id].closed = True
- try: self.c.close_circuit(id)
- except TorCtl.ErrorReply, e:
- plog("ERROR", "Failed closing circuit " + str(id) + ": " + str(e))
-
- def print_circuits(self, list=None):
- """ Print out the circuits + some info, optionally pass a (sorted) list """
- if list: circs = list
- else: circs = self.circuits.values()
- plog("INFO", "We have " + str(len(circs)) + " circuits:")
- for c in circs:
- print("+ " + c.to_string())
-
- def build_idle_circuit(self):
- """ Build an idle circuit """
- circ = None
- while circ == None:
- try:
- # Configure which port to use here
- self.selmgr.set_target("255.255.255.255", 80)
- circ = self.c.build_circuit(self.selmgr.pathlen,
- self.selmgr.path_selector)
- self.circuits[circ.circ_id] = circ
- except TorCtl.ErrorReply, e:
- # FIXME: How come some routers are non-existant? Shouldn't
- # we have gotten an NS event to notify us they disappeared?
- plog("NOTICE", "Error building circuit: " + str(e.args))
-
- def circ_status_event(self, c):
- """ Handle circuit status events """
- # Construct output for logging
- output = [c.event_name, str(c.circ_id), c.status]
- if c.path: output.append(",".join(c.path))
- if c.reason: output.append("REASON=" + c.reason)
- if c.remote_reason: output.append("REMOTE_REASON=" + c.remote_reason)
- plog("DEBUG", " ".join(output))
-
- # Circuits we don't control get built by Tor
- if c.circ_id not in self.circuits:
- plog("DEBUG", "Ignoring circuit " + str(c.circ_id) +
- " (controlled by Tor or not yet in the list)")
- return
-
- # EXTENDED
- if c.status == "EXTENDED":
- # Compute elapsed time
- extend_time = c.arrived_at - self.circuits[c.circ_id].last_extended_at
- # Add to the list
- self.circuits[c.circ_id].extend_times.append(extend_time)
- plog("DEBUG", "Circuit " + str(c.circ_id) + " extended in " +
- str(extend_time) + " sec")
- self.circuits[c.circ_id].last_extended_at = c.arrived_at
-
- # FAILED & CLOSED
- elif c.status == "FAILED" or c.status == "CLOSED":
- # XXX: Can still get a STREAM FAILED for this circ after this
- circ = self.circuits[c.circ_id]
-
- # Logging and statistics
- if not circ.built:
- message = ["FAILED"]
- if c.reason: message.append("REASON=" + c.reason)
- if c.remote_reason: message.append("REMOTE_REASON=" + c.remote_reason)
- self.setup_logger.append(" ".join(message) + ": " +
- str(circ.extend_times))
- # Increase counter and write circ_stats to file
- self.circ_stats.failures += 1
- self.stats_logger.write(self.circ_stats.to_string())
-
- # Actual removal of the circ
- del self.circuits[c.circ_id]
- # Give away pending streams
- for stream in circ.pending_streams:
- plog("DEBUG", "Finding new circ for " + str(stream.strm_id))
- self.attach_stream_any(stream, stream.detached_from)
- # Check if there are enough circs
- self.check_circuit_pool()
- return
-
- # BUILT
- elif c.status == "BUILT":
- self.circuits[c.circ_id].built = True
- for stream in self.circuits[c.circ_id].pending_streams:
- try:
- self.c.attach_stream(stream.strm_id, c.circ_id)
- except TorCtl.ErrorReply, e:
- # No need to retry here. We should get the failed
- # event for either the circ or stream next
- plog("WARN", "Error attaching stream: " + str(e.args))
-
- # Log setup durations to file
- self.setup_logger.append(str(self.circuits[c.circ_id].extend_times))
- # Compute duration by summing up extend_times
- duration = reduce(lambda x, y: x+y,
- self.circuits[c.circ_id].extend_times, 0.0)
- plog("DEBUG", "Circuit " + str(c.circ_id) + " needed " +
- str(duration) + " seconds to be built")
- # Add duration to circ_stats and write file
- self.circ_stats.add_value(duration)
- self.stats_logger.write(self.circ_stats.to_string())
- # Save the duration to the circuit for later use
- self.circuits[c.circ_id].setup_duration = duration
-
- # OTHER?
+ def print_info(self):
+ """ Create a string holding info and the proposals for printing """
+ out = str(self.graph.info())
+ for p in self.proposals:
+ out += "\nPath Proposal: " + p.to_string()
+ # Only print them out if there are not too much
+ if len(self.proposals) < 100:
+ print(out)
else:
- # If this was e.g. a LAUNCHED
- pass
+ plog("INFO", "More than 100 proposals!")
+ print(self.graph.info())
+ # But log all of them to the file if it exists
+ if self.logfile: self.logfile.write(out)
-## StreamHandler ##############################################################
-
-# TODO: Move to PathSupport
-class StreamHandler(CircuitHandler):
- """ This is a StreamHandler that extends from the CircuitHandler """
- def __init__(self, c, selmgr, num_circs):
- # Call constructor of superclass
- CircuitHandler.__init__(self, c, selmgr, num_circs)
- self.sorted_circs = None # optional
- #self.new_nym = True
-
- def clear_dns_cache(self):
- """ Send signal CLEARDNSCACHE """
- lines = self.c.sendAndRecv("SIGNAL CLEARDNSCACHE\r\n")
- for _, msg, more in lines:
- plog("DEBUG", "CLEARDNSCACHE: " + msg)
-
- def close_stream(self, id, reason):
- """ Close a stream with given id and reason """
- self.c.close_stream(id, reason)
-
- def create_and_attach(self, stream, unattached_streams):
- """ Create a new circuit and attach (stream + unattached_streams) """
- circ = None
- self.selmgr.set_target(stream.host, stream.port)
- while circ == None:
- try:
- circ = self.c.build_circuit(self.selmgr.pathlen, self.selmgr.path_selector)
- except TorCtl.ErrorReply, e:
- plog("NOTICE", "Error building circ: " + str(e.args))
- for u in unattached_streams:
- plog("DEBUG", "Attaching " + str(u.strm_id) + " pending build of circuit " + str(circ.circ_id))
- u.pending_circ = circ
- circ.pending_streams.extend(unattached_streams)
- self.circuits[circ.circ_id] = circ
- self.last_exit = circ.exit
-
- def attach_stream_any(self, stream, badcircs):
- """ Attach a regular user stream """
- unattached_streams = [stream]
- if self.new_nym:
- self.new_nym = False
- plog("DEBUG", "Obeying new nym")
- for key in self.circuits.keys():
- if (not self.circuits[key].dirty
- and len(self.circuits[key].pending_streams)):
- plog("WARN", "New nym called, destroying circuit "+str(key)
- +" with "+str(len(self.circuits[key].pending_streams))
- +" pending streams")
- unattached_streams.extend(self.circuits[key].pending_streams)
- del self.circuits[key].pending_streams[:]
- # FIXME: Consider actually closing circs if no streams
- self.circuits[key].dirty = True
-
- # Check if there is a sorted list of circs
- if self.sorted_circs: list = self.sorted_circs
- else: list = self.circuits.values()
- for circ in list:
- # Check each circuit
- if circ.built and not circ.closed and circ.circ_id not in badcircs and not circ.dirty:
- if circ.exit.will_exit_to(stream.host, stream.port):
- try:
- self.c.attach_stream(stream.strm_id, circ.circ_id)
- stream.pending_circ = circ # Only one possible here
- circ.pending_streams.append(stream)
- # Clear cache after the attach?
- #self.clear_dns_cache()
- self.last_exit = circ.exit
- except TorCtl.ErrorReply, e:
- # No need to retry here. We should get the failed
- # event for either the circ or stream next
- plog("WARN", "Error attaching stream: " + str(e.args))
- return
- break
- else:
- plog("DEBUG", "Circuit " + str(circ.circ_id) + " won't exit")
- else:
- self.create_and_attach(stream, unattached_streams)
-
- def stream_status_event(self, s):
- """ Catch user stream events """
- # Construct debugging output
- output = [s.event_name, str(s.strm_id), s.status, str(s.circ_id), s.target_host, str(s.target_port)]
- if s.reason: output.append("REASON=" + s.reason)
- if s.remote_reason: output.append("REMOTE_REASON=" + s.remote_reason)
- plog("DEBUG", " ".join(output))
-
- # If target_host is not an IP-address
- if not re.match(r"\d+.\d+.\d+.\d+", s.target_host):
- s.target_host = "255.255.255.255" # ignore DNS for exit policy check
-
- # NEW or NEWRESOLVE
- if s.status == "NEW" or s.status == "NEWRESOLVE":
- if s.status == "NEWRESOLVE" and not s.target_port:
- s.target_port = self.resolve_port
- # Set up the new stream
- stream = Stream(s.strm_id, s.target_host, s.target_port, s.status)
- self.streams[s.strm_id] = stream
- self.attach_stream_any(self.streams[s.strm_id], self.streams[s.strm_id].detached_from)
-
- # DETACHED
- elif s.status == "DETACHED":
- # Stream not found
- if s.strm_id not in self.streams:
- plog("WARN", "Detached stream " + str(s.strm_id) + " not found")
- self.streams[s.strm_id] = Stream(s.strm_id, s.target_host, s.target_port, "NEW")
- # Circuit not found
- if not s.circ_id:
- plog("WARN", "Stream " + str(s.strm_id) + " detached from no circuit!")
- else:
- self.streams[s.strm_id].detached_from.append(s.circ_id)
- # Detect timeouts on user streams
- if s.reason == "TIMEOUT":
- # TODO: Count timeouts on the stream?
- #self.streams[s.strm_id].timeout_counter += 1
- plog("DEBUG", "User stream timed out on circuit " + str(s.circ_id))
- # Stream was pending
- if self.streams[s.strm_id] in self.streams[s.strm_id].pending_circ.pending_streams:
- self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
- # Attach to another circ
- self.streams[s.strm_id].pending_circ = None
- self.attach_stream_any(self.streams[s.strm_id], self.streams[s.strm_id].detached_from)
-
- # SUCCEEDED
- if s.status == "SUCCEEDED":
- if s.strm_id not in self.streams:
- plog("NOTICE", "Succeeded stream " + str(s.strm_id) + " not found")
- return
- if s.circ_id and self.streams[s.strm_id].pending_circ.circ_id != s.circ_id:
- # Hrmm.. this can happen on a new-nym.. Very rare, putting warn
- # in because I'm still not sure this is correct
- plog("WARN", "Mismatch of pending: "
- + str(self.streams[s.strm_id].pending_circ.circ_id) + " vs "
- + str(s.circ_id))
- self.streams[s.strm_id].circ = self.circuits[s.circ_id]
- else:
- self.streams[s.strm_id].circ = self.streams[s.strm_id].pending_circ
- self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
- self.streams[s.strm_id].pending_circ = None
- self.streams[s.strm_id].attached_at = s.arrived_at
-
- # FAILED or CLOSED
- elif s.status == "FAILED" or s.status == "CLOSED":
- if s.strm_id not in self.streams:
- plog("NOTICE", "Failed stream " + str(s.strm_id) + " not found")
- return
- #if not s.circ_id: plog("WARN", "Stream " + str(s.strm_id) + " closed/failed from no circuit")
- # We get failed and closed for each stream. OK to return and let the CLOSED do the cleanup
- if s.status == "FAILED":
- # Avoid busted circuits that will not resolve or carry traffic
- self.streams[s.strm_id].failed = True
- if s.circ_id in self.circuits: self.circuits[s.circ_id].dirty = True
- elif self.streams[s.strm_id].attached_at != 0:
- plog("WARN", "Failed stream on unknown circuit " + str(s.circ_id))
- return
- # CLOSED
- if self.streams[s.strm_id].pending_circ:
- self.streams[s.strm_id].pending_circ.pending_streams.remove(self.streams[s.strm_id])
- # Actual removal of the stream
- del self.streams[s.strm_id]
-
- # REMAP
- elif s.status == "REMAP":
- if s.strm_id not in self.streams:
- plog("WARN", "Remap id "+str(s.strm_id)+" not found")
- else:
- if not re.match(r"\d+.\d+.\d+.\d+", s.target_host):
- s.target_host = "255.255.255.255"
- plog("NOTICE", "Non-IP remap for "+str(s.strm_id) + " to " + s.target_host)
- self.streams[s.strm_id].host = s.target_host
- self.streams[s.strm_id].port = s.target_port
-
## PingHandler ################################################################
-class PingHandler(StreamHandler):
+class PingHandler(PathSupport.StreamHandler):
""" This class extends the general StreamHandler to handle ping-requests """
- def __init__(self, c, selmgr, num_circs, router, partial=False):
+ def __init__(self, c, selmgr, num_circs, RouterClass, partial=False):
# Anything ping-related
self.ping_queue = Queue.Queue() # (circ_id, hop)-pairs
- self.start_times = {} # dict mapping (circ_id, hop):start_time
+ self.start_times = {} # dict (circ_id, hop):start_time
+
+ # Handle global testing_mode
+ if testing_mode:
+ self.latency_logger = FileHandler(data_dir + "mean-latencies")
+
# Additional stuff for measuring single links
self.partial_circs = partial
if self.partial_circs:
- self.router = router # object that represents this OR
- self.model = NetworkModel(self.router) # model for recording link-RTTs
- # Handle testing_mode
- if testing_mode:
- self.latency_logger= FileHandler("data/op-addon/mean-latencies")
+ self.model_logger = FileHandler(data_dir + "proposals")
+ self.model = NetworkModel(self.model_logger)
+
+ # Stuff for recording statistics
+ self.circ_stats = CircuitBuildingStats() # record setup-durations
+ self.stats_logger = FileHandler(data_dir + "circ-setup-stats")
+ self.setup_logger = FileHandler(data_dir + "circ-setup-durations")
+
# Init the StreamHandler
- StreamHandler.__init__(self, c, selmgr, num_circs)
+ PathSupport.StreamHandler.__init__(self, c, selmgr, num_circs, RouterClass)
+
# Sorted circuit list
self.sorted_circs = [] # list of circs sorted by current RTT
# Start the Pinger that triggers the connections
@@ -785,11 +500,19 @@
self.sorted_circs = sort_list(self.circuits.values(), notlambda)
plog("DEBUG", "Refreshed sorted list of circuits")
+ def print_circuits(self, list=None):
+ """ Print out the circuits + some info, optionally pass a (sorted) list """
+ if list: circs = list
+ else: circs = self.circuits.values()
+ plog("INFO", "We have " + str(len(circs)) + " circuits:")
+ for c in circs:
+ print("+ " + c.to_string())
+
def enqueue_pings(self):
- """ To be schedule_immediated by pinger before the initial connection is triggered """
+ """ schedule_immediate from pinger before triggering the initial ping """
print("")
self.refresh_sorted_list()
- # XXX: Check if there are any, else let the Pinger wait a bit?
+ # XXX: Check if there are any, else let the Pinger wait?
circs = self.circuits.values()
for c in circs:
if c.built:
@@ -806,7 +529,7 @@
plog("DEBUG", "Enqueued circuit " + str(id) + " hop None")
def established(self, circ_list):
- """ Check if there is at least one circuit built """
+ """ Check if there is at least one circuit established """
for c in circ_list:
if c.built:
return True
@@ -824,8 +547,8 @@
self.print_circuits(self.sorted_circs)
if self.partial_circs:
# Print out the model
- self.model.print_graph()
- self.model.find_circuits()
+ self.model.find_all_proposals()
+ self.model.print_info()
# Enqueue again all circs
self.enqueue_pings()
@@ -917,12 +640,14 @@
def stream_status_event(self, s):
""" Separate pings from regular streams directly """
- if not (s.target_host == ping_dummy_host and s.target_port == ping_dummy_port):
+ if not (s.target_host == ping_dummy_host and
+ s.target_port == ping_dummy_port):
# This is no ping, call the other method
- return StreamHandler.stream_status_event(self, s)
+ return PathSupport.StreamHandler.stream_status_event(self, s)
# Construct debugging output
- output = [s.event_name, str(s.strm_id), s.status, str(s.circ_id), s.target_host, str(s.target_port)]
+ output = [s.event_name, str(s.strm_id), s.status, str(s.circ_id),
+ s.target_host, str(s.target_port)]
if s.reason: output.append("REASON=" + s.reason)
if s.remote_reason: output.append("REMOTE_REASON=" + s.remote_reason)
plog("DEBUG", " ".join(output))
@@ -944,11 +669,13 @@
if (s.reason == "TIMEOUT"):
self.circuits[s.circ_id].timeout_counter += 1
self.circuits[s.circ_id].slowness_counter += 1
- plog("DEBUG", str(self.circuits[s.circ_id].timeout_counter) + " timeout(s) on circuit " + str(s.circ_id))
+ plog("DEBUG", str(self.circuits[s.circ_id].timeout_counter) +
+ " timeout(s) on circuit " + str(s.circ_id))
if timeout_limit > 0:
if self.circuits[s.circ_id].timeout_counter >= timeout_limit and not self.circuits[s.circ_id].closed:
# Close the circuit
- plog("DEBUG", "Reached limit on timeouts --> closing circuit " + str(s.circ_id))
+ plog("DEBUG", "Reached limit on timeouts --> closing circuit "
+ + str(s.circ_id))
self.close_circuit(s.circ_id)
# Set RTT for this circ to None
self.circuits[s.circ_id].current_rtt = None
@@ -965,6 +692,48 @@
# Only record
self.record_ping(s)
+ def circ_status_event(self, c):
+ """ Override to record statistics on circuit-setups and -failures """
+ if c.circ_id not in self.circuits:
+ return PathSupport.CircuitHandler.circ_status_event(self, c)
+
+ # Catch FAILED/CLOSED now since circ will be removed
+ elif c.status == "FAILED" or c.status == "CLOSED":
+ circ = self.circuits[c.circ_id]
+ # Do logging and statistics
+ # TODO: Log failures on built circuits
+ if not circ.built:
+ message = ["FAILED"]
+ if c.reason: message.append("REASON=" + c.reason)
+ if c.remote_reason: message.append("REMOTE_REASON=" + c.remote_reason)
+ self.setup_logger.append(" ".join(message) + ": " +
+ str(circ.extend_times))
+ # Increase counter and write circ_stats to file
+ if self.partial_circs:
+ if circ.rtt_created:
+ self.circ_stats.failures += 1
+ self.stats_logger.write(self.circ_stats.to_string())
+ else:
+ self.circ_stats.failures += 1
+ self.stats_logger.write(self.circ_stats.to_string())
+
+ # Call the underlying method in any case
+ PathSupport.CircuitHandler.circ_status_event(self, c)
+ self.refresh_sorted_list()
+
+ # Log something on BUILT
+ if c.status == "BUILT":
+ circ = self.circuits[c.circ_id]
+ self.setup_logger.append(str(circ.extend_times))
+ # Add duration to circ_stats and write file
+ if self.partial_circs:
+ if circ.rtt_created:
+ self.circ_stats.add_value(circ.setup_duration)
+ self.stats_logger.write(self.circ_stats.to_string())
+ else:
+ self.circ_stats.add_value(circ.setup_duration)
+ self.stats_logger.write(self.circ_stats.to_string())
+
def get_trad_circs(self):
""" Count the circuits with rtt_created == False """
trad_circs = 0
@@ -977,22 +746,45 @@
""" Check if we currently do not have (TODO: had?) a circuit
with the given path (= Routers) """
for c in self.circuits.values():
- if c.path == path: return False
- # XXX: Check if this path can exit?
+ if c.path == path:
+ plog("ERROR", "Proposed circuit already exists")
+ return False
+ # Additionally check if this path can exit
if not path[len(path)-1].will_exit_to("255.255.255.255", 80):
plog("ERROR", "Proposed circuit would not exit")
return False
return True
+ def keys_to_routers(self, keys):
+ """ See if we know the routers spec. by keys and return them """
+ routers = []
+ for k in keys:
+ if k in self.routers:
+ routers.append(self.routers[k])
+ else:
+ plog("WARN", "We do not know about a router having ID " + k)
+ return
+ if len(routers) == len(keys):
+ return routers
+
+ def probabilistic_selection(self, proposals):
+ """ Select a proposal in a probabilistic way """
+ # TODO:
+ # Compute the sum of all expected RTTs
+ # Choose a random number from range(0,sum)
+ # Go through the proposals and subtract
+ pass
+
def build_idle_circuit(self):
- """ Override from CircuitHandler to support circuit-creation from the NetworkModel """
+ """ Override from CircuitHandler to support circuit-creation from model """
if self.partial_circs:
circ = None
# This is to ensure expansion of the explored subnet
# Check if ratio would be ok when adding new rtt_created circ
trad = float(self.get_trad_circs())
ratio = trad/(len(self.circuits.values())+1)
- plog("DEBUG","Expected Ratio = " + str(ratio) + " >= " + str(min_ratio) + " ?")
+ plog("DEBUG","Expected Ratio = " + str(ratio) +
+ " >= " + str(min_ratio) + " ?")
if ratio >= min_ratio:
# Get the proposals RTT <= slow
proposals = self.model.get_proposals(slow)
@@ -1001,31 +793,35 @@
proposals = sort_list(proposals, lambda x: x.rtt)
# Check them out
while len(proposals) >= 1:
- # Random choice or choose the fastest!
-
- choice = random.choice(proposals)
- #choice = proposals[0]
- # TODO: Probabilistic selection
-
- # Check if we already have a circ with this path
- if self.path_is_ok(choice.path):
- plog("INFO", "Chosen proposal: " + choice.to_string())
- try:
- circ = self.c.build_circuit_from_path(choice.path)
- circ.rtt_created = True
- self.circuits[circ.circ_id] = circ
- return
- except TorCtl.ErrorReply, e:
- plog("NOTICE", "Error building circuit: " + str(e.args))
- else:
- # Remove this proposals
- plog("DEBUG", "Proposed circuit already exists")
- proposals.remove(choice)
-
+
+ # Random choice
+ choice = random.choice(proposals)
+ # Choose the fastest
+ #choice = proposals[0]
+
+ # Convert the chosen ids to routers
+ r_path = self.keys_to_routers(choice.path)
+ if r_path:
+ # Check if we already have a circ with this path
+ if self.path_is_ok(r_path):
+ plog("INFO", "Chosen proposal: " + choice.to_string())
+ try:
+ circ = self.c.build_circuit_from_path(r_path)
+ circ.rtt_created = True
+ self.circuits[circ.circ_id] = circ
+ return
+ except TorCtl.ErrorReply, e:
+ plog("NOTICE", "Error building circuit: " + str(e.args))
+ else:
+ # Remove this proposals
+ proposals.remove(choice)
+ else:
+ plog("WARN", "keys_to_routers() did not work!")
+
# Build a circuit with the standard method
plog("DEBUG", "Falling back to normal path selection")
- CircuitHandler.build_idle_circuit(self)
-
+ PathSupport.CircuitHandler.build_idle_circuit(self)
+
## Pinger #####################################################################
class Pinger(threading.Thread):
@@ -1072,20 +868,18 @@
global path_config
ip = None
try:
- # Try to get our IP
+ # Try to determine our IP
info = conn.get_info("address")
ip = info["address"]
+ # Get the country_code
+ country_code = GeoIPSupport.get_country(ip)
+ plog("INFO", "Our IP address is " + str(ip) + " [" + str(country_code) + "]")
except:
- plog("ERROR", "Could not get our IP")
- ip = "127.0.0.1"
- # Set up a router object
- router = GeoIPSupport.GeoIPRouter(TorCtl.Router(None,"ROOT",None,False,None,None,ip,None,None))
- # TODO: Check if ip == None?
- plog("INFO", "Our IP address is " + router.get_ip_dotted() + " [" + str(router.country_code) + "]")
- # Set entry_country here?
+ plog("ERROR", "Could not get our IP and country")
+ return False
# path_config.entry_country = router.country_code
- return router
-
+ return True
+
def configure(conn):
""" Set events and options """
conn.set_events([TorCtl.EVENT_TYPE.STREAM,
@@ -1106,8 +900,8 @@
except socket.error, e:
plog("ERROR", "Could not connect to Tor process .. running?")
return
- # Setup a router instance here
- router = setup_location(conn)
+ # Setup our location
+ setup_location(conn)
# Configure myself
configure(conn)
# Get the size of the circuit-pool from config
@@ -1116,19 +910,26 @@
if measure_circs:
# We measure latencies
if measure_partial_circs:
- handler = PingHandler(conn, __selmgr, num_circs, router, True)
+ handler = PingHandler(conn, __selmgr, num_circs,
+ GeoIPSupport.GeoIPRouter, True)
else:
- handler = PingHandler(conn, __selmgr, num_circs, router)
+ handler = PingHandler(conn, __selmgr, num_circs, GeoIPSupport.GeoIPRouter)
else:
# No pings, only a StreamHandler
- handler = StreamHandler(conn, __selmgr, num_circs)
+ handler = PathSupport.StreamHandler(conn, __selmgr, num_circs,
+ GeoIPSupport.GeoIPRouter)
conn.set_event_handler(handler)
# Go to sleep to be able to get killed from the commandline
- # TODO: Do this only if not in testing_mode?
+ # TODO: Do this only if _not_ in testing_mode?
try:
while True:
time.sleep(60)
except KeyboardInterrupt:
+ # XXX: Schedule this
+ if measure_circs:
+ if measure_partial_circs:
+ handler.model.save_graph()
+ # Stop other threads?
cleanup(conn)
sys.exit(1)