commit 5331580a114442db79a7af3fd767107e98b4969a
Author: Mike Perry <mikeperry-git(a)fscked.org>
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__":
