nominatim/db/async_connection.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """ Database helper functions for the indexer.
   8 """
   9 import logging
  10 import select
  11 import time
  12
  13 import psycopg2
  14 from psycopg2.extras import wait_select
  15
  16 # psycopg2 emits different exceptions pre and post 2.8. Detect if the new error
  17 # module is available and adapt the error handling accordingly.
  18 try:
  19     import psycopg2.errors # pylint: disable=no-name-in-module,import-error
  20     __has_psycopg2_errors__ = True
  21 except ImportError:
  22     __has_psycopg2_errors__ = False
  23
  24 LOG = logging.getLogger()
  25
  26 class DeadlockHandler:
  27     """ Context manager that catches deadlock exceptions and calls
  28         the given handler function. All other exceptions are passed on
  29         normally.
  30     """
  31
  32     def __init__(self, handler, ignore_sql_errors=False):
  33         self.handler = handler
  34         self.ignore_sql_errors = ignore_sql_errors
  35
  36     def __enter__(self):
  37         return self
  38
  39     def __exit__(self, exc_type, exc_value, traceback):
  40         if __has_psycopg2_errors__:
  41             if exc_type == psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
  42                 self.handler()
  43                 return True
  44         elif exc_type == psycopg2.extensions.TransactionRollbackError \
  45              and exc_value.pgcode == '40P01':
  46             self.handler()
  47             return True
  48
  49         if self.ignore_sql_errors and isinstance(exc_value, psycopg2.Error):
  50             LOG.info("SQL error ignored: %s", exc_value)
  51             return True
  52
  53         return False
  54
  55
  56 class DBConnection:
  57     """ A single non-blocking database connection.
  58     """
  59
  60     def __init__(self, dsn, cursor_factory=None, ignore_sql_errors=False):
  61         self.current_query = None
  62         self.current_params = None
  63         self.dsn = dsn
  64         self.ignore_sql_errors = ignore_sql_errors
  65
  66         self.conn = None
  67         self.cursor = None
  68         self.connect(cursor_factory=cursor_factory)
  69
  70     def close(self):
  71         """ Close all open connections. Does not wait for pending requests.
  72         """
  73         if self.conn is not None:
  74             self.cursor.close()
  75             self.conn.close()
  76
  77         self.conn = None
  78
  79     def connect(self, cursor_factory=None):
  80         """ (Re)connect to the database. Creates an asynchronous connection
  81             with JIT and parallel processing disabled. If a connection was
  82             already open, it is closed and a new connection established.
  83             The caller must ensure that no query is pending before reconnecting.
  84         """
  85         self.close()
  86
  87         # Use a dict to hand in the parameters because async is a reserved
  88         # word in Python3.
  89         self.conn = psycopg2.connect(**{'dsn': self.dsn, 'async': True})
  90         self.wait()
  91
  92         self.cursor = self.conn.cursor(cursor_factory=cursor_factory)
  93         # Disable JIT and parallel workers as they are known to cause problems.
  94         # Update pg_settings instead of using SET because it does not yield
  95         # errors on older versions of Postgres where the settings are not
  96         # implemented.
  97         self.perform(
  98             """ UPDATE pg_settings SET setting = -1 WHERE name = 'jit_above_cost';
  99                 UPDATE pg_settings SET setting = 0
 100                    WHERE name = 'max_parallel_workers_per_gather';""")
 101         self.wait()
 102
 103     def _deadlock_handler(self):
 104         LOG.info("Deadlock detected (params = %s), retry.", str(self.current_params))
 105         self.cursor.execute(self.current_query, self.current_params)
 106
 107     def wait(self):
 108         """ Block until any pending operation is done.
 109         """
 110         while True:
 111             with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors):
 112                 wait_select(self.conn)
 113                 self.current_query = None
 114                 return
 115
 116     def perform(self, sql, args=None):
 117         """ Send SQL query to the server. Returns immediately without
 118             blocking.
 119         """
 120         self.current_query = sql
 121         self.current_params = args
 122         self.cursor.execute(sql, args)
 123
 124     def fileno(self):
 125         """ File descriptor to wait for. (Makes this class select()able.)
 126         """
 127         return self.conn.fileno()
 128
 129     def is_done(self):
 130         """ Check if the connection is available for a new query.
 131
 132             Also checks if the previous query has run into a deadlock.
 133             If so, then the previous query is repeated.
 134         """
 135         if self.current_query is None:
 136             return True
 137
 138         with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors):
 139             if self.conn.poll() == psycopg2.extensions.POLL_OK:
 140                 self.current_query = None
 141                 return True
 142
 143         return False
 144
 145
 146 class WorkerPool:
 147     """ A pool of asynchronous database connections.
 148
 149         The pool may be used as a context manager.
 150     """
 151     REOPEN_CONNECTIONS_AFTER = 100000
 152
 153     def __init__(self, dsn, pool_size, ignore_sql_errors=False):
 154         self.threads = [DBConnection(dsn, ignore_sql_errors=ignore_sql_errors)
 155                         for _ in range(pool_size)]
 156         self.free_workers = self._yield_free_worker()
 157         self.wait_time = 0
 158
 159
 160     def finish_all(self):
 161         """ Wait for all connection to finish.
 162         """
 163         for thread in self.threads:
 164             while not thread.is_done():
 165                 thread.wait()
 166
 167         self.free_workers = self._yield_free_worker()
 168
 169     def close(self):
 170         """ Close all connections and clear the pool.
 171         """
 172         for thread in self.threads:
 173             thread.close()
 174         self.threads = []
 175         self.free_workers = None
 176
 177
 178     def next_free_worker(self):
 179         """ Get the next free connection.
 180         """
 181         return next(self.free_workers)
 182
 183
 184     def _yield_free_worker(self):
 185         ready = self.threads
 186         command_stat = 0
 187         while True:
 188             for thread in ready:
 189                 if thread.is_done():
 190                     command_stat += 1
 191                     yield thread
 192
 193             if command_stat > self.REOPEN_CONNECTIONS_AFTER:
 194                 self._reconnect_threads()
 195                 ready = self.threads
 196                 command_stat = 0
 197             else:
 198                 tstart = time.time()
 199                 _, ready, _ = select.select([], self.threads, [])
 200                 self.wait_time += time.time() - tstart
 201
 202
 203     def _reconnect_threads(self):
 204         for thread in self.threads:
 205             while not thread.is_done():
 206                 thread.wait()
 207             thread.connect()
 208
 209
 210     def __enter__(self):
 211         return self
 212
 213
 214     def __exit__(self, exc_type, exc_value, traceback):
 215         self.finish_all()
 216         self.close()