src/nominatim_db/db/async_connection.py

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