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

wip

Browse source
This commit is contained in:
Victor Shyba 2019-10-06 05:25:01 -03:00 • committed by Lex Berezhny
parent 9965801258
commit ea2a583803
3 changed files with 20 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
torba/torba/client/baseledger.py
View file

@ -2,6 +2,7 @@ import base64
import os import os
import asyncio import asyncio
import logging import logging
import zlib
from functools import partial from functools import partial
from binascii import hexlify, unhexlify from binascii import hexlify, unhexlify
from io import StringIO from io import StringIO
@ -309,13 +310,21 @@ class BaseLedger(metaclass=LedgerRegistry):
async def initial_headers_sync(self): async def initial_headers_sync(self):
target = self.network.remote_height target = self.network.remote_height
current = len(self.headers) current = len(self.headers)
get_chunk = partial(self.network.retriable_call, self.network.get_headers, count=2016, b64=True) get_chunk = partial(self.network.retriable_call, self.network.get_headers, count=4096, b64=True)
chunks = [asyncio.ensure_future(get_chunk(height)) for height in range(current, target, 2016)] chunks = [asyncio.ensure_future(get_chunk(height)) for height in range(current, target, 4096)]
import time
start = time.time()
total = 0
async with self.headers.checkpointed_connector() as connector: async with self.headers.checkpointed_connector() as connector:
for chunk in chunks: for chunk in chunks:
headers = await chunk headers = await chunk
connector.connect(len(self.headers), base64.b64decode(headers['base64'])) total += len(headers['base64'])
log.info("Headers sync: %s / %s", connector.tell() // self.headers.header_size, target) try:
connector.connect(len(self.headers), zlib.decompress(base64.b64decode(headers['base64']), wbits=-15, bufsize=600_000))
except BaseException:
log.exception("ops")
log.info("Headers sync: %s / %s -- %s", connector.tell() // self.headers.header_size, target, total)
print(time.time() - start)
async def update_headers(self, height=None, headers=None, subscription_update=False): async def update_headers(self, height=None, headers=None, subscription_update=False):
rewound = 0 rewound = 0

1
torba/torba/client/basenetwork.py
View file

@ -1,5 +1,6 @@
import logging import logging
import asyncio import asyncio
import zlib
from operator import itemgetter from operator import itemgetter
from typing import Dict, Optional, Tuple from typing import Dict, Optional, Tuple
from time import perf_counter from time import perf_counter

8
torba/torba/server/session.py
View file

@ -14,6 +14,8 @@ import datetime
import itertools import itertools
import json import json
import os import os
import zlib
import pylru import pylru
import ssl import ssl
import time import time
@ -616,7 +618,7 @@ class SessionBase(RPCSession):
sessions. sessions.
""" """
MAX_CHUNK_SIZE = 2016 MAX_CHUNK_SIZE = 40960
session_counter = itertools.count() session_counter = itertools.count()
request_handlers: typing.Dict[str, typing.Callable] = {} request_handlers: typing.Dict[str, typing.Callable] = {}
version = '0.5.7' version = '0.5.7'
@ -1022,8 +1024,10 @@ class ElectrumX(SessionBase):
max_size = self.MAX_CHUNK_SIZE max_size = self.MAX_CHUNK_SIZE
count = min(count, max_size) count = min(count, max_size)
headers, count = await self.db.read_headers(start_height, count) headers, count = await self.db.read_headers(start_height, count)
compressobj = zlib.compressobj(wbits=-15, level=1, memLevel=9)
headers = base64.b64encode(compressobj.compress(headers) + compressobj.flush()).decode() if b64 else headers.hex()
result = { result = {
'base64' if b64 else 'hex': base64.b64encode(headers).decode() if b64 else headers.hex(), 'base64' if b64 else 'hex': headers,
'count': count, 'count': count,
'max': max_size 'max': max_size
} }