RanchiMall/electrumx
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve group handling.

Browse Source 
I believe this fixes #94
This commit is contained in:
Neil Booth 2017-01-07 15:11:39 +09:00
parent fef14aab1f
commit 30b0ebd2dd
1 changed files with 24 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
server/controller.py
View File

@ -53,8 +53,9 @@ class Controller(util.LoggedClass):
self.irc = IRC(env) self.irc = IRC(env)
self.env = env self.env = env
self.servers = {} self.servers = {}
# Map of session to the key of its list in self.groups
self.sessions = {} self.sessions = {}
self.groups = defaultdict(set) self.groups = defaultdict(list)
self.txs_sent = 0 self.txs_sent = 0
self.next_log_sessions = 0 self.next_log_sessions = 0
self.state = self.CATCHING_UP self.state = self.CATCHING_UP
@ -108,9 +109,10 @@ class Controller(util.LoggedClass):
def session_priority(self, session): def session_priority(self, session):
if isinstance(session, LocalRPC): if isinstance(session, LocalRPC):
return 0 return 0
group_bandwidth = sum(s.bandwidth_used for s in self.sessions[session]) gid = self.sessions[session]
group_bandwidth = sum(s.bandwidth_used for s in self.groups[gid])
return 1 + (bisect_left(self.bands, session.bandwidth_used) return 1 + (bisect_left(self.bands, session.bandwidth_used)
+ bisect_left(self.bands, group_bandwidth) + 1) // 2 + bisect_left(self.bands, group_bandwidth)) // 2
def is_deprioritized(self, session): def is_deprioritized(self, session):
return self.session_priority(session) > self.BANDS return self.session_priority(session) > self.BANDS
@ -333,9 +335,9 @@ class Controller(util.LoggedClass):
if now > self.next_stale_check: if now > self.next_stale_check:
self.next_stale_check = now + 300 self.next_stale_check = now + 300
self.clear_stale_sessions() self.clear_stale_sessions()
group = self.groups[int(session.start - self.start) // 900] gid = int(session.start - self.start) // 900
group.add(session) self.groups[gid].append(session)
self.sessions[session] = group self.sessions[session] = gid
session.log_info('{} {}, {:,d} total' session.log_info('{} {}, {:,d} total'
.format(session.kind, session.peername(), .format(session.kind, session.peername(),
len(self.sessions))) len(self.sessions)))
@ -350,8 +352,9 @@ class Controller(util.LoggedClass):
def remove_session(self, session): def remove_session(self, session):
'''Remove a session from our sessions list if there.''' '''Remove a session from our sessions list if there.'''
if session in self.sessions: if session in self.sessions:
group = self.sessions.pop(session) gid = self.sessions.pop(session)
group.remove(session) assert gid in self.groups
self.groups[gid].remove(session)
self.subscription_count -= session.sub_count() self.subscription_count -= session.sub_count()
def close_session(self, session): def close_session(self, session):
@ -385,13 +388,16 @@ class Controller(util.LoggedClass):
self.logger.info('closing stale connections {}'.format(stale)) self.logger.info('closing stale connections {}'.format(stale))
# Consolidate small groups # Consolidate small groups
keys = [k for k, v in self.groups.items() if len(v) <= 4 gids = [gid for gid, l in self.groups.items() if len(l) <= 4
and sum(session.bandwidth_used for session in v) < 10000] and sum(session.bandwidth_used for session in l) < 10000]
if len(keys) > 1: if len(gids) > 1:
group = set.union(*(self.groups[key] for key in keys)) sessions = sum([self.groups[gid] for gid in gids], [])
for key in keys: new_gid = max(gids)
del self.groups[key] for gid in gids:
self.groups[max(keys)] = group del self.groups[gid]
for session in sessions:
self.sessions[session] = new_gid
self.groups[new_gid] = sessions
def new_subscription(self): def new_subscription(self):
if self.subscription_count >= self.max_subs: if self.subscription_count >= self.max_subs:
@ -457,9 +463,9 @@ class Controller(util.LoggedClass):
def group_data(self): def group_data(self):
'''Returned to the RPC 'groups' call.''' '''Returned to the RPC 'groups' call.'''
result = [] result = []
for group_id in sorted(self.groups.keys()): for gid in sorted(self.groups.keys()):
sessions = self.groups[group_id] sessions = self.groups[gid]
result.append([group_id, result.append([gid,
len(sessions), len(sessions),
sum(s.bandwidth_used for s in sessions), sum(s.bandwidth_used for s in sessions),
sum(s.requests_remaining() for s in sessions), sum(s.requests_remaining() for s in sessions),