Merge pull request #2171 from jomuel/revisit-gevent-exception-handling

Revisit gevent exception handling

Author: jomuel

CONTRIBUTING.md

@@ -428,25 +428,6 @@ def foo(a: typing.Optional[int] = None)
 ```
 
 
-
-##### Particuliarities
-
-**Concurrency/Usage of gevent**
-
-Raiden uses a patched version of gevent for concurrency.
-
-The patching is necessary to enforce our security policy: Unhandled exceptions that occur inside a
-`gevent.Greenlet` are by default not propagated outside of it. However, we require Raiden to halt
-at every unhandled exception. To achieve this, the internal error handler of gevent is patched
-and the `Greenlet` class replaced with a customized `RaidenGreenlet` throughout the project.
-
-**The methods `link`, `link_exception`, `link_value` and `rawlink` of the `Greenlet` class are
-broken by the patching and must never be used.** For example, an exception handler linked to a
-greenlet by `link_exception` will not be called; the exception will instead be bubbled up and
-cause Raiden to crash. `RaidenGreenlet` provides the equivalent methods `link_safe`,
-`link_exception_safe` and `rawlink_safe` to be used instead.
-
-
 #### Solidity
 
 For solidity we generally follow the style guide as shown in the [solidity

raiden/app.py

@@ -26,14 +26,11 @@
     INITIAL_PORT,
 )
 from raiden.utils import pex
-from raiden.utils.gevent_utils import configure_gevent
 from raiden.utils import typing
 
 
 log = structlog.get_logger(__name__)
 
-configure_gevent()
-
 
 class App:  # pylint: disable=too-few-public-methods
     DEFAULT_CONFIG = {

raiden/connection_manager.py

@@ -1,6 +1,7 @@
 from binascii import unhexlify
 
 import gevent
+from gevent.event import AsyncResult
 from gevent.lock import Semaphore
 from random import shuffle
 
@@ -18,7 +19,6 @@
 )
 from raiden.transfer import views
 from raiden.utils.typing import Address
-from raiden.utils.gevent_utils import RaidenAsyncResult
 
 log = structlog.get_logger(__name__)  # pylint: disable=invalid-name
 
@@ -147,8 +147,8 @@ def connect(
 
     def leave_async(self):
         """ Async version of `leave()` """
-        leave_result = RaidenAsyncResult()
-        gevent.spawn(self.leave).link_safe(leave_result)
+        leave_result = AsyncResult()
+        gevent.spawn(self.leave).link(leave_result)
         return leave_result
 
     def leave(self, registry_address):

raiden/exceptions.py

@@ -191,11 +191,3 @@ class ChannelOutdatedError(RaidenError):
 class InsufficientGasReserve(RaidenError):
     """ Raised when an action cannot be done because the available balance
     is not sufficient for the lifecycles of all active channels. """
-
-
-class UnhandledExceptionInGreenlet(RaidenError):
-    """Raised when a Greenlet has been linked but the
-    linked functions failed to handle an exception."""
-
-    def __init__(self, exception):
-        super().__init__(f'{type(exception)} ({str(exception) or "-"})')

raiden/network/proxies/secret_registry.py

@@ -1,4 +1,5 @@
 import structlog
+from gevent.event import AsyncResult
 from typing import List
 
 from web3.exceptions import BadFunctionCallOutput
@@ -34,7 +35,6 @@
     privatekey_to_address,
     compare_versions,
 )
-from raiden.utils.gevent_utils import RaidenAsyncResult
 
 log = structlog.get_logger(__name__)  # pylint: disable=invalid-name
 
@@ -75,7 +75,7 @@ def register_secret(self, secret: typing.Secret):
 
     def register_secret_batch(self, secrets: List[typing.Secret]):
         secret_batch = list()
-        secret_registry_transaction = RaidenAsyncResult()
+        secret_registry_transaction = AsyncResult()
 
         for secret in secrets:
             secrethash = sha3(secret)

raiden/network/proxies/token_network.py

@@ -1,4 +1,5 @@
 from collections import defaultdict
+from gevent.event import AsyncResult
 from typing import List, NamedTuple, Optional
 
 import structlog
@@ -46,7 +47,6 @@
     privatekey_to_address,
     typing,
 )
-from raiden.utils.gevent_utils import RaidenAsyncResult
 from raiden.transfer.utils import hash_balance_data
 
 log = structlog.get_logger(__name__)  # pylint: disable=invalid-name
@@ -162,7 +162,7 @@ def new_netting_channel(
         # Prevent concurrent attempts to open a channel with the same token and
         # partner address.
         if partner not in self.open_channel_transactions:
-            new_open_channel_transaction = RaidenAsyncResult()
+            new_open_channel_transaction = AsyncResult()
             self.open_channel_transactions[partner] = new_open_channel_transaction
 
             try:

raiden/network/transport/matrix.py

@@ -76,7 +76,6 @@
     Type,
     Iterable,
 )
-from raiden.utils.gevent_utils import RaidenAsyncResult
 from raiden_libs.network.matrix import GMatrixClient, Room
 
 
@@ -200,7 +199,6 @@ def start(
 
         # TODO: Add (better) error handling strategy
         self._client.start_listener_thread()
-        self._client.sync_thread.link_exception_safe(self._client_exception_handler)
 
         # TODO: Add greenlet that regularly refreshes our presence state
         self._client.set_presence_state(UserPresence.ONLINE.value)
@@ -253,7 +251,7 @@ def send_async(
             )
 
         message_id = message.message_identifier
-        async_result = RaidenAsyncResult()
+        async_result = AsyncResult()
         if isinstance(message, Processed):
             async_result.set(True)  # processed messages shouldn't get a Delivered reply
             self._send_immediate(receiver_address, json.dumps(message.to_dict()))
@@ -845,18 +843,6 @@ def _get_rtt(server_name):
         )
         return best_server
 
-    def _client_exception_handler(self, greenlet):
-        self._running = False
-        try:
-            greenlet.get()
-        except MatrixError as ex:
-            gevent.get_hub().handle_system_error(
-                TransportError,
-                TransportError(
-                    f'Unexpected error while communicating with Matrix homeserver: {ex}',
-                ),
-            )
-
     def _sign(self, data: bytes) -> bytes:
         """ Use eth_sign compatible hasher to sign matrix data """
         return signing.sign(

raiden/network/transport/udp/udp_transport.py

@@ -38,7 +38,6 @@
     retry_with_recovery,
 )
 from raiden.raiden_service import RaidenService
-from raiden.utils.gevent_utils import RaidenAsyncResult, RaidenGreenletEvent
 
 log = structlog.get_logger(__name__)  # pylint: disable=invalid-name
 
@@ -159,7 +158,7 @@ def __init__(self, discovery, udpsocket, throttle_policy, config):
         self.nat_keepalive_timeout = config['nat_keepalive_timeout']
         self.nat_invitation_timeout = config['nat_invitation_timeout']
 
-        self.event_stop = RaidenGreenletEvent()
+        self.event_stop = Event()
 
         self.greenlets = list()
         self.addresses_events = dict()
@@ -253,8 +252,8 @@ def start_health_check(self, recipient):
             )
 
             events = healthcheck.HealthEvents(
-                event_healthy=RaidenGreenletEvent(),
-                event_unhealthy=RaidenGreenletEvent(),
+                event_healthy=Event(),
+                event_unhealthy=Event(),
             )
 
             self.addresses_events[recipient] = events
@@ -365,7 +364,7 @@ def send_async(
 
         # ignore duplicates
         if message_id not in self.messageids_to_asyncresults:
-            self.messageids_to_asyncresults[message_id] = RaidenAsyncResult()
+            self.messageids_to_asyncresults[message_id] = AsyncResult()
 
             queue = self.get_queue_for(queue_identifier)
             queue.put((messagedata, message_id))
@@ -403,7 +402,7 @@ def maybe_sendraw_with_result(
         """
         async_result = self.messageids_to_asyncresults.get(message_id)
         if async_result is None:
-            async_result = RaidenAsyncResult()
+            async_result = AsyncResult()
             self.messageids_to_asyncresults[message_id] = async_result
 
         host_port = self.get_host_port(recipient)

raiden/network/transport/udp/udp_utils.py

@@ -5,7 +5,6 @@
     _AbstractLinkable,
     Event,
 )
-from raiden.utils.gevent_utils import RaidenGreenletEvent
 from raiden.utils import typing
 # type alias to avoid both circular dependencies and flake8 errors
 UDPTransport = 'UDPTransport'
@@ -17,13 +16,13 @@ def event_first_of(*events: _AbstractLinkable) -> Event:
     The event returned is /not/ cleared with any of the `events`, this value
     must not be reused if the clearing behavior is used.
     """
-    first_finished = RaidenGreenletEvent()
+    first_finished = Event()
 
     if not all(isinstance(e, _AbstractLinkable) for e in events):
         raise ValueError('all events must be linkable')
 
     for event in events:
-        event.rawlink_safe(lambda _: first_finished.set())
+        event.rawlink(lambda _: first_finished.set())
 
     return first_finished
 

raiden/raiden_service.py

@@ -8,6 +8,7 @@
 
 from coincurve import PrivateKey
 from eth_utils import is_binary_address
+from gevent.event import AsyncResult, Event
 from gevent.lock import Semaphore
 
 from raiden import constants, routing, waiting
@@ -51,7 +52,6 @@
     create_default_identifier,
     typing,
 )
-from raiden.utils.gevent_utils import RaidenAsyncResult, RaidenGreenletEvent
 
 log = structlog.get_logger(__name__)  # pylint: disable=invalid-name
 
@@ -145,7 +145,7 @@ def __init__(
         self.tokennetworkids_to_connectionmanagers = dict()
         self.identifier_to_results: typing.Dict[
             typing.PaymentIdentifier,
-            RaidenAsyncResult,
+            AsyncResult,
         ] = dict()
 
         self.chain: BlockChainService = chain
@@ -164,8 +164,8 @@ def __init__(
         self.blockchain_events = BlockchainEvents()
         self.alarm = AlarmTask(chain)
         self.shutdown_timeout = config['shutdown_timeout']
-        self.stop_event = RaidenGreenletEvent()
-        self.start_event = RaidenGreenletEvent()
+        self.stop_event = Event()
+        self.start_event = Event()
         self.chain.client.inject_stop_event(self.stop_event)
 
         self.wal = None
@@ -194,7 +194,7 @@ def __init__(
 
         self.event_poll_lock = gevent.lock.Semaphore()
 
-    def start_async(self) -> RaidenGreenletEvent:
+    def start_async(self) -> Event:
         """ Start the node asynchronously. """
         self.start_event.clear()
         self.stop_event.clear()
@@ -299,13 +299,13 @@ def start_async(self) -> RaidenGreenletEvent:
             def set_start_on_registration(_):
                 self.start_event.set()
 
-            endpoint_registration_greenlet.link_safe(set_start_on_registration)
+            endpoint_registration_greenlet.link(set_start_on_registration)
         else:
             self.start_event.set()
 
         return self.start_event
 
-    def start(self) -> RaidenGreenletEvent:
+    def start(self) -> Event:
         """ Start the node. """
         self.start_async().wait()
 
@@ -592,7 +592,7 @@ def start_mediated_transfer(
         if identifier in self.identifier_to_results:
             return self.identifier_to_results[identifier]
 
-        async_result = RaidenAsyncResult()
+        async_result = AsyncResult()
         self.identifier_to_results[identifier] = async_result
 
         secret = random_secret()