nominatim/db/async_connection.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim.
   4 # Copyright (C) 2021 by the Nominatim developer community.
   5 # For a full list of authors see the git log.
   6 """ Database helper functions for the indexer.
   7 """
   8 import logging
   9 import psycopg2
  10 from psycopg2.extras import wait_select
  11
  12 LOG = logging.getLogger()
  13
  14 class DBConnection:
  15     """ A single non-blocking database connection.
  16     """
  17
  18     def __init__(self, dsn):
  19         self.current_query = None
  20         self.current_params = None
  21         self.dsn = dsn
  22
  23         self.conn = None
  24         self.cursor = None
  25         self.connect()
  26
  27     def connect(self):
  28         """ (Re)connect to the database. Creates an asynchronous connection
  29             with JIT and parallel processing disabled. If a connection was
  30             already open, it is closed and a new connection established.
  31             The caller must ensure that no query is pending before reconnecting.
  32         """
  33         if self.conn is not None:
  34             self.cursor.close()
  35             self.conn.close()
  36
  37         # Use a dict to hand in the parameters because async is a reserved
  38         # word in Python3.
  39         self.conn = psycopg2.connect(**{'dsn' : self.dsn, 'async' : True})
  40         self.wait()
  41
  42         self.cursor = self.conn.cursor()
  43         # Disable JIT and parallel workers as they are known to cause problems.
  44         # Update pg_settings instead of using SET because it does not yield
  45         # errors on older versions of Postgres where the settings are not
  46         # implemented.
  47         self.perform(
  48             """ UPDATE pg_settings SET setting = -1 WHERE name = 'jit_above_cost';
  49                 UPDATE pg_settings SET setting = 0
  50                    WHERE name = 'max_parallel_workers_per_gather';""")
  51         self.wait()
  52
  53     def wait(self):
  54         """ Block until any pending operation is done.
  55         """
  56         while True:
  57             try:
  58                 wait_select(self.conn)
  59                 self.current_query = None
  60                 return
  61             except psycopg2.extensions.TransactionRollbackError as error:
  62                 if error.pgcode == '40P01':
  63                     LOG.info("Deadlock detected (params = %s), retry.",
  64                              str(self.current_params))
  65                     self.cursor.execute(self.current_query, self.current_params)
  66                 else:
  67                     raise
  68             except psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
  69                 self.cursor.execute(self.current_query, self.current_params)
  70
  71     def perform(self, sql, args=None):
  72         """ Send SQL query to the server. Returns immediately without
  73             blocking.
  74         """
  75         self.current_query = sql
  76         self.current_params = args
  77         self.cursor.execute(sql, args)
  78
  79     def fileno(self):
  80         """ File descriptor to wait for. (Makes this class select()able.)
  81         """
  82         return self.conn.fileno()
  83
  84     def is_done(self):
  85         """ Check if the connection is available for a new query.
  86
  87             Also checks if the previous query has run into a deadlock.
  88             If so, then the previous query is repeated.
  89         """
  90         if self.current_query is None:
  91             return True
  92
  93         try:
  94             if self.conn.poll() == psycopg2.extensions.POLL_OK:
  95                 self.current_query = None
  96                 return True
  97         except psycopg2.extensions.TransactionRollbackError as error:
  98             if error.pgcode == '40P01':
  99                 LOG.info("Deadlock detected (params = %s), retry.", str(self.current_params))
 100                 self.cursor.execute(self.current_query, self.current_params)
 101             else:
 102                 raise
 103         except psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
 104             self.cursor.execute(self.current_query, self.current_params)
 105
 106         return False