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

stop trapping CancelledError

Browse source
This commit is contained in:
Victor Shyba 2019-02-21 23:07:45 -03:00 committed by Jack Robison
parent 13c7026b20
commit 7a96e742f2
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
3 changed files with 9 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lbrynet/stream/managed_stream.py
View file

@ -206,7 +206,7 @@ class ManagedStream:
for blob_hash in we_have: for blob_hash in we_have:
await protocol.send_blob(blob_hash) await protocol.send_blob(blob_hash)
sent.append(blob_hash) sent.append(blob_hash)
except (asyncio.CancelledError, asyncio.TimeoutError, ValueError): except (asyncio.TimeoutError, ValueError):
return sent return sent
except ConnectionRefusedError: except ConnectionRefusedError:
return sent return sent

13
lbrynet/stream/reflector/server.py
View file

@ -71,17 +71,14 @@ class ReflectorServerProtocol(asyncio.Protocol):
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
) )
self.incoming.clear()
self.writer.close_handle()
self.writer = None
self.send_response({"received_sd_blob": True}) self.send_response({"received_sd_blob": True})
except (asyncio.TimeoutError, asyncio.CancelledError): except asyncio.TimeoutError:
self.send_response({"received_sd_blob": False})
self.transport.close()
finally:
self.incoming.clear() self.incoming.clear()
self.writer.close_handle() self.writer.close_handle()
self.writer = None self.writer = None
self.transport.close()
self.send_response({"received_sd_blob": False})
return
else: else:
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
@ -111,7 +108,7 @@ class ReflectorServerProtocol(asyncio.Protocol):
try: try:
await asyncio.wait_for(blob.finished_writing.wait(), 30, loop=self.loop) await asyncio.wait_for(blob.finished_writing.wait(), 30, loop=self.loop)
self.send_response({"received_blob": True}) self.send_response({"received_blob": True})
except (asyncio.TimeoutError, asyncio.CancelledError): except (asyncio.TimeoutError):
self.send_response({"received_blob": False}) self.send_response({"received_blob": False})
self.incoming.clear() self.incoming.clear()
self.writer.close_handle() self.writer.close_handle()

10
lbrynet/stream/stream_manager.py
View file

@ -275,11 +275,8 @@ class StreamManager:
def wait_for_stream_finished(self, stream: ManagedStream): def wait_for_stream_finished(self, stream: ManagedStream):
async def _wait_for_stream_finished(): async def _wait_for_stream_finished():
if stream.downloader and stream.running: if stream.downloader and stream.running:
try: await stream.downloader.stream_finished_event.wait()
await stream.downloader.stream_finished_event.wait() stream.update_status(ManagedStream.STATUS_FINISHED)
stream.update_status(ManagedStream.STATUS_FINISHED)
except asyncio.CancelledError:
pass
task = self.loop.create_task(_wait_for_stream_finished()) task = self.loop.create_task(_wait_for_stream_finished())
self.update_stream_finished_futs.append(task) self.update_stream_finished_futs.append(task)
task.add_done_callback( task.add_done_callback(
@ -358,10 +355,9 @@ class StreamManager:
stream.tx = await self.wallet.send_amount_to_address( stream.tx = await self.wallet.send_amount_to_address(
lbc_to_dewies(str(fee_amount)), fee_address.encode('latin1')) lbc_to_dewies(str(fee_amount)), fee_address.encode('latin1'))
return stream return stream
except (asyncio.TimeoutError, asyncio.CancelledError) as e: except asyncio.TimeoutError as e:
if stream_task.exception(): if stream_task.exception():
raise stream_task.exception() raise stream_task.exception()
return
finally: finally:
if sd_hash in self.starting_streams: if sd_hash in self.starting_streams:
del self.starting_streams[sd_hash] del self.starting_streams[sd_hash]