tor-commits July 2020

tor-commits@lists.torproject.org

17 participants
2100 discussions

[stem/master] Stop using a thread for processing asynchronous events
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit ac9311c275b1b81904f1d0f72efff919ee590330 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun Apr 12 19:20:48 2020 +0300 Stop using a thread for processing asynchronous events --- stem/control.py | 34 ++++++++++------------------------ test/unit/control/controller.py | 2 +- 2 files changed, 11 insertions(+), 25 deletions(-) diff --git a/stem/control.py b/stem/control.py index 2d33bd54..a3b7d253 100644 --- a/stem/control.py +++ b/stem/control.py @@ -560,12 +560,7 @@ class BaseController(object): self._reply_queue = asyncio.Queue() # type: asyncio.Queue[Union[stem.response.ControlMessage, stem.ControllerError]] self._event_queue = asyncio.Queue() # type: asyncio.Queue[stem.response.ControlMessage] - # thread to continually pull from the control socket - self._reader_thread = None # type: Optional[threading.Thread] - - # thread to pull from the _event_queue and call handle_event - self._event_notice = threading.Event() - self._event_thread = None # type: Optional[threading.Thread] + self._event_notice = asyncio.Event() # saves our socket's prior _connect() and _close() methods so they can be # called along with ours @@ -582,7 +577,7 @@ class BaseController(object): self._state_change_threads = [] # type: List[threading.Thread] # threads we've spawned to notify of state changes if self._socket.is_alive(): - self._launch_threads() + self._create_loop_tasks() if is_authenticated: self._post_authentication() @@ -837,7 +832,7 @@ class BaseController(object): pass async def _connect(self) -> None: - self._launch_threads() + self._create_loop_tasks() self._notify_status_listeners(State.INIT) await self._socket_connect() self._is_authenticated = False @@ -852,10 +847,6 @@ class BaseController(object): # joins on our threads if it's safe to do so - for t in (self._reader_thread, self._event_thread): - if t and t.is_alive() and threading.current_thread() != t: - t.join() - self._notify_status_listeners(State.CLOSED) await self._socket_close() @@ -909,22 +900,15 @@ class BaseController(object): else: listener(self, state, change_timestamp) - def _launch_threads(self) -> None: + def _create_loop_tasks(self) -> None: """ Initializes daemon threads. Threads can't be reused so we need to recreate them if we're restarted. """ self._asyncio_loop.create_task(self._reader_loop()) + self._asyncio_loop.create_task(self._event_loop()) - # In theory concurrent calls could result in multiple start() calls on a - # single thread, which would cause an unexpected exception. Best be safe. - - with self._socket._get_send_lock(): - if not self._event_thread or not self._event_thread.is_alive(): - self._event_thread = threading.Thread(target = self._event_loop, name = 'Event notifier') - self._event_thread.setDaemon(True) - self._event_thread.start() async def _reader_loop(self) -> None: """ @@ -955,7 +939,7 @@ class BaseController(object): await self._reply_queue.put(exc) - def _event_loop(self) -> None: + async def _event_loop(self) -> None: """ Continually pulls messages from the _event_queue and sends them to our handle_event callback. This is done via its own thread so subclasses with a @@ -982,8 +966,10 @@ class BaseController(object): if not self.is_alive(): break - self._event_notice.wait(0.05) - self._event_notice.clear() + try: + await asyncio.wait_for(self._event_notice.wait(), timeout=0.05) + except asyncio.TimeoutError: + self._event_notice.clear() class Controller(BaseController): diff --git a/test/unit/control/controller.py b/test/unit/control/controller.py index 02ed2774..c0a07e2a 100644 --- a/test/unit/control/controller.py +++ b/test/unit/control/controller.py @@ -707,7 +707,7 @@ class TestControl(unittest.TestCase): with patch('time.time', Mock(return_value = TEST_TIMESTAMP)): with patch('stem.control.Controller.is_alive') as is_alive_mock: is_alive_mock.return_value = True - self.controller._launch_threads() + self.controller._create_loop_tasks() try: # Converting an event back into an uncast ControlMessage, then feeding it

1 0

[stem/master] Fix closing conections
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 330c79ac7025dcaf15470e993b8d5c3607f07f78 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 14 22:00:20 2020 +0300 Fix closing conections --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index d592c4af..0cc187b4 100644 --- a/stem/control.py +++ b/stem/control.py @@ -659,8 +659,8 @@ class BaseController(object): # provide an assurance to the caller that when we raise a SocketClosed # exception we are shut down afterward for realz. - self.close() - raise + await self.close() + raise def is_alive(self) -> bool: """

1 0

[stem/master] Use awaiting instead of `create_task` while sending a message
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit e3d54409cfec1cf6e6cd88f0f64a2b15be84c69f Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 14 23:31:56 2020 +0300 Use awaiting instead of `create_task` while sending a message --- stem/control.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 0cc187b4..0559ee39 100644 --- a/stem/control.py +++ b/stem/control.py @@ -642,8 +642,7 @@ class BaseController(object): break try: - self._asyncio_loop.create_task(self._socket.send(message)) - + await self._socket.send(message) response = await self._reply_queue.get() # If the message we received back had an exception then re-raise it to the

1 0

[stem/master] Make the `with_default` decorator compatible with asynchronous functions
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit d865c718242146c9ef14901482ebe940be100bb2 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Wed Apr 15 21:49:02 2020 +0300 Make the `with_default` decorator compatible with asynchronous functions --- stem/control.py | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 0bf1b35c..d5e0d0ba 100644 --- a/stem/control.py +++ b/stem/control.py @@ -458,6 +458,12 @@ def with_default(yields: bool = False) -> Callable: """ def decorator(func: Callable) -> Callable: + is_coroutine_func = asyncio.iscoroutinefunction(func) + def coroutine_if_needed(func: Callable) -> Callable: + if is_coroutine_func: + return asyncio.coroutine(func) + return func + def get_default(func: Callable, args: Any, kwargs: Any) -> Any: arg_names = inspect.getfullargspec(func).args[1:] # drop 'self' default_position = arg_names.index('default') if 'default' in arg_names else None @@ -469,9 +475,13 @@ def with_default(yields: bool = False) -> Callable: if not yields: @functools.wraps(func) + @coroutine_if_needed def wrapped(self, *args: Any, **kwargs: Any) -> Any: try: - return func(self, *args, **kwargs) + result = func(self, *args, **kwargs) + if is_coroutine_func: + result = yield from result + return result except: default = get_default(func, args, kwargs) @@ -481,9 +491,13 @@ def with_default(yields: bool = False) -> Callable: return default else: @functools.wraps(func) + @coroutine_if_needed def wrapped(self, *args: Any, **kwargs: Any) -> Any: try: - for val in func(self, *args, **kwargs): + result = func(self, *args, **kwargs) + if is_coroutine_func: + result = yield from result + for val in result: yield val except: default = get_default(func, args, kwargs)

1 0

[stem/master] Start canceling pending tasks while closing a controller
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 1072a8eef7b6d6327e0b329488d868439fc8d6df Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 14 23:59:32 2020 +0300 Start canceling pending tasks while closing a controller --- stem/control.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 435fb741..b7ebe740 100644 --- a/stem/control.py +++ b/stem/control.py @@ -546,6 +546,7 @@ class BaseController(object): self._socket = control_socket self._asyncio_loop = asyncio.get_event_loop() + self._asyncio_loop_tasks = [] self._msg_lock = threading.RLock() @@ -724,6 +725,9 @@ class BaseController(object): await self._socket.close() + for task in self._asyncio_loop_tasks: + task.cancel() + # Join on any outstanding state change listeners. Closing is a state change # of its own, so if we have any listeners it's quite likely there's some # work in progress. @@ -898,8 +902,8 @@ class BaseController(object): them if we're restarted. """ - self._asyncio_loop.create_task(self._reader_loop()) - self._asyncio_loop.create_task(self._event_loop()) + for coroutine in (self._reader_loop(), self._event_loop()): + self._asyncio_loop_tasks.append(self._asyncio_loop.create_task(coroutine)) async def _reader_loop(self) -> None:

1 0

[stem/master] Make reconnecting asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 1447f538903a3f11ba5dfd8fffc65e6136b7a715 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Wed Apr 15 21:02:01 2020 +0300 Make reconnecting asynchronous --- stem/control.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/stem/control.py b/stem/control.py index b7ebe740..0bf1b35c 100644 --- a/stem/control.py +++ b/stem/control.py @@ -1084,7 +1084,7 @@ class Controller(BaseController): import stem.connection await stem.connection.authenticate(self, *args, **kwargs) - def reconnect(self, *args: Any, **kwargs: Any) -> None: + async def reconnect(self, *args: Any, **kwargs: Any) -> None: """ Reconnects and authenticates to our control socket. @@ -1096,9 +1096,9 @@ class Controller(BaseController): """ with self._msg_lock: - self.connect() + await self.connect() self.clear_cache() - self.authenticate(*args, **kwargs) + await self.authenticate(*args, **kwargs) @with_default() def get_info(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, get_bytes: bool = False) -> Union[str, Dict[str, str]]:

1 0

[stem/master] Make the authentication process asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 7d9a8d057a83eea3b12bfe1a25c45d31b061c72f Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 14 23:42:59 2020 +0300 Make the authentication process asynchronous --- stem/connection.py | 62 +++++++++++++++++++++++++++--------------------------- stem/control.py | 54 +++++++++++++++++++++++------------------------ 2 files changed, 58 insertions(+), 58 deletions(-) diff --git a/stem/connection.py b/stem/connection.py index 0ce2a153..eac38e2f 100644 --- a/stem/connection.py +++ b/stem/connection.py @@ -366,7 +366,7 @@ def _connect_auth(control_socket: stem.socket.ControlSocket, password: str, pass return None -def authenticate(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], password: Optional[str] = None, chroot_path: Optional[str] = None, protocolinfo_response: Optional[stem.response.protocolinfo.ProtocolInfoResponse] = None) -> None: +async def authenticate(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], password: Optional[str] = None, chroot_path: Optional[str] = None, protocolinfo_response: Optional[stem.response.protocolinfo.ProtocolInfoResponse] = None) -> None: """ Authenticates to a control socket using the information provided by a PROTOCOLINFO response. In practice this will often be all we need to @@ -477,7 +477,7 @@ def authenticate(controller: Union[stem.control.BaseController, stem.socket.Cont if not protocolinfo_response: try: - protocolinfo_response = get_protocolinfo(controller) + protocolinfo_response = await get_protocolinfo(controller) except stem.ProtocolError: raise IncorrectSocketType('unable to use the control socket') except stem.SocketError as exc: @@ -524,9 +524,9 @@ def authenticate(controller: Union[stem.control.BaseController, stem.socket.Cont try: if auth_type == AuthMethod.NONE: - authenticate_none(controller, False) + await authenticate_none(controller, False) elif auth_type == AuthMethod.PASSWORD: - authenticate_password(controller, password, False) + await authenticate_password(controller, password, False) elif auth_type in (AuthMethod.COOKIE, AuthMethod.SAFECOOKIE): cookie_path = protocolinfo_response.cookie_path @@ -534,12 +534,12 @@ def authenticate(controller: Union[stem.control.BaseController, stem.socket.Cont cookie_path = os.path.join(chroot_path, cookie_path.lstrip(os.path.sep)) if auth_type == AuthMethod.SAFECOOKIE: - authenticate_safecookie(controller, cookie_path, False) + await authenticate_safecookie(controller, cookie_path, False) else: - authenticate_cookie(controller, cookie_path, False) + await authenticate_cookie(controller, cookie_path, False) if isinstance(controller, stem.control.BaseController): - controller._post_authentication() + await controller._post_authentication() return # success! except OpenAuthRejected as exc: @@ -580,7 +580,7 @@ def authenticate(controller: Union[stem.control.BaseController, stem.socket.Cont raise AssertionError('BUG: Authentication failed without providing a recognized exception: %s' % str(auth_exceptions)) -def authenticate_none(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], suppress_ctl_errors: bool = True) -> None: +async def authenticate_none(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], suppress_ctl_errors: bool = True) -> None: """ Authenticates to an open control socket. All control connections need to authenticate before they can be used, even if tor hasn't been configured to @@ -605,19 +605,19 @@ def authenticate_none(controller: Union[stem.control.BaseController, stem.socket """ try: - auth_response = _msg(controller, 'AUTHENTICATE') + auth_response = await _msg(controller, 'AUTHENTICATE') # if we got anything but an OK response then error if str(auth_response) != 'OK': try: - controller.connect() + await controller.connect() except: pass raise OpenAuthRejected(str(auth_response), auth_response) except stem.ControllerError as exc: try: - controller.connect() + await controller.connect() except: pass @@ -627,7 +627,7 @@ def authenticate_none(controller: Union[stem.control.BaseController, stem.socket raise OpenAuthRejected('Socket failed (%s)' % exc) -def authenticate_password(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], password: str, suppress_ctl_errors: bool = True) -> None: +async def authenticate_password(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], password: str, suppress_ctl_errors: bool = True) -> None: """ Authenticates to a control socket that uses a password (via the HashedControlPassword torrc option). Quotes in the password are escaped. @@ -668,12 +668,12 @@ def authenticate_password(controller: Union[stem.control.BaseController, stem.so password = password.replace('"', '\\"') try: - auth_response = _msg(controller, 'AUTHENTICATE "%s"' % password) + auth_response = await _msg(controller, 'AUTHENTICATE "%s"' % password) # if we got anything but an OK response then error if str(auth_response) != 'OK': try: - controller.connect() + await controller.connect() except: pass @@ -687,7 +687,7 @@ def authenticate_password(controller: Union[stem.control.BaseController, stem.so raise PasswordAuthRejected(str(auth_response), auth_response) except stem.ControllerError as exc: try: - controller.connect() + await controller.connect() except: pass @@ -697,7 +697,7 @@ def authenticate_password(controller: Union[stem.control.BaseController, stem.so raise PasswordAuthRejected('Socket failed (%s)' % exc) -def authenticate_cookie(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], cookie_path: str, suppress_ctl_errors: bool = True) -> None: +async def authenticate_cookie(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], cookie_path: str, suppress_ctl_errors: bool = True) -> None: """ Authenticates to a control socket that uses the contents of an authentication cookie (generated via the CookieAuthentication torrc option). This does basic @@ -757,12 +757,12 @@ def authenticate_cookie(controller: Union[stem.control.BaseController, stem.sock auth_token_hex = binascii.b2a_hex(stem.util.str_tools._to_bytes(cookie_data)) msg = 'AUTHENTICATE %s' % stem.util.str_tools._to_unicode(auth_token_hex) - auth_response = _msg(controller, msg) + auth_response = await _msg(controller, msg) # if we got anything but an OK response then error if str(auth_response) != 'OK': try: - controller.connect() + await controller.connect() except: pass @@ -777,7 +777,7 @@ def authenticate_cookie(controller: Union[stem.control.BaseController, stem.sock raise CookieAuthRejected(str(auth_response), cookie_path, False, auth_response) except stem.ControllerError as exc: try: - controller.connect() + await controller.connect() except: pass @@ -787,7 +787,7 @@ def authenticate_cookie(controller: Union[stem.control.BaseController, stem.sock raise CookieAuthRejected('Socket failed (%s)' % exc, cookie_path, False) -def authenticate_safecookie(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], cookie_path: str, suppress_ctl_errors: bool = True) -> None: +async def authenticate_safecookie(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], cookie_path: str, suppress_ctl_errors: bool = True) -> None: """ Authenticates to a control socket using the safe cookie method, which is enabled by setting the CookieAuthentication torrc option on Tor client's which @@ -853,11 +853,11 @@ def authenticate_safecookie(controller: Union[stem.control.BaseController, stem. try: client_nonce_hex = stem.util.str_tools._to_unicode(binascii.b2a_hex(client_nonce)) - authchallenge_response = _msg(controller, 'AUTHCHALLENGE SAFECOOKIE %s' % client_nonce_hex) # type: ignore + authchallenge_response = await _msg(controller, 'AUTHCHALLENGE SAFECOOKIE %s' % client_nonce_hex) # type: ignore if not authchallenge_response.is_ok(): try: - controller.connect() + await controller.connect() except: pass @@ -880,7 +880,7 @@ def authenticate_safecookie(controller: Union[stem.control.BaseController, stem. raise AuthChallengeFailed(authchallenge_response_str, cookie_path) except stem.ControllerError as exc: try: - controller.connect() + await controller.connect() except: pass @@ -912,7 +912,7 @@ def authenticate_safecookie(controller: Union[stem.control.BaseController, stem. CLIENT_HASH_CONSTANT, cookie_data + client_nonce + authchallenge_response.server_nonce) - auth_response = _msg(controller, 'AUTHENTICATE %s' % stem.util.str_tools._to_unicode(binascii.b2a_hex(client_hash))) + auth_response = await _msg(controller, 'AUTHENTICATE %s' % stem.util.str_tools._to_unicode(binascii.b2a_hex(client_hash))) except stem.ControllerError as exc: try: controller.connect() @@ -942,7 +942,7 @@ def authenticate_safecookie(controller: Union[stem.control.BaseController, stem. raise CookieAuthRejected(str(auth_response), cookie_path, True, auth_response) -def get_protocolinfo(controller: Union[stem.control.BaseController, stem.socket.ControlSocket]) -> stem.response.protocolinfo.ProtocolInfoResponse: +async def get_protocolinfo(controller: Union[stem.control.BaseController, stem.socket.ControlSocket]) -> stem.response.protocolinfo.ProtocolInfoResponse: """ Issues a PROTOCOLINFO query to a control socket, getting information about the tor process running on it. If the socket is already closed then it is @@ -963,7 +963,7 @@ def get_protocolinfo(controller: Union[stem.control.BaseController, stem.socket. """ try: - protocolinfo_response = _msg(controller, 'PROTOCOLINFO 1') + protocolinfo_response = await _msg(controller, 'PROTOCOLINFO 1') except: protocolinfo_response = None @@ -974,7 +974,7 @@ def get_protocolinfo(controller: Union[stem.control.BaseController, stem.socket. controller.connect() try: - protocolinfo_response = _msg(controller, 'PROTOCOLINFO 1') + protocolinfo_response = await _msg(controller, 'PROTOCOLINFO 1') except stem.SocketClosed as exc: raise stem.SocketError(exc) @@ -982,17 +982,17 @@ def get_protocolinfo(controller: Union[stem.control.BaseController, stem.socket. return protocolinfo_response # type: ignore -def _msg(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], message: str) -> stem.response.ControlMessage: +async def _msg(controller: Union[stem.control.BaseController, stem.socket.ControlSocket], message: str) -> stem.response.ControlMessage: """ Sends and receives a message with either a :class:`~stem.socket.ControlSocket` or :class:`~stem.control.BaseController`. """ if isinstance(controller, stem.socket.ControlSocket): - controller.send(message) - return controller.recv() + await controller.send(message) + return await controller.recv() else: - return controller.msg(message) + return await controller.msg(message) def _connection_for_default_port(address: str) -> stem.socket.ControlPort: diff --git a/stem/control.py b/stem/control.py index 0559ee39..435fb741 100644 --- a/stem/control.py +++ b/stem/control.py @@ -576,7 +576,7 @@ class BaseController(object): self._create_loop_tasks() if is_authenticated: - self._post_authentication() + self._asyncio_loop.create_task(self._post_authentication()) async def msg(self, message: str) -> stem.response.ControlMessage: """ @@ -843,7 +843,7 @@ class BaseController(object): await self._socket_close() - def _post_authentication(self) -> None: + async def _post_authentication(self) -> None: # actions to be taken after we have a newly authenticated connection self._is_authenticated = True @@ -1042,7 +1042,7 @@ class Controller(BaseController): self.clear_cache() self._notify_status_listeners(State.RESET) - self.add_event_listener(_sighup_listener, EventType.SIGNAL) # type: ignore + self._asyncio_loop.create_task(self.add_event_listener(_sighup_listener, EventType.SIGNAL)) def _confchanged_listener(event: stem.response.events.ConfChangedEvent) -> None: if self.is_caching_enabled(): @@ -1057,7 +1057,7 @@ class Controller(BaseController): self._confchanged_cache_invalidation(to_cache) - self.add_event_listener(_confchanged_listener, EventType.CONF_CHANGED) # type: ignore + self._asyncio_loop.create_task(self.add_event_listener(_confchanged_listener, EventType.CONF_CHANGED)) def _address_changed_listener(event: stem.response.events.StatusEvent) -> None: if event.action in ('EXTERNAL_ADDRESS', 'DNS_USELESS'): @@ -1065,20 +1065,20 @@ class Controller(BaseController): self._set_cache({'address': None}, 'getinfo') self._last_address_exc = None - self.add_event_listener(_address_changed_listener, EventType.STATUS_SERVER) # type: ignore + self._asyncio_loop.create_task(self.add_event_listener(_address_changed_listener, EventType.STATUS_SERVER)) - def close(self) -> None: + async def close(self) -> None: self.clear_cache() - super(Controller, self).close() + await super(Controller, self).close() - def authenticate(self, *args: Any, **kwargs: Any) -> None: + async def authenticate(self, *args: Any, **kwargs: Any) -> None: """ A convenience method to authenticate the controller. This is just a pass-through to :func:`stem.connection.authenticate`. """ import stem.connection - stem.connection.authenticate(self, *args, **kwargs) + await stem.connection.authenticate(self, *args, **kwargs) def reconnect(self, *args: Any, **kwargs: Any) -> None: """ @@ -2073,7 +2073,7 @@ class Controller(BaseController): if hs_desc_content_listener: self.remove_event_listener(hs_desc_content_listener) - def get_conf(self, param: str, default: Any = UNDEFINED, multiple: bool = False) -> Union[str, Sequence[str]]: + async def get_conf(self, param: str, default: Any = UNDEFINED, multiple: bool = False) -> Union[str, Sequence[str]]: """ get_conf(param, default = UNDEFINED, multiple = False) @@ -2119,18 +2119,18 @@ class Controller(BaseController): if not param: return default if default != UNDEFINED else None - entries = self.get_conf_map(param, default, multiple) + entries = await self.get_conf_map(param, default, multiple) return _case_insensitive_lookup(entries, param, default) # TODO: temporary aliases until we have better type support in our API - def _get_conf_single(self, param: str, default: Any = UNDEFINED) -> str: - return self.get_conf(param, default) # type: ignore + async def _get_conf_single(self, param: str, default: Any = UNDEFINED) -> str: + return await self.get_conf(param, default) # type: ignore - def _get_conf_multiple(self, param: str, default: Any = UNDEFINED) -> List[str]: - return self.get_conf(param, default, multiple = True) # type: ignore + async def _get_conf_multiple(self, param: str, default: Any = UNDEFINED) -> List[str]: + return await self.get_conf(param, default, multiple = True) # type: ignore - def get_conf_map(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, multiple: bool = True) -> Dict[str, Union[str, Sequence[str]]]: + await def get_conf_map(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, multiple: bool = True) -> Dict[str, Union[str, Sequence[str]]]: """ get_conf_map(params, default = UNDEFINED, multiple = True) @@ -2211,7 +2211,7 @@ class Controller(BaseController): return self._get_conf_dict_to_response(reply, default, multiple) try: - response = stem.response._convert_to_getconf(self.msg('GETCONF %s' % ' '.join(lookup_params))) + response = stem.response._convert_to_getconf(await self.msg('GETCONF %s' % ' '.join(lookup_params))) reply.update(response.entries) if self.is_caching_enabled(): @@ -3001,7 +3001,7 @@ class Controller(BaseController): else: raise stem.ProtocolError('DEL_ONION returned unexpected response code: %s' % response.code) - def add_event_listener(self, listener: Callable[[stem.response.events.Event], None], *events: 'stem.control.EventType') -> None: + async def add_event_listener(self, listener: Callable[[stem.response.events.Event], None], *events: 'stem.control.EventType') -> None: """ Directs further tor controller events to a given function. The function is expected to take a single argument, which is a @@ -3050,7 +3050,7 @@ class Controller(BaseController): for event_type in events: self._event_listeners.setdefault(event_type, []).append(listener) - failed_events = self._attach_listeners()[1] + failed_events = (await self._attach_listeners())[1] # restricted the failures to just things we requested @@ -3732,14 +3732,14 @@ class Controller(BaseController): self.msg('DROPGUARDS') - def _post_authentication(self) -> None: - super(Controller, self)._post_authentication() + async def _post_authentication(self) -> None: + await super(Controller, self)._post_authentication() # try to re-attach event listeners to the new instance with self._event_listeners_lock: try: - failed_events = self._attach_listeners()[1] + failed_events = (await self._attach_listeners())[1] if failed_events: # remove our listeners for these so we don't keep failing @@ -3753,10 +3753,10 @@ class Controller(BaseController): # issue TAKEOWNERSHIP if we're the owning process for this tor instance - owning_pid = self.get_conf('__OwningControllerProcess', None) + owning_pid = await self.get_conf('__OwningControllerProcess', None) if owning_pid == str(os.getpid()) and self.is_localhost(): - response = stem.response._convert_to_single_line(self.msg('TAKEOWNERSHIP')) + response = stem.response._convert_to_single_line(await self.msg('TAKEOWNERSHIP')) if response.is_ok(): # Now that tor is tracking our ownership of the process via the control @@ -3793,7 +3793,7 @@ class Controller(BaseController): except Exception as exc: log.warn('Event listener raised an uncaught exception (%s): %s' % (exc, event)) - def _attach_listeners(self) -> Tuple[Sequence[str], Sequence[str]]: + async def _attach_listeners(self) -> Tuple[Sequence[str], Sequence[str]]: """ Attempts to subscribe to the self._event_listeners events from tor. This is a no-op if we're not currently authenticated. @@ -3808,7 +3808,7 @@ class Controller(BaseController): with self._event_listeners_lock: if self.is_authenticated(): # try to set them all - response = self.msg('SETEVENTS %s' % ' '.join(self._event_listeners.keys())) + response = await self.msg('SETEVENTS %s' % ' '.join(self._event_listeners.keys())) if response.is_ok(): set_events = list(self._event_listeners.keys()) @@ -3827,7 +3827,7 @@ class Controller(BaseController): # See if we can set some subset of our events. for event in list(self._event_listeners.keys()): - response = self.msg('SETEVENTS %s' % ' '.join(set_events + [event])) + response = await self.msg('SETEVENTS %s' % ' '.join(set_events + [event])) if response.is_ok(): set_events.append(event)

1 0

[stem/master] Make `Controller.get_uptime` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit b05a662283397fe0f58986dfda9f087bea3ba159 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Wed Apr 15 22:13:53 2020 +0300 Make `Controller.get_uptime` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index f847207f..0e03edfb 100644 --- a/stem/control.py +++ b/stem/control.py @@ -1650,7 +1650,7 @@ class Controller(BaseController): raise ValueError("Unable to resolve when tor began" if self.is_localhost() else "Tor isn't running locally") @with_default() - def get_uptime(self, default: Any = UNDEFINED) -> float: + async def get_uptime(self, default: Any = UNDEFINED) -> float: """ get_uptime(default = UNDEFINED) @@ -1666,7 +1666,7 @@ class Controller(BaseController): was provided """ - return time.time() - self.get_start_time() + return time.time() - (await self.get_start_time()) def is_user_traffic_allowed(self) -> 'stem.control.UserTrafficAllowed': """

1 0

[stem/master] Make `Controller.get_version` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 9864b4838437f69bd1c61874d35c0b0c518e50ba Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Wed Apr 15 21:52:07 2020 +0300 Make `Controller.get_version` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index dd723db4..c55d7015 100644 --- a/stem/control.py +++ b/stem/control.py @@ -1238,7 +1238,7 @@ class Controller(BaseController): raise @with_default() - def get_version(self, default: Any = UNDEFINED) -> stem.version.Version: + async def get_version(self, default: Any = UNDEFINED) -> stem.version.Version: """ get_version(default = UNDEFINED) @@ -1260,7 +1260,7 @@ class Controller(BaseController): version = self._get_cache('version') if not version: - version_str = self.get_info('version') + version_str = await self.get_info('version') version = stem.version.Version(version_str[4:] if version_str.startswith('Tor ') else version_str) self._set_cache({'version': version})

1 0

[stem/master] Fix the `with_default` decorator for synchronous functions
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit b3f59acd750f10b11af76c52575b56208aad80dd Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Fri Apr 17 18:41:12 2020 +0300 Fix the `with_default` decorator for synchronous functions --- stem/control.py | 95 +++++++++++++++++++++++++++++++++------------------------ 1 file changed, 55 insertions(+), 40 deletions(-) diff --git a/stem/control.py b/stem/control.py index 0e03edfb..cc9ef964 100644 --- a/stem/control.py +++ b/stem/control.py @@ -458,12 +458,6 @@ def with_default(yields: bool = False) -> Callable: """ def decorator(func: Callable) -> Callable: - is_coroutine_func = asyncio.iscoroutinefunction(func) - def coroutine_if_needed(func: Callable) -> Callable: - if is_coroutine_func: - return asyncio.coroutine(func) - return func - def get_default(func: Callable, args: Any, kwargs: Any) -> Any: arg_names = inspect.getfullargspec(func).args[1:] # drop 'self' default_position = arg_names.index('default') if 'default' in arg_names else None @@ -473,41 +467,62 @@ def with_default(yields: bool = False) -> Callable: else: return kwargs.get('default', UNDEFINED) - if not yields: - @functools.wraps(func) - @coroutine_if_needed - def wrapped(self, *args: Any, **kwargs: Any) -> Any: - try: - result = func(self, *args, **kwargs) - if is_coroutine_func: - result = yield from result - return result - except: - default = get_default(func, args, kwargs) - - if default == UNDEFINED: - raise - else: - return default + if asyncio.iscoroutinefunction(func): + if not yields: + @functools.wraps(func) + async def wrapped(self, *args: Any, **kwargs: Any) -> Any: + try: + return await func(self, *args, **kwargs) + except: + default = get_default(func, args, kwargs) + + if default == UNDEFINED: + raise + else: + return default + else: + @functools.wraps(func) + async def wrapped(self, *args: Any, **kwargs: Any) -> Any: + try: + for val in await func(self, *args, **kwargs): + yield val + except: + default = get_default(func, args, kwargs) + + if default == UNDEFINED: + raise + else: + if default is not None: + for val in default: + yield val else: - @functools.wraps(func) - @coroutine_if_needed - def wrapped(self, *args: Any, **kwargs: Any) -> Any: - try: - result = func(self, *args, **kwargs) - if is_coroutine_func: - result = yield from result - for val in result: - yield val - except: - default = get_default(func, args, kwargs) - - if default == UNDEFINED: - raise - else: - if default is not None: - for val in default: - yield val + if not yields: + @functools.wraps(func) + def wrapped(self, *args: Any, **kwargs: Any) -> Any: + try: + return func(self, *args, **kwargs) + except: + default = get_default(func, args, kwargs) + + if default == UNDEFINED: + raise + else: + return default + else: + @functools.wraps(func) + def wrapped(self, *args: Any, **kwargs: Any) -> Any: + try: + for val in func(self, *args, **kwargs): + yield val + except: + default = get_default(func, args, kwargs) + + if default == UNDEFINED: + raise + else: + if default is not None: + for val in default: + yield val return wrapped

1 0

Jump to page:

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

tor-commits July 2020