Nominatim/nominatim/db/async_connection.py

# SPDX-License-Identifier: GPL-2.0-only
#
# This file is part of Nominatim. (https://nominatim.org)
#
# Copyright (C) 2022 by the Nominatim developer community.
# For a full list of authors see the git log.
""" Non-blocking database connections.
"""
from typing import Callable, Any, Optional, Iterator, Sequence
import logging
import select
import time

import psycopg2
from psycopg2.extras import wait_select

# psycopg2 emits different exceptions pre and post 2.8. Detect if the new error
# module is available and adapt the error handling accordingly.
try:
    import psycopg2.errors # pylint: disable=no-name-in-module,import-error
    __has_psycopg2_errors__ = True
except ImportError:
    __has_psycopg2_errors__ = False

from nominatim.typing import T_cursor, Query

LOG = logging.getLogger()

class DeadlockHandler:
    """ Context manager that catches deadlock exceptions and calls
        the given handler function. All other exceptions are passed on
        normally.
    """

    def __init__(self, handler: Callable[[], None], ignore_sql_errors: bool = False) -> None:
        self.handler = handler
        self.ignore_sql_errors = ignore_sql_errors

    def __enter__(self) -> 'DeadlockHandler':
        return self

    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> bool:
        if __has_psycopg2_errors__:
            if exc_type == psycopg2.errors.DeadlockDetected: # pylint: disable=E1101
                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


class DBConnection:
    """ A single non-blocking database connection.
    """

    def __init__(self, dsn: str,
                 cursor_factory: Optional[Callable[..., T_cursor]] = None,
                 ignore_sql_errors: bool = False) -> None:
        self.dsn = dsn

        self.current_query: Optional[Query] = None
        self.current_params: Optional[Sequence[Any]] = None
        self.ignore_sql_errors = ignore_sql_errors

        self.conn: Optional['psycopg2._psycopg.connection'] = None
        self.cursor: Optional['psycopg2._psycopg.cursor'] = None
        self.connect(cursor_factory=cursor_factory)

    def close(self) -> None:
        """ Close all open connections. Does not wait for pending requests.
        """
        if self.conn is not None:
            if self.cursor is not None:
                self.cursor.close()
                self.cursor = None
            self.conn.close()

        self.conn = None

    def connect(self, cursor_factory: Optional[Callable[..., T_cursor]] = None) -> None:
        """ (Re)connect to the database. Creates an asynchronous connection
            with JIT and parallel processing disabled. If a connection was
            already open, it is closed and a new connection established.
            The caller must ensure that no query is pending before reconnecting.
        """
        self.close()

        # 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}) # type: ignore
        assert self.conn
        self.wait()

        if cursor_factory is not None:
            self.cursor = self.conn.cursor(cursor_factory=cursor_factory)
        else:
            self.cursor = self.conn.cursor()
        # Disable JIT and parallel workers as they are known to cause problems.
        # Update pg_settings instead of using SET because it does not yield
        # errors on older versions of Postgres where the settings are not
        # implemented.
        self.perform(
            """ UPDATE pg_settings SET setting = -1 WHERE name = 'jit_above_cost';
                UPDATE pg_settings SET setting = 0
                   WHERE name = 'max_parallel_workers_per_gather';""")
        self.wait()

    def _deadlock_handler(self) -> None:
        LOG.info("Deadlock detected (params = %s), retry.", str(self.current_params))
        assert self.cursor is not None
        assert self.current_query is not None
        assert self.current_params is not None

        self.cursor.execute(self.current_query, self.current_params)

    def wait(self) -> None:
        """ Block until any pending operation is done.
        """
        while True:
            with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors):
                wait_select(self.conn)
                self.current_query = None
                return

    def perform(self, sql: Query, args: Optional[Sequence[Any]] = None) -> None:
        """ Send SQL query to the server. Returns immediately without
            blocking.
        """
        assert self.cursor is not None
        self.current_query = sql
        self.current_params = args
        self.cursor.execute(sql, args)

    def fileno(self) -> int:
        """ File descriptor to wait for. (Makes this class select()able.)
        """
        assert self.conn is not None
        return self.conn.fileno()

    def is_done(self) -> bool:
        """ Check if the connection is available for a new query.

            Also checks if the previous query has run into a deadlock.
            If so, then the previous query is repeated.
        """
        assert self.conn is not None

        if self.current_query is None:
            return True

        with DeadlockHandler(self._deadlock_handler, self.ignore_sql_errors):
            if self.conn.poll() == psycopg2.extensions.POLL_OK:
                self.current_query = None
                return True

        return False


class WorkerPool:
    """ A pool of asynchronous database connections.

        The pool may be used as a context manager.
    """
    REOPEN_CONNECTIONS_AFTER = 100000

    def __init__(self, dsn: str, pool_size: int, ignore_sql_errors: bool = False) -> None:
        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.0


    def finish_all(self) -> None:
        """ Wait for all connection to finish.
        """
        for thread in self.threads:
            while not thread.is_done():
                thread.wait()

        self.free_workers = self._yield_free_worker()

    def close(self) -> None:
        """ Close all connections and clear the pool.
        """
        for thread in self.threads:
            thread.close()
        self.threads = []
        self.free_workers = iter([])


    def next_free_worker(self) -> DBConnection:
        """ Get the next free connection.
        """
        return next(self.free_workers)


    def _yield_free_worker(self) -> Iterator[DBConnection]:
        ready = self.threads
        command_stat = 0
        while True:
            for thread in ready:
                if thread.is_done():
                    command_stat += 1
                    yield thread

            if command_stat > self.REOPEN_CONNECTIONS_AFTER:
                self._reconnect_threads()
                ready = self.threads
                command_stat = 0
            else:
                tstart = time.time()
                _, ready, _ = select.select([], self.threads, [])
                self.wait_time += time.time() - tstart


    def _reconnect_threads(self) -> None:
        for thread in self.threads:
            while not thread.is_done():
                thread.wait()
            thread.connect()


    def __enter__(self) -> 'WorkerPool':
        return self


    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
        self.finish_all()
        self.close()