From ea2a583803158c4cb583a6d4665a037b22ca01e5 Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Sun, 6 Oct 2019 05:25:01 -0300 Subject: [PATCH] wip --- torba/torba/client/baseledger.py | 17 +++++++++++++---- torba/torba/client/basenetwork.py | 1 + torba/torba/server/session.py | 8 ++++++-- 3 files changed, 20 insertions(+), 6 deletions(-) diff --git a/torba/torba/client/baseledger.py b/torba/torba/client/baseledger.py index 7f55cf88f..e19e4acbc 100644 --- a/torba/torba/client/baseledger.py +++ b/torba/torba/client/baseledger.py @@ -2,6 +2,7 @@ import base64 import os import asyncio import logging +import zlib from functools import partial from binascii import hexlify, unhexlify from io import StringIO @@ -309,13 +310,21 @@ class BaseLedger(metaclass=LedgerRegistry): async def initial_headers_sync(self): target = self.network.remote_height current = len(self.headers) - get_chunk = partial(self.network.retriable_call, self.network.get_headers, count=2016, b64=True) - chunks = [asyncio.ensure_future(get_chunk(height)) for height in range(current, target, 2016)] + 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, 4096)] + import time + start = time.time() + total = 0 async with self.headers.checkpointed_connector() as connector: for chunk in chunks: headers = await chunk - connector.connect(len(self.headers), base64.b64decode(headers['base64'])) - log.info("Headers sync: %s / %s", connector.tell() // self.headers.header_size, target) + total += len(headers['base64']) + 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): rewound = 0 diff --git a/torba/torba/client/basenetwork.py b/torba/torba/client/basenetwork.py index b2a5b6878..77de1925e 100644 --- a/torba/torba/client/basenetwork.py +++ b/torba/torba/client/basenetwork.py @@ -1,5 +1,6 @@ import logging import asyncio +import zlib from operator import itemgetter from typing import Dict, Optional, Tuple from time import perf_counter diff --git a/torba/torba/server/session.py b/torba/torba/server/session.py index a45f7ede1..d9252d350 100644 --- a/torba/torba/server/session.py +++ b/torba/torba/server/session.py @@ -14,6 +14,8 @@ import datetime import itertools import json import os +import zlib + import pylru import ssl import time @@ -616,7 +618,7 @@ class SessionBase(RPCSession): sessions. """ - MAX_CHUNK_SIZE = 2016 + MAX_CHUNK_SIZE = 40960 session_counter = itertools.count() request_handlers: typing.Dict[str, typing.Callable] = {} version = '0.5.7' @@ -1022,8 +1024,10 @@ class ElectrumX(SessionBase): max_size = self.MAX_CHUNK_SIZE count = min(count, max_size) 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 = { - 'base64' if b64 else 'hex': base64.b64encode(headers).decode() if b64 else headers.hex(), + 'base64' if b64 else 'hex': headers, 'count': count, 'max': max_size }