X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/b9a09129fa20a901b20f03ddf2d2547d289b7225..bb175cc95898de420781867973b38d033c187e81:/nominatim/db/async_connection.py diff --git a/nominatim/db/async_connection.py b/nominatim/db/async_connection.py index fe8b3006..a86c5bdc 100644 --- a/nominatim/db/async_connection.py +++ b/nominatim/db/async_connection.py @@ -28,22 +28,27 @@ class DeadlockHandler: normally. """ - def __init__(self, handler): + def __init__(self, handler, ignore_sql_errors=False): self.handler = handler + self.ignore_sql_errors = ignore_sql_errors def __enter__(self): - pass + return self def __exit__(self, exc_type, exc_value, traceback): if __has_psycopg2_errors__: if exc_type == psycopg2.errors.DeadlockDetected: # pylint: disable=E1101 self.handler() return True - else: - if exc_type == psycopg2.extensions.TransactionRollbackError: - if exc_value.pgcode == '40P01': - self.handler() - return True + elif exc_type == psycopg2.extensions.TransactionRollbackError \ + and exc_value.pgcode == '40P01': + self.handler() + return True + + if self.ignore_sql_errors and isinstance(exc_value, psycopg2.Error): + LOG.info("SQL error ignored: %s", exc_value) + return True + return False @@ -51,10 +56,11 @@ class DBConnection: """ A single non-blocking database connection. """ - def __init__(self, dsn, cursor_factory=None): + def __init__(self, dsn, cursor_factory=None, ignore_sql_errors=False): self.current_query = None self.current_params = None self.dsn = dsn + self.ignore_sql_errors = ignore_sql_errors self.conn = None self.cursor = None @@ -79,7 +85,7 @@ class DBConnection: # Use a dict to hand in the parameters because async is a reserved # word in Python3. - self.conn = psycopg2.connect(**{'dsn' : self.dsn, 'async' : True}) + self.conn = psycopg2.connect(**{'dsn': self.dsn, 'async': True}) self.wait() self.cursor = self.conn.cursor(cursor_factory=cursor_factory) @@ -101,7 +107,7 @@ class DBConnection: """ Block until any pending operation is done. """ while True: - with DeadlockHandler(self._deadlock_handler): + with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors): wait_select(self.conn) self.current_query = None return @@ -128,7 +134,7 @@ class DBConnection: if self.current_query is None: return True - with DeadlockHandler(self._deadlock_handler): + with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors): if self.conn.poll() == psycopg2.extensions.POLL_OK: self.current_query = None return True @@ -143,8 +149,9 @@ class WorkerPool: """ REOPEN_CONNECTIONS_AFTER = 100000 - def __init__(self, dsn, pool_size): - self.threads = [DBConnection(dsn) for _ in range(pool_size)] + def __init__(self, dsn, pool_size, ignore_sql_errors=False): + self.threads = [DBConnection(dsn, ignore_sql_errors=ignore_sql_errors) + for _ in range(pool_size)] self.free_workers = self._yield_free_worker() self.wait_time = 0 @@ -183,10 +190,7 @@ class WorkerPool: yield thread if command_stat > self.REOPEN_CONNECTIONS_AFTER: - for thread in self.threads: - while not thread.is_done(): - thread.wait() - thread.connect() + self._reconnect_threads() ready = self.threads command_stat = 0 else: @@ -195,6 +199,13 @@ class WorkerPool: self.wait_time += time.time() - tstart + def _reconnect_threads(self): + for thread in self.threads: + while not thread.is_done(): + thread.wait() + thread.connect() + + def __enter__(self): return self