Revert of Revert buildbot 361b81b43fc7793b66facd4bac7abe0adb30f3d5.

third_party/buildbot_8_4p1/buildbot/db/pool.py

 # This file is part of Buildbot.  Buildbot is free software: you can
 # redistribute it and/or modify it under the terms of the GNU General Public
 # License as published by the Free Software Foundation, version 2.
 #
 # This program is distributed in the hope that it will be useful, but WITHOUT
 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 # details.
 #
 # You should have received a copy of the GNU General Public License along with
 # this program; if not, write to the Free Software Foundation, Inc., 51
 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 #
 # Copyright Buildbot Team Members
 
 import os
 import sqlalchemy as sa
 import twisted
-from twisted.internet import reactor, threads, defer
+from twisted.internet import reactor, threads, defer, task
 from twisted.python import threadpool, failure, versions, log
 
+# Hack for bug #1992.  In as-yet-unknown circumstances, select() fails to
+# notice that a selfpipe has been written to, thus causing callFromThread, as
+# used in deferToThreadPool, to hang indefinitely.  The workaround is to wake
+# up the select loop every second by ensuring that there is an event occuring
+# every second, with this busy loop:
+def bug1992hack(f):
+    def w(*args, **kwargs):
+        busyloop = task.LoopingCall(lambda : None)
+        busyloop.start(1)
+        d = f(*args, **kwargs)
+        def stop_loop(r):
+            busyloop.stop()
+            return r
+        d.addBoth(stop_loop)
+        return d
+    w.__name__ = f.__name__
+    w.__doc__ = f.__doc__
+    return w
+
 
 class DBThreadPool(threadpool.ThreadPool):
     """
     A pool of threads ready and waiting to execute queries.
 
     If the engine has an C{optimal_thread_pool_size} attribute, then the
     maxthreads of the thread pool will be set to that value.  This is most
     useful for SQLite in-memory connections, where exactly one connection
     (and thus thread) should be used.
     """
@@ skipping 38 matching lines @@
 
     def shutdown(self):
         """Manually stop the pool.  This is only necessary from tests, as the
         pool will stop itself when the reactor stops under normal
         circumstances."""
         if not self._stop_evt:
             return # pool is already stopped
         reactor.removeSystemEventTrigger(self._stop_evt)
         self._stop()
 
+    @bug1992hack
     def do(self, callable, *args, **kwargs):
         """
         Call C{callable} in a thread, with a Connection as first argument.
         Returns a deferred that will indicate the results of the callable.
 
         Note: do not return any SQLAlchemy objects via this deferred!
         """
         def thd():
             conn = self.engine.contextual_connect()
             if self.__broken_sqlite: # see bug #1810
                 conn.execute("select * from sqlite_master")
             try:
                 rv = callable(conn, *args, **kwargs)
                 assert not isinstance(rv, sa.engine.ResultProxy), \
                         "do not return ResultProxy objects!"
             finally:
                 conn.close()
             return rv
         return threads.deferToThreadPool(reactor, self, thd)
 
+    @bug1992hack
     def do_with_engine(self, callable, *args, **kwargs):
         """
         Like L{do}, but with an SQLAlchemy Engine as the first argument.  This
         is only used for schema manipulation, and is not used at master
         runtime.
         """
         def thd():
             if self.__broken_sqlite: # see bug #1810
                 self.engine.execute("select * from sqlite_master")
             rv = callable(self.engine, *args, **kwargs)
             assert not isinstance(rv, sa.engine.ResultProxy), \
                     "do not return ResultProxy objects!"
             return rv
         return threads.deferToThreadPool(reactor, self, thd)
 
     # older implementations for twisted < 0.8.2, which does not have
     # deferToThreadPool; this basically re-implements it, although it gets some
     # of the synchronization wrong - the thread may still be "in use" when the
     # deferred fires in the parent, which can lead to database accesses hopping
     # between threads.  In practice, this should not cause any difficulty.
+    @bug1992hack
     def do_081(self, callable, *args, **kwargs): # pragma: no cover
         d = defer.Deferred()
         def thd():
             try:
                 conn = self.engine.contextual_connect()
                 if self.__broken_sqlite: # see bug #1810
                     conn.execute("select * from sqlite_master")
                 try:
                     rv = callable(conn, *args, **kwargs)
                     assert not isinstance(rv, sa.engine.ResultProxy), \
                             "do not return ResultProxy objects!"
                 finally:
                     conn.close()
                 reactor.callFromThread(d.callback, rv)
             except:
                 reactor.callFromThread(d.errback, failure.Failure())
         self.callInThread(thd)
         return d
 
+    @bug1992hack
     def do_with_engine_081(self, callable, *args, **kwargs): # pragma: no cover
         d = defer.Deferred()
         def thd():
             try:
                 conn = self.engine
                 if self.__broken_sqlite: # see bug #1810
                     conn.execute("select * from sqlite_master")
                 rv = callable(conn, *args, **kwargs)
                 assert not isinstance(rv, sa.engine.ResultProxy), \
                         "do not return ResultProxy objects!"
@@ skipping 36 matching lines @@
 
         try:
             test()
         except sqlite.OperationalError:
             # this is the expected error indicating it's broken
             return True
 
         # but this version should not fail..
         test(select_from_sqlite_master=True)
         return False # not broken - no workaround required