148 lines
6.5 KiB
Python
148 lines
6.5 KiB
Python
import asyncio
|
|
import typing
|
|
import logging
|
|
import binascii
|
|
|
|
from lbry.dht.peer import make_kademlia_peer
|
|
from lbry.error import DownloadSDTimeoutError
|
|
from lbry.utils import resolve_host, lru_cache_concurrent
|
|
from lbry.stream.descriptor import StreamDescriptor
|
|
from lbry.blob_exchange.downloader import BlobDownloader
|
|
if typing.TYPE_CHECKING:
|
|
from lbry.conf import Config
|
|
from lbry.dht.node import Node
|
|
from lbry.blob.blob_manager import BlobManager
|
|
from lbry.blob.blob_file import AbstractBlob
|
|
from lbry.blob.blob_info import BlobInfo
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class StreamDownloader:
|
|
def __init__(self, loop: asyncio.AbstractEventLoop, config: 'Config', blob_manager: 'BlobManager', sd_hash: str,
|
|
descriptor: typing.Optional[StreamDescriptor] = None):
|
|
self.loop = loop
|
|
self.config = config
|
|
self.blob_manager = blob_manager
|
|
self.sd_hash = sd_hash
|
|
self.search_queue = asyncio.Queue(loop=loop) # blob hashes to feed into the iterative finder
|
|
self.peer_queue = asyncio.Queue(loop=loop) # new peers to try
|
|
self.blob_downloader = BlobDownloader(self.loop, self.config, self.blob_manager, self.peer_queue)
|
|
self.descriptor: typing.Optional[StreamDescriptor] = descriptor
|
|
self.node: typing.Optional['Node'] = None
|
|
self.accumulate_task: typing.Optional[asyncio.Task] = None
|
|
self.fixed_peers_handle: typing.Optional[asyncio.Handle] = None
|
|
self.fixed_peers_delay: typing.Optional[float] = None
|
|
self.added_fixed_peers = False
|
|
self.time_to_descriptor: typing.Optional[float] = None
|
|
self.time_to_first_bytes: typing.Optional[float] = None
|
|
|
|
async def cached_read_blob(blob_info: 'BlobInfo') -> bytes:
|
|
return await self.read_blob(blob_info, 2)
|
|
|
|
if self.blob_manager.decrypted_blob_lru_cache:
|
|
cached_read_blob = lru_cache_concurrent(override_lru_cache=self.blob_manager.decrypted_blob_lru_cache)(
|
|
cached_read_blob
|
|
)
|
|
|
|
self.cached_read_blob = cached_read_blob
|
|
|
|
async def add_fixed_peers(self):
|
|
def _delayed_add_fixed_peers():
|
|
self.added_fixed_peers = True
|
|
self.peer_queue.put_nowait([
|
|
make_kademlia_peer(None, address, None, tcp_port=port + 1, allow_localhost=True)
|
|
for address, port in addresses
|
|
])
|
|
|
|
if not self.config.reflector_servers:
|
|
return
|
|
addresses = [
|
|
(await resolve_host(url, port + 1, proto='tcp'), port)
|
|
for url, port in self.config.reflector_servers
|
|
]
|
|
if 'dht' in self.config.components_to_skip or not self.node or not \
|
|
len(self.node.protocol.routing_table.get_peers()) > 0:
|
|
self.fixed_peers_delay = 0.0
|
|
else:
|
|
self.fixed_peers_delay = self.config.fixed_peer_delay
|
|
|
|
self.fixed_peers_handle = self.loop.call_later(self.fixed_peers_delay, _delayed_add_fixed_peers)
|
|
|
|
async def load_descriptor(self, connection_id: int = 0):
|
|
# download or get the sd blob
|
|
sd_blob = self.blob_manager.get_blob(self.sd_hash)
|
|
if not sd_blob.get_is_verified():
|
|
try:
|
|
now = self.loop.time()
|
|
sd_blob = await asyncio.wait_for(
|
|
self.blob_downloader.download_blob(self.sd_hash, connection_id),
|
|
self.config.blob_download_timeout, loop=self.loop
|
|
)
|
|
log.info("downloaded sd blob %s", self.sd_hash)
|
|
self.time_to_descriptor = self.loop.time() - now
|
|
except asyncio.TimeoutError:
|
|
raise DownloadSDTimeoutError(self.sd_hash)
|
|
|
|
# parse the descriptor
|
|
self.descriptor = await StreamDescriptor.from_stream_descriptor_blob(
|
|
self.loop, self.blob_manager.blob_dir, sd_blob
|
|
)
|
|
log.info("loaded stream manifest %s", self.sd_hash)
|
|
|
|
async def start(self, node: typing.Optional['Node'] = None, connection_id: int = 0):
|
|
# set up peer accumulation
|
|
if node:
|
|
self.node = node
|
|
if self.accumulate_task and not self.accumulate_task.done():
|
|
self.accumulate_task.cancel()
|
|
_, self.accumulate_task = self.node.accumulate_peers(self.search_queue, self.peer_queue)
|
|
await self.add_fixed_peers()
|
|
# start searching for peers for the sd hash
|
|
self.search_queue.put_nowait(self.sd_hash)
|
|
log.info("searching for peers for stream %s", self.sd_hash)
|
|
|
|
if not self.descriptor:
|
|
await self.load_descriptor(connection_id)
|
|
|
|
# add the head blob to the peer search
|
|
self.search_queue.put_nowait(self.descriptor.blobs[0].blob_hash)
|
|
log.info("added head blob to peer search for stream %s", self.sd_hash)
|
|
|
|
if not await self.blob_manager.storage.stream_exists(self.sd_hash):
|
|
await self.blob_manager.storage.store_stream(
|
|
self.blob_manager.get_blob(self.sd_hash, length=self.descriptor.length), self.descriptor
|
|
)
|
|
|
|
async def download_stream_blob(self, blob_info: 'BlobInfo', connection_id: int = 0) -> 'AbstractBlob':
|
|
if not filter(lambda b: b.blob_hash == blob_info.blob_hash, self.descriptor.blobs[:-1]):
|
|
raise ValueError(f"blob {blob_info.blob_hash} is not part of stream with sd hash {self.sd_hash}")
|
|
blob = await asyncio.wait_for(
|
|
self.blob_downloader.download_blob(blob_info.blob_hash, blob_info.length, connection_id),
|
|
self.config.blob_download_timeout * 10, loop=self.loop
|
|
)
|
|
return blob
|
|
|
|
def decrypt_blob(self, blob_info: 'BlobInfo', blob: 'AbstractBlob') -> bytes:
|
|
return blob.decrypt(
|
|
binascii.unhexlify(self.descriptor.key.encode()), binascii.unhexlify(blob_info.iv.encode())
|
|
)
|
|
|
|
async def read_blob(self, blob_info: 'BlobInfo', connection_id: int = 0) -> bytes:
|
|
start = None
|
|
if self.time_to_first_bytes is None:
|
|
start = self.loop.time()
|
|
blob = await self.download_stream_blob(blob_info, connection_id)
|
|
decrypted = self.decrypt_blob(blob_info, blob)
|
|
if start:
|
|
self.time_to_first_bytes = self.loop.time() - start
|
|
return decrypted
|
|
|
|
def stop(self):
|
|
if self.accumulate_task:
|
|
self.accumulate_task.cancel()
|
|
self.accumulate_task = None
|
|
if self.fixed_peers_handle:
|
|
self.fixed_peers_handle.cancel()
|
|
self.fixed_peers_handle = None
|
|
self.blob_downloader.close()
|