LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

fix mempool notification bloat

Browse source
This commit is contained in:
Jack Robison 2020-12-07 13:16:37 -05:00
parent 561368570e
commit eae0290978
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
1 changed files with 15 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
lbry/wallet/server/session.py
View file

@ -640,6 +640,7 @@ class SessionManager:
height_changed = height != self.notified_height height_changed = height != self.notified_height
if height_changed: if height_changed:
await self._refresh_hsub_results(height) await self._refresh_hsub_results(height)
if not self.sessions: if not self.sessions:
return return
@ -649,25 +650,22 @@ class SessionManager:
for session in self.sessions.values() if session.subscribe_headers for session in self.sessions.values() if session.subscribe_headers
] ]
if header_tasks: if header_tasks:
self.logger.info(f'notify {len(header_tasks)} sessions of new header')
await asyncio.wait(header_tasks) await asyncio.wait(header_tasks)
for hashX in touched.intersection(self.mempool_statuses.keys()):
self.mempool_statuses.pop(hashX, None)
touched = touched.intersection(self.hashx_subscriptions_by_session.keys()) touched.intersection_update(self.hashx_subscriptions_by_session.keys())
if touched or (height_changed and self.mempool_statuses): if touched or (height_changed and self.mempool_statuses):
mempool_hashxs = set(self.mempool_statuses.keys()) notified_hashxs = 0
notified = set() notified_sessions = 0
for hashX in touched: for hashX in touched.union(self.mempool_statuses.keys()):
for session_id in self.hashx_subscriptions_by_session[hashX]: for session_id in self.hashx_subscriptions_by_session[hashX]:
asyncio.create_task(self.sessions[session_id].send_history_notification(hashX)) asyncio.create_task(self.sessions[session_id].send_history_notification(hashX))
notified.add(hashX) notified_sessions += 1
for hashX in mempool_hashxs.difference(touched): notified_hashxs += 1
for session_id in self.hashx_subscriptions_by_session[hashX]: self.logger.info(f'notified {notified_sessions} sessions/{notified_hashxs:,d} touched addresses')
asyncio.create_task(self.sessions[session_id].send_history_notification(hashX))
notified.add(hashX)
if touched:
es = '' if len(touched) == 1 else 'es'
self.logger.info(f'notified {len(notified)} mempool/{len(touched):,d} touched address{es}')
def add_session(self, session): def add_session(self, session):
self.sessions[id(session)] = session self.sessions[id(session)] = session
@ -1231,8 +1229,10 @@ class LBRYElectrumX(SessionBase):
return await self.address_status(hashX) return await self.address_status(hashX)
async def hashX_unsubscribe(self, hashX, alias): async def hashX_unsubscribe(self, hashX, alias):
self.session_mgr.hashx_subscriptions_by_session[hashX].remove(id(self)) sessions = self.session_mgr.hashx_subscriptions_by_session[hashX]
self.hashX_subs.pop(hashX, None) sessions.remove(id(self))
if not sessions:
self.hashX_subs.pop(hashX, None)
def address_to_hashX(self, address): def address_to_hashX(self, address):
try: try: