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

connector

Browse Source
This commit is contained in:
4tochka 2019-05-02 12:55:34 +04:00
parent a4c0c03dec
commit e37f623534
1 changed files with 59 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
pybtc/connector.py
View File

@ -77,7 +77,7 @@ class Connector:
# cache and system # cache and system
self.preload = preload self.preload = preload
self.block_preload = Cache(max_size=50000) self.block_preload = Cache(max_size=500)
self.block_hashes = Cache(max_size=100000) self.block_hashes = Cache(max_size=100000)
self.block_hashes_preload_mutex = False self.block_hashes_preload_mutex = False
self.tx_cache = Cache(max_size=50000) self.tx_cache = Cache(max_size=50000)
@ -300,7 +300,8 @@ class Connector:
h = self.block_hashes.get(self.last_block_height + 1) h = self.block_hashes.get(self.last_block_height + 1)
if h is None: if h is None:
h = await self.rpc.getblockhash(self.last_block_height + 1) h = await self.rpc.getblockhash(self.last_block_height + 1)
self.loop.create_task(self.preload_block_hashes()) if not self.block_hashes_preload_mutex:
self.loop.create_task(self.preload_block_hashes())
else: else:
h = await self.rpc.getblockhash(self.last_block_height + 1) h = await self.rpc.getblockhash(self.last_block_height + 1)
self.blocks_download_time += time.time() - q self.blocks_download_time += time.time() - q
@ -315,7 +316,9 @@ class Connector:
try: try:
if self.deep_synchronization: if self.deep_synchronization:
q = time.time() q = time.time()
raw_block = await self.rpc.getblock(hash, 0) raw_block = self.block_preload.get(hash)
if not raw_block:
raw_block = await self.rpc.getblock(hash, 0)
self.blocks_download_time += time.time() - q self.blocks_download_time += time.time() - q
q = time.time() q = time.time()
block = decode_block_tx(raw_block) block = decode_block_tx(raw_block)
@ -607,35 +610,56 @@ class Connector:
return return
try: try:
self.block_hashes_preload_mutex = True self.block_hashes_preload_mutex = True
self.block_hashes = Cache(max_size=100000)
self.block_hashes = Cache(max_size=500)
max_height = self.node_last_block - self.deep_synchronization max_height = self.node_last_block - self.deep_synchronization
height = self.last_block_height + 1 height = self.last_block_height + 1
lh = height processed_height = self.last_block_height
l = height + 100000
if l > max_height: while height < max_height:
l = max_height if self.last_block_height - height < 400:
while height < l: try:
try: batch = list()
batch = list() h_list = list()
while True: while True:
batch.append(["getblockhash", height]) batch.append(["getblockhash", height])
if len(batch) >= self.batch_limit or height >= l: h_list.append(height)
height += 1 if len(batch) >= self.batch_limit or height >= max_height:
break height += 1
height += 1 break
result = await self.rpc.batch(batch) height += 1
for r in result: result = await self.rpc.batch(batch)
try: headers = list()
self.block_hashes.set(lh, r["result"]) batch = list()
except: for lh, r in zip(h_list, result):
pass try:
lh += 1 self.block_hashes.set(lh, r["result"])
batch.append(["getblock", r["result"], 0])
headers.append(r["result"])
except:
pass
blocks = await self.rpc.batch(batch)
for x,y in zip(headers, blocks):
try:
self.block_preload.set(x, y)
except:
pass
except asyncio.CancelledError:
self.log.info("connector preload_block_hashes failed")
break
except:
pass
if self.block_preload.len() < 500:
continue
await asyncio.sleep(10)
# remove unused items
[self.block_preload.remove(i) for i in range(processed_height,
self.last_block_height)]
processed_height = self.last_block_height
except asyncio.CancelledError:
self.log.info("connector preload_block_hashes failed")
break
except:
pass
finally: finally:
self.block_hashes_preload_mutex = False self.block_hashes_preload_mutex = False
@ -932,6 +956,12 @@ class Cache():
except: except:
return None return None
def remove(self, key):
try:
del self._store[key]
except:
pass
def pop_last(self): def pop_last(self):
try: try:
i = next(reversed(self._store)) i = next(reversed(self._store))