From 47190d7d5521295e3c0b4e002b1bc3f4e1b2a94d Mon Sep 17 00:00:00 2001 From: Lonami Exo Date: Tue, 21 Aug 2018 11:31:14 +0200 Subject: [PATCH] Fix event loop not being passed into many asyncio calls --- telethon/client/telegrambaseclient.py | 6 +++--- telethon/events/inlinequery.py | 10 ++++++---- telethon/network/mtprotosender.py | 24 ++++++++++++++---------- telethon/network/mtprotostate.py | 7 ++++--- telethon/tl/core/messagecontainer.py | 2 +- telethon/tl/core/tlmessage.py | 7 +++++-- telethon/tl/custom/conversation.py | 8 ++++---- 7 files changed, 37 insertions(+), 27 deletions(-) diff --git a/telethon/client/telegrambaseclient.py b/telethon/client/telegrambaseclient.py index 65c20177..b3d74fa6 100644 --- a/telethon/client/telegrambaseclient.py +++ b/telethon/client/telegrambaseclient.py @@ -244,7 +244,7 @@ class TelegramBaseClient(abc.ABC): # being ``n`` the amount of borrows a given sender has; once ``n`` # reaches ``0`` it should be disconnected and removed. self._borrowed_senders = {} - self._borrow_sender_lock = asyncio.Lock() + self._borrow_sender_lock = asyncio.Lock(loop=self._loop) # Save whether the user is authorized here (a.k.a. logged in) self._authorized = None # None = We don't know yet @@ -258,8 +258,8 @@ class TelegramBaseClient(abc.ABC): self._channel_pts = {} if sequential_updates: - self._updates_queue = asyncio.Queue() - self._dispatching_updates_queue = asyncio.Event() + self._updates_queue = asyncio.Queue(loop=self._loop) + self._dispatching_updates_queue = asyncio.Event(loop=self._loop) else: self._updates_queue = None self._dispatching_updates_queue = None diff --git a/telethon/events/inlinequery.py b/telethon/events/inlinequery.py index 3c163137..3eb503b1 100644 --- a/telethon/events/inlinequery.py +++ b/telethon/events/inlinequery.py @@ -162,8 +162,10 @@ class InlineQuery(EventBuilder): if self._answered: return - results = [self._as_awaitable(x) for x in results] - done, _ = await asyncio.wait(results) + results = [self._as_awaitable(x, self._client.loop) + for x in results] + + done, _ = await asyncio.wait(results, loop=self._client.loop) results = [x.result() for x in done] if switch_pm: @@ -181,10 +183,10 @@ class InlineQuery(EventBuilder): ) @staticmethod - def _as_awaitable(obj): + def _as_awaitable(obj, loop): if inspect.isawaitable(obj): return obj - f = asyncio.Future() + f = asyncio.Future(loop=loop) f.set_result(obj) return f diff --git a/telethon/network/mtprotosender.py b/telethon/network/mtprotosender.py index 16e00b18..44bd4c75 100644 --- a/telethon/network/mtprotosender.py +++ b/telethon/network/mtprotosender.py @@ -205,14 +205,16 @@ class MTProtoSender: result = [] after = None for r in request: - message = self.state.create_message(r, after=after) + message = self.state.create_message( + r, loop=self._loop, after=after) + self._pending_messages[message.msg_id] = message self._send_queue.put_nowait(message) result.append(message.future) after = ordered and message return result else: - message = self.state.create_message(request) + message = self.state.create_message(request, loop=self._loop) self._pending_messages[message.msg_id] = message self._send_queue.put_nowait(message) return message.future @@ -280,7 +282,7 @@ class MTProtoSender: # First connection or manual reconnection after a failure if self._disconnected is None or self._disconnected.done(): - self._disconnected = asyncio.Future() + self._disconnected = asyncio.Future(loop=self._loop) __log__.info('Connection to {} complete!'.format(self._ip)) async def _reconnect(self): @@ -352,7 +354,7 @@ class MTProtoSender: while self._user_connected and not self._reconnecting: if self._pending_ack: self._last_ack = self.state.create_message( - MsgsAck(list(self._pending_ack)) + MsgsAck(list(self._pending_ack)), loop=self._loop ) self._send_queue.put_nowait(self._last_ack) self._pending_ack.clear() @@ -365,7 +367,9 @@ class MTProtoSender: continue if isinstance(messages, list): - message = self.state.create_message(MessageContainer(messages)) + message = self.state.create_message( + MessageContainer(messages), loop=self._loop) + self._pending_messages[message.msg_id] = message self._pending_containers.append(message) else: @@ -394,7 +398,7 @@ class MTProtoSender: __log__.warning('OSError while sending %s', e) else: __log__.exception('Unhandled exception while receiving') - await asyncio.sleep(1) + await asyncio.sleep(1, loop=self._loop) self._start_reconnect() break @@ -433,7 +437,7 @@ class MTProtoSender: __log__.warning('OSError while receiving %s', e) else: __log__.exception('Unhandled exception while receiving') - await asyncio.sleep(1) + await asyncio.sleep(1, loop=self._loop) self._start_reconnect() break @@ -471,7 +475,7 @@ class MTProtoSender: return except Exception as e: __log__.exception('Unhandled exception while unpacking %s',e) - await asyncio.sleep(1) + await asyncio.sleep(1, loop=self._loop) else: try: await self._process_message(message) @@ -480,7 +484,7 @@ class MTProtoSender: except Exception as e: __log__.exception('Unhandled exception while ' 'processing %s', message) - await asyncio.sleep(1) + await asyncio.sleep(1, loop=self._loop) # Response Handlers @@ -525,7 +529,7 @@ class MTProtoSender: if rpc_result.error: error = rpc_message_to_error(rpc_result.error) self._send_queue.put_nowait(self.state.create_message( - MsgsAck([message.msg_id]) + MsgsAck([message.msg_id]), loop=self._loop )) if not message.future.cancelled(): diff --git a/telethon/network/mtprotostate.py b/telethon/network/mtprotostate.py index e7bde1de..579d26cd 100644 --- a/telethon/network/mtprotostate.py +++ b/telethon/network/mtprotostate.py @@ -37,7 +37,7 @@ class MTProtoState: self._sequence = 0 self._last_msg_id = 0 - def create_message(self, obj, after=None): + def create_message(self, obj, *, loop, after=None): """ Creates a new `telethon.tl.tl_message.TLMessage` from the given `telethon.tl.tlobject.TLObject` instance. @@ -47,7 +47,8 @@ class MTProtoState: seq_no=self._get_seq_no(isinstance(obj, TLRequest)), obj=obj, after_id=after.msg_id if after else None, - out=True # Pre-convert the request into bytes + out=True, # Pre-convert the request into bytes + loop=loop ) def update_message_id(self, message): @@ -135,7 +136,7 @@ class MTProtoState: # reader isn't used for anything else after this, it's unnecessary. obj = reader.tgread_object() - return TLMessage(remote_msg_id, remote_sequence, obj) + return TLMessage(remote_msg_id, remote_sequence, obj, loop=None) def _get_new_msg_id(self): """ diff --git a/telethon/tl/core/messagecontainer.py b/telethon/tl/core/messagecontainer.py index fc36fd5e..de304424 100644 --- a/telethon/tl/core/messagecontainer.py +++ b/telethon/tl/core/messagecontainer.py @@ -43,5 +43,5 @@ class MessageContainer(TLObject): before = reader.tell_position() obj = reader.tgread_object() # May over-read e.g. RpcResult reader.set_position(before + length) - messages.append(TLMessage(msg_id, seq_no, obj)) + messages.append(TLMessage(msg_id, seq_no, obj, loop=None)) return MessageContainer(messages) diff --git a/telethon/tl/core/tlmessage.py b/telethon/tl/core/tlmessage.py index e4284eaf..406632db 100644 --- a/telethon/tl/core/tlmessage.py +++ b/telethon/tl/core/tlmessage.py @@ -24,10 +24,13 @@ class TLMessage(TLObject): sent `TLMessage`, and this result can be represented as a `Future` that will eventually be set with either a result, error or cancelled. """ - def __init__(self, msg_id, seq_no, obj, out=False, after_id=0): + def __init__(self, msg_id, seq_no, obj, *, loop, out=False, after_id=0): self.obj = obj self.container_msg_id = None - self.future = asyncio.Future() + + # If no loop is given then it is an incoming message. + # Only outgoing messages need the future to await them. + self.future = asyncio.Future(loop=loop) if loop else None # After which message ID this one should run. We do this so # InvokeAfterMsgRequest is transparent to the user and we can diff --git a/telethon/tl/custom/conversation.py b/telethon/tl/custom/conversation.py index 577220ca..bc32633a 100644 --- a/telethon/tl/custom/conversation.py +++ b/telethon/tl/custom/conversation.py @@ -181,7 +181,7 @@ class Conversation(ChatGetter): return incoming # Otherwise the next incoming response will be the one to use - future = asyncio.Future() + future = asyncio.Future(loop=self._client.loop) pending[target_id] = future return self._get_result(future, start_time, timeout) @@ -209,7 +209,7 @@ class Conversation(ChatGetter): return earliest_edit # Otherwise the next incoming response will be the one to use - future = asyncio.Future() + future = asyncio.Future(loop=self._client.loop) self._pending_edits[target_id] = future return await self._get_result(future, start_time, timeout) @@ -220,7 +220,7 @@ class Conversation(ChatGetter): will also trigger even without a response. """ start_time = time.time() - future = asyncio.Future() + future = asyncio.Future(loop=self._client.loop) target_id = self._get_message_id(message) if self._last_read is None: @@ -265,7 +265,7 @@ class Conversation(ChatGetter): counter = Conversation._custom_counter Conversation._custom_counter += 1 - future = asyncio.Future() + future = asyncio.Future(loop=self._client.loop) async def result(): try: return await self._get_result(future, start_time, timeout)