Multiple resource leaks fixed in proxy and dealer

Author: DZabavchik

crossbar/__init__.py

@@ -22,6 +22,10 @@
 if not hasattr(eth_abi, 'encode_single') and hasattr(eth_abi, 'encode'):
     eth_abi.encode_single = eth_abi.encode
 
+import eth_typing
+if not hasattr(eth_typing, 'ChainID') and hasattr(eth_typing, 'ChainId'):
+    eth_typing.ChainID = eth_typing.ChainId
+
 # monkey patch web3 for master branch / upcoming v6 (which we need for python 3.11)
 # AttributeError: type object 'Web3' has no attribute 'toChecksumAddress'. Did you mean: 'to_checksum_address'?
 import web3

crossbar/router/dealer.py

@@ -40,6 +40,7 @@ class InvocationRequest(object):
         'id',
         'registration',
         'caller',
+        'caller_session_id',
         'call',
         'callee',
         'forward_for',
@@ -53,6 +54,7 @@ def __init__(self, id, registration, caller, call, callee, forward_for, authoriz
         self.id = id
         self.registration = registration
         self.caller = caller
+        self.caller_session_id = caller._session_id
         self.call = call
         self.callee = callee
         self.forward_for = forward_for
@@ -185,6 +187,7 @@ def detach(self, session):
         is_rlink_session = (session._authrole == "rlink")
         if session in self._caller_to_invocations:
 
+            # this needs to update all four places where we track invocations similar to _remove_invoke_request
             outstanding = self._caller_to_invocations.get(session, [])
             for invoke in outstanding:  # type: InvocationRequest
                 if invoke.canceled:
@@ -207,11 +210,26 @@ def detach(self, session):
                     request=invoke.id,
                     session=session._session_id,
                 )
+
+                if invoke.timeout_call:
+                    invoke.timeout_call.cancel()
+                    invoke.timeout_call = None
+
+                invokes = self._callee_to_invocations[callee]
+                invokes.remove(invoke)
+                if not invokes:
+                    del self._callee_to_invocations[callee]
+
+                del self._invocations[invoke.id]
+                del self._invocations_by_call[(invoke.caller_session_id, invoke.call.request)]
+
                 self._router.send(invoke.callee, message.Interrupt(
                     request=invoke.id,
                     mode=message.Cancel.KILLNOWAIT,
                 ))
 
+            del self._caller_to_invocations[session]
+
         if session in self._session_to_registrations:
 
             # send out Errors for any in-flight calls we have
@@ -235,6 +253,21 @@ def detach(self, session):
                 if invoke.caller._transport:
                     invoke.caller._transport.send(reply)
 
+                if invoke.timeout_call:
+                    invoke.timeout_call.cancel()
+                    invoke.timeout_call = None
+
+                invokes = self._caller_to_invocations[invoke.caller]
+                invokes.remove(invoke)
+                if not invokes:
+                    del self._caller_to_invocations[invoke.caller]
+
+                del self._invocations[invoke.id]
+                del self._invocations_by_call[(invoke.caller_session_id, invoke.call.request)]
+
+            if outstanding:
+                del self._callee_to_invocations[session]
+
             for registration in self._session_to_registrations[session]:
                 was_registered, was_last_callee = self._registration_map.drop_observer(session, registration)
 
@@ -1120,23 +1153,20 @@ def _remove_invoke_request(self, invocation_request):
             invocation_request.timeout_call.cancel()
             invocation_request.timeout_call = None
 
-        invokes = self._callee_to_invocations[invocation_request.callee]
-        invokes.remove(invocation_request)
-        if not invokes:
-            del self._callee_to_invocations[invocation_request.callee]
-
-        invokes = self._caller_to_invocations[invocation_request.caller]
-        invokes.remove(invocation_request)
-        if not invokes:
-            del self._caller_to_invocations[invocation_request.caller]
+        # all four places should always be updated together
+        if invocation_request.id in self._invocations:
+            del self._invocations[invocation_request.id]
+            invokes = self._callee_to_invocations[invocation_request.callee]
+            invokes.remove(invocation_request)
+            if not invokes:
+                del self._callee_to_invocations[invocation_request.callee]
 
-        del self._invocations[invocation_request.id]
+            invokes = self._caller_to_invocations[invocation_request.caller]
+            invokes.remove(invocation_request)
+            if not invokes:
+                del self._caller_to_invocations[invocation_request.caller]
 
-        # the session_id will be None if the caller session has
-        # already vanished
-        caller_id = invocation_request.caller._session_id
-        if caller_id is not None:
-            del self._invocations_by_call[caller_id, invocation_request.call.request]
+            del self._invocations_by_call[invocation_request.caller_session_id, invocation_request.call.request]
 
     # noinspection PyUnusedLocal
     def processCancel(self, session, cancel):

crossbar/router/session.py

@@ -736,12 +736,13 @@ def onClose(self, wasClean):
                 self.onLeave(CloseDetails())
             except Exception:
                 self.log.failure("Exception raised in onLeave callback")
+                self.log.warn("{tb}".format(tb=Failure().getTraceback()))
 
             try:
                 self._router.detach(self)
             except Exception as e:
                 self.log.error("Failed to detach session '{}': {}".format(self._session_id, e))
-                self.log.debug("{tb}".format(tb=Failure().getTraceback()))
+                self.log.warn("{tb}".format(tb=Failure().getTraceback()))
 
             self._session_id = None
 

crossbar/router/test/test_dealer.py

@@ -111,10 +111,17 @@ def test_outstanding_invoke_but_caller_gone(self):
         outstanding = mock.Mock()
         outstanding.call.request = 1
 
+        # there was a bug where timeout calls were not getting cancelled
+        # mock has non-null timeout_call, so we need to set it to None
+        outstanding.timeout_call = None
         dealer = self.router._dealer
         dealer.attach(session)
 
+        # All four maps involved in invocation tracking must be updated atomically
+        dealer._caller_to_invocations[outstanding.caller] = [outstanding]
         dealer._callee_to_invocations[session] = [outstanding]
+        dealer._invocations[outstanding.id] = outstanding
+        dealer._invocations_by_call[(outstanding.caller_session_id, outstanding.call.request)] = outstanding
         # pretend we've disconnected already
         outstanding.caller._transport = None
 

crossbar/worker/main.py

@@ -123,7 +123,7 @@ def _run_command_exec_worker(options, reactor=None, personality=None):
 
     # we use an Autobahn utility to import the "best" available Twisted reactor
     from autobahn.twisted.choosereactor import install_reactor
-    reactor = install_reactor(options.reactor)
+    reactor = install_reactor(explicit_reactor=options.reactor or os.environ.get('CROSSBAR_REACTOR', None))
 
     # make sure logging to something else than stdio is setup _first_
     from crossbar._logging import make_JSON_observer, cb_logging_aware