Merge pull request #4715 from golemfactory/b0.21-locked-monkey

Protect asyncio add/remove Reader/Writer against multithreading

golem/core/common.py

@@ -3,6 +3,7 @@
 import os
 import subprocess
 import sys
+import threading
 from calendar import timegm
 from datetime import datetime
 from functools import wraps
@@ -11,6 +12,7 @@
 import pytz
 
 from golem.core import simpleenv
+from golem.decorators import locked
 
 F = TypeVar('F', bound=Callable[..., Any])
 
@@ -322,27 +324,20 @@ def install_reactor():
     asyncioreactor.install(asyncio.get_event_loop())
 
     from twisted.internet import reactor
-    _patch_remove_writer(reactor)
+    _patch_modify_io(reactor)
 
     from golem.core.variables import REACTOR_THREAD_POOL_SIZE
     reactor.suggestThreadPoolSize(REACTOR_THREAD_POOL_SIZE)
     return reactor
 
 
-def _patch_remove_writer(reactor):
-    import logging
-    import types
+def _patch_modify_io(reactor):
+    lock = threading.Lock()
 
-    logger = logging.getLogger('golem.core')
-
-    def patched_remove_writer(_, writer):
-        try:
-            reactor_remove_writer(writer)
-        except KeyError as err:
-            logger.debug("Reactor removeWriter error: %r", err)
-
-    reactor_remove_writer = reactor.removeWriter
-    reactor.removeWriter = types.MethodType(patched_remove_writer, reactor)
+    reactor.removeWriter = locked(lock)(reactor.removeWriter)
+    reactor.addWriter = locked(lock)(reactor.addWriter)
+    reactor.removeReader = locked(lock)(reactor.removeReader)
+    reactor.addReader = locked(lock)(reactor.addReader)
 
 
 if is_windows():

golem/decorators.py

@@ -1,13 +1,14 @@
 import functools
 import logging
 
-from golem import model
 
 log = logging.getLogger('golem.decorators')
 
 
 def run_with_db():
     """Run only when DB is active"""
+    from golem import model
+
     def wrapped(f):
         @functools.wraps(f)
         def curry(*args, **kwargs):
@@ -21,3 +22,14 @@ def curry(*args, **kwargs):
             return f(*args, **kwargs)
         return curry
     return wrapped
+
+
+def locked(lock):
+    """Run under a threadsafe lock"""
+    def wrapped(f):
+        @functools.wraps(f)
+        def curry(*args, **kwargs):
+            with lock:
+                return f(*args, **kwargs)
+        return curry
+    return wrapped