misc: monkeypatch thread objects to be multitenant-aware (#18663)

wcs/middleware.py

@@ -14,10 +14,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, see <http://www.gnu.org/licenses/>.
 
-import copy
 import json
 import thread
-import threading
 import urllib
 
 from django.http import HttpResponseBadRequest, HttpResponseRedirect
@@ -113,11 +111,8 @@ class AfterJobsMiddleware(object):
             http_response = HTTPResponse()
             http_response.after_jobs = response.after_jobs
             if self.ASYNC:
-                publisher = copy.copy(get_publisher())
-                publisher.pgconn = None
                 thread.start_new_thread(
-                        http_response.process_after_jobs,
-                        (publisher,))
+                        http_response.process_after_jobs, ())
             else:
                 http_response.process_after_jobs()
         return response

wcs/qommon/__init__.py

@@ -15,6 +15,9 @@
 # along with this program; if not, see <http://www.gnu.org/licenses/>.
 
 import ConfigParser
+import copy
+import threading
+
 import django.apps
 from django.conf import settings
 from quixote import get_publisher
@@ -41,6 +44,95 @@ import publisher
 publisher._ = _
 
 
+class TenantAwareThread(threading.Thread):
+    def __init__(self, *args, **kwargs):
+        self.publisher = copy.copy(get_publisher())
+        if self.publisher:
+            self.publisher.pgconn = None
+        super(TenantAwareThread, self).__init__(*args, **kwargs)
+
+    def run(self):
+        if self.publisher:
+            self.publisher.set_in_thread()
+        super(TenantAwareThread, self).run()
+
+
+class _Timer(TenantAwareThread):
+    """Call a function after a specified number of seconds:
+
+            t = Timer(30.0, f, args=[], kwargs={})
+            t.start()
+            t.cancel()     # stop the timer's action if it's still waiting
+
+    """
+
+    def __init__(self, interval, function, args=[], kwargs={}):
+        super(_Timer, self).__init__()
+        self.interval = interval
+        self.function = function
+        self.args = args
+        self.kwargs = kwargs
+        self.finished = threading.Event()
+
+    def cancel(self):
+        """Stop the timer if it hasn't finished yet"""
+        self.finished.set()
+
+    def run(self):
+        self.finished.wait(self.interval)
+        if not self.finished.is_set():
+            self.function(*self.args, **self.kwargs)
+        self.finished.set()
+
+
+class _MainThread(TenantAwareThread):
+
+    def __init__(self):
+        super(_MainThread, self).__init__(name="MainThread")
+        self._Thread__started.set()
+        self._set_ident()
+        with threading._active_limbo_lock:
+            threading._active[threading._get_ident()] = self
+
+    def _set_daemon(self):
+        return False
+
+    def _exitfunc(self):
+        self._Thread__stop()
+        t = threading._pickSomeNonDaemonThread()
+        if t:
+            if __debug__:
+                self._note("%s: waiting for other threads", self)
+        while t:
+            t.join()
+            t = threading._pickSomeNonDaemonThread()
+        if __debug__:
+            self._note("%s: exiting", self)
+        self._Thread__delete()
+
+
+class _DummyThread(TenantAwareThread):
+
+    def __init__(self):
+        super(_DummyThread, self).__init__(name=threading._newname("Dummy-%d"))
+
+        # Thread.__block consumes an OS-level locking primitive, which
+        # can never be used by a _DummyThread.  Since a _DummyThread
+        # instance is immortal, that's bad, so release this resource.
+        del self._Thread__block
+
+        self._Thread__started.set()
+        self._set_ident()
+        with threading._active_limbo_lock:
+            threading._active[threading._get_ident()] = self
+
+    def _set_daemon(self):
+        return True
+
+    def join(self, timeout=None):
+        assert False, "cannot join a dummy thread"
+
+
 class AppConfig(django.apps.AppConfig):
     name = 'wcs.qommon'
 
@@ -54,6 +146,11 @@ class AppConfig(django.apps.AppConfig):
             for i, extra in enumerate(settings.WCS_EXTRA_MODULES):
                 config.set('extra', 'cmd_line_extra_%d' % i, extra)
 
+        threading.Thread = TenantAwareThread
+        threading._DummyThread = _DummyThread
+        threading._MainThread = _MainThread
+        threading._Timer = _Timer
+
         get_publisher_class().configure(config)
         get_publisher_class().register_tld_names = True
         get_publisher_class().init_publisher_class()

wcs/qommon/http_response.py

@@ -144,13 +144,10 @@ class HTTPResponse(quixote.http_response.HTTPResponse):
         self.after_jobs.append((job, cmd))
         return job
 
-    def process_after_jobs(self, publisher=None):
+    def process_after_jobs(self):
         if not self.after_jobs:
             return
 
-        if publisher:
-            publisher.set_in_thread()
-
         for job, job_function in self.after_jobs:
             if job.completion_time:
                 continue