[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [torflow/master] Make way for fix for #1984.



commit 5331580a114442db79a7af3fd767107e98b4969a
Author: Mike Perry <mikeperry-git@xxxxxxxxxx>
Date:   Mon Oct 31 20:07:45 2011 -0700

    Make way for fix for #1984.
    
    Clean out some annoying garbage from a previous filtering method.
---
 NetworkScanners/BwAuthority/aggregate.py |  181 +++++++----------------------
 1 files changed, 44 insertions(+), 137 deletions(-)

diff --git a/NetworkScanners/BwAuthority/aggregate.py b/NetworkScanners/BwAuthority/aggregate.py
index 8658d33..0e20c5b 100755
--- a/NetworkScanners/BwAuthority/aggregate.py
+++ b/NetworkScanners/BwAuthority/aggregate.py
@@ -69,34 +69,11 @@ def base10_round(bw_val):
       return 1
     return ret
 
-
-
-def closest_to_one(ratio_list):
-  min_dist = 0x7fffffff
-  min_item = -1
-  for i in xrange(len(ratio_list)):
-    if abs(1.0-ratio_list[i]) < min_dist:
-      min_dist = abs(1.0-ratio_list[i])
-      min_item = i
-  return min_item
-
-class NodeData:
-  def __init__(self, timestamp):
-    self.strm_bw = []
-    self.filt_bw = []
-    self.ns_bw = []
-    self.desc_bw = []
-    self.timestamp = timestamp
-
 class Node:
   def __init__(self):
-    self.node_data = {}
     self.ignore = False
     self.idhex = None
     self.nick = None
-    self.chosen_time = None
-    self.chosen_sbw = None
-    self.chosen_fbw = None
     self.sbw_ratio = None
     self.fbw_ratio = None
     self.pid_bw = 0
@@ -108,17 +85,20 @@ class Node:
     self.ratio = None
     self.new_bw = None
     self.change = None
-    self.bw_idx = 0
-    self.strm_bw = []
-    self.filt_bw = []
-    self.ns_bw = []
-    self.desc_bw = []
-    self.timestamps = []
+
+    # measurement vars from bwauth lines
+    self.measured_at = 0
+    self.strm_bw = 0
+    self.filt_bw = 0
+    self.ns_bw = 0
+    self.desc_bw = 0
+    self.circ_fail_rate = 0
+    self.strm_fail_rate = 0
 
   def revert_to_vote(self, vote):
     self.new_bw = vote.bw
     self.pid_error = vote.pid_error
-    self.chosen_time = vote.measured_at
+    self.measured_at = vote.measured_at
 
   # Derivative of error for pid control
   def get_pid_bw(self, prev_vote, kp):
@@ -128,10 +108,10 @@ class Node:
     # This is non-standard
     self.pid_error_sum = prev_vote.pid_error_sum*(1 - 1.0/T_i) + self.pid_error
 
-    self.pid_bw = self.ns_bw[self.bw_idx] \
-             + kp*(self.ns_bw[self.bw_idx]*self.pid_error \
-             +     self.ns_bw[self.bw_idx]*self.integral_error()/T_i \
-             +     self.ns_bw[self.bw_idx]*self.d_error_dt()*T_d)
+    self.pid_bw = self.ns_bw \
+             + kp*(self.ns_bw*self.pid_error \
+             +     self.ns_bw*self.integral_error()/T_i \
+             +     self.ns_bw*self.d_error_dt()*T_d)
     return self.pid_bw
 
   # Time-weighted sum of error per unit of time (measurement sample)
@@ -153,74 +133,14 @@ class Node:
       raise Exception("Line mismatch")
     self.idhex = line.idhex
     self.nick = line.nick
-    if line.slice_file not in self.node_data \
-      or self.node_data[line.slice_file].timestamp < line.timestamp:
-      self.node_data[line.slice_file] = NodeData(line.timestamp)
-
-    # TODO: This is kinda nutty. Can we simplify? For instance,
-    # do these really need to be lists inside the nd?
-    nd = self.node_data[line.slice_file]
-    nd.strm_bw.append(line.strm_bw)
-    nd.filt_bw.append(line.filt_bw)
-    nd.ns_bw.append(line.ns_bw)
-    nd.desc_bw.append(line.desc_bw)
-
-    self.strm_bw = []
-    self.filt_bw = []
-    self.ns_bw = []
-    self.desc_bw = []
-    self.timestamps = []
-
-    for nd in self.node_data.itervalues():
-      self.strm_bw.extend(nd.strm_bw)
-      self.filt_bw.extend(nd.filt_bw)
-      self.ns_bw.extend(nd.ns_bw)
-      self.desc_bw.extend(nd.desc_bw)
-      for i in xrange(len(nd.ns_bw)):
-        self.timestamps.append(nd.timestamp)
-
-  def avg_strm_bw(self):
-    return sum(self.strm_bw)/float(len(self.strm_bw))
-
-  def avg_filt_bw(self):
-    return sum(self.filt_bw)/float(len(self.filt_bw))
-
-  def avg_ns_bw(self):
-    return sum(self.ns_bw)/float(len(self.ns_bw))
-
-  def avg_desc_bw(self):
-    return sum(self.desc_bw)/float(len(self.desc_bw))
-
-  # This can be bad for bootstrapping or highly bw-variant nodes... 
-  # we will choose an old measurement in that case.. We need
-  # to build some kind of time-bias here..
-  def _choose_strm_bw_one(self, net_avg):
-    i = closest_to_one(map(lambda f: f/net_avg, self.strm_bw))
-    self.chosen_sbw = i
-    return self.chosen_sbw
-
-  def _choose_filt_bw_one(self, net_avg):
-    i = closest_to_one(map(lambda f: f/net_avg, self.filt_bw))
-    self.chosen_fbw = i
-    return self.chosen_fbw
-
-  # Simply return the most recent one instead of this
-  # closest-to-one stuff
-  def choose_filt_bw(self, net_avg):
-    max_idx = 0
-    for i in xrange(len(self.timestamps)):
-      if self.timestamps[i] > self.timestamps[max_idx]:
-        max_idx = i
-    self.chosen_fbw = max_idx
-    return self.chosen_fbw
-
-  def choose_strm_bw(self, net_avg):
-    max_idx = 0
-    for i in xrange(len(self.timestamps)):
-      if self.timestamps[i] > self.timestamps[max_idx]:
-        max_idx = i
-    self.chosen_sbw = max_idx
-    return self.chosen_sbw
+    if line.measured_at > self.measured_at:
+      self.measured_at = line.measured_at
+      self.strm_bw = line.strm_bw
+      self.filt_bw = line.filt_bw
+      self.ns_bw = line.ns_bw
+      self.desc_bw = line.desc_bw
+      self.circ_fail_rate = line.circ_fail_rate
+      self.strm_fail_rate = line.strm_fail_rate
 
 class Line:
   def __init__(self, line, slice_file, timestamp):
@@ -231,7 +151,13 @@ class Line:
     self.ns_bw = int(re.search("[\s]*ns_bw=([\S]+)[\s]*", line).group(1))
     self.desc_bw = int(re.search("[\s]*desc_bw=([\S]+)[\s]*", line).group(1))
     self.slice_file = slice_file
-    self.timestamp = timestamp
+    self.measured_at = timestamp
+    try:
+      self.circ_fail_rate = float(re.search("[\s]*circ_fail_rate=([\S]+)[\s]*", line).group(1))
+      self.strm_fail_rate = float(re.search("[\s]*strm_fail_rate=([\S]+)[\s]*", line).group(1))
+    except:
+      self.circ_fail_rate = 0
+      self.strm_fail_rate = 0
 
 class Vote:
   def __init__(self, line):
@@ -391,25 +317,10 @@ def main(argv):
   if len(nodes) == 0:
     plog("NOTICE", "No scan results yet.")
     sys.exit(1)
- 
-  pre_strm_avg = sum(map(lambda n: n.avg_strm_bw(), nodes.itervalues()))/ \
-                  float(len(nodes))
-  pre_filt_avg = sum(map(lambda n: n.avg_filt_bw(), nodes.itervalues()))/ \
-                  float(len(nodes))
-
-  plog("DEBUG", "Network pre_strm_avg: "+str(pre_strm_avg))
-  plog("DEBUG", "Network pre_filt_avg: "+str(pre_filt_avg))
 
-  for n in nodes.itervalues():
-    n.choose_strm_bw(pre_strm_avg)
-    n.choose_filt_bw(pre_filt_avg)
-    plog("DEBUG", "Node "+n.nick+" chose sbw: "+\
-                str(n.strm_bw[n.chosen_sbw])+" fbw: "+\
-                str(n.filt_bw[n.chosen_fbw]))
-
-  true_strm_avg = sum(map(lambda n: n.strm_bw[n.chosen_sbw],
+  true_strm_avg = sum(map(lambda n: n.strm_bw,
                        nodes.itervalues()))/float(len(nodes))
-  true_filt_avg = sum(map(lambda n: n.filt_bw[n.chosen_fbw],
+  true_filt_avg = sum(map(lambda n: n.filt_bw,
                        nodes.itervalues()))/float(len(nodes))
 
   plog("DEBUG", "Network true_strm_avg: "+str(true_strm_avg))
@@ -427,11 +338,11 @@ def main(argv):
       if n.idhex in prev_votes.vote_map and n.idhex in prev_consensus:
         if "Guard" in prev_consensus[n.idhex].flags:
           guard_cnt += 1
-          guard_measure_time += (n.timestamps[n.chosen_fbw] - \
+          guard_measure_time += (n.measured_at - \
                                   prev_votes.vote_map[n.idhex].measured_at)
         else:
           node_cnt += 1
-          node_measure_time += (n.timestamps[n.chosen_fbw] - \
+          node_measure_time += (n.measured_at - \
                                   prev_votes.vote_map[n.idhex].measured_at)
 
   # There is a difference between measure period and sample rate.
@@ -443,25 +354,21 @@ def main(argv):
 
   tot_net_bw = 0
   for n in nodes.itervalues():
-    n.fbw_ratio = n.filt_bw[n.chosen_fbw]/true_filt_avg
-    n.sbw_ratio = n.strm_bw[n.chosen_sbw]/true_strm_avg
+    n.fbw_ratio = n.filt_bw/true_filt_avg
+    n.sbw_ratio = n.strm_bw/true_strm_avg
     if n.sbw_ratio > n.fbw_ratio:
       # Does this ever happen?
       plog("NOTICE", "sbw > fbw for "+n.nick)
       n.ratio = n.sbw_ratio
-      n.bw_idx = n.chosen_sbw
-      n.pid_error = (n.strm_bw[n.chosen_sbw] - true_strm_avg)/true_strm_avg
+      n.pid_error = (n.strm_bw - true_strm_avg)/true_strm_avg
     else:
       n.ratio = n.fbw_ratio
-      n.bw_idx = n.chosen_fbw
-      n.pid_error = (n.filt_bw[n.chosen_fbw] - true_filt_avg)/true_filt_avg
-
-    n.chosen_time = n.timestamps[n.bw_idx]
+      n.pid_error = (n.filt_bw - true_filt_avg)/true_filt_avg
 
     if cs_junk.bwauth_pid_control:
       if n.idhex in prev_votes.vote_map:
         # If there is a new sample, let's use it for all but guards
-        if n.chosen_time > prev_votes.vote_map[n.idhex].measured_at:
+        if n.measured_at > prev_votes.vote_map[n.idhex].measured_at:
           # Nodes with the Guard flag will respond slowly to feedback,
           # so they should be sampled less often, and in proportion to
           # the appropriate Wgx weight.
@@ -470,7 +377,7 @@ def main(argv):
              and "Exit" not in prev_consensus[n.idhex].flags):
             # Do full feedback if our previous vote > 2.5 weeks old
             if n.idhex not in prev_votes.vote_map or \
-                n.chosen_time - prev_votes.vote_map[n.idhex].measured_at > GUARD_SAMPLE_RATE:
+                n.measured_at - prev_votes.vote_map[n.idhex].measured_at > GUARD_SAMPLE_RATE:
               n.new_bw = n.get_pid_bw(prev_votes.vote_map[n.idhex], K_p)
             else:
               pid_error = n.pid_error
@@ -497,7 +404,7 @@ def main(argv):
           # Reset values. Don't vote/sample this measurement round.
           n.revert_to_vote(prev_votes.vote_map[n.idhex])
       else: # No prev vote, pure consensus feedback this round
-        n.new_bw = n.ns_bw[n.bw_idx] + K_p*n.ns_bw[n.bw_idx]*n.pid_error
+        n.new_bw = n.ns_bw + K_p*n.ns_bw*n.pid_error
         n.pid_error_sum = n.pid_error
         n.pid_bw = n.new_bw
         plog("INFO", "No prev vote for node "+n.nick+": Consensus feedback")
@@ -505,9 +412,9 @@ def main(argv):
       n.pid_bw = 0
       n.pid_error = 0
       n.pid_error_sum = 0
-      n.new_bw = n.desc_bw[n.bw_idx]*n.ratio
+      n.new_bw = n.desc_bw*n.ratio
 
-    n.change = n.new_bw - n.desc_bw[n.bw_idx]
+    n.change = n.new_bw - n.desc_bw
 
     if n.idhex in prev_consensus:
       if prev_consensus[n.idhex].bandwidth != None:
@@ -535,7 +442,7 @@ def main(argv):
       n.pid_error_sum = 0 # Don't let unused error accumulate...
 
   # WTF is going on here?
-  oldest_timestamp = min(map(lambda n: n.chosen_time,
+  oldest_timestamp = min(map(lambda n: n.measured_at,
              filter(lambda n: n.idhex in prev_consensus,
                        nodes.itervalues())))
   plog("INFO", "Oldest measured node: "+time.ctime(oldest_timestamp))
@@ -577,7 +484,7 @@ def main(argv):
   # FIXME: Split out debugging data
   for n in n_print:
     if not n.ignore:
-      out.write("node_id="+n.idhex+" bw="+str(base10_round(n.new_bw))+" nick="+n.nick+ " measured_at="+str(int(n.chosen_time))+" pid_error="+str(n.pid_error)+" pid_error_sum="+str(n.pid_error_sum)+" pid_bw="+str(n.pid_bw)+" pid_delta="+str(n.derror_dt)+"\n")
+      out.write("node_id="+n.idhex+" bw="+str(base10_round(n.new_bw))+" nick="+n.nick+ " measured_at="+str(int(n.measured_at))+" pid_error="+str(n.pid_error)+" pid_error_sum="+str(n.pid_error_sum)+" pid_bw="+str(n.pid_bw)+" pid_delta="+str(n.derror_dt)+"\n")
   out.close()
 
 if __name__ == "__main__":



_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits