Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / db / async_connection.py

diff --git a/nominatim/db/async_connection.py b/nominatim/db/async_connection.py
index db4b89ce2fa74b6a03763dccac1f00df5f8ef22e..a86c5bdcee13eb37c4874f29be286d2b06beecc0 100644 (file)
--- a/nominatim/db/async_connection.py
+++ b/nominatim/db/async_connection.py
@@ -33,18 +33,17 @@ class DeadlockHandler:
         self.ignore_sql_errors = ignore_sql_errors
 
     def __enter__(self):
         self.ignore_sql_errors = ignore_sql_errors
 
     def __enter__(self):

-        pass
+        return self

 
     def __exit__(self, exc_type, exc_value, traceback):
         if __has_psycopg2_errors__:
             if exc_type == psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
                 self.handler()
                 return True
 
     def __exit__(self, exc_type, exc_value, traceback):
         if __has_psycopg2_errors__:
             if exc_type == psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
                 self.handler()
                 return True

-        else:
-            if exc_type == psycopg2.extensions.TransactionRollbackError:
-                if exc_value.pgcode == '40P01':
-                    self.handler()
-                    return True
+        elif exc_type == psycopg2.extensions.TransactionRollbackError \
+             and exc_value.pgcode == '40P01':
+            self.handler()
+            return True

 
         if self.ignore_sql_errors and isinstance(exc_value, psycopg2.Error):
             LOG.info("SQL error ignored: %s", exc_value)
 
         if self.ignore_sql_errors and isinstance(exc_value, psycopg2.Error):
             LOG.info("SQL error ignored: %s", exc_value)


@@ -86,7 +85,7 @@ class DBConnection:
 
         # Use a dict to hand in the parameters because async is a reserved
         # word in Python3.
 
         # Use a dict to hand in the parameters because async is a reserved
         # word in Python3.

-        self.conn = psycopg2.connect(**{'dsn' : self.dsn, 'async' : True})
+        self.conn = psycopg2.connect(**{'dsn': self.dsn, 'async': True})

         self.wait()
 
         self.cursor = self.conn.cursor(cursor_factory=cursor_factory)
         self.wait()
 
         self.cursor = self.conn.cursor(cursor_factory=cursor_factory)


@@ -191,10 +190,7 @@ class WorkerPool:
                     yield thread
 
             if command_stat > self.REOPEN_CONNECTIONS_AFTER:
                     yield thread
 
             if command_stat > self.REOPEN_CONNECTIONS_AFTER:

-                for thread in self.threads:
-                    while not thread.is_done():
-                        thread.wait()
-                    thread.connect()
+                self._reconnect_threads()

                 ready = self.threads
                 command_stat = 0
             else:
                 ready = self.threads
                 command_stat = 0
             else:


@@ -203,6 +199,13 @@ class WorkerPool:
                 self.wait_time += time.time() - tstart
 
 
                 self.wait_time += time.time() - tstart
 
 

+    def _reconnect_threads(self):
+        for thread in self.threads:
+            while not thread.is_done():
+                thread.wait()
+            thread.connect()
+
+

     def __enter__(self):
         return self
 
     def __enter__(self):
         return self