Re-land https://codereview.chromium.org/2103053002/

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 time
 import os
 import sqlalchemy as sa
 import twisted
+from buildbot.process import metrics
 from twisted.internet import reactor, threads, defer
 from twisted.python import threadpool, failure, versions, log
 
 
 class DBThreadPool(threadpool.ThreadPool):
 
     running = False
 
     # Some versions of SQLite incorrectly cache metadata about which tables are
     # and are not present on a per-connection basis.  This cache can be flushed
@@ skipping 51 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()
 
-    def do(self, callable, *args, **kwargs):
-        def thd():
-            conn = self.engine.contextual_connect()
+    # Try about 170 times over the space of a day, with the last few tries
+    # being about an hour apart.  This is designed to span a reasonable amount
+    # of time for repairing a broken database server, while still failing
+    # actual problematic queries eventually
+    BACKOFF_START = 1.0
+    BACKOFF_MULT = 1.05
+    MAX_OPERATIONALERROR_TIME = 3600*24 # one day
+    def __thd(self, with_engine, callable, args, kwargs):
+        # try to call callable(arg, *args, **kwargs) repeatedly until no
+        # OperationalErrors occur, where arg is either the engine (with_engine)
+        # or a connection (not with_engine)
+        backoff = self.BACKOFF_START
+        start = time.time()
+        while True:
+            if with_engine:
+                arg = self.engine
+            else:
+                arg = self.engine.contextual_connect()
+
             if self.__broken_sqlite: # see bug #1810
-                conn.execute("select * from sqlite_master")
+                arg.execute("select * from sqlite_master")
             try:
-                rv = callable(conn, *args, **kwargs)
+                rv = callable(arg, *args, **kwargs)
                 assert not isinstance(rv, sa.engine.ResultProxy), \
                         "do not return ResultProxy objects!"
+            except sa.exc.OperationalError, e:
+                text = e.orig.args[0]
+                if "Lost connection" in text \
+                    or "database is locked" in text:
+
+                    # see if we've retried too much
+                    elapsed = time.time() - start
+                    if elapsed > self.MAX_OPERATIONALERROR_TIME:
+                        raise
+
+                    metrics.MetricCountEvent.log(
+                            "DBThreadPool.retry-on-OperationalError")
+                    log.msg("automatically retrying query after "
+                            "OperationalError (%ss sleep)" % backoff)
+
+                    # sleep (remember, we're in a thread..)
+                    time.sleep(backoff)
+                    backoff *= self.BACKOFF_MULT
+
+                    # and re-try
+                    continue
+                else:
+                    raise
             finally:
-                conn.close()
-            return rv
-        return threads.deferToThreadPool(reactor, self, thd)
+                if not with_engine:
+                    arg.close()
+            break
+        return rv
+
+    def do(self, callable, *args, **kwargs):
+        return threads.deferToThreadPool(reactor, self,
+                self.__thd, False, callable, args, kwargs)
 
     def do_with_engine(self, callable, *args, **kwargs):
-        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)
+        return threads.deferToThreadPool(reactor, self,
+                self.__thd, True, callable, args, kwargs)
 
     # 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.
-    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")
+    if twisted.version < versions.Version('twisted', 8, 2, 0):
+        def __081_wrap(self, with_engine, callable, args, kwargs): # pragma: no cover
+            d = defer.Deferred()
+            def thd():
                 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
+                    reactor.callFromThread(d.callback,
+                            self.__thd(with_engine, callable, args, kwargs))
+                except:
+                    reactor.callFromThread(d.errback,
+                            failure.Failure())
+            self.callInThread(thd)
+            return d
 
-    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!"
-                reactor.callFromThread(d.callback, rv)
-            except:
-                reactor.callFromThread(d.errback, failure.Failure())
-        self.callInThread(thd)
-        return d
-    if twisted.version < versions.Version('twisted', 8, 2, 0):
+        def do_081(self, callable, *args, **kwargs): # pragma: no cover
+            return self.__081_wrap(False, callable, args, kwargs)
+
+        def do_with_engine_081(self, callable, *args, **kwargs): # pragma: no cover
+            return self.__081_wrap(True, callable, args, kwargs)
+
         do = do_081
         do_with_engine = do_with_engine_081
 
     def detect_bug1810(self):
         # detect buggy SQLite implementations; call only for a known-sqlite
         # dialect
         try:
             import pysqlite2.dbapi2 as sqlite
             sqlite = sqlite
         except ImportError:
@@ skipping 39 matching lines @@
         except:
             return (0,)
 
         if vers_row:
             try:
                 return tuple(map(int, vers_row[0].split('.')))
             except (TypeError, ValueError):
                 return (0,)
         else:
             return (0,)