]> git.openstreetmap.org Git - nominatim.git/blob - nominatim/db/async_connection.py
add Python package configuration
[nominatim.git] / 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 """ 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 nominatim.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()