Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

improve wallet server address history cache

Browse source
This commit is contained in:
Jack Robison 2020-06-04 09:15:21 -04:00
parent 48502961cf
commit 8d93dd5adc
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 7 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lbry/wallet/server/block_processor.py
View file

@ -183,6 +183,7 @@ class BlockProcessor:
self.state_lock = asyncio.Lock() self.state_lock = asyncio.Lock()
self.search_cache = {} self.search_cache = {}
self.history_cache = {}
async def run_in_thread_with_lock(self, func, *args): async def run_in_thread_with_lock(self, func, *args):
# Run in a thread to prevent blocking. Shielded so that # Run in a thread to prevent blocking. Shielded so that
@ -213,6 +214,7 @@ class BlockProcessor:
await self.run_in_thread_with_lock(self.advance_blocks, blocks) await self.run_in_thread_with_lock(self.advance_blocks, blocks)
for cache in self.search_cache.values(): for cache in self.search_cache.values():
cache.clear() cache.clear()
self.history_cache.clear()
await self._maybe_flush() await self._maybe_flush()
processed_time = time.perf_counter() - start processed_time = time.perf_counter() - start
self.block_count_metric.set(self.height) self.block_count_metric.set(self.height)

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

@ -135,7 +135,7 @@ class SessionManager:
"docker_tag": DOCKER_TAG, "docker_tag": DOCKER_TAG,
'version': lbry.__version__, 'version': lbry.__version__,
"min_version": util.version_string(VERSION.PROTOCOL_MIN), "min_version": util.version_string(VERSION.PROTOCOL_MIN),
"cpu_count": os.cpu_count() "cpu_count": str(os.cpu_count())
}) })
session_count_metric = Gauge("session_count", "Number of connected client sessions", namespace=NAMESPACE, session_count_metric = Gauge("session_count", "Number of connected client sessions", namespace=NAMESPACE,
labelnames=("version",)) labelnames=("version",))
@ -177,7 +177,7 @@ class SessionManager:
self.cur_group = SessionGroup(0) self.cur_group = SessionGroup(0)
self.txs_sent = 0 self.txs_sent = 0
self.start_time = time.time() self.start_time = time.time()
self.history_cache = pylru.lrucache(256) self.history_cache = self.bp.history_cache
self.notified_height: typing.Optional[int] = None self.notified_height: typing.Optional[int] = None
# Cache some idea of room to avoid recounting on each subscription # Cache some idea of room to avoid recounting on each subscription
self.subs_room = 0 self.subs_room = 0
@ -608,26 +608,20 @@ class SessionManager:
async def limited_history(self, hashX): async def limited_history(self, hashX):
"""A caching layer.""" """A caching layer."""
hc = self.history_cache if hashX not in self.history_cache:
if hashX not in hc:
# History DoS limit. Each element of history is about 99 # History DoS limit. Each element of history is about 99
# bytes when encoded as JSON. This limits resource usage # bytes when encoded as JSON. This limits resource usage
# on bloated history requests, and uses a smaller divisor # on bloated history requests, and uses a smaller divisor
# so large requests are logged before refusing them. # so large requests are logged before refusing them.
limit = self.env.max_send // 97 limit = self.env.max_send // 97
hc[hashX] = await self.db.limited_history(hashX, limit=limit) self.history_cache[hashX] = await self.db.limited_history(hashX, limit=limit)
return hc[hashX] return self.history_cache[hashX]
async def _notify_sessions(self, height, touched): async def _notify_sessions(self, height, touched):
"""Notify sessions about height changes and touched addresses.""" """Notify sessions about height changes and touched addresses."""
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)
# Invalidate our history cache for touched hashXs
hc = self.history_cache
for hashX in set(hc).intersection(touched):
del hc[hashX]
if self.sessions: if self.sessions:
await asyncio.wait([ await asyncio.wait([
session.notify(touched, height_changed) for session in self.sessions session.notify(touched, height_changed) for session in self.sessions