Compare commits
55 commits
Author | SHA1 | Date | |
---|---|---|---|
eb5da9511e | |||
8722ef840e | |||
6e75a1a89b | |||
ef3189de1d | |||
c2d2080034 | |||
d0b5a0a8fd | |||
1d0e17be21 | |||
4ef03bb1f4 | |||
4bd4bcdc27 | |||
e5ca967fa2 | |||
eed7d02e8b | |||
02aecad52b | |||
585962d930 | |||
ea4fba39a6 | |||
7a86406746 | |||
c8a3eb97a4 | |||
20213628d7 | |||
2d1649f972 | |||
5cb04b86a0 | |||
93ab6b3be3 | |||
b9762c3e64 | |||
82592d00ef | |||
c118174c1a | |||
d284acd8b8 | |||
235c98372d | |||
d2f5073ef4 | |||
84e5e43117 | |||
7bd025ae54 | |||
8f28ce65b0 | |||
d36e305129 | |||
2609dee8fb | |||
a2da86d4b5 | |||
aa16c7fee5 | |||
3266f72b82 | |||
77cd2a3f8a | |||
308e586e9a | |||
Philip Ahlqvist | 84beddfd77 | ||
6258651650 | |||
cc5f0b6630 | |||
f64d507d39 | |||
001819d5c2 | |||
8b4c046d28 | |||
2c20ad6c43 | |||
9e610cc54c | |||
b9d25c6d01 | |||
419b5b45f2 | |||
516c2dd5d0 | |||
b99102f9c9 | |||
8c6c7b655c | |||
48c6873fc4 | |||
15dc52bd9a | |||
52d555078f | |||
cc976bd010 | |||
9cc6992011 | |||
a1b87460c5 |
66
.github/workflows/main.yml
vendored
66
.github/workflows/main.yml
vendored
|
@ -1,18 +1,18 @@
|
||||||
name: ci
|
name: ci
|
||||||
on: ["push", "pull_request"]
|
on: ["push", "pull_request", "workflow_dispatch"]
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
|
|
||||||
lint:
|
lint:
|
||||||
name: lint
|
name: lint
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-20.04
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v2
|
- uses: actions/checkout@v3
|
||||||
- uses: actions/setup-python@v1
|
- uses: actions/setup-python@v4
|
||||||
with:
|
with:
|
||||||
python-version: '3.7'
|
python-version: '3.9'
|
||||||
- name: extract pip cache
|
- name: extract pip cache
|
||||||
uses: actions/cache@v2
|
uses: actions/cache@v3
|
||||||
with:
|
with:
|
||||||
path: ~/.cache/pip
|
path: ~/.cache/pip
|
||||||
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
||||||
|
@ -26,26 +26,26 @@ jobs:
|
||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
os:
|
os:
|
||||||
- ubuntu-latest
|
- ubuntu-20.04
|
||||||
- macos-latest
|
- macos-latest
|
||||||
- windows-latest
|
- windows-latest
|
||||||
runs-on: ${{ matrix.os }}
|
runs-on: ${{ matrix.os }}
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v2
|
- uses: actions/checkout@v3
|
||||||
- uses: actions/setup-python@v1
|
- uses: actions/setup-python@v4
|
||||||
with:
|
with:
|
||||||
python-version: '3.7'
|
python-version: '3.9'
|
||||||
- name: set pip cache dir
|
- name: set pip cache dir
|
||||||
id: pip-cache
|
shell: bash
|
||||||
run: echo "::set-output name=dir::$(pip cache dir)"
|
run: echo "PIP_CACHE_DIR=$(pip cache dir)" >> $GITHUB_ENV
|
||||||
- name: extract pip cache
|
- name: extract pip cache
|
||||||
uses: actions/cache@v2
|
uses: actions/cache@v3
|
||||||
with:
|
with:
|
||||||
path: ${{ steps.pip-cache.outputs.dir }}
|
path: ${{ env.PIP_CACHE_DIR }}
|
||||||
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
||||||
restore-keys: ${{ runner.os }}-pip-
|
restore-keys: ${{ runner.os }}-pip-
|
||||||
- id: os-name
|
- id: os-name
|
||||||
uses: ASzc/change-string-case-action@v1
|
uses: ASzc/change-string-case-action@v5
|
||||||
with:
|
with:
|
||||||
string: ${{ runner.os }}
|
string: ${{ runner.os }}
|
||||||
- run: python -m pip install --user --upgrade pip wheel
|
- run: python -m pip install --user --upgrade pip wheel
|
||||||
|
@ -72,7 +72,7 @@ jobs:
|
||||||
|
|
||||||
tests-integration:
|
tests-integration:
|
||||||
name: "tests / integration"
|
name: "tests / integration"
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-20.04
|
||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
test:
|
test:
|
||||||
|
@ -93,16 +93,16 @@ jobs:
|
||||||
uses: elastic/elastic-github-actions/elasticsearch@master
|
uses: elastic/elastic-github-actions/elasticsearch@master
|
||||||
with:
|
with:
|
||||||
stack-version: 7.12.1
|
stack-version: 7.12.1
|
||||||
- uses: actions/checkout@v2
|
- uses: actions/checkout@v3
|
||||||
- uses: actions/setup-python@v1
|
- uses: actions/setup-python@v4
|
||||||
with:
|
with:
|
||||||
python-version: '3.7'
|
python-version: '3.9'
|
||||||
- if: matrix.test == 'other'
|
- if: matrix.test == 'other'
|
||||||
run: |
|
run: |
|
||||||
sudo apt-get update
|
sudo apt-get update
|
||||||
sudo apt-get install -y --no-install-recommends ffmpeg
|
sudo apt-get install -y --no-install-recommends ffmpeg
|
||||||
- name: extract pip cache
|
- name: extract pip cache
|
||||||
uses: actions/cache@v2
|
uses: actions/cache@v3
|
||||||
with:
|
with:
|
||||||
path: ./.tox
|
path: ./.tox
|
||||||
key: tox-integration-${{ matrix.test }}-${{ hashFiles('setup.py') }}
|
key: tox-integration-${{ matrix.test }}-${{ hashFiles('setup.py') }}
|
||||||
|
@ -123,7 +123,7 @@ jobs:
|
||||||
|
|
||||||
coverage:
|
coverage:
|
||||||
needs: ["tests-unit", "tests-integration"]
|
needs: ["tests-unit", "tests-integration"]
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-20.04
|
||||||
steps:
|
steps:
|
||||||
- name: finalize coverage report submission
|
- name: finalize coverage report submission
|
||||||
env:
|
env:
|
||||||
|
@ -138,29 +138,29 @@ jobs:
|
||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
os:
|
os:
|
||||||
- ubuntu-18.04
|
- ubuntu-20.04
|
||||||
- macos-latest
|
- macos-latest
|
||||||
- windows-latest
|
- windows-latest
|
||||||
runs-on: ${{ matrix.os }}
|
runs-on: ${{ matrix.os }}
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v2
|
- uses: actions/checkout@v3
|
||||||
- uses: actions/setup-python@v1
|
- uses: actions/setup-python@v4
|
||||||
with:
|
with:
|
||||||
python-version: '3.7'
|
python-version: '3.9'
|
||||||
- id: os-name
|
- id: os-name
|
||||||
uses: ASzc/change-string-case-action@v1
|
uses: ASzc/change-string-case-action@v5
|
||||||
with:
|
with:
|
||||||
string: ${{ runner.os }}
|
string: ${{ runner.os }}
|
||||||
- name: set pip cache dir
|
- name: set pip cache dir
|
||||||
id: pip-cache
|
shell: bash
|
||||||
run: echo "::set-output name=dir::$(pip cache dir)"
|
run: echo "PIP_CACHE_DIR=$(pip cache dir)" >> $GITHUB_ENV
|
||||||
- name: extract pip cache
|
- name: extract pip cache
|
||||||
uses: actions/cache@v2
|
uses: actions/cache@v3
|
||||||
with:
|
with:
|
||||||
path: ${{ steps.pip-cache.outputs.dir }}
|
path: ${{ env.PIP_CACHE_DIR }}
|
||||||
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
|
||||||
restore-keys: ${{ runner.os }}-pip-
|
restore-keys: ${{ runner.os }}-pip-
|
||||||
- run: pip install pyinstaller==4.4
|
- run: pip install pyinstaller==4.6
|
||||||
- run: pip install -e .
|
- run: pip install -e .
|
||||||
- if: startsWith(github.ref, 'refs/tags/v')
|
- if: startsWith(github.ref, 'refs/tags/v')
|
||||||
run: python docker/set_build.py
|
run: python docker/set_build.py
|
||||||
|
@ -175,7 +175,7 @@ jobs:
|
||||||
pip install pywin32==301
|
pip install pywin32==301
|
||||||
pyinstaller --additional-hooks-dir=scripts/. --icon=icons/lbry256.ico --onefile --name lbrynet lbry/extras/cli.py
|
pyinstaller --additional-hooks-dir=scripts/. --icon=icons/lbry256.ico --onefile --name lbrynet lbry/extras/cli.py
|
||||||
dist/lbrynet.exe --version
|
dist/lbrynet.exe --version
|
||||||
- uses: actions/upload-artifact@v2
|
- uses: actions/upload-artifact@v3
|
||||||
with:
|
with:
|
||||||
name: lbrynet-${{ steps.os-name.outputs.lowercase }}
|
name: lbrynet-${{ steps.os-name.outputs.lowercase }}
|
||||||
path: dist/
|
path: dist/
|
||||||
|
@ -184,7 +184,7 @@ jobs:
|
||||||
name: "release"
|
name: "release"
|
||||||
if: startsWith(github.ref, 'refs/tags/v')
|
if: startsWith(github.ref, 'refs/tags/v')
|
||||||
needs: ["build"]
|
needs: ["build"]
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-20.04
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v1
|
- uses: actions/checkout@v1
|
||||||
- uses: actions/download-artifact@v2
|
- uses: actions/download-artifact@v2
|
||||||
|
|
2
.github/workflows/release.yml
vendored
2
.github/workflows/release.yml
vendored
|
@ -7,7 +7,7 @@ on:
|
||||||
jobs:
|
jobs:
|
||||||
release:
|
release:
|
||||||
name: "slack notification"
|
name: "slack notification"
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-20.04
|
||||||
steps:
|
steps:
|
||||||
- uses: LoveToKnow/slackify-markdown-action@v1.0.0
|
- uses: LoveToKnow/slackify-markdown-action@v1.0.0
|
||||||
id: markdown
|
id: markdown
|
||||||
|
|
|
@ -1,2 +1,2 @@
|
||||||
__version__ = "0.111.0"
|
__version__ = "0.113.0"
|
||||||
version = tuple(map(int, __version__.split('.'))) # pylint: disable=invalid-name
|
version = tuple(map(int, __version__.split('.'))) # pylint: disable=invalid-name
|
||||||
|
|
|
@ -87,8 +87,8 @@ class AbstractBlob:
|
||||||
self.blob_completed_callback = blob_completed_callback
|
self.blob_completed_callback = blob_completed_callback
|
||||||
self.blob_directory = blob_directory
|
self.blob_directory = blob_directory
|
||||||
self.writers: typing.Dict[typing.Tuple[typing.Optional[str], typing.Optional[int]], HashBlobWriter] = {}
|
self.writers: typing.Dict[typing.Tuple[typing.Optional[str], typing.Optional[int]], HashBlobWriter] = {}
|
||||||
self.verified: asyncio.Event = asyncio.Event(loop=self.loop)
|
self.verified: asyncio.Event = asyncio.Event()
|
||||||
self.writing: asyncio.Event = asyncio.Event(loop=self.loop)
|
self.writing: asyncio.Event = asyncio.Event()
|
||||||
self.readers: typing.List[typing.BinaryIO] = []
|
self.readers: typing.List[typing.BinaryIO] = []
|
||||||
self.added_on = added_on or time.time()
|
self.added_on = added_on or time.time()
|
||||||
self.is_mine = is_mine
|
self.is_mine = is_mine
|
||||||
|
@ -222,7 +222,7 @@ class AbstractBlob:
|
||||||
peer_port: typing.Optional[int] = None) -> HashBlobWriter:
|
peer_port: typing.Optional[int] = None) -> HashBlobWriter:
|
||||||
if (peer_address, peer_port) in self.writers and not self.writers[(peer_address, peer_port)].closed():
|
if (peer_address, peer_port) in self.writers and not self.writers[(peer_address, peer_port)].closed():
|
||||||
raise OSError(f"attempted to download blob twice from {peer_address}:{peer_port}")
|
raise OSError(f"attempted to download blob twice from {peer_address}:{peer_port}")
|
||||||
fut = asyncio.Future(loop=self.loop)
|
fut = asyncio.Future()
|
||||||
writer = HashBlobWriter(self.blob_hash, self.get_length, fut)
|
writer = HashBlobWriter(self.blob_hash, self.get_length, fut)
|
||||||
self.writers[(peer_address, peer_port)] = writer
|
self.writers[(peer_address, peer_port)] = writer
|
||||||
|
|
||||||
|
|
|
@ -32,7 +32,7 @@ class BlobExchangeClientProtocol(asyncio.Protocol):
|
||||||
self.buf = b''
|
self.buf = b''
|
||||||
|
|
||||||
# this is here to handle the race when the downloader is closed right as response_fut gets a result
|
# this is here to handle the race when the downloader is closed right as response_fut gets a result
|
||||||
self.closed = asyncio.Event(loop=self.loop)
|
self.closed = asyncio.Event()
|
||||||
|
|
||||||
def data_received(self, data: bytes):
|
def data_received(self, data: bytes):
|
||||||
if self.connection_manager:
|
if self.connection_manager:
|
||||||
|
@ -111,7 +111,7 @@ class BlobExchangeClientProtocol(asyncio.Protocol):
|
||||||
self.transport.write(msg)
|
self.transport.write(msg)
|
||||||
if self.connection_manager:
|
if self.connection_manager:
|
||||||
self.connection_manager.sent_data(f"{self.peer_address}:{self.peer_port}", len(msg))
|
self.connection_manager.sent_data(f"{self.peer_address}:{self.peer_port}", len(msg))
|
||||||
response: BlobResponse = await asyncio.wait_for(self._response_fut, self.peer_timeout, loop=self.loop)
|
response: BlobResponse = await asyncio.wait_for(self._response_fut, self.peer_timeout)
|
||||||
availability_response = response.get_availability_response()
|
availability_response = response.get_availability_response()
|
||||||
price_response = response.get_price_response()
|
price_response = response.get_price_response()
|
||||||
blob_response = response.get_blob_response()
|
blob_response = response.get_blob_response()
|
||||||
|
@ -151,7 +151,7 @@ class BlobExchangeClientProtocol(asyncio.Protocol):
|
||||||
f" timeout in {self.peer_timeout}"
|
f" timeout in {self.peer_timeout}"
|
||||||
log.debug(msg)
|
log.debug(msg)
|
||||||
msg = f"downloaded {self.blob.blob_hash[:8]} from {self.peer_address}:{self.peer_port}"
|
msg = f"downloaded {self.blob.blob_hash[:8]} from {self.peer_address}:{self.peer_port}"
|
||||||
await asyncio.wait_for(self.writer.finished, self.peer_timeout, loop=self.loop)
|
await asyncio.wait_for(self.writer.finished, self.peer_timeout)
|
||||||
# wait for the io to finish
|
# wait for the io to finish
|
||||||
await self.blob.verified.wait()
|
await self.blob.verified.wait()
|
||||||
log.info("%s at %fMB/s", msg,
|
log.info("%s at %fMB/s", msg,
|
||||||
|
@ -187,7 +187,7 @@ class BlobExchangeClientProtocol(asyncio.Protocol):
|
||||||
try:
|
try:
|
||||||
self._blob_bytes_received = 0
|
self._blob_bytes_received = 0
|
||||||
self.blob, self.writer = blob, blob.get_blob_writer(self.peer_address, self.peer_port)
|
self.blob, self.writer = blob, blob.get_blob_writer(self.peer_address, self.peer_port)
|
||||||
self._response_fut = asyncio.Future(loop=self.loop)
|
self._response_fut = asyncio.Future()
|
||||||
return await self._download_blob()
|
return await self._download_blob()
|
||||||
except OSError:
|
except OSError:
|
||||||
# i'm not sure how to fix this race condition - jack
|
# i'm not sure how to fix this race condition - jack
|
||||||
|
@ -244,7 +244,7 @@ async def request_blob(loop: asyncio.AbstractEventLoop, blob: Optional['Abstract
|
||||||
try:
|
try:
|
||||||
if not connected_protocol:
|
if not connected_protocol:
|
||||||
await asyncio.wait_for(loop.create_connection(lambda: protocol, address, tcp_port),
|
await asyncio.wait_for(loop.create_connection(lambda: protocol, address, tcp_port),
|
||||||
peer_connect_timeout, loop=loop)
|
peer_connect_timeout)
|
||||||
connected_protocol = protocol
|
connected_protocol = protocol
|
||||||
if blob is None or blob.get_is_verified() or not blob.is_writeable():
|
if blob is None or blob.get_is_verified() or not blob.is_writeable():
|
||||||
# blob is None happens when we are just opening a connection
|
# blob is None happens when we are just opening a connection
|
||||||
|
|
|
@ -30,7 +30,7 @@ class BlobDownloader:
|
||||||
self.failures: typing.Dict['KademliaPeer', int] = {}
|
self.failures: typing.Dict['KademliaPeer', int] = {}
|
||||||
self.connection_failures: typing.Set['KademliaPeer'] = set()
|
self.connection_failures: typing.Set['KademliaPeer'] = set()
|
||||||
self.connections: typing.Dict['KademliaPeer', 'BlobExchangeClientProtocol'] = {}
|
self.connections: typing.Dict['KademliaPeer', 'BlobExchangeClientProtocol'] = {}
|
||||||
self.is_running = asyncio.Event(loop=self.loop)
|
self.is_running = asyncio.Event()
|
||||||
|
|
||||||
def should_race_continue(self, blob: 'AbstractBlob'):
|
def should_race_continue(self, blob: 'AbstractBlob'):
|
||||||
max_probes = self.config.max_connections_per_download * (1 if self.connections else 10)
|
max_probes = self.config.max_connections_per_download * (1 if self.connections else 10)
|
||||||
|
@ -64,8 +64,8 @@ class BlobDownloader:
|
||||||
self.scores[peer] = bytes_received / elapsed if bytes_received and elapsed else 1
|
self.scores[peer] = bytes_received / elapsed if bytes_received and elapsed else 1
|
||||||
|
|
||||||
async def new_peer_or_finished(self):
|
async def new_peer_or_finished(self):
|
||||||
active_tasks = list(self.active_connections.values()) + [asyncio.sleep(1)]
|
active_tasks = list(self.active_connections.values()) + [asyncio.create_task(asyncio.sleep(1))]
|
||||||
await asyncio.wait(active_tasks, loop=self.loop, return_when='FIRST_COMPLETED')
|
await asyncio.wait(active_tasks, return_when='FIRST_COMPLETED')
|
||||||
|
|
||||||
def cleanup_active(self):
|
def cleanup_active(self):
|
||||||
if not self.active_connections and not self.connections:
|
if not self.active_connections and not self.connections:
|
||||||
|
@ -126,7 +126,7 @@ class BlobDownloader:
|
||||||
|
|
||||||
async def download_blob(loop, config: 'Config', blob_manager: 'BlobManager', dht_node: 'Node',
|
async def download_blob(loop, config: 'Config', blob_manager: 'BlobManager', dht_node: 'Node',
|
||||||
blob_hash: str) -> 'AbstractBlob':
|
blob_hash: str) -> 'AbstractBlob':
|
||||||
search_queue = asyncio.Queue(loop=loop, maxsize=config.max_connections_per_download)
|
search_queue = asyncio.Queue(maxsize=config.max_connections_per_download)
|
||||||
search_queue.put_nowait(blob_hash)
|
search_queue.put_nowait(blob_hash)
|
||||||
peer_queue, accumulate_task = dht_node.accumulate_peers(search_queue)
|
peer_queue, accumulate_task = dht_node.accumulate_peers(search_queue)
|
||||||
fixed_peers = None if not config.fixed_peers else await get_kademlia_peers_from_hosts(config.fixed_peers)
|
fixed_peers = None if not config.fixed_peers else await get_kademlia_peers_from_hosts(config.fixed_peers)
|
||||||
|
|
|
@ -25,19 +25,19 @@ class BlobServerProtocol(asyncio.Protocol):
|
||||||
self.idle_timeout = idle_timeout
|
self.idle_timeout = idle_timeout
|
||||||
self.transfer_timeout = transfer_timeout
|
self.transfer_timeout = transfer_timeout
|
||||||
self.server_task: typing.Optional[asyncio.Task] = None
|
self.server_task: typing.Optional[asyncio.Task] = None
|
||||||
self.started_listening = asyncio.Event(loop=self.loop)
|
self.started_listening = asyncio.Event()
|
||||||
self.buf = b''
|
self.buf = b''
|
||||||
self.transport: typing.Optional[asyncio.Transport] = None
|
self.transport: typing.Optional[asyncio.Transport] = None
|
||||||
self.lbrycrd_address = lbrycrd_address
|
self.lbrycrd_address = lbrycrd_address
|
||||||
self.peer_address_and_port: typing.Optional[str] = None
|
self.peer_address_and_port: typing.Optional[str] = None
|
||||||
self.started_transfer = asyncio.Event(loop=self.loop)
|
self.started_transfer = asyncio.Event()
|
||||||
self.transfer_finished = asyncio.Event(loop=self.loop)
|
self.transfer_finished = asyncio.Event()
|
||||||
self.close_on_idle_task: typing.Optional[asyncio.Task] = None
|
self.close_on_idle_task: typing.Optional[asyncio.Task] = None
|
||||||
|
|
||||||
async def close_on_idle(self):
|
async def close_on_idle(self):
|
||||||
while self.transport:
|
while self.transport:
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(self.started_transfer.wait(), self.idle_timeout, loop=self.loop)
|
await asyncio.wait_for(self.started_transfer.wait(), self.idle_timeout)
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
log.debug("closing idle connection from %s", self.peer_address_and_port)
|
log.debug("closing idle connection from %s", self.peer_address_and_port)
|
||||||
return self.close()
|
return self.close()
|
||||||
|
@ -101,7 +101,7 @@ class BlobServerProtocol(asyncio.Protocol):
|
||||||
log.debug("send %s to %s:%i", blob_hash, peer_address, peer_port)
|
log.debug("send %s to %s:%i", blob_hash, peer_address, peer_port)
|
||||||
self.started_transfer.set()
|
self.started_transfer.set()
|
||||||
try:
|
try:
|
||||||
sent = await asyncio.wait_for(blob.sendfile(self), self.transfer_timeout, loop=self.loop)
|
sent = await asyncio.wait_for(blob.sendfile(self), self.transfer_timeout)
|
||||||
if sent and sent > 0:
|
if sent and sent > 0:
|
||||||
self.blob_manager.connection_manager.sent_data(self.peer_address_and_port, sent)
|
self.blob_manager.connection_manager.sent_data(self.peer_address_and_port, sent)
|
||||||
log.info("sent %s (%i bytes) to %s:%i", blob_hash, sent, peer_address, peer_port)
|
log.info("sent %s (%i bytes) to %s:%i", blob_hash, sent, peer_address, peer_port)
|
||||||
|
@ -157,7 +157,7 @@ class BlobServer:
|
||||||
self.loop = loop
|
self.loop = loop
|
||||||
self.blob_manager = blob_manager
|
self.blob_manager = blob_manager
|
||||||
self.server_task: typing.Optional[asyncio.Task] = None
|
self.server_task: typing.Optional[asyncio.Task] = None
|
||||||
self.started_listening = asyncio.Event(loop=self.loop)
|
self.started_listening = asyncio.Event()
|
||||||
self.lbrycrd_address = lbrycrd_address
|
self.lbrycrd_address = lbrycrd_address
|
||||||
self.idle_timeout = idle_timeout
|
self.idle_timeout = idle_timeout
|
||||||
self.transfer_timeout = transfer_timeout
|
self.transfer_timeout = transfer_timeout
|
||||||
|
|
11
lbry/conf.py
11
lbry/conf.py
|
@ -688,6 +688,9 @@ class Config(CLIConfig):
|
||||||
tracker_servers = Servers("BitTorrent-compatible (BEP15) UDP trackers for helping P2P discovery", [
|
tracker_servers = Servers("BitTorrent-compatible (BEP15) UDP trackers for helping P2P discovery", [
|
||||||
('tracker.lbry.com', 9252),
|
('tracker.lbry.com', 9252),
|
||||||
('tracker.lbry.grin.io', 9252),
|
('tracker.lbry.grin.io', 9252),
|
||||||
|
('tracker.lbry.pigg.es', 9252),
|
||||||
|
('tracker.lizard.technology', 9252),
|
||||||
|
('s1.lbry.network', 9252),
|
||||||
])
|
])
|
||||||
|
|
||||||
lbryum_servers = Servers("SPV wallet servers", [
|
lbryum_servers = Servers("SPV wallet servers", [
|
||||||
|
@ -700,14 +703,20 @@ class Config(CLIConfig):
|
||||||
('spv17.lbry.com', 50001),
|
('spv17.lbry.com', 50001),
|
||||||
('spv18.lbry.com', 50001),
|
('spv18.lbry.com', 50001),
|
||||||
('spv19.lbry.com', 50001),
|
('spv19.lbry.com', 50001),
|
||||||
|
('hub.lbry.grin.io', 50001),
|
||||||
|
('hub.lizard.technology', 50001),
|
||||||
|
('s1.lbry.network', 50001),
|
||||||
])
|
])
|
||||||
known_dht_nodes = Servers("Known nodes for bootstrapping connection to the DHT", [
|
known_dht_nodes = Servers("Known nodes for bootstrapping connection to the DHT", [
|
||||||
('dht.lbry.grin.io', 4444), # Grin
|
('dht.lbry.grin.io', 4444), # Grin
|
||||||
('dht.lbry.madiator.com', 4444), # Madiator
|
('dht.lbry.madiator.com', 4444), # Madiator
|
||||||
|
('dht.lbry.pigg.es', 4444), # Pigges
|
||||||
('lbrynet1.lbry.com', 4444), # US EAST
|
('lbrynet1.lbry.com', 4444), # US EAST
|
||||||
('lbrynet2.lbry.com', 4444), # US WEST
|
('lbrynet2.lbry.com', 4444), # US WEST
|
||||||
('lbrynet3.lbry.com', 4444), # EU
|
('lbrynet3.lbry.com', 4444), # EU
|
||||||
('lbrynet4.lbry.com', 4444) # ASIA
|
('lbrynet4.lbry.com', 4444), # ASIA
|
||||||
|
('dht.lizard.technology', 4444), # Jack
|
||||||
|
('s2.lbry.network', 4444),
|
||||||
])
|
])
|
||||||
|
|
||||||
# blockchain
|
# blockchain
|
||||||
|
|
|
@ -67,7 +67,7 @@ class ConnectionManager:
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
last = time.perf_counter()
|
last = time.perf_counter()
|
||||||
await asyncio.sleep(0.1, loop=self.loop)
|
await asyncio.sleep(0.1)
|
||||||
self._status['incoming_bps'].clear()
|
self._status['incoming_bps'].clear()
|
||||||
self._status['outgoing_bps'].clear()
|
self._status['outgoing_bps'].clear()
|
||||||
now = time.perf_counter()
|
now = time.perf_counter()
|
||||||
|
|
|
@ -42,15 +42,13 @@ class BlobAnnouncer:
|
||||||
log.debug("failed to announce %s, could only find %d peers, retrying soon.", blob_hash[:8], peers)
|
log.debug("failed to announce %s, could only find %d peers, retrying soon.", blob_hash[:8], peers)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
self.announcements_sent_metric.labels(peers=0, error=True).inc()
|
self.announcements_sent_metric.labels(peers=0, error=True).inc()
|
||||||
if isinstance(err, asyncio.CancelledError): # TODO: remove when updated to 3.8
|
|
||||||
raise err
|
|
||||||
log.warning("error announcing %s: %s", blob_hash[:8], str(err))
|
log.warning("error announcing %s: %s", blob_hash[:8], str(err))
|
||||||
|
|
||||||
async def _announce(self, batch_size: typing.Optional[int] = 10):
|
async def _announce(self, batch_size: typing.Optional[int] = 10):
|
||||||
while batch_size:
|
while batch_size:
|
||||||
if not self.node.joined.is_set():
|
if not self.node.joined.is_set():
|
||||||
await self.node.joined.wait()
|
await self.node.joined.wait()
|
||||||
await asyncio.sleep(60, loop=self.loop)
|
await asyncio.sleep(60)
|
||||||
if not self.node.protocol.routing_table.get_peers():
|
if not self.node.protocol.routing_table.get_peers():
|
||||||
log.warning("No peers in DHT, announce round skipped")
|
log.warning("No peers in DHT, announce round skipped")
|
||||||
continue
|
continue
|
||||||
|
@ -59,7 +57,7 @@ class BlobAnnouncer:
|
||||||
log.debug("announcer task wake up, %d blobs to announce", len(self.announce_queue))
|
log.debug("announcer task wake up, %d blobs to announce", len(self.announce_queue))
|
||||||
while len(self.announce_queue) > 0:
|
while len(self.announce_queue) > 0:
|
||||||
log.info("%i blobs to announce", len(self.announce_queue))
|
log.info("%i blobs to announce", len(self.announce_queue))
|
||||||
await asyncio.gather(*[self._run_consumer() for _ in range(batch_size)], loop=self.loop)
|
await asyncio.gather(*[self._run_consumer() for _ in range(batch_size)])
|
||||||
announced = list(filter(None, self.announced))
|
announced = list(filter(None, self.announced))
|
||||||
if announced:
|
if announced:
|
||||||
await self.storage.update_last_announced_blobs(announced)
|
await self.storage.update_last_announced_blobs(announced)
|
||||||
|
|
|
@ -37,7 +37,7 @@ class Node:
|
||||||
self.protocol = KademliaProtocol(loop, peer_manager, node_id, external_ip, udp_port, peer_port, rpc_timeout,
|
self.protocol = KademliaProtocol(loop, peer_manager, node_id, external_ip, udp_port, peer_port, rpc_timeout,
|
||||||
split_buckets_under_index, is_bootstrap_node)
|
split_buckets_under_index, is_bootstrap_node)
|
||||||
self.listening_port: asyncio.DatagramTransport = None
|
self.listening_port: asyncio.DatagramTransport = None
|
||||||
self.joined = asyncio.Event(loop=self.loop)
|
self.joined = asyncio.Event()
|
||||||
self._join_task: asyncio.Task = None
|
self._join_task: asyncio.Task = None
|
||||||
self._refresh_task: asyncio.Task = None
|
self._refresh_task: asyncio.Task = None
|
||||||
self._storage = storage
|
self._storage = storage
|
||||||
|
@ -79,7 +79,7 @@ class Node:
|
||||||
else:
|
else:
|
||||||
if force_once:
|
if force_once:
|
||||||
break
|
break
|
||||||
fut = asyncio.Future(loop=self.loop)
|
fut = asyncio.Future()
|
||||||
self.loop.call_later(constants.REFRESH_INTERVAL // 4, fut.set_result, None)
|
self.loop.call_later(constants.REFRESH_INTERVAL // 4, fut.set_result, None)
|
||||||
await fut
|
await fut
|
||||||
continue
|
continue
|
||||||
|
@ -93,7 +93,7 @@ class Node:
|
||||||
if force_once:
|
if force_once:
|
||||||
break
|
break
|
||||||
|
|
||||||
fut = asyncio.Future(loop=self.loop)
|
fut = asyncio.Future()
|
||||||
self.loop.call_later(constants.REFRESH_INTERVAL, fut.set_result, None)
|
self.loop.call_later(constants.REFRESH_INTERVAL, fut.set_result, None)
|
||||||
await fut
|
await fut
|
||||||
|
|
||||||
|
@ -108,7 +108,7 @@ class Node:
|
||||||
for peer in peers:
|
for peer in peers:
|
||||||
log.debug("store to %s %s %s", peer.address, peer.udp_port, peer.tcp_port)
|
log.debug("store to %s %s %s", peer.address, peer.udp_port, peer.tcp_port)
|
||||||
stored_to_tup = await asyncio.gather(
|
stored_to_tup = await asyncio.gather(
|
||||||
*(self.protocol.store_to_peer(hash_value, peer) for peer in peers), loop=self.loop
|
*(self.protocol.store_to_peer(hash_value, peer) for peer in peers)
|
||||||
)
|
)
|
||||||
stored_to = [node_id for node_id, contacted in stored_to_tup if contacted]
|
stored_to = [node_id for node_id, contacted in stored_to_tup if contacted]
|
||||||
if stored_to:
|
if stored_to:
|
||||||
|
@ -182,14 +182,14 @@ class Node:
|
||||||
for address, udp_port in known_node_urls or []
|
for address, udp_port in known_node_urls or []
|
||||||
]))
|
]))
|
||||||
except socket.gaierror:
|
except socket.gaierror:
|
||||||
await asyncio.sleep(30, loop=self.loop)
|
await asyncio.sleep(30)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
self.protocol.peer_manager.reset()
|
self.protocol.peer_manager.reset()
|
||||||
self.protocol.ping_queue.enqueue_maybe_ping(*seed_peers, delay=0.0)
|
self.protocol.ping_queue.enqueue_maybe_ping(*seed_peers, delay=0.0)
|
||||||
await self.peer_search(self.protocol.node_id, shortlist=seed_peers, count=32)
|
await self.peer_search(self.protocol.node_id, shortlist=seed_peers, count=32)
|
||||||
|
|
||||||
await asyncio.sleep(1, loop=self.loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
def start(self, interface: str, known_node_urls: typing.Optional[typing.List[typing.Tuple[str, int]]] = None):
|
def start(self, interface: str, known_node_urls: typing.Optional[typing.List[typing.Tuple[str, int]]] = None):
|
||||||
self._join_task = self.loop.create_task(self.join_network(interface, known_node_urls))
|
self._join_task = self.loop.create_task(self.join_network(interface, known_node_urls))
|
||||||
|
@ -271,7 +271,7 @@ class Node:
|
||||||
def accumulate_peers(self, search_queue: asyncio.Queue,
|
def accumulate_peers(self, search_queue: asyncio.Queue,
|
||||||
peer_queue: typing.Optional[asyncio.Queue] = None
|
peer_queue: typing.Optional[asyncio.Queue] = None
|
||||||
) -> typing.Tuple[asyncio.Queue, asyncio.Task]:
|
) -> typing.Tuple[asyncio.Queue, asyncio.Task]:
|
||||||
queue = peer_queue or asyncio.Queue(loop=self.loop)
|
queue = peer_queue or asyncio.Queue()
|
||||||
return queue, self.loop.create_task(self._accumulate_peers_for_value(search_queue, queue))
|
return queue, self.loop.create_task(self._accumulate_peers_for_value(search_queue, queue))
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -83,7 +83,7 @@ class IterativeFinder(AsyncIterator):
|
||||||
self.contacted: typing.Set['KademliaPeer'] = set()
|
self.contacted: typing.Set['KademliaPeer'] = set()
|
||||||
self.distance = Distance(key)
|
self.distance = Distance(key)
|
||||||
|
|
||||||
self.iteration_queue = asyncio.Queue(loop=self.loop)
|
self.iteration_queue = asyncio.Queue()
|
||||||
|
|
||||||
self.running_probes: typing.Dict['KademliaPeer', asyncio.Task] = {}
|
self.running_probes: typing.Dict['KademliaPeer', asyncio.Task] = {}
|
||||||
self.iteration_count = 0
|
self.iteration_count = 0
|
||||||
|
|
|
@ -253,7 +253,7 @@ class PingQueue:
|
||||||
del self._pending_contacts[peer]
|
del self._pending_contacts[peer]
|
||||||
self.maybe_ping(peer)
|
self.maybe_ping(peer)
|
||||||
break
|
break
|
||||||
await asyncio.sleep(1, loop=self._loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
def start(self):
|
def start(self):
|
||||||
assert not self._running
|
assert not self._running
|
||||||
|
@ -319,10 +319,10 @@ class KademliaProtocol(DatagramProtocol):
|
||||||
self.ping_queue = PingQueue(self.loop, self)
|
self.ping_queue = PingQueue(self.loop, self)
|
||||||
self.node_rpc = KademliaRPC(self, self.loop, self.peer_port)
|
self.node_rpc = KademliaRPC(self, self.loop, self.peer_port)
|
||||||
self.rpc_timeout = rpc_timeout
|
self.rpc_timeout = rpc_timeout
|
||||||
self._split_lock = asyncio.Lock(loop=self.loop)
|
self._split_lock = asyncio.Lock()
|
||||||
self._to_remove: typing.Set['KademliaPeer'] = set()
|
self._to_remove: typing.Set['KademliaPeer'] = set()
|
||||||
self._to_add: typing.Set['KademliaPeer'] = set()
|
self._to_add: typing.Set['KademliaPeer'] = set()
|
||||||
self._wakeup_routing_task = asyncio.Event(loop=self.loop)
|
self._wakeup_routing_task = asyncio.Event()
|
||||||
self.maintaing_routing_task: typing.Optional[asyncio.Task] = None
|
self.maintaing_routing_task: typing.Optional[asyncio.Task] = None
|
||||||
|
|
||||||
@functools.lru_cache(128)
|
@functools.lru_cache(128)
|
||||||
|
@ -385,7 +385,7 @@ class KademliaProtocol(DatagramProtocol):
|
||||||
while self._to_add:
|
while self._to_add:
|
||||||
async with self._split_lock:
|
async with self._split_lock:
|
||||||
await self._add_peer(self._to_add.pop())
|
await self._add_peer(self._to_add.pop())
|
||||||
await asyncio.gather(self._wakeup_routing_task.wait(), asyncio.sleep(.1, loop=self.loop), loop=self.loop)
|
await asyncio.gather(self._wakeup_routing_task.wait(), asyncio.sleep(.1))
|
||||||
self._wakeup_routing_task.clear()
|
self._wakeup_routing_task.clear()
|
||||||
|
|
||||||
def _handle_rpc(self, sender_contact: 'KademliaPeer', message: RequestDatagram):
|
def _handle_rpc(self, sender_contact: 'KademliaPeer', message: RequestDatagram):
|
||||||
|
|
|
@ -8,6 +8,7 @@ from prometheus_client import Gauge
|
||||||
|
|
||||||
from lbry import utils
|
from lbry import utils
|
||||||
from lbry.dht import constants
|
from lbry.dht import constants
|
||||||
|
from lbry.dht.error import RemoteException
|
||||||
from lbry.dht.protocol.distance import Distance
|
from lbry.dht.protocol.distance import Distance
|
||||||
if typing.TYPE_CHECKING:
|
if typing.TYPE_CHECKING:
|
||||||
from lbry.dht.peer import KademliaPeer, PeerManager
|
from lbry.dht.peer import KademliaPeer, PeerManager
|
||||||
|
@ -395,7 +396,7 @@ class TreeRoutingTable:
|
||||||
try:
|
try:
|
||||||
await probe(to_replace)
|
await probe(to_replace)
|
||||||
return False
|
return False
|
||||||
except asyncio.TimeoutError:
|
except (asyncio.TimeoutError, RemoteException):
|
||||||
log.debug("Replacing dead contact in bucket %i: %s:%i with %s:%i ", bucket_index,
|
log.debug("Replacing dead contact in bucket %i: %s:%i with %s:%i ", bucket_index,
|
||||||
to_replace.address, to_replace.udp_port, peer.address, peer.udp_port)
|
to_replace.address, to_replace.udp_port, peer.address, peer.udp_port)
|
||||||
if to_replace in self.buckets[bucket_index]:
|
if to_replace in self.buckets[bucket_index]:
|
||||||
|
|
|
@ -37,7 +37,7 @@ class Component(metaclass=ComponentType):
|
||||||
def running(self):
|
def running(self):
|
||||||
return self._running
|
return self._running
|
||||||
|
|
||||||
async def get_status(self):
|
async def get_status(self): # pylint: disable=no-self-use
|
||||||
return
|
return
|
||||||
|
|
||||||
async def start(self):
|
async def start(self):
|
||||||
|
|
|
@ -42,7 +42,7 @@ class ComponentManager:
|
||||||
self.analytics_manager = analytics_manager
|
self.analytics_manager = analytics_manager
|
||||||
self.component_classes = {}
|
self.component_classes = {}
|
||||||
self.components = set()
|
self.components = set()
|
||||||
self.started = asyncio.Event(loop=self.loop)
|
self.started = asyncio.Event()
|
||||||
self.peer_manager = peer_manager or PeerManager(asyncio.get_event_loop_policy().get_event_loop())
|
self.peer_manager = peer_manager or PeerManager(asyncio.get_event_loop_policy().get_event_loop())
|
||||||
|
|
||||||
for component_name, component_class in self.default_component_classes.items():
|
for component_name, component_class in self.default_component_classes.items():
|
||||||
|
@ -118,7 +118,7 @@ class ComponentManager:
|
||||||
component._setup() for component in stage if not component.running
|
component._setup() for component in stage if not component.running
|
||||||
]
|
]
|
||||||
if needing_start:
|
if needing_start:
|
||||||
await asyncio.wait(needing_start)
|
await asyncio.wait(map(asyncio.create_task, needing_start))
|
||||||
self.started.set()
|
self.started.set()
|
||||||
|
|
||||||
async def stop(self):
|
async def stop(self):
|
||||||
|
@ -131,7 +131,7 @@ class ComponentManager:
|
||||||
component._stop() for component in stage if component.running
|
component._stop() for component in stage if component.running
|
||||||
]
|
]
|
||||||
if needing_stop:
|
if needing_stop:
|
||||||
await asyncio.wait(needing_stop)
|
await asyncio.wait(map(asyncio.create_task, needing_stop))
|
||||||
|
|
||||||
def all_components_running(self, *component_names):
|
def all_components_running(self, *component_names):
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -374,7 +374,7 @@ class FileManagerComponent(Component):
|
||||||
log.info('Done setting up file manager')
|
log.info('Done setting up file manager')
|
||||||
|
|
||||||
async def stop(self):
|
async def stop(self):
|
||||||
self.file_manager.stop()
|
await self.file_manager.stop()
|
||||||
|
|
||||||
|
|
||||||
class BackgroundDownloaderComponent(Component):
|
class BackgroundDownloaderComponent(Component):
|
||||||
|
@ -551,7 +551,7 @@ class UPnPComponent(Component):
|
||||||
while True:
|
while True:
|
||||||
if now:
|
if now:
|
||||||
await self._maintain_redirects()
|
await self._maintain_redirects()
|
||||||
await asyncio.sleep(360, loop=self.component_manager.loop)
|
await asyncio.sleep(360)
|
||||||
|
|
||||||
async def _maintain_redirects(self):
|
async def _maintain_redirects(self):
|
||||||
# setup the gateway if necessary
|
# setup the gateway if necessary
|
||||||
|
@ -560,8 +560,6 @@ class UPnPComponent(Component):
|
||||||
self.upnp = await UPnP.discover(loop=self.component_manager.loop)
|
self.upnp = await UPnP.discover(loop=self.component_manager.loop)
|
||||||
log.info("found upnp gateway: %s", self.upnp.gateway.manufacturer_string)
|
log.info("found upnp gateway: %s", self.upnp.gateway.manufacturer_string)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
if isinstance(err, asyncio.CancelledError): # TODO: remove when updated to 3.8
|
|
||||||
raise
|
|
||||||
log.warning("upnp discovery failed: %s", err)
|
log.warning("upnp discovery failed: %s", err)
|
||||||
self.upnp = None
|
self.upnp = None
|
||||||
|
|
||||||
|
@ -673,7 +671,7 @@ class UPnPComponent(Component):
|
||||||
log.info("Removing upnp redirects: %s", self.upnp_redirects)
|
log.info("Removing upnp redirects: %s", self.upnp_redirects)
|
||||||
await asyncio.wait([
|
await asyncio.wait([
|
||||||
self.upnp.delete_port_mapping(port, protocol) for protocol, port in self.upnp_redirects.items()
|
self.upnp.delete_port_mapping(port, protocol) for protocol, port in self.upnp_redirects.items()
|
||||||
], loop=self.component_manager.loop)
|
])
|
||||||
if self._maintain_redirects_task and not self._maintain_redirects_task.done():
|
if self._maintain_redirects_task and not self._maintain_redirects_task.done():
|
||||||
self._maintain_redirects_task.cancel()
|
self._maintain_redirects_task.cancel()
|
||||||
|
|
||||||
|
|
|
@ -614,7 +614,8 @@ class Daemon(metaclass=JSONRPCServerType):
|
||||||
content_type='application/json'
|
content_type='application/json'
|
||||||
)
|
)
|
||||||
|
|
||||||
async def handle_metrics_get_request(self, request: web.Request):
|
@staticmethod
|
||||||
|
async def handle_metrics_get_request(request: web.Request):
|
||||||
try:
|
try:
|
||||||
return web.Response(
|
return web.Response(
|
||||||
text=prom_generate_latest().decode(),
|
text=prom_generate_latest().decode(),
|
||||||
|
@ -2943,19 +2944,21 @@ class Daemon(metaclass=JSONRPCServerType):
|
||||||
|
|
||||||
@requires(WALLET_COMPONENT)
|
@requires(WALLET_COMPONENT)
|
||||||
async def jsonrpc_channel_sign(
|
async def jsonrpc_channel_sign(
|
||||||
self, channel_name=None, channel_id=None, hexdata=None, channel_account_id=None, wallet_id=None):
|
self, channel_name=None, channel_id=None, hexdata=None, salt=None,
|
||||||
|
channel_account_id=None, wallet_id=None):
|
||||||
"""
|
"""
|
||||||
Signs data using the specified channel signing key.
|
Signs data using the specified channel signing key.
|
||||||
|
|
||||||
Usage:
|
Usage:
|
||||||
channel_sign [<channel_name> | --channel_name=<channel_name>]
|
channel_sign [<channel_name> | --channel_name=<channel_name>] [<channel_id> | --channel_id=<channel_id>]
|
||||||
[<channel_id> | --channel_id=<channel_id>] [<hexdata> | --hexdata=<hexdata>]
|
[<hexdata> | --hexdata=<hexdata>] [<salt> | --salt=<salt>]
|
||||||
[--channel_account_id=<channel_account_id>...] [--wallet_id=<wallet_id>]
|
[--channel_account_id=<channel_account_id>...] [--wallet_id=<wallet_id>]
|
||||||
|
|
||||||
Options:
|
Options:
|
||||||
--channel_name=<channel_name> : (str) name of channel used to sign (or use channel id)
|
--channel_name=<channel_name> : (str) name of channel used to sign (or use channel id)
|
||||||
--channel_id=<channel_id> : (str) claim id of channel used to sign (or use channel name)
|
--channel_id=<channel_id> : (str) claim id of channel used to sign (or use channel name)
|
||||||
--hexdata=<hexdata> : (str) data to sign, encoded as hexadecimal
|
--hexdata=<hexdata> : (str) data to sign, encoded as hexadecimal
|
||||||
|
--salt=<salt> : (str) salt to use for signing, default is to use timestamp
|
||||||
--channel_account_id=<channel_account_id>: (str) one or more account ids for accounts to look in
|
--channel_account_id=<channel_account_id>: (str) one or more account ids for accounts to look in
|
||||||
for channel certificates, defaults to all accounts.
|
for channel certificates, defaults to all accounts.
|
||||||
--wallet_id=<wallet_id> : (str) restrict operation to specific wallet
|
--wallet_id=<wallet_id> : (str) restrict operation to specific wallet
|
||||||
|
@ -2972,11 +2975,13 @@ class Daemon(metaclass=JSONRPCServerType):
|
||||||
signing_channel = await self.get_channel_or_error(
|
signing_channel = await self.get_channel_or_error(
|
||||||
wallet, channel_account_id, channel_id, channel_name, for_signing=True
|
wallet, channel_account_id, channel_id, channel_name, for_signing=True
|
||||||
)
|
)
|
||||||
timestamp = str(int(time.time()))
|
if salt is None:
|
||||||
signature = signing_channel.sign_data(unhexlify(str(hexdata)), timestamp)
|
salt = str(int(time.time()))
|
||||||
|
signature = signing_channel.sign_data(unhexlify(str(hexdata)), salt)
|
||||||
return {
|
return {
|
||||||
'signature': signature,
|
'signature': signature,
|
||||||
'signing_ts': timestamp
|
'signing_ts': salt, # DEPRECATED
|
||||||
|
'salt': salt,
|
||||||
}
|
}
|
||||||
|
|
||||||
@requires(WALLET_COMPONENT)
|
@requires(WALLET_COMPONENT)
|
||||||
|
|
|
@ -80,8 +80,6 @@ class MarketFeed:
|
||||||
self.rate = ExchangeRate(self.market, rate, int(time.time()))
|
self.rate = ExchangeRate(self.market, rate, int(time.time()))
|
||||||
self.last_check = time.time()
|
self.last_check = time.time()
|
||||||
return self.rate
|
return self.rate
|
||||||
except asyncio.CancelledError:
|
|
||||||
raise
|
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
log.warning("Timed out fetching exchange rate from %s.", self.name)
|
log.warning("Timed out fetching exchange rate from %s.", self.name)
|
||||||
except json.JSONDecodeError as e:
|
except json.JSONDecodeError as e:
|
||||||
|
|
|
@ -793,7 +793,7 @@ class SQLiteStorage(SQLiteMixin):
|
||||||
|
|
||||||
await self.db.run(_save_claims)
|
await self.db.run(_save_claims)
|
||||||
if update_file_callbacks:
|
if update_file_callbacks:
|
||||||
await asyncio.wait(update_file_callbacks)
|
await asyncio.wait(map(asyncio.create_task, update_file_callbacks))
|
||||||
if claim_id_to_supports:
|
if claim_id_to_supports:
|
||||||
await self.save_supports(claim_id_to_supports)
|
await self.save_supports(claim_id_to_supports)
|
||||||
|
|
||||||
|
|
|
@ -13,11 +13,12 @@ from lbry.schema.url import URL
|
||||||
from lbry.wallet.dewies import dewies_to_lbc
|
from lbry.wallet.dewies import dewies_to_lbc
|
||||||
from lbry.file.source_manager import SourceManager
|
from lbry.file.source_manager import SourceManager
|
||||||
from lbry.file.source import ManagedDownloadSource
|
from lbry.file.source import ManagedDownloadSource
|
||||||
|
from lbry.extras.daemon.storage import StoredContentClaim
|
||||||
if typing.TYPE_CHECKING:
|
if typing.TYPE_CHECKING:
|
||||||
from lbry.conf import Config
|
from lbry.conf import Config
|
||||||
from lbry.extras.daemon.analytics import AnalyticsManager
|
from lbry.extras.daemon.analytics import AnalyticsManager
|
||||||
from lbry.extras.daemon.storage import SQLiteStorage
|
from lbry.extras.daemon.storage import SQLiteStorage
|
||||||
from lbry.wallet import WalletManager, Output
|
from lbry.wallet import WalletManager
|
||||||
from lbry.extras.daemon.exchange_rate_manager import ExchangeRateManager
|
from lbry.extras.daemon.exchange_rate_manager import ExchangeRateManager
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
@ -49,10 +50,10 @@ class FileManager:
|
||||||
await manager.started.wait()
|
await manager.started.wait()
|
||||||
self.started.set()
|
self.started.set()
|
||||||
|
|
||||||
def stop(self):
|
async def stop(self):
|
||||||
for manager in self.source_managers.values():
|
for manager in self.source_managers.values():
|
||||||
# fixme: pop or not?
|
# fixme: pop or not?
|
||||||
manager.stop()
|
await manager.stop()
|
||||||
self.started.clear()
|
self.started.clear()
|
||||||
|
|
||||||
@cache_concurrent
|
@cache_concurrent
|
||||||
|
@ -98,8 +99,6 @@ class FileManager:
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
raise ResolveTimeoutError(uri)
|
raise ResolveTimeoutError(uri)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
if isinstance(err, asyncio.CancelledError):
|
|
||||||
raise
|
|
||||||
log.exception("Unexpected error resolving stream:")
|
log.exception("Unexpected error resolving stream:")
|
||||||
raise ResolveError(f"Unexpected error resolving stream: {str(err)}")
|
raise ResolveError(f"Unexpected error resolving stream: {str(err)}")
|
||||||
if 'error' in resolved_result:
|
if 'error' in resolved_result:
|
||||||
|
@ -194,21 +193,24 @@ class FileManager:
|
||||||
####################
|
####################
|
||||||
# make downloader and wait for start
|
# make downloader and wait for start
|
||||||
####################
|
####################
|
||||||
|
# temporary with fields we know so downloader can start. Missing fields are populated later.
|
||||||
|
stored_claim = StoredContentClaim(outpoint=outpoint, claim_id=txo.claim_id, name=txo.claim_name,
|
||||||
|
amount=txo.amount, height=txo.tx_ref.height,
|
||||||
|
serialized=claim.to_bytes().hex())
|
||||||
|
|
||||||
if not claim.stream.source.bt_infohash:
|
if not claim.stream.source.bt_infohash:
|
||||||
# fixme: this shouldnt be here
|
# fixme: this shouldnt be here
|
||||||
stream = ManagedStream(
|
stream = ManagedStream(
|
||||||
self.loop, self.config, source_manager.blob_manager, claim.stream.source.sd_hash,
|
self.loop, self.config, source_manager.blob_manager, claim.stream.source.sd_hash,
|
||||||
download_directory, file_name, ManagedStream.STATUS_RUNNING, content_fee=payment,
|
download_directory, file_name, ManagedStream.STATUS_RUNNING, content_fee=payment,
|
||||||
analytics_manager=self.analytics_manager
|
analytics_manager=self.analytics_manager, claim=stored_claim
|
||||||
)
|
)
|
||||||
stream.downloader.node = source_manager.node
|
stream.downloader.node = source_manager.node
|
||||||
else:
|
else:
|
||||||
stream = TorrentSource(
|
stream = TorrentSource(
|
||||||
self.loop, self.config, self.storage, identifier=claim.stream.source.bt_infohash,
|
self.loop, self.config, self.storage, identifier=claim.stream.source.bt_infohash,
|
||||||
file_name=file_name, download_directory=download_directory or self.config.download_dir,
|
file_name=file_name, download_directory=download_directory or self.config.download_dir,
|
||||||
status=ManagedStream.STATUS_RUNNING,
|
status=ManagedStream.STATUS_RUNNING, claim=stored_claim, analytics_manager=self.analytics_manager,
|
||||||
analytics_manager=self.analytics_manager,
|
|
||||||
torrent_session=source_manager.torrent_session
|
torrent_session=source_manager.torrent_session
|
||||||
)
|
)
|
||||||
log.info("starting download for %s", uri)
|
log.info("starting download for %s", uri)
|
||||||
|
@ -240,13 +242,12 @@ class FileManager:
|
||||||
claim_info = await self.storage.get_content_claim_for_torrent(stream.identifier)
|
claim_info = await self.storage.get_content_claim_for_torrent(stream.identifier)
|
||||||
stream.set_claim(claim_info, claim)
|
stream.set_claim(claim_info, claim)
|
||||||
if save_file:
|
if save_file:
|
||||||
await asyncio.wait_for(stream.save_file(), timeout - (self.loop.time() - before_download),
|
await asyncio.wait_for(stream.save_file(), timeout - (self.loop.time() - before_download))
|
||||||
loop=self.loop)
|
|
||||||
return stream
|
return stream
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
error = DownloadDataTimeoutError(stream.sd_hash)
|
error = DownloadDataTimeoutError(stream.sd_hash)
|
||||||
raise error
|
raise error
|
||||||
except Exception as err: # forgive data timeout, don't delete stream
|
except (Exception, asyncio.CancelledError) as err: # forgive data timeout, don't delete stream
|
||||||
expected = (DownloadSDTimeoutError, DownloadDataTimeoutError, InsufficientFundsError,
|
expected = (DownloadSDTimeoutError, DownloadDataTimeoutError, InsufficientFundsError,
|
||||||
KeyFeeAboveMaxAllowedError, ResolveError, InvalidStreamURLError)
|
KeyFeeAboveMaxAllowedError, ResolveError, InvalidStreamURLError)
|
||||||
if isinstance(err, expected):
|
if isinstance(err, expected):
|
||||||
|
|
|
@ -47,10 +47,10 @@ class ManagedDownloadSource:
|
||||||
self.analytics_manager = analytics_manager
|
self.analytics_manager = analytics_manager
|
||||||
self.downloader = None
|
self.downloader = None
|
||||||
|
|
||||||
self.saving = asyncio.Event(loop=self.loop)
|
self.saving = asyncio.Event()
|
||||||
self.finished_writing = asyncio.Event(loop=self.loop)
|
self.finished_writing = asyncio.Event()
|
||||||
self.started_writing = asyncio.Event(loop=self.loop)
|
self.started_writing = asyncio.Event()
|
||||||
self.finished_write_attempt = asyncio.Event(loop=self.loop)
|
self.finished_write_attempt = asyncio.Event()
|
||||||
|
|
||||||
# @classmethod
|
# @classmethod
|
||||||
# async def create(cls, loop: asyncio.AbstractEventLoop, config: 'Config', file_path: str,
|
# async def create(cls, loop: asyncio.AbstractEventLoop, config: 'Config', file_path: str,
|
||||||
|
@ -67,7 +67,7 @@ class ManagedDownloadSource:
|
||||||
async def save_file(self, file_name: Optional[str] = None, download_directory: Optional[str] = None):
|
async def save_file(self, file_name: Optional[str] = None, download_directory: Optional[str] = None):
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
def stop_tasks(self):
|
async def stop_tasks(self):
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
def set_claim(self, claim_info: typing.Dict, claim: 'Claim'):
|
def set_claim(self, claim_info: typing.Dict, claim: 'Claim'):
|
||||||
|
|
|
@ -54,16 +54,16 @@ class SourceManager:
|
||||||
self.storage = storage
|
self.storage = storage
|
||||||
self.analytics_manager = analytics_manager
|
self.analytics_manager = analytics_manager
|
||||||
self._sources: typing.Dict[str, ManagedDownloadSource] = {}
|
self._sources: typing.Dict[str, ManagedDownloadSource] = {}
|
||||||
self.started = asyncio.Event(loop=self.loop)
|
self.started = asyncio.Event()
|
||||||
|
|
||||||
def add(self, source: ManagedDownloadSource):
|
def add(self, source: ManagedDownloadSource):
|
||||||
self._sources[source.identifier] = source
|
self._sources[source.identifier] = source
|
||||||
|
|
||||||
def remove(self, source: ManagedDownloadSource):
|
async def remove(self, source: ManagedDownloadSource):
|
||||||
if source.identifier not in self._sources:
|
if source.identifier not in self._sources:
|
||||||
return
|
return
|
||||||
self._sources.pop(source.identifier)
|
self._sources.pop(source.identifier)
|
||||||
source.stop_tasks()
|
await source.stop_tasks()
|
||||||
|
|
||||||
async def initialize_from_database(self):
|
async def initialize_from_database(self):
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
@ -72,10 +72,10 @@ class SourceManager:
|
||||||
await self.initialize_from_database()
|
await self.initialize_from_database()
|
||||||
self.started.set()
|
self.started.set()
|
||||||
|
|
||||||
def stop(self):
|
async def stop(self):
|
||||||
while self._sources:
|
while self._sources:
|
||||||
_, source = self._sources.popitem()
|
_, source = self._sources.popitem()
|
||||||
source.stop_tasks()
|
await source.stop_tasks()
|
||||||
self.started.clear()
|
self.started.clear()
|
||||||
|
|
||||||
async def create(self, file_path: str, key: Optional[bytes] = None,
|
async def create(self, file_path: str, key: Optional[bytes] = None,
|
||||||
|
@ -83,7 +83,7 @@ class SourceManager:
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):
|
async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):
|
||||||
self.remove(source)
|
await self.remove(source)
|
||||||
if delete_file and source.output_file_exists:
|
if delete_file and source.output_file_exists:
|
||||||
os.remove(source.full_path)
|
os.remove(source.full_path)
|
||||||
|
|
||||||
|
|
|
@ -23,6 +23,7 @@ class BackgroundDownloader:
|
||||||
except ValueError:
|
except ValueError:
|
||||||
return
|
return
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
|
log.debug("Cancelled background downloader")
|
||||||
raise
|
raise
|
||||||
except Exception:
|
except Exception:
|
||||||
log.error("Unexpected download error on background downloader")
|
log.error("Unexpected download error on background downloader")
|
||||||
|
|
|
@ -27,8 +27,8 @@ class StreamDownloader:
|
||||||
self.config = config
|
self.config = config
|
||||||
self.blob_manager = blob_manager
|
self.blob_manager = blob_manager
|
||||||
self.sd_hash = sd_hash
|
self.sd_hash = sd_hash
|
||||||
self.search_queue = asyncio.Queue(loop=loop) # blob hashes to feed into the iterative finder
|
self.search_queue = asyncio.Queue() # blob hashes to feed into the iterative finder
|
||||||
self.peer_queue = asyncio.Queue(loop=loop) # new peers to try
|
self.peer_queue = asyncio.Queue() # new peers to try
|
||||||
self.blob_downloader = BlobDownloader(self.loop, self.config, self.blob_manager, self.peer_queue)
|
self.blob_downloader = BlobDownloader(self.loop, self.config, self.blob_manager, self.peer_queue)
|
||||||
self.descriptor: typing.Optional[StreamDescriptor] = descriptor
|
self.descriptor: typing.Optional[StreamDescriptor] = descriptor
|
||||||
self.node: typing.Optional['Node'] = None
|
self.node: typing.Optional['Node'] = None
|
||||||
|
@ -72,7 +72,7 @@ class StreamDownloader:
|
||||||
now = self.loop.time()
|
now = self.loop.time()
|
||||||
sd_blob = await asyncio.wait_for(
|
sd_blob = await asyncio.wait_for(
|
||||||
self.blob_downloader.download_blob(self.sd_hash, connection_id),
|
self.blob_downloader.download_blob(self.sd_hash, connection_id),
|
||||||
self.config.blob_download_timeout, loop=self.loop
|
self.config.blob_download_timeout
|
||||||
)
|
)
|
||||||
log.info("downloaded sd blob %s", self.sd_hash)
|
log.info("downloaded sd blob %s", self.sd_hash)
|
||||||
self.time_to_descriptor = self.loop.time() - now
|
self.time_to_descriptor = self.loop.time() - now
|
||||||
|
@ -111,7 +111,7 @@ class StreamDownloader:
|
||||||
raise ValueError(f"blob {blob_info.blob_hash} is not part of stream with sd hash {self.sd_hash}")
|
raise ValueError(f"blob {blob_info.blob_hash} is not part of stream with sd hash {self.sd_hash}")
|
||||||
blob = await asyncio.wait_for(
|
blob = await asyncio.wait_for(
|
||||||
self.blob_downloader.download_blob(blob_info.blob_hash, blob_info.length, connection_id),
|
self.blob_downloader.download_blob(blob_info.blob_hash, blob_info.length, connection_id),
|
||||||
self.config.blob_download_timeout * 10, loop=self.loop
|
self.config.blob_download_timeout * 10
|
||||||
)
|
)
|
||||||
return blob
|
return blob
|
||||||
|
|
||||||
|
|
|
@ -60,9 +60,9 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
self.file_output_task: typing.Optional[asyncio.Task] = None
|
self.file_output_task: typing.Optional[asyncio.Task] = None
|
||||||
self.delayed_stop_task: typing.Optional[asyncio.Task] = None
|
self.delayed_stop_task: typing.Optional[asyncio.Task] = None
|
||||||
self.streaming_responses: typing.List[typing.Tuple[Request, StreamResponse]] = []
|
self.streaming_responses: typing.List[typing.Tuple[Request, StreamResponse]] = []
|
||||||
self.fully_reflected = asyncio.Event(loop=self.loop)
|
self.fully_reflected = asyncio.Event()
|
||||||
self.streaming = asyncio.Event(loop=self.loop)
|
self.streaming = asyncio.Event()
|
||||||
self._running = asyncio.Event(loop=self.loop)
|
self._running = asyncio.Event()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def sd_hash(self) -> str:
|
def sd_hash(self) -> str:
|
||||||
|
@ -161,7 +161,7 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
log.info("start downloader for stream (sd hash: %s)", self.sd_hash)
|
log.info("start downloader for stream (sd hash: %s)", self.sd_hash)
|
||||||
self._running.set()
|
self._running.set()
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(self.downloader.start(), timeout, loop=self.loop)
|
await asyncio.wait_for(self.downloader.start(), timeout)
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
self._running.clear()
|
self._running.clear()
|
||||||
raise DownloadSDTimeoutError(self.sd_hash)
|
raise DownloadSDTimeoutError(self.sd_hash)
|
||||||
|
@ -191,7 +191,7 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
Stop any running save/stream tasks as well as the downloader and update the status in the database
|
Stop any running save/stream tasks as well as the downloader and update the status in the database
|
||||||
"""
|
"""
|
||||||
|
|
||||||
self.stop_tasks()
|
await self.stop_tasks()
|
||||||
if (finished and self.status != self.STATUS_FINISHED) or self.status == self.STATUS_RUNNING:
|
if (finished and self.status != self.STATUS_FINISHED) or self.status == self.STATUS_RUNNING:
|
||||||
await self.update_status(self.STATUS_FINISHED if finished else self.STATUS_STOPPED)
|
await self.update_status(self.STATUS_FINISHED if finished else self.STATUS_STOPPED)
|
||||||
|
|
||||||
|
@ -279,7 +279,7 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
log.info("finished saving file for lbry://%s#%s (sd hash %s...) -> %s", self.claim_name, self.claim_id,
|
log.info("finished saving file for lbry://%s#%s (sd hash %s...) -> %s", self.claim_name, self.claim_id,
|
||||||
self.sd_hash[:6], self.full_path)
|
self.sd_hash[:6], self.full_path)
|
||||||
await self.blob_manager.storage.set_saved_file(self.stream_hash)
|
await self.blob_manager.storage.set_saved_file(self.stream_hash)
|
||||||
except Exception as err:
|
except (Exception, asyncio.CancelledError) as err:
|
||||||
if os.path.isfile(output_path):
|
if os.path.isfile(output_path):
|
||||||
log.warning("removing incomplete download %s for %s", output_path, self.sd_hash)
|
log.warning("removing incomplete download %s for %s", output_path, self.sd_hash)
|
||||||
os.remove(output_path)
|
os.remove(output_path)
|
||||||
|
@ -321,15 +321,16 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
await self.update_status(ManagedStream.STATUS_RUNNING)
|
await self.update_status(ManagedStream.STATUS_RUNNING)
|
||||||
self.file_output_task = self.loop.create_task(self._save_file(self.full_path))
|
self.file_output_task = self.loop.create_task(self._save_file(self.full_path))
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(self.started_writing.wait(), self.config.download_timeout, loop=self.loop)
|
await asyncio.wait_for(self.started_writing.wait(), self.config.download_timeout)
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
log.warning("timeout starting to write data for lbry://%s#%s", self.claim_name, self.claim_id)
|
log.warning("timeout starting to write data for lbry://%s#%s", self.claim_name, self.claim_id)
|
||||||
self.stop_tasks()
|
await self.stop_tasks()
|
||||||
await self.update_status(ManagedStream.STATUS_STOPPED)
|
await self.update_status(ManagedStream.STATUS_STOPPED)
|
||||||
|
|
||||||
def stop_tasks(self):
|
async def stop_tasks(self):
|
||||||
if self.file_output_task and not self.file_output_task.done():
|
if self.file_output_task and not self.file_output_task.done():
|
||||||
self.file_output_task.cancel()
|
self.file_output_task.cancel()
|
||||||
|
await asyncio.gather(self.file_output_task, return_exceptions=True)
|
||||||
self.file_output_task = None
|
self.file_output_task = None
|
||||||
while self.streaming_responses:
|
while self.streaming_responses:
|
||||||
req, response = self.streaming_responses.pop()
|
req, response = self.streaming_responses.pop()
|
||||||
|
@ -366,7 +367,7 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
return sent
|
return sent
|
||||||
except ConnectionError:
|
except ConnectionError:
|
||||||
return sent
|
return sent
|
||||||
except (OSError, Exception) as err:
|
except (OSError, Exception, asyncio.CancelledError) as err:
|
||||||
if isinstance(err, asyncio.CancelledError):
|
if isinstance(err, asyncio.CancelledError):
|
||||||
log.warning("stopped uploading %s#%s to reflector", self.claim_name, self.claim_id)
|
log.warning("stopped uploading %s#%s to reflector", self.claim_name, self.claim_id)
|
||||||
elif isinstance(err, OSError):
|
elif isinstance(err, OSError):
|
||||||
|
@ -401,7 +402,7 @@ class ManagedStream(ManagedDownloadSource):
|
||||||
self.sd_hash[:6])
|
self.sd_hash[:6])
|
||||||
await self.stop()
|
await self.stop()
|
||||||
return
|
return
|
||||||
await asyncio.sleep(1, loop=self.loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
def _prepare_range_response_headers(self, get_range: str) -> typing.Tuple[typing.Dict[str, str], int, int, int]:
|
def _prepare_range_response_headers(self, get_range: str) -> typing.Tuple[typing.Dict[str, str], int, int, int]:
|
||||||
if '=' in get_range:
|
if '=' in get_range:
|
||||||
|
|
|
@ -21,7 +21,7 @@ class ReflectorServerProtocol(asyncio.Protocol):
|
||||||
self.loop = asyncio.get_event_loop()
|
self.loop = asyncio.get_event_loop()
|
||||||
self.blob_manager = blob_manager
|
self.blob_manager = blob_manager
|
||||||
self.server_task: asyncio.Task = None
|
self.server_task: asyncio.Task = None
|
||||||
self.started_listening = asyncio.Event(loop=self.loop)
|
self.started_listening = asyncio.Event()
|
||||||
self.buf = b''
|
self.buf = b''
|
||||||
self.transport: asyncio.StreamWriter = None
|
self.transport: asyncio.StreamWriter = None
|
||||||
self.writer: typing.Optional['HashBlobWriter'] = None
|
self.writer: typing.Optional['HashBlobWriter'] = None
|
||||||
|
@ -29,9 +29,9 @@ class ReflectorServerProtocol(asyncio.Protocol):
|
||||||
self.descriptor: typing.Optional['StreamDescriptor'] = None
|
self.descriptor: typing.Optional['StreamDescriptor'] = None
|
||||||
self.sd_blob: typing.Optional['BlobFile'] = None
|
self.sd_blob: typing.Optional['BlobFile'] = None
|
||||||
self.received = []
|
self.received = []
|
||||||
self.incoming = incoming_event or asyncio.Event(loop=self.loop)
|
self.incoming = incoming_event or asyncio.Event()
|
||||||
self.not_incoming = not_incoming_event or asyncio.Event(loop=self.loop)
|
self.not_incoming = not_incoming_event or asyncio.Event()
|
||||||
self.stop_event = stop_event or asyncio.Event(loop=self.loop)
|
self.stop_event = stop_event or asyncio.Event()
|
||||||
self.chunk_size = response_chunk_size
|
self.chunk_size = response_chunk_size
|
||||||
self.wait_for_stop_task: typing.Optional[asyncio.Task] = None
|
self.wait_for_stop_task: typing.Optional[asyncio.Task] = None
|
||||||
self.partial_event = partial_event
|
self.partial_event = partial_event
|
||||||
|
@ -94,7 +94,7 @@ class ReflectorServerProtocol(asyncio.Protocol):
|
||||||
self.incoming.set()
|
self.incoming.set()
|
||||||
self.send_response({"send_sd_blob": True})
|
self.send_response({"send_sd_blob": True})
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(self.sd_blob.verified.wait(), 30, loop=self.loop)
|
await asyncio.wait_for(self.sd_blob.verified.wait(), 30)
|
||||||
self.descriptor = await StreamDescriptor.from_stream_descriptor_blob(
|
self.descriptor = await StreamDescriptor.from_stream_descriptor_blob(
|
||||||
self.loop, self.blob_manager.blob_dir, self.sd_blob
|
self.loop, self.blob_manager.blob_dir, self.sd_blob
|
||||||
)
|
)
|
||||||
|
@ -140,7 +140,7 @@ class ReflectorServerProtocol(asyncio.Protocol):
|
||||||
self.incoming.set()
|
self.incoming.set()
|
||||||
self.send_response({"send_blob": True})
|
self.send_response({"send_blob": True})
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(blob.verified.wait(), 30, loop=self.loop)
|
await asyncio.wait_for(blob.verified.wait(), 30)
|
||||||
self.send_response({"received_blob": True})
|
self.send_response({"received_blob": True})
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
self.send_response({"received_blob": False})
|
self.send_response({"received_blob": False})
|
||||||
|
@ -162,10 +162,10 @@ class ReflectorServer:
|
||||||
self.loop = asyncio.get_event_loop()
|
self.loop = asyncio.get_event_loop()
|
||||||
self.blob_manager = blob_manager
|
self.blob_manager = blob_manager
|
||||||
self.server_task: typing.Optional[asyncio.Task] = None
|
self.server_task: typing.Optional[asyncio.Task] = None
|
||||||
self.started_listening = asyncio.Event(loop=self.loop)
|
self.started_listening = asyncio.Event()
|
||||||
self.stopped_listening = asyncio.Event(loop=self.loop)
|
self.stopped_listening = asyncio.Event()
|
||||||
self.incoming_event = incoming_event or asyncio.Event(loop=self.loop)
|
self.incoming_event = incoming_event or asyncio.Event()
|
||||||
self.not_incoming_event = not_incoming_event or asyncio.Event(loop=self.loop)
|
self.not_incoming_event = not_incoming_event or asyncio.Event()
|
||||||
self.response_chunk_size = response_chunk_size
|
self.response_chunk_size = response_chunk_size
|
||||||
self.stop_event = stop_event
|
self.stop_event = stop_event
|
||||||
self.partial_needs = partial_needs # for testing cases where it doesn't know what it wants
|
self.partial_needs = partial_needs # for testing cases where it doesn't know what it wants
|
||||||
|
|
|
@ -54,7 +54,7 @@ class StreamManager(SourceManager):
|
||||||
self.re_reflect_task: Optional[asyncio.Task] = None
|
self.re_reflect_task: Optional[asyncio.Task] = None
|
||||||
self.update_stream_finished_futs: typing.List[asyncio.Future] = []
|
self.update_stream_finished_futs: typing.List[asyncio.Future] = []
|
||||||
self.running_reflector_uploads: typing.Dict[str, asyncio.Task] = {}
|
self.running_reflector_uploads: typing.Dict[str, asyncio.Task] = {}
|
||||||
self.started = asyncio.Event(loop=self.loop)
|
self.started = asyncio.Event()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def streams(self):
|
def streams(self):
|
||||||
|
@ -150,7 +150,7 @@ class StreamManager(SourceManager):
|
||||||
file_info['added_on'], file_info['fully_reflected']
|
file_info['added_on'], file_info['fully_reflected']
|
||||||
)))
|
)))
|
||||||
if add_stream_tasks:
|
if add_stream_tasks:
|
||||||
await asyncio.gather(*add_stream_tasks, loop=self.loop)
|
await asyncio.gather(*add_stream_tasks)
|
||||||
log.info("Started stream manager with %i files", len(self._sources))
|
log.info("Started stream manager with %i files", len(self._sources))
|
||||||
if not self.node:
|
if not self.node:
|
||||||
log.info("no DHT node given, resuming downloads trusting that we can contact reflector")
|
log.info("no DHT node given, resuming downloads trusting that we can contact reflector")
|
||||||
|
@ -159,14 +159,11 @@ class StreamManager(SourceManager):
|
||||||
self.resume_saving_task = asyncio.ensure_future(asyncio.gather(
|
self.resume_saving_task = asyncio.ensure_future(asyncio.gather(
|
||||||
*(self._sources[sd_hash].save_file(file_name, download_directory)
|
*(self._sources[sd_hash].save_file(file_name, download_directory)
|
||||||
for (file_name, download_directory, sd_hash) in to_resume_saving),
|
for (file_name, download_directory, sd_hash) in to_resume_saving),
|
||||||
loop=self.loop
|
|
||||||
))
|
))
|
||||||
|
|
||||||
async def reflect_streams(self):
|
async def reflect_streams(self):
|
||||||
try:
|
try:
|
||||||
return await self._reflect_streams()
|
return await self._reflect_streams()
|
||||||
except asyncio.CancelledError:
|
|
||||||
raise
|
|
||||||
except Exception:
|
except Exception:
|
||||||
log.exception("reflector task encountered an unexpected error!")
|
log.exception("reflector task encountered an unexpected error!")
|
||||||
|
|
||||||
|
@ -186,21 +183,21 @@ class StreamManager(SourceManager):
|
||||||
batch.append(self.reflect_stream(stream))
|
batch.append(self.reflect_stream(stream))
|
||||||
if len(batch) >= self.config.concurrent_reflector_uploads:
|
if len(batch) >= self.config.concurrent_reflector_uploads:
|
||||||
log.debug("waiting for batch of %s reflecting streams", len(batch))
|
log.debug("waiting for batch of %s reflecting streams", len(batch))
|
||||||
await asyncio.gather(*batch, loop=self.loop)
|
await asyncio.gather(*batch)
|
||||||
log.debug("done processing %s streams", len(batch))
|
log.debug("done processing %s streams", len(batch))
|
||||||
batch = []
|
batch = []
|
||||||
if batch:
|
if batch:
|
||||||
log.debug("waiting for batch of %s reflecting streams", len(batch))
|
log.debug("waiting for batch of %s reflecting streams", len(batch))
|
||||||
await asyncio.gather(*batch, loop=self.loop)
|
await asyncio.gather(*batch)
|
||||||
log.debug("done processing %s streams", len(batch))
|
log.debug("done processing %s streams", len(batch))
|
||||||
await asyncio.sleep(300, loop=self.loop)
|
await asyncio.sleep(300)
|
||||||
|
|
||||||
async def start(self):
|
async def start(self):
|
||||||
await super().start()
|
await super().start()
|
||||||
self.re_reflect_task = self.loop.create_task(self.reflect_streams())
|
self.re_reflect_task = self.loop.create_task(self.reflect_streams())
|
||||||
|
|
||||||
def stop(self):
|
async def stop(self):
|
||||||
super().stop()
|
await super().stop()
|
||||||
if self.resume_saving_task and not self.resume_saving_task.done():
|
if self.resume_saving_task and not self.resume_saving_task.done():
|
||||||
self.resume_saving_task.cancel()
|
self.resume_saving_task.cancel()
|
||||||
if self.re_reflect_task and not self.re_reflect_task.done():
|
if self.re_reflect_task and not self.re_reflect_task.done():
|
||||||
|
@ -227,7 +224,8 @@ class StreamManager(SourceManager):
|
||||||
)
|
)
|
||||||
return task
|
return task
|
||||||
|
|
||||||
async def _retriable_reflect_stream(self, stream, host, port):
|
@staticmethod
|
||||||
|
async def _retriable_reflect_stream(stream, host, port):
|
||||||
sent = await stream.upload_to_reflector(host, port)
|
sent = await stream.upload_to_reflector(host, port)
|
||||||
while not stream.is_fully_reflected and stream.reflector_progress > 0 and len(sent) > 0:
|
while not stream.is_fully_reflected and stream.reflector_progress > 0 and len(sent) > 0:
|
||||||
stream.reflector_progress = 0
|
stream.reflector_progress = 0
|
||||||
|
@ -262,7 +260,7 @@ class StreamManager(SourceManager):
|
||||||
return
|
return
|
||||||
if source.identifier in self.running_reflector_uploads:
|
if source.identifier in self.running_reflector_uploads:
|
||||||
self.running_reflector_uploads[source.identifier].cancel()
|
self.running_reflector_uploads[source.identifier].cancel()
|
||||||
source.stop_tasks()
|
await source.stop_tasks()
|
||||||
if source.identifier in self.streams:
|
if source.identifier in self.streams:
|
||||||
del self.streams[source.identifier]
|
del self.streams[source.identifier]
|
||||||
blob_hashes = [source.identifier] + [b.blob_hash for b in source.descriptor.blobs[:-1]]
|
blob_hashes = [source.identifier] + [b.blob_hash for b in source.descriptor.blobs[:-1]]
|
||||||
|
|
|
@ -87,7 +87,7 @@ class TorrentHandle:
|
||||||
self._show_status()
|
self._show_status()
|
||||||
if self.finished.is_set():
|
if self.finished.is_set():
|
||||||
break
|
break
|
||||||
await asyncio.sleep(0.1, loop=self._loop)
|
await asyncio.sleep(0.1)
|
||||||
|
|
||||||
async def pause(self):
|
async def pause(self):
|
||||||
await self._loop.run_in_executor(
|
await self._loop.run_in_executor(
|
||||||
|
@ -150,7 +150,7 @@ class TorrentSession:
|
||||||
await self._loop.run_in_executor(
|
await self._loop.run_in_executor(
|
||||||
self._executor, self._pop_alerts
|
self._executor, self._pop_alerts
|
||||||
)
|
)
|
||||||
await asyncio.sleep(1, loop=self._loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
async def pause(self):
|
async def pause(self):
|
||||||
await self._loop.run_in_executor(
|
await self._loop.run_in_executor(
|
||||||
|
|
|
@ -36,7 +36,7 @@ class Torrent:
|
||||||
def __init__(self, loop, handle):
|
def __init__(self, loop, handle):
|
||||||
self._loop = loop
|
self._loop = loop
|
||||||
self._handle = handle
|
self._handle = handle
|
||||||
self.finished = asyncio.Event(loop=loop)
|
self.finished = asyncio.Event()
|
||||||
|
|
||||||
def _threaded_update_status(self):
|
def _threaded_update_status(self):
|
||||||
status = self._handle.status()
|
status = self._handle.status()
|
||||||
|
@ -58,7 +58,7 @@ class Torrent:
|
||||||
log.info("finished downloading torrent!")
|
log.info("finished downloading torrent!")
|
||||||
await self.pause()
|
await self.pause()
|
||||||
break
|
break
|
||||||
await asyncio.sleep(1, loop=self._loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
async def pause(self):
|
async def pause(self):
|
||||||
log.info("pause torrent")
|
log.info("pause torrent")
|
||||||
|
|
|
@ -74,7 +74,7 @@ class TorrentSource(ManagedDownloadSource):
|
||||||
def bt_infohash(self):
|
def bt_infohash(self):
|
||||||
return self.identifier
|
return self.identifier
|
||||||
|
|
||||||
def stop_tasks(self):
|
async def stop_tasks(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
@ -118,8 +118,8 @@ class TorrentManager(SourceManager):
|
||||||
async def start(self):
|
async def start(self):
|
||||||
await super().start()
|
await super().start()
|
||||||
|
|
||||||
def stop(self):
|
async def stop(self):
|
||||||
super().stop()
|
await super().stop()
|
||||||
log.info("finished stopping the torrent manager")
|
log.info("finished stopping the torrent manager")
|
||||||
|
|
||||||
async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):
|
async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):
|
||||||
|
|
|
@ -450,8 +450,8 @@ def is_running_from_bundle():
|
||||||
|
|
||||||
|
|
||||||
class LockWithMetrics(asyncio.Lock):
|
class LockWithMetrics(asyncio.Lock):
|
||||||
def __init__(self, acquire_metric, held_time_metric, loop=None):
|
def __init__(self, acquire_metric, held_time_metric):
|
||||||
super().__init__(loop=loop)
|
super().__init__()
|
||||||
self._acquire_metric = acquire_metric
|
self._acquire_metric = acquire_metric
|
||||||
self._lock_held_time_metric = held_time_metric
|
self._lock_held_time_metric = held_time_metric
|
||||||
self._lock_acquired_time = None
|
self._lock_acquired_time = None
|
||||||
|
|
|
@ -141,7 +141,7 @@ class CoinSelector:
|
||||||
_) -> List[OutputEffectiveAmountEstimator]:
|
_) -> List[OutputEffectiveAmountEstimator]:
|
||||||
""" Accumulate UTXOs at random until there is enough to cover the target. """
|
""" Accumulate UTXOs at random until there is enough to cover the target. """
|
||||||
target = self.target + self.cost_of_change
|
target = self.target + self.cost_of_change
|
||||||
self.random.shuffle(txos, self.random.random)
|
self.random.shuffle(txos, random=self.random.random) # pylint: disable=deprecated-argument
|
||||||
selection = []
|
selection = []
|
||||||
amount = 0
|
amount = 0
|
||||||
for coin in txos:
|
for coin in txos:
|
||||||
|
|
|
@ -329,10 +329,10 @@ class Ledger(metaclass=LedgerRegistry):
|
||||||
async def start(self):
|
async def start(self):
|
||||||
if not os.path.exists(self.path):
|
if not os.path.exists(self.path):
|
||||||
os.mkdir(self.path)
|
os.mkdir(self.path)
|
||||||
await asyncio.wait([
|
await asyncio.wait(map(asyncio.create_task, [
|
||||||
self.db.open(),
|
self.db.open(),
|
||||||
self.headers.open()
|
self.headers.open()
|
||||||
])
|
]))
|
||||||
fully_synced = self.on_ready.first
|
fully_synced = self.on_ready.first
|
||||||
asyncio.create_task(self.network.start())
|
asyncio.create_task(self.network.start())
|
||||||
await self.network.on_connected.first
|
await self.network.on_connected.first
|
||||||
|
@ -466,9 +466,9 @@ class Ledger(metaclass=LedgerRegistry):
|
||||||
async def subscribe_accounts(self):
|
async def subscribe_accounts(self):
|
||||||
if self.network.is_connected and self.accounts:
|
if self.network.is_connected and self.accounts:
|
||||||
log.info("Subscribe to %i accounts", len(self.accounts))
|
log.info("Subscribe to %i accounts", len(self.accounts))
|
||||||
await asyncio.wait([
|
await asyncio.wait(map(asyncio.create_task, [
|
||||||
self.subscribe_account(a) for a in self.accounts
|
self.subscribe_account(a) for a in self.accounts
|
||||||
])
|
]))
|
||||||
|
|
||||||
async def subscribe_account(self, account: Account):
|
async def subscribe_account(self, account: Account):
|
||||||
for address_manager in account.address_managers.values():
|
for address_manager in account.address_managers.values():
|
||||||
|
@ -938,9 +938,7 @@ class Ledger(metaclass=LedgerRegistry):
|
||||||
"%d change addresses (gap: %d), %d channels, %d certificates and %d claims. ",
|
"%d change addresses (gap: %d), %d channels, %d certificates and %d claims. ",
|
||||||
account.id, balance, total_receiving, account.receiving.gap, total_change,
|
account.id, balance, total_receiving, account.receiving.gap, total_change,
|
||||||
account.change.gap, channel_count, len(account.channel_keys), claim_count)
|
account.change.gap, channel_count, len(account.channel_keys), claim_count)
|
||||||
except Exception as err:
|
except Exception:
|
||||||
if isinstance(err, asyncio.CancelledError): # TODO: remove when updated to 3.8
|
|
||||||
raise
|
|
||||||
log.exception(
|
log.exception(
|
||||||
'Failed to display wallet state, please file issue '
|
'Failed to display wallet state, please file issue '
|
||||||
'for this bug along with the traceback you see below:')
|
'for this bug along with the traceback you see below:')
|
||||||
|
@ -963,9 +961,7 @@ class Ledger(metaclass=LedgerRegistry):
|
||||||
claim_ids = [p.purchased_claim_id for p in purchases]
|
claim_ids = [p.purchased_claim_id for p in purchases]
|
||||||
try:
|
try:
|
||||||
resolved, _, _, _ = await self.claim_search([], claim_ids=claim_ids)
|
resolved, _, _, _ = await self.claim_search([], claim_ids=claim_ids)
|
||||||
except Exception as err:
|
except Exception:
|
||||||
if isinstance(err, asyncio.CancelledError): # TODO: remove when updated to 3.8
|
|
||||||
raise
|
|
||||||
log.exception("Resolve failed while looking up purchased claim ids:")
|
log.exception("Resolve failed while looking up purchased claim ids:")
|
||||||
resolved = []
|
resolved = []
|
||||||
lookup = {claim.claim_id: claim for claim in resolved}
|
lookup = {claim.claim_id: claim for claim in resolved}
|
||||||
|
@ -1045,9 +1041,7 @@ class Ledger(metaclass=LedgerRegistry):
|
||||||
claim_ids = collection.claim.collection.claims.ids[offset:page_size + offset]
|
claim_ids = collection.claim.collection.claims.ids[offset:page_size + offset]
|
||||||
try:
|
try:
|
||||||
resolve_results, _, _, _ = await self.claim_search([], claim_ids=claim_ids)
|
resolve_results, _, _, _ = await self.claim_search([], claim_ids=claim_ids)
|
||||||
except Exception as err:
|
except Exception:
|
||||||
if isinstance(err, asyncio.CancelledError): # TODO: remove when updated to 3.8
|
|
||||||
raise
|
|
||||||
log.exception("Resolve failed while looking up collection claim ids:")
|
log.exception("Resolve failed while looking up collection claim ids:")
|
||||||
return []
|
return []
|
||||||
claims = []
|
claims = []
|
||||||
|
|
|
@ -117,7 +117,7 @@ class ClientSession(BaseClientSession):
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
await asyncio.sleep(max(0, max_idle - (now - self.last_send)))
|
await asyncio.sleep(max(0, max_idle - (now - self.last_send)))
|
||||||
except Exception as err:
|
except (Exception, asyncio.CancelledError) as err:
|
||||||
if isinstance(err, asyncio.CancelledError):
|
if isinstance(err, asyncio.CancelledError):
|
||||||
log.info("closing connection to %s:%i", *self.server)
|
log.info("closing connection to %s:%i", *self.server)
|
||||||
else:
|
else:
|
||||||
|
@ -214,7 +214,7 @@ class Network:
|
||||||
def loop_task_done_callback(f):
|
def loop_task_done_callback(f):
|
||||||
try:
|
try:
|
||||||
f.result()
|
f.result()
|
||||||
except Exception:
|
except (Exception, asyncio.CancelledError):
|
||||||
if self.running:
|
if self.running:
|
||||||
log.exception("wallet server connection loop crashed")
|
log.exception("wallet server connection loop crashed")
|
||||||
|
|
||||||
|
@ -312,7 +312,8 @@ class Network:
|
||||||
sleep_delay = 30
|
sleep_delay = 30
|
||||||
while self.running:
|
while self.running:
|
||||||
await asyncio.wait(
|
await asyncio.wait(
|
||||||
[asyncio.sleep(30), self._urgent_need_reconnect.wait()], return_when=asyncio.FIRST_COMPLETED
|
map(asyncio.create_task, [asyncio.sleep(30), self._urgent_need_reconnect.wait()]),
|
||||||
|
return_when=asyncio.FIRST_COMPLETED
|
||||||
)
|
)
|
||||||
if self._urgent_need_reconnect.is_set():
|
if self._urgent_need_reconnect.is_set():
|
||||||
sleep_delay = 30
|
sleep_delay = 30
|
||||||
|
@ -338,7 +339,7 @@ class Network:
|
||||||
try:
|
try:
|
||||||
if not self._urgent_need_reconnect.is_set():
|
if not self._urgent_need_reconnect.is_set():
|
||||||
await asyncio.wait(
|
await asyncio.wait(
|
||||||
[self._keepalive_task, self._urgent_need_reconnect.wait()],
|
[self._keepalive_task, asyncio.create_task(self._urgent_need_reconnect.wait())],
|
||||||
return_when=asyncio.FIRST_COMPLETED
|
return_when=asyncio.FIRST_COMPLETED
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
|
|
|
@ -214,6 +214,7 @@ class SPVNode:
|
||||||
self.port = 50001 + node_number # avoid conflict with default daemon
|
self.port = 50001 + node_number # avoid conflict with default daemon
|
||||||
self.udp_port = self.port
|
self.udp_port = self.port
|
||||||
self.elastic_notifier_port = 19080 + node_number
|
self.elastic_notifier_port = 19080 + node_number
|
||||||
|
self.elastic_services = f'localhost:9200/localhost:{self.elastic_notifier_port}'
|
||||||
self.session_timeout = 600
|
self.session_timeout = 600
|
||||||
self.stopped = True
|
self.stopped = True
|
||||||
self.index_name = uuid4().hex
|
self.index_name = uuid4().hex
|
||||||
|
@ -235,7 +236,7 @@ class SPVNode:
|
||||||
'host': self.hostname,
|
'host': self.hostname,
|
||||||
'tcp_port': self.port,
|
'tcp_port': self.port,
|
||||||
'udp_port': self.udp_port,
|
'udp_port': self.udp_port,
|
||||||
'elastic_notifier_port': self.elastic_notifier_port,
|
'elastic_services': self.elastic_services,
|
||||||
'session_timeout': self.session_timeout,
|
'session_timeout': self.session_timeout,
|
||||||
'max_query_workers': 0,
|
'max_query_workers': 0,
|
||||||
'es_index_prefix': self.index_name,
|
'es_index_prefix': self.index_name,
|
||||||
|
@ -263,7 +264,6 @@ class SPVNode:
|
||||||
await self.server.start()
|
await self.server.start()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.stopped = True
|
self.stopped = True
|
||||||
if not isinstance(e, asyncio.CancelledError):
|
|
||||||
log.exception("failed to start spv node")
|
log.exception("failed to start spv node")
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
|
|
|
@ -395,8 +395,8 @@ class RPCSession(SessionBase):
|
||||||
namespace=NAMESPACE, labelnames=("version",)
|
namespace=NAMESPACE, labelnames=("version",)
|
||||||
)
|
)
|
||||||
|
|
||||||
def __init__(self, *, framer=None, loop=None, connection=None):
|
def __init__(self, *, framer=None, connection=None):
|
||||||
super().__init__(framer=framer, loop=loop)
|
super().__init__(framer=framer)
|
||||||
self.connection = connection or self.default_connection()
|
self.connection = connection or self.default_connection()
|
||||||
self.client_version = 'unknown'
|
self.client_version = 'unknown'
|
||||||
|
|
||||||
|
|
|
@ -182,6 +182,8 @@ class Wallet:
|
||||||
raise InvalidPasswordError()
|
raise InvalidPasswordError()
|
||||||
if "unknown compression method" in e.args[0].lower():
|
if "unknown compression method" in e.args[0].lower():
|
||||||
raise InvalidPasswordError()
|
raise InvalidPasswordError()
|
||||||
|
if "invalid window size" in e.args[0].lower():
|
||||||
|
raise InvalidPasswordError()
|
||||||
raise
|
raise
|
||||||
return json.loads(decompressed)
|
return json.loads(decompressed)
|
||||||
|
|
||||||
|
|
|
@ -28,6 +28,7 @@ disable=
|
||||||
no-else-return,
|
no-else-return,
|
||||||
cyclic-import,
|
cyclic-import,
|
||||||
missing-docstring,
|
missing-docstring,
|
||||||
|
consider-using-f-string,
|
||||||
duplicate-code,
|
duplicate-code,
|
||||||
expression-not-assigned,
|
expression-not-assigned,
|
||||||
inconsistent-return-statements,
|
inconsistent-return-statements,
|
||||||
|
|
8
setup.py
8
setup.py
|
@ -18,7 +18,7 @@ setup(
|
||||||
long_description_content_type="text/markdown",
|
long_description_content_type="text/markdown",
|
||||||
keywords="lbry protocol media",
|
keywords="lbry protocol media",
|
||||||
license='MIT',
|
license='MIT',
|
||||||
python_requires='>=3.7',
|
python_requires='>=3.8',
|
||||||
packages=find_packages(exclude=('tests',)),
|
packages=find_packages(exclude=('tests',)),
|
||||||
zip_safe=False,
|
zip_safe=False,
|
||||||
entry_points={
|
entry_points={
|
||||||
|
@ -36,7 +36,7 @@ setup(
|
||||||
'distro==1.4.0',
|
'distro==1.4.0',
|
||||||
'base58==1.0.0',
|
'base58==1.0.0',
|
||||||
'cffi==1.13.2',
|
'cffi==1.13.2',
|
||||||
'cryptography==2.5',
|
'cryptography==3.4.7',
|
||||||
'protobuf==3.17.2',
|
'protobuf==3.17.2',
|
||||||
'prometheus_client==0.7.1',
|
'prometheus_client==0.7.1',
|
||||||
'ecdsa==0.13.3',
|
'ecdsa==0.13.3',
|
||||||
|
@ -50,14 +50,14 @@ setup(
|
||||||
],
|
],
|
||||||
extras_require={
|
extras_require={
|
||||||
'lint': [
|
'lint': [
|
||||||
'pylint==2.10.0'
|
'pylint==2.13.9'
|
||||||
],
|
],
|
||||||
'test': [
|
'test': [
|
||||||
'coverage',
|
'coverage',
|
||||||
'jsonschema==4.4.0',
|
'jsonschema==4.4.0',
|
||||||
],
|
],
|
||||||
'hub': [
|
'hub': [
|
||||||
'hub@git+https://github.com/lbryio/hub.git@024aceda53fe6d1ab8d519b73584437c25de6975'
|
'hub@git+https://github.com/lbryio/hub.git@929448d64bcbe6c5e476757ec78456beaa85e56a'
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
classifiers=[
|
classifiers=[
|
||||||
|
|
|
@ -51,7 +51,8 @@ def mock_network_loop(loop: asyncio.AbstractEventLoop,
|
||||||
return rx.datagram_received(data, from_addr)
|
return rx.datagram_received(data, from_addr)
|
||||||
|
|
||||||
protocol = proto_lam()
|
protocol = proto_lam()
|
||||||
transport = asyncio.DatagramTransport(extra={'socket': mock_sock})
|
transport = mock.Mock(spec=asyncio.DatagramTransport)
|
||||||
|
transport.get_extra_info = lambda k: {'socket': mock_sock}[k]
|
||||||
transport.is_closing = lambda: False
|
transport.is_closing = lambda: False
|
||||||
transport.close = lambda: mock_sock.close()
|
transport.close = lambda: mock_sock.close()
|
||||||
mock_sock.sendto = sendto
|
mock_sock.sendto = sendto
|
||||||
|
@ -60,7 +61,6 @@ def mock_network_loop(loop: asyncio.AbstractEventLoop,
|
||||||
dht_network[from_addr] = protocol
|
dht_network[from_addr] = protocol
|
||||||
return transport, protocol
|
return transport, protocol
|
||||||
|
|
||||||
with mock.patch('socket.socket') as mock_socket:
|
|
||||||
mock_sock = mock.Mock(spec=socket.socket)
|
mock_sock = mock.Mock(spec=socket.socket)
|
||||||
mock_sock.setsockopt = lambda *_: None
|
mock_sock.setsockopt = lambda *_: None
|
||||||
mock_sock.bind = lambda *_: None
|
mock_sock.bind = lambda *_: None
|
||||||
|
@ -70,6 +70,5 @@ def mock_network_loop(loop: asyncio.AbstractEventLoop,
|
||||||
mock_sock.close = lambda: None
|
mock_sock.close = lambda: None
|
||||||
mock_sock.type = socket.SOCK_DGRAM
|
mock_sock.type = socket.SOCK_DGRAM
|
||||||
mock_sock.fileno = lambda: 7
|
mock_sock.fileno = lambda: 7
|
||||||
mock_socket.return_value = mock_sock
|
|
||||||
loop.create_datagram_endpoint = create_datagram_endpoint
|
loop.create_datagram_endpoint = create_datagram_endpoint
|
||||||
yield
|
yield
|
||||||
|
|
|
@ -31,7 +31,7 @@ STREAM_TYPES = {
|
||||||
|
|
||||||
def verify(channel, data, signature, channel_hash=None):
|
def verify(channel, data, signature, channel_hash=None):
|
||||||
pieces = [
|
pieces = [
|
||||||
signature['signing_ts'].encode(),
|
signature['salt'].encode(),
|
||||||
channel_hash or channel.claim_hash,
|
channel_hash or channel.claim_hash,
|
||||||
data
|
data
|
||||||
]
|
]
|
||||||
|
@ -1239,8 +1239,13 @@ class ChannelCommands(CommandTestCase):
|
||||||
channel = channel_tx.outputs[0]
|
channel = channel_tx.outputs[0]
|
||||||
signature1 = await self.out(self.daemon.jsonrpc_channel_sign(channel_name='@signer', hexdata=data_to_sign))
|
signature1 = await self.out(self.daemon.jsonrpc_channel_sign(channel_name='@signer', hexdata=data_to_sign))
|
||||||
signature2 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=data_to_sign))
|
signature2 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=data_to_sign))
|
||||||
|
signature3 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=data_to_sign, salt='beef'))
|
||||||
|
signature4 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=data_to_sign, salt='beef'))
|
||||||
|
self.assertNotEqual(signature2, signature3)
|
||||||
|
self.assertEqual(signature3, signature4)
|
||||||
self.assertTrue(verify(channel, unhexlify(data_to_sign), signature1))
|
self.assertTrue(verify(channel, unhexlify(data_to_sign), signature1))
|
||||||
self.assertTrue(verify(channel, unhexlify(data_to_sign), signature2))
|
self.assertTrue(verify(channel, unhexlify(data_to_sign), signature2))
|
||||||
|
self.assertTrue(verify(channel, unhexlify(data_to_sign), signature3))
|
||||||
signature3 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=99))
|
signature3 = await self.out(self.daemon.jsonrpc_channel_sign(channel_id=channel.claim_id, hexdata=99))
|
||||||
self.assertTrue(verify(channel, unhexlify('99'), signature3))
|
self.assertTrue(verify(channel, unhexlify('99'), signature3))
|
||||||
|
|
||||||
|
|
|
@ -89,6 +89,21 @@ class FileCommands(CommandTestCase):
|
||||||
await self.reflector.blob_manager.delete_blobs(all_except_sd)
|
await self.reflector.blob_manager.delete_blobs(all_except_sd)
|
||||||
self.assertEqual(all_except_sd, await self.daemon.jsonrpc_file_reflect(sd_hash=sd_hash))
|
self.assertEqual(all_except_sd, await self.daemon.jsonrpc_file_reflect(sd_hash=sd_hash))
|
||||||
|
|
||||||
|
async def test_sd_blob_fields_fallback(self):
|
||||||
|
claim_id = self.get_claim_id(await self.stream_create('foo', '0.01', suffix='.txt'))
|
||||||
|
stream = (await self.daemon.jsonrpc_file_list())["items"][0]
|
||||||
|
stream.descriptor.suggested_file_name = ' '
|
||||||
|
stream.descriptor.stream_name = ' '
|
||||||
|
stream.descriptor.stream_hash = stream.descriptor.get_stream_hash()
|
||||||
|
sd_hash = stream.descriptor.sd_hash = stream.descriptor.calculate_sd_hash()
|
||||||
|
await stream.descriptor.make_sd_blob()
|
||||||
|
await self.daemon.jsonrpc_file_delete(claim_name='foo')
|
||||||
|
await self.stream_update(claim_id=claim_id, sd_hash=sd_hash)
|
||||||
|
file_dict = await self.out(self.daemon.jsonrpc_get('lbry://foo', save_file=True))
|
||||||
|
self.assertEqual(file_dict['suggested_file_name'], stream.file_name)
|
||||||
|
self.assertEqual(file_dict['stream_name'], stream.file_name)
|
||||||
|
self.assertEqual(file_dict['mime_type'], 'text/plain')
|
||||||
|
|
||||||
async def test_file_management(self):
|
async def test_file_management(self):
|
||||||
await self.stream_create('foo', '0.01')
|
await self.stream_create('foo', '0.01')
|
||||||
await self.stream_create('foo2', '0.01')
|
await self.stream_create('foo2', '0.01')
|
||||||
|
@ -339,7 +354,7 @@ class FileCommands(CommandTestCase):
|
||||||
await self.daemon.jsonrpc_get('lbry://foo')
|
await self.daemon.jsonrpc_get('lbry://foo')
|
||||||
with open(original_path, 'wb') as handle:
|
with open(original_path, 'wb') as handle:
|
||||||
handle.write(b'some other stuff was there instead')
|
handle.write(b'some other stuff was there instead')
|
||||||
self.daemon.file_manager.stop()
|
await self.daemon.file_manager.stop()
|
||||||
await self.daemon.file_manager.start()
|
await self.daemon.file_manager.start()
|
||||||
await asyncio.wait_for(self.wait_files_to_complete(), timeout=5) # if this hangs, file didn't get set completed
|
await asyncio.wait_for(self.wait_files_to_complete(), timeout=5) # if this hangs, file didn't get set completed
|
||||||
# check that internal state got through up to the file list API
|
# check that internal state got through up to the file list API
|
||||||
|
@ -367,8 +382,7 @@ class FileCommands(CommandTestCase):
|
||||||
resp = await self.out(self.daemon.jsonrpc_get('lbry://foo', timeout=2))
|
resp = await self.out(self.daemon.jsonrpc_get('lbry://foo', timeout=2))
|
||||||
self.assertNotIn('error', resp)
|
self.assertNotIn('error', resp)
|
||||||
self.assertTrue(os.path.isfile(path))
|
self.assertTrue(os.path.isfile(path))
|
||||||
self.daemon.file_manager.stop()
|
await self.daemon.file_manager.stop()
|
||||||
await asyncio.sleep(0.01, loop=self.loop) # FIXME: this sleep should not be needed
|
|
||||||
self.assertFalse(os.path.isfile(path))
|
self.assertFalse(os.path.isfile(path))
|
||||||
|
|
||||||
async def test_incomplete_downloads_retry(self):
|
async def test_incomplete_downloads_retry(self):
|
||||||
|
@ -463,7 +477,7 @@ class FileCommands(CommandTestCase):
|
||||||
|
|
||||||
# restart the daemon and make sure the fee is still there
|
# restart the daemon and make sure the fee is still there
|
||||||
|
|
||||||
self.daemon.file_manager.stop()
|
await self.daemon.file_manager.stop()
|
||||||
await self.daemon.file_manager.start()
|
await self.daemon.file_manager.start()
|
||||||
self.assertItemCount(await self.daemon.jsonrpc_file_list(), 1)
|
self.assertItemCount(await self.daemon.jsonrpc_file_list(), 1)
|
||||||
self.assertEqual((await self.daemon.jsonrpc_file_list())['items'][0].content_fee.raw, raw_content_fee)
|
self.assertEqual((await self.daemon.jsonrpc_file_list())['items'][0].content_fee.raw, raw_content_fee)
|
||||||
|
|
|
@ -3,7 +3,9 @@ import hashlib
|
||||||
import aiohttp
|
import aiohttp
|
||||||
import aiohttp.web
|
import aiohttp.web
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import contextlib
|
||||||
|
|
||||||
|
from lbry.file.source import ManagedDownloadSource
|
||||||
from lbry.utils import aiohttp_request
|
from lbry.utils import aiohttp_request
|
||||||
from lbry.blob.blob_file import MAX_BLOB_SIZE
|
from lbry.blob.blob_file import MAX_BLOB_SIZE
|
||||||
from lbry.testcase import CommandTestCase
|
from lbry.testcase import CommandTestCase
|
||||||
|
@ -21,7 +23,7 @@ def get_random_bytes(n: int) -> bytes:
|
||||||
|
|
||||||
class RangeRequests(CommandTestCase):
|
class RangeRequests(CommandTestCase):
|
||||||
async def _restart_stream_manager(self):
|
async def _restart_stream_manager(self):
|
||||||
self.daemon.file_manager.stop()
|
await self.daemon.file_manager.stop()
|
||||||
await self.daemon.file_manager.start()
|
await self.daemon.file_manager.start()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -352,13 +354,20 @@ class RangeRequests(CommandTestCase):
|
||||||
path = stream.full_path
|
path = stream.full_path
|
||||||
self.assertIsNotNone(path)
|
self.assertIsNotNone(path)
|
||||||
if wait_for_start_writing:
|
if wait_for_start_writing:
|
||||||
|
with contextlib.suppress(asyncio.CancelledError):
|
||||||
await stream.started_writing.wait()
|
await stream.started_writing.wait()
|
||||||
self.assertTrue(os.path.isfile(path))
|
self.assertTrue(os.path.isfile(path))
|
||||||
await self._restart_stream_manager()
|
await self.daemon.file_manager.stop()
|
||||||
|
# while stopped, we get no response to query and no file is present
|
||||||
|
self.assertEqual((await self.daemon.jsonrpc_file_list())['items'], [])
|
||||||
|
self.assertEqual(os.path.isfile(path), stream.status == ManagedDownloadSource.STATUS_FINISHED)
|
||||||
|
await self.daemon.file_manager.start()
|
||||||
|
# after restart, we get a response to query and same file path
|
||||||
stream = (await self.daemon.jsonrpc_file_list())['items'][0]
|
stream = (await self.daemon.jsonrpc_file_list())['items'][0]
|
||||||
self.assertIsNotNone(stream.full_path)
|
self.assertIsNotNone(stream.full_path)
|
||||||
self.assertFalse(os.path.isfile(path))
|
self.assertEqual(stream.full_path, path)
|
||||||
if wait_for_start_writing:
|
if wait_for_start_writing:
|
||||||
|
with contextlib.suppress(asyncio.CancelledError):
|
||||||
await stream.started_writing.wait()
|
await stream.started_writing.wait()
|
||||||
self.assertTrue(os.path.isfile(path))
|
self.assertTrue(os.path.isfile(path))
|
||||||
|
|
||||||
|
@ -414,6 +423,6 @@ class RangeRequestsLRUCache(CommandTestCase):
|
||||||
|
|
||||||
# running with cache size 0 gets through without errors without
|
# running with cache size 0 gets through without errors without
|
||||||
# this since the server doesn't stop immediately
|
# this since the server doesn't stop immediately
|
||||||
await asyncio.sleep(1, loop=self.loop)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
await self._request_stream()
|
await self._request_stream()
|
||||||
|
|
|
@ -1508,27 +1508,27 @@ class ResolveClaimTakeovers(BaseResolveTestCase):
|
||||||
COIN = int(1E8)
|
COIN = int(1E8)
|
||||||
|
|
||||||
self.assertEqual(self.conductor.spv_node.writer.height, 207)
|
self.assertEqual(self.conductor.spv_node.writer.height, 207)
|
||||||
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stage_put(
|
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stash_put(
|
||||||
(208, bytes.fromhex(claim_id1)), (0, 10 * COIN)
|
(208, bytes.fromhex(claim_id1)), (0, 10 * COIN)
|
||||||
)
|
)
|
||||||
await self.generate(1)
|
await self.generate(1)
|
||||||
self.assertEqual(self.conductor.spv_node.writer.height, 208)
|
self.assertEqual(self.conductor.spv_node.writer.height, 208)
|
||||||
|
|
||||||
self.assertEqual(1.7090807854206793, await get_trending_score(claim_id1))
|
self.assertEqual(1.7090807854206793, await get_trending_score(claim_id1))
|
||||||
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stage_put(
|
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stash_put(
|
||||||
(209, bytes.fromhex(claim_id1)), (10 * COIN, 100 * COIN)
|
(209, bytes.fromhex(claim_id1)), (10 * COIN, 100 * COIN)
|
||||||
)
|
)
|
||||||
await self.generate(1)
|
await self.generate(1)
|
||||||
self.assertEqual(self.conductor.spv_node.writer.height, 209)
|
self.assertEqual(self.conductor.spv_node.writer.height, 209)
|
||||||
self.assertEqual(2.2437974397778886, await get_trending_score(claim_id1))
|
self.assertEqual(2.2437974397778886, await get_trending_score(claim_id1))
|
||||||
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stage_put(
|
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stash_put(
|
||||||
(309, bytes.fromhex(claim_id1)), (100 * COIN, 1000000 * COIN)
|
(309, bytes.fromhex(claim_id1)), (100 * COIN, 1000000 * COIN)
|
||||||
)
|
)
|
||||||
await self.generate(100)
|
await self.generate(100)
|
||||||
self.assertEqual(self.conductor.spv_node.writer.height, 309)
|
self.assertEqual(self.conductor.spv_node.writer.height, 309)
|
||||||
self.assertEqual(5.157053472135866, await get_trending_score(claim_id1))
|
self.assertEqual(5.157053472135866, await get_trending_score(claim_id1))
|
||||||
|
|
||||||
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stage_put(
|
self.conductor.spv_node.writer.db.prefix_db.trending_notification.stash_put(
|
||||||
(409, bytes.fromhex(claim_id1)), (1000000 * COIN, 1 * COIN)
|
(409, bytes.fromhex(claim_id1)), (1000000 * COIN, 1 * COIN)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
@ -2,7 +2,7 @@ import asyncio
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from lbry.testcase import CommandTestCase
|
from lbry.testcase import CommandTestCase
|
||||||
|
from lbry.wallet import Transaction
|
||||||
|
|
||||||
class TransactionCommandsTestCase(CommandTestCase):
|
class TransactionCommandsTestCase(CommandTestCase):
|
||||||
|
|
||||||
|
@ -29,17 +29,42 @@ class TransactionCommandsTestCase(CommandTestCase):
|
||||||
# someone's tx
|
# someone's tx
|
||||||
change_address = await self.blockchain.get_raw_change_address()
|
change_address = await self.blockchain.get_raw_change_address()
|
||||||
sendtxid = await self.blockchain.send_to_address(change_address, 10)
|
sendtxid = await self.blockchain.send_to_address(change_address, 10)
|
||||||
await asyncio.sleep(0.2)
|
# After a few tries, Hub should have the transaction (in mempool).
|
||||||
|
for i in range(5):
|
||||||
tx = await self.daemon.jsonrpc_transaction_show(sendtxid)
|
tx = await self.daemon.jsonrpc_transaction_show(sendtxid)
|
||||||
self.assertEqual(tx.id, sendtxid)
|
# Retry if Hub is not aware of the transaction.
|
||||||
self.assertEqual(tx.height, -1)
|
if isinstance(tx, dict):
|
||||||
|
# Fields: 'success', 'code', 'message'
|
||||||
|
self.assertFalse(tx['success'], tx)
|
||||||
|
self.assertEqual(tx['code'], 404, tx)
|
||||||
|
self.assertEqual(tx['message'], "transaction not found", tx)
|
||||||
|
await asyncio.sleep(0.1)
|
||||||
|
continue
|
||||||
|
break
|
||||||
|
# verify transaction show (in mempool)
|
||||||
|
self.assertTrue(isinstance(tx, Transaction), str(tx))
|
||||||
|
# Fields: 'txid', 'raw', 'height', 'position', 'is_verified', and more.
|
||||||
|
self.assertEqual(tx.id, sendtxid, vars(tx))
|
||||||
|
self.assertEqual(tx.height, -1, vars(tx))
|
||||||
|
self.assertEqual(tx.is_verified, False, vars(tx))
|
||||||
|
|
||||||
|
# transaction is confirmed and leaves mempool
|
||||||
await self.generate(1)
|
await self.generate(1)
|
||||||
|
|
||||||
|
# verify transaction show
|
||||||
tx = await self.daemon.jsonrpc_transaction_show(sendtxid)
|
tx = await self.daemon.jsonrpc_transaction_show(sendtxid)
|
||||||
self.assertEqual(tx.height, self.ledger.headers.height)
|
self.assertTrue(isinstance(tx, Transaction), str(tx))
|
||||||
|
self.assertEqual(tx.id, sendtxid, vars(tx))
|
||||||
|
self.assertEqual(tx.height, self.ledger.headers.height, vars(tx))
|
||||||
|
self.assertEqual(tx.is_verified, True, vars(tx))
|
||||||
|
|
||||||
# inexistent
|
# inexistent
|
||||||
result = await self.daemon.jsonrpc_transaction_show('0'*64)
|
result = await self.daemon.jsonrpc_transaction_show('0'*64)
|
||||||
self.assertFalse(result['success'])
|
self.assertTrue(isinstance(result, dict), result)
|
||||||
|
# Fields: 'success', 'code', 'message'
|
||||||
|
self.assertFalse(result['success'], result)
|
||||||
|
self.assertEqual(result['code'], 404, result)
|
||||||
|
self.assertEqual(result['message'], "transaction not found", result)
|
||||||
|
|
||||||
async def test_utxo_release(self):
|
async def test_utxo_release(self):
|
||||||
await self.send_to_address_and_wait(
|
await self.send_to_address_and_wait(
|
||||||
|
|
|
@ -36,7 +36,7 @@ class TestBlob(AsyncioTestCase):
|
||||||
writer.write(self.blob_bytes)
|
writer.write(self.blob_bytes)
|
||||||
await blob.verified.wait()
|
await blob.verified.wait()
|
||||||
self.assertTrue(blob.get_is_verified())
|
self.assertTrue(blob.get_is_verified())
|
||||||
await asyncio.sleep(0, loop=self.loop) # wait for the db save task
|
await asyncio.sleep(0) # wait for the db save task
|
||||||
return blob
|
return blob
|
||||||
|
|
||||||
async def _test_close_writers_on_finished(self, blob_class=AbstractBlob, blob_directory=None):
|
async def _test_close_writers_on_finished(self, blob_class=AbstractBlob, blob_directory=None):
|
||||||
|
@ -48,7 +48,7 @@ class TestBlob(AsyncioTestCase):
|
||||||
with self.assertRaises(InvalidDataError):
|
with self.assertRaises(InvalidDataError):
|
||||||
writers[1].write(self.blob_bytes * 2)
|
writers[1].write(self.blob_bytes * 2)
|
||||||
await writers[1].finished
|
await writers[1].finished
|
||||||
await asyncio.sleep(0, loop=self.loop)
|
await asyncio.sleep(0)
|
||||||
self.assertEqual(4, len(blob.writers))
|
self.assertEqual(4, len(blob.writers))
|
||||||
|
|
||||||
# write the blob
|
# write the blob
|
||||||
|
@ -208,7 +208,7 @@ class TestBlob(AsyncioTestCase):
|
||||||
async def read_blob_buffer():
|
async def read_blob_buffer():
|
||||||
with reader as read_handle:
|
with reader as read_handle:
|
||||||
self.assertEqual(1, len(blob.readers))
|
self.assertEqual(1, len(blob.readers))
|
||||||
await asyncio.sleep(2, loop=self.loop)
|
await asyncio.sleep(2)
|
||||||
self.assertEqual(0, len(blob.readers))
|
self.assertEqual(0, len(blob.readers))
|
||||||
return read_handle.read()
|
return read_handle.read()
|
||||||
|
|
||||||
|
|
|
@ -183,7 +183,7 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
writer.write(mock_blob_bytes)
|
writer.write(mock_blob_bytes)
|
||||||
return self.loop.create_task(_inner())
|
return self.loop.create_task(_inner())
|
||||||
|
|
||||||
await asyncio.gather(write_task(writer1), write_task(writer2), loop=self.loop)
|
await asyncio.gather(write_task(writer1), write_task(writer2))
|
||||||
|
|
||||||
self.assertDictEqual({1: mock_blob_bytes, 2: mock_blob_bytes}, results)
|
self.assertDictEqual({1: mock_blob_bytes, 2: mock_blob_bytes}, results)
|
||||||
self.assertEqual(1, write_called_count)
|
self.assertEqual(1, write_called_count)
|
||||||
|
@ -239,7 +239,8 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
async def test_server_chunked_request(self):
|
async def test_server_chunked_request(self):
|
||||||
blob_hash = "7f5ab2def99f0ddd008da71db3a3772135f4002b19b7605840ed1034c8955431bd7079549e65e6b2a3b9c17c773073ed"
|
blob_hash = "7f5ab2def99f0ddd008da71db3a3772135f4002b19b7605840ed1034c8955431bd7079549e65e6b2a3b9c17c773073ed"
|
||||||
server_protocol = BlobServerProtocol(self.loop, self.server_blob_manager, self.server.lbrycrd_address)
|
server_protocol = BlobServerProtocol(self.loop, self.server_blob_manager, self.server.lbrycrd_address)
|
||||||
transport = asyncio.Transport(extra={'peername': ('ip', 90)})
|
transport = mock.Mock(spec=asyncio.Transport)
|
||||||
|
transport.get_extra_info = lambda k: {'peername': ('ip', 90)}[k]
|
||||||
received_data = BytesIO()
|
received_data = BytesIO()
|
||||||
transport.is_closing = lambda: received_data.closed
|
transport.is_closing = lambda: received_data.closed
|
||||||
transport.write = received_data.write
|
transport.write = received_data.write
|
||||||
|
@ -269,7 +270,7 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
client_blob.delete()
|
client_blob.delete()
|
||||||
|
|
||||||
# wait for less than the idle timeout
|
# wait for less than the idle timeout
|
||||||
await asyncio.sleep(0.5, loop=self.loop)
|
await asyncio.sleep(0.5)
|
||||||
|
|
||||||
# download the blob again
|
# download the blob again
|
||||||
downloaded, protocol2 = await request_blob(self.loop, client_blob, self.server_from_client.address,
|
downloaded, protocol2 = await request_blob(self.loop, client_blob, self.server_from_client.address,
|
||||||
|
@ -283,10 +284,10 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
client_blob.delete()
|
client_blob.delete()
|
||||||
|
|
||||||
# check that the connection times out from the server side
|
# check that the connection times out from the server side
|
||||||
await asyncio.sleep(0.9, loop=self.loop)
|
await asyncio.sleep(0.9)
|
||||||
self.assertFalse(protocol.transport.is_closing())
|
self.assertFalse(protocol.transport.is_closing())
|
||||||
self.assertIsNotNone(protocol.transport._sock)
|
self.assertIsNotNone(protocol.transport._sock)
|
||||||
await asyncio.sleep(0.1, loop=self.loop)
|
await asyncio.sleep(0.1)
|
||||||
self.assertIsNone(protocol.transport)
|
self.assertIsNone(protocol.transport)
|
||||||
|
|
||||||
def test_max_request_size(self):
|
def test_max_request_size(self):
|
||||||
|
@ -322,7 +323,7 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
server_blob = self.server_blob_manager.get_blob(blob_hash)
|
server_blob = self.server_blob_manager.get_blob(blob_hash)
|
||||||
|
|
||||||
async def sendfile(writer):
|
async def sendfile(writer):
|
||||||
await asyncio.sleep(2, loop=self.loop)
|
await asyncio.sleep(2)
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
server_blob.sendfile = sendfile
|
server_blob.sendfile = sendfile
|
||||||
|
@ -346,7 +347,7 @@ class TestBlobExchange(BlobExchangeTestBase):
|
||||||
def _mock_accumulate_peers(q1, q2=None):
|
def _mock_accumulate_peers(q1, q2=None):
|
||||||
async def _task():
|
async def _task():
|
||||||
pass
|
pass
|
||||||
q2 = q2 or asyncio.Queue(loop=self.loop)
|
q2 = q2 or asyncio.Queue()
|
||||||
return q2, self.loop.create_task(_task())
|
return q2, self.loop.create_task(_task())
|
||||||
|
|
||||||
mock_node.accumulate_peers = _mock_accumulate_peers
|
mock_node.accumulate_peers = _mock_accumulate_peers
|
||||||
|
|
|
@ -72,14 +72,14 @@ class CacheConcurrentDecoratorTests(AsyncioTestCase):
|
||||||
@utils.cache_concurrent
|
@utils.cache_concurrent
|
||||||
async def foo(self, arg1, arg2=None, delay=1):
|
async def foo(self, arg1, arg2=None, delay=1):
|
||||||
self.called.append((arg1, arg2, delay))
|
self.called.append((arg1, arg2, delay))
|
||||||
await asyncio.sleep(delay, loop=self.loop)
|
await asyncio.sleep(delay)
|
||||||
self.counter += 1
|
self.counter += 1
|
||||||
self.finished.append((arg1, arg2, delay))
|
self.finished.append((arg1, arg2, delay))
|
||||||
return object()
|
return object()
|
||||||
|
|
||||||
async def test_gather_duplicates(self):
|
async def test_gather_duplicates(self):
|
||||||
result = await asyncio.gather(
|
result = await asyncio.gather(
|
||||||
self.loop.create_task(self.foo(1)), self.loop.create_task(self.foo(1)), loop=self.loop
|
self.loop.create_task(self.foo(1)), self.loop.create_task(self.foo(1))
|
||||||
)
|
)
|
||||||
self.assertEqual(1, len(self.called))
|
self.assertEqual(1, len(self.called))
|
||||||
self.assertEqual(1, len(self.finished))
|
self.assertEqual(1, len(self.finished))
|
||||||
|
@ -93,7 +93,7 @@ class CacheConcurrentDecoratorTests(AsyncioTestCase):
|
||||||
|
|
||||||
with self.assertRaises(asyncio.CancelledError):
|
with self.assertRaises(asyncio.CancelledError):
|
||||||
await asyncio.gather(
|
await asyncio.gather(
|
||||||
t1, self.loop.create_task(self.foo(1)), loop=self.loop
|
t1, self.loop.create_task(self.foo(1))
|
||||||
)
|
)
|
||||||
self.assertEqual(1, len(self.called))
|
self.assertEqual(1, len(self.called))
|
||||||
self.assertEqual(0, len(self.finished))
|
self.assertEqual(0, len(self.finished))
|
||||||
|
|
|
@ -128,7 +128,7 @@ class TestBlobAnnouncer(AsyncioTestCase):
|
||||||
await self.chain_peer(constants.generate_id(current + 4), '1.2.3.13')
|
await self.chain_peer(constants.generate_id(current + 4), '1.2.3.13')
|
||||||
last = await self.chain_peer(constants.generate_id(current + 5), '1.2.3.14')
|
last = await self.chain_peer(constants.generate_id(current + 5), '1.2.3.14')
|
||||||
|
|
||||||
search_q, peer_q = asyncio.Queue(loop=self.loop), asyncio.Queue(loop=self.loop)
|
search_q, peer_q = asyncio.Queue(), asyncio.Queue()
|
||||||
search_q.put_nowait(blob1)
|
search_q.put_nowait(blob1)
|
||||||
|
|
||||||
_, task = last.accumulate_peers(search_q, peer_q)
|
_, task = last.accumulate_peers(search_q, peer_q)
|
||||||
|
|
|
@ -2,7 +2,6 @@ import unittest
|
||||||
from unittest import mock
|
from unittest import mock
|
||||||
import json
|
import json
|
||||||
|
|
||||||
from lbry.conf import Config
|
|
||||||
from lbry.extras.daemon.storage import SQLiteStorage
|
from lbry.extras.daemon.storage import SQLiteStorage
|
||||||
from lbry.extras.daemon.componentmanager import ComponentManager
|
from lbry.extras.daemon.componentmanager import ComponentManager
|
||||||
from lbry.extras.daemon.components import DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT
|
from lbry.extras.daemon.components import DATABASE_COMPONENT, DHT_COMPONENT, WALLET_COMPONENT
|
||||||
|
@ -11,6 +10,7 @@ from lbry.extras.daemon.components import UPNP_COMPONENT, BLOB_COMPONENT
|
||||||
from lbry.extras.daemon.components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT
|
from lbry.extras.daemon.components import PEER_PROTOCOL_SERVER_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT
|
||||||
from lbry.extras.daemon.daemon import Daemon as LBRYDaemon
|
from lbry.extras.daemon.daemon import Daemon as LBRYDaemon
|
||||||
from lbry.wallet import WalletManager, Wallet
|
from lbry.wallet import WalletManager, Wallet
|
||||||
|
from lbry.conf import Config
|
||||||
|
|
||||||
from tests import test_utils
|
from tests import test_utils
|
||||||
# from tests.mocks import mock_conf_settings, FakeNetwork, FakeFileManager
|
# from tests.mocks import mock_conf_settings, FakeNetwork, FakeFileManager
|
||||||
|
|
|
@ -109,9 +109,9 @@ class TestManagedStream(BlobExchangeTestBase):
|
||||||
await self._test_transfer_stream(10, stop_when_done=False)
|
await self._test_transfer_stream(10, stop_when_done=False)
|
||||||
self.assertEqual(self.stream.status, "finished")
|
self.assertEqual(self.stream.status, "finished")
|
||||||
self.assertTrue(self.stream._running.is_set())
|
self.assertTrue(self.stream._running.is_set())
|
||||||
await asyncio.sleep(0.5, loop=self.loop)
|
await asyncio.sleep(0.5)
|
||||||
self.assertTrue(self.stream._running.is_set())
|
self.assertTrue(self.stream._running.is_set())
|
||||||
await asyncio.sleep(2, loop=self.loop)
|
await asyncio.sleep(2)
|
||||||
self.assertEqual(self.stream.status, "finished")
|
self.assertEqual(self.stream.status, "finished")
|
||||||
self.assertFalse(self.stream._running.is_set())
|
self.assertFalse(self.stream._running.is_set())
|
||||||
|
|
||||||
|
|
|
@ -86,13 +86,13 @@ class TestReflector(AsyncioTestCase):
|
||||||
self.assertListEqual(sent, [])
|
self.assertListEqual(sent, [])
|
||||||
|
|
||||||
async def test_reflect_stream(self):
|
async def test_reflect_stream(self):
|
||||||
return await asyncio.wait_for(self._test_reflect_stream(response_chunk_size=50), 3, loop=self.loop)
|
return await asyncio.wait_for(self._test_reflect_stream(response_chunk_size=50), 3)
|
||||||
|
|
||||||
async def test_reflect_stream_but_reflector_changes_its_mind(self):
|
async def test_reflect_stream_but_reflector_changes_its_mind(self):
|
||||||
return await asyncio.wait_for(self._test_reflect_stream(partial_needs=True), 3, loop=self.loop)
|
return await asyncio.wait_for(self._test_reflect_stream(partial_needs=True), 3)
|
||||||
|
|
||||||
async def test_reflect_stream_small_response_chunks(self):
|
async def test_reflect_stream_small_response_chunks(self):
|
||||||
return await asyncio.wait_for(self._test_reflect_stream(response_chunk_size=30), 3, loop=self.loop)
|
return await asyncio.wait_for(self._test_reflect_stream(response_chunk_size=30), 3)
|
||||||
|
|
||||||
async def test_announces(self):
|
async def test_announces(self):
|
||||||
to_announce = await self.storage.get_blobs_to_announce()
|
to_announce = await self.storage.get_blobs_to_announce()
|
||||||
|
|
|
@ -174,7 +174,7 @@ class TestStreamManager(BlobExchangeTestBase):
|
||||||
await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
||||||
else:
|
else:
|
||||||
await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
||||||
await asyncio.sleep(0, loop=self.loop)
|
await asyncio.sleep(0)
|
||||||
self.assertTrue(checked_analytics_event)
|
self.assertTrue(checked_analytics_event)
|
||||||
|
|
||||||
async def test_time_to_first_bytes(self):
|
async def test_time_to_first_bytes(self):
|
||||||
|
@ -317,7 +317,7 @@ class TestStreamManager(BlobExchangeTestBase):
|
||||||
stream.downloader.node = self.stream_manager.node
|
stream.downloader.node = self.stream_manager.node
|
||||||
await stream.save_file()
|
await stream.save_file()
|
||||||
await stream.finished_writing.wait()
|
await stream.finished_writing.wait()
|
||||||
await asyncio.sleep(0, loop=self.loop)
|
await asyncio.sleep(0)
|
||||||
self.assertTrue(stream.finished)
|
self.assertTrue(stream.finished)
|
||||||
self.assertFalse(stream.running)
|
self.assertFalse(stream.running)
|
||||||
self.assertTrue(os.path.isfile(os.path.join(self.client_dir, "test_file")))
|
self.assertTrue(os.path.isfile(os.path.join(self.client_dir, "test_file")))
|
||||||
|
@ -355,7 +355,7 @@ class TestStreamManager(BlobExchangeTestBase):
|
||||||
|
|
||||||
self.stream_manager.analytics_manager._post = check_post
|
self.stream_manager.analytics_manager._post = check_post
|
||||||
await self._test_download_error_on_start(expected_error, timeout)
|
await self._test_download_error_on_start(expected_error, timeout)
|
||||||
await asyncio.sleep(0, loop=self.loop)
|
await asyncio.sleep(0)
|
||||||
self.assertListEqual([expected_error.__name__], received)
|
self.assertListEqual([expected_error.__name__], received)
|
||||||
|
|
||||||
async def test_insufficient_funds(self):
|
async def test_insufficient_funds(self):
|
||||||
|
@ -424,7 +424,7 @@ class TestStreamManager(BlobExchangeTestBase):
|
||||||
self.assertIsNone(stream.full_path)
|
self.assertIsNone(stream.full_path)
|
||||||
self.assertEqual(0, stream.written_bytes)
|
self.assertEqual(0, stream.written_bytes)
|
||||||
|
|
||||||
self.stream_manager.stop()
|
await self.stream_manager.stop()
|
||||||
await self.stream_manager.start()
|
await self.stream_manager.start()
|
||||||
self.assertEqual(1, len(self.stream_manager.streams))
|
self.assertEqual(1, len(self.stream_manager.streams))
|
||||||
stream = list(self.stream_manager.streams.values())[0]
|
stream = list(self.stream_manager.streams.values())[0]
|
||||||
|
@ -448,8 +448,8 @@ class TestStreamManager(BlobExchangeTestBase):
|
||||||
self.assertDictEqual(self.stream_manager.streams, {})
|
self.assertDictEqual(self.stream_manager.streams, {})
|
||||||
stream = await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
stream = await self.file_manager.download_from_uri(self.uri, self.exchange_rate_manager)
|
||||||
await stream.finished_writing.wait()
|
await stream.finished_writing.wait()
|
||||||
await asyncio.sleep(0, loop=self.loop)
|
await asyncio.sleep(0)
|
||||||
self.stream_manager.stop()
|
await self.stream_manager.stop()
|
||||||
self.client_blob_manager.stop()
|
self.client_blob_manager.stop()
|
||||||
# partial removal, only sd blob is missing.
|
# partial removal, only sd blob is missing.
|
||||||
# in this case, we recover the sd blob while the other blobs are kept untouched as 'finished'
|
# in this case, we recover the sd blob while the other blobs are kept untouched as 'finished'
|
||||||
|
|
|
@ -470,7 +470,7 @@ class TestUpgrade(AsyncioTestCase):
|
||||||
|
|
||||||
|
|
||||||
class TestSQLiteRace(AsyncioTestCase):
|
class TestSQLiteRace(AsyncioTestCase):
|
||||||
max_misuse_attempts = 80000
|
max_misuse_attempts = 120000
|
||||||
|
|
||||||
def setup_db(self):
|
def setup_db(self):
|
||||||
self.db = sqlite3.connect(":memory:", isolation_level=None)
|
self.db = sqlite3.connect(":memory:", isolation_level=None)
|
||||||
|
|
Loading…
Reference in a new issue