Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / nominatim.py

diff --git a/nominatim/nominatim.py b/nominatim/nominatim.py
index 848846ae45123b28b69b53513c378d2ab0435231..b29bf343e035e02cdfd62a43e0e0d3a01ebebcd5 100755 (executable)
--- a/nominatim/nominatim.py
+++ b/nominatim/nominatim.py
@@ -35,9 +35,14 @@ import select
 log = logging.getLogger()
 
 def make_connection(options, asynchronous=False):
-    return psycopg2.connect(dbname=options.dbname, user=options.user,
-                            password=options.password, host=options.host,
-                            port=options.port, async_=asynchronous)
+    params = {'dbname' : options.dbname,
+              'user' : options.user,
+              'password' : options.password,
+              'host' : options.host,
+              'port' : options.port,
+              'async' : asynchronous}
+
+    return psycopg2.connect(**params)
 
 
 class RankRunner(object):
@@ -119,12 +124,33 @@ class DBConnection(object):
         self.wait()
 
         self.cursor = self.conn.cursor()
+        # Disable JIT and parallel workers as they are known to cause problems.
+        # Update pg_settings instead of using SET because it does not yield
+        # errors on older versions of Postgres where the settings are not
+        # implemented.
+        self.perform(
+            """ UPDATE pg_settings SET setting = -1 WHERE name = 'jit_above_cost';
+                UPDATE pg_settings SET setting = 0 
+                   WHERE name = 'max_parallel_workers_per_gather';""")
+        self.wait()
 
     def wait(self):
         """ Block until any pending operation is done.
         """
-        wait_select(self.conn)
-        self.current_query = None
+        while True:
+            try:
+                wait_select(self.conn)
+                self.current_query = None
+                return
+            except psycopg2.extensions.TransactionRollbackError as e:
+                if e.pgcode == '40P01':
+                    log.info("Deadlock detected (params = {}), retry."
+                              .format(self.current_params))
+                    self.cursor.execute(self.current_query, self.current_params)
+                else:
+                    raise
+            except psycopg2.errors.DeadlockDetected:
+                self.cursor.execute(self.current_query, self.current_params)
 
     def perform(self, sql, args=None):
         """ Send SQL query to the server. Returns immediately without
@@ -158,6 +184,8 @@ class DBConnection(object):
                 self.cursor.execute(self.current_query, self.current_params)
             else:
                 raise
+        except psycopg2.errors.DeadlockDetected:
+            self.cursor.execute(self.current_query, self.current_params)
 
         return False
 
@@ -278,14 +306,14 @@ class Indexer(object):
             if command_stat > 100000:
                 for t in self.threads:
                     while not t.is_done():
-                        wait_select(t.conn)
+                        t.wait()
                     t.connect()
                 command_stat = 0
                 ready = self.threads
             else:
                 ready, _, _ = select.select(self.threads, [], [])
 
-        assert(False, "Unreachable code")
+        assert False, "Unreachable code"
 
 
 def nominatim_arg_parser():