Fix threadsafety in ThreadedMemoryReactorClock

changelog.d/8497.misc

@@ -0,0 +1 @@
+Fix a threadsafety bug in unit tests.

tests/server.py

@@ -1,6 +1,8 @@
 import json
 import logging
+from collections import deque
 from io import SEEK_END, BytesIO
+from typing import Any, Callable, Deque, List
 
 import attr
 from zope.interface import implementer
@@ -251,6 +253,7 @@ def __init__(self):
         self._tcp_callbacks = {}
         self._udp = []
         lookups = self.lookups = {}
+        self._thread_callbacks = deque()  # type: Deque[Callable[[], None]]()
 
         @implementer(IResolverSimple)
         class FakeResolver:
@@ -272,10 +275,10 @@ def callFromThread(self, callback, *args, **kwargs):
         """
         Make the callback fire in the next reactor iteration.
         """
-        d = Deferred()
-        d.addCallback(lambda x: callback(*args, **kwargs))
-        self.callLater(0, d.callback, True)
-        return d
+        cb = lambda: callback(*args, **kwargs)
+        # it's not safe to call callLater() here, so we append the callback to a
+        # separate queue.
+        self._thread_callbacks.append(cb)
 
     def getThreadPool(self):
         return self.threadpool
@@ -303,6 +306,18 @@ def connectTCP(self, host, port, factory, timeout=30, bindAddress=None):
 
         return conn
 
+    def advance(self, amount):
+        # run any "callFromThread" callbacks before anything registered with
+        # "callLater"
+        while True:
+            try:
+                callback = self._thread_callbacks.popleft()
+            except IndexError:
+                break
+            callback()
+
+        super().advance(amount)
+
 
 class ThreadPool:
     """