diff --git a/CHANGELOG.md b/CHANGELOG.md index 88e4748e2..b44638dd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # **Upcoming release** - #859 Fix `find_definition`/Rename not resolving `async def` methods called via `self` (@gadievron) +- #886 Track and close SQLite connections created in worker threads in AutoImport (@mcepl) - #895 patchedast Starred and keyword now consumes their syntactically expected and ** (@lieryan) - #896 patchedast cleanup and refactoring (@lieryan) - #905 Ignore `node_modules` directories by default when scanning project files and source folders. (@marcoeangeli) diff --git a/docs/library.rst b/docs/library.rst index 6a62c3e00..68beb0663 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -853,7 +853,18 @@ by passing memory=False to AutoImport constructor. This default will change in the future, if you want to always store the autoimport cache in-memory, then you need to explicitly pass memory=True. -It must be closed when done with the ``AutoImport.close()`` method. +It must be closed when done with the ``AutoImport.close()`` method. When +sharing an instance across threads, wait for worker threads to finish before +calling ``close()``. It raises ``RopeError`` rather than closing a connection +owned by a live worker thread. A worker can close its own connection sooner +with ``close_thread_connection()``. Connections left by finished workers are +closed when another thread creates a connection or when ``close()`` runs. +If committing or closing a connection fails, ``close()`` attempts to close +all remaining connections before raising the first SQLite error. + +Assigning to ``AutoImport.connection`` replaces the current thread's +connection without closing the old one. The caller must close the old +connection; ``AutoImport`` only tracks and closes the replacement. AutoImport can search for a name from both modules and statements you can import from them. diff --git a/rope/contrib/autoimport/sqlite.py b/rope/contrib/autoimport/sqlite.py index 7ae8fb718..a261e3201 100644 --- a/rope/contrib/autoimport/sqlite.py +++ b/rope/contrib/autoimport/sqlite.py @@ -15,9 +15,10 @@ from hashlib import sha256 from itertools import chain from pathlib import Path -from threading import local +from threading import Lock, Thread, current_thread, local from typing import ( TYPE_CHECKING, + Dict, Generator, Iterable, Iterator, @@ -142,6 +143,10 @@ def __init__( "`AutoImport(memory=True)` explicitly.", DeprecationWarning, ) + self._closed = False + self._connections: Dict[Thread, sqlite3.Connection] = {} + self._connections_lock = Lock() + self._observer: Optional[resourceobserver.ResourceObserver] = None self.thread_local = local() self.connection = self.create_database_connection( project=project, @@ -149,10 +154,10 @@ def __init__( ) self._setup_db() if observe: - observer = resourceobserver.ResourceObserver( + self._observer = resourceobserver.ResourceObserver( changed=self._changed, moved=self._moved, removed=self._removed ) - project.add_observer(observer) + project.add_observer(self._observer) @classmethod def create_database_connection( @@ -186,10 +191,60 @@ def calculate_project_hash(data: str) -> str: else: project_hash = calculate_project_hash(project.ropefolder.real_path) return sqlite3.connect( - f"file:rope-{project_hash}:?mode=memory&cache=shared", uri=True + f"file:rope-{project_hash}:?mode=memory&cache=shared", + uri=True, + check_same_thread=False, ) else: - return sqlite3.connect(project.ropefolder.pathlib / "autoimport.db") + return sqlite3.connect( + project.ropefolder.pathlib / "autoimport.db", + check_same_thread=False, + ) + + @staticmethod + def _commit_and_close(conn: sqlite3.Connection) -> None: + try: + conn.commit() + except sqlite3.Error: + with contextlib.suppress(sqlite3.Error): + conn.close() + raise + else: + conn.close() + + @classmethod + def _close_connections(cls, connections: Iterable[sqlite3.Connection]) -> None: + first_error: Optional[sqlite3.Error] = None + for conn in connections: + try: + cls._commit_and_close(conn) + except sqlite3.Error as error: + if first_error is None: + first_error = error + if first_error is not None: + raise first_error + + def _close_finished_thread_connections(self) -> None: + """Close finished threads' connections while holding the registry lock.""" + connections = [] + for thread, conn in list(self._connections.items()): + if not thread.is_alive(): + del self._connections[thread] + connections.append(conn) + self._close_connections(connections) + + def _register_connection(self, conn: sqlite3.Connection) -> None: + thread = current_thread() + with self._connections_lock: + if self._closed: + self._commit_and_close(conn) + raise exceptions.RopeError("AutoImport instance has been closed") + try: + self._close_finished_thread_connections() + except sqlite3.Error: + conn.close() + raise + self._connections[thread] = conn @property def connection(self) -> sqlite3.Connection: @@ -198,16 +253,32 @@ def connection(self) -> sqlite3.Connection: This makes sure AutoImport can be shared across threads. """ + if self._closed: + raise exceptions.RopeError("AutoImport instance has been closed") if not hasattr(self.thread_local, "connection"): - self.thread_local.connection = self.create_database_connection( + conn = self.create_database_connection( project=self.project, memory=self.memory, ) + self._register_connection(conn) + self.thread_local.connection = conn return self.thread_local.connection @connection.setter def connection(self, value: sqlite3.Connection): - self.thread_local.connection = value + """Replace this thread's connection; the caller must close the old one.""" + thread = current_thread() + with self._connections_lock: + if self._closed: + self._commit_and_close(value) + raise exceptions.RopeError("AutoImport instance has been closed") + self._close_finished_thread_connections() + old_conn = getattr(self.thread_local, "connection", None) + if old_conn is not None and old_conn is not value: + if self._connections.get(thread) is old_conn: + del self._connections[thread] + self._connections[thread] = value + self.thread_local.connection = value def _setup_db(self): models.Metadata.create_table(self.connection) @@ -457,10 +528,54 @@ def update_module(self, module: str): self._del_if_exist(module) self.generate_modules_cache([module]) + def close_thread_connection(self): + """Close the SQLite connection owned by the current thread.""" + conn = getattr(self.thread_local, "connection", None) + if conn is not None: + thread = current_thread() + with self._connections_lock: + if self._connections.get(thread) is conn: + del self._connections[thread] + with contextlib.suppress(AttributeError): + del self.thread_local.connection + self._commit_and_close(conn) + def close(self): - """Close the autoimport database.""" - self.connection.commit() - self.connection.close() + """Close the database after all worker threads have stopped. + + Raises ``RopeError`` if another live thread still owns a connection. + SQLite errors are raised after attempting to close every connection. + """ + thread = current_thread() + with self._connections_lock: + if self._closed: + return + active_threads = [ + owner + for owner in self._connections + if owner is not thread and owner.is_alive() + ] + if active_threads: + raise exceptions.RopeError( + "Cannot close AutoImport while other threads own connections; " + "wait for them to finish first" + ) + self._closed = True + connections = list(self._connections.values()) + self._connections.clear() + + if self._observer is not None: + with contextlib.suppress(Exception): + self.project.remove_observer(self._observer) + self._observer = None + + self._close_connections(connections) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() def get_name_locations(self, name): """Return a list of ``(resource, lineno)`` tuples.""" diff --git a/ropetest/contrib/autoimport/autoimporttest.py b/ropetest/contrib/autoimport/autoimporttest.py index 072ab1ed8..8ef474c19 100644 --- a/ropetest/contrib/autoimport/autoimporttest.py +++ b/ropetest/contrib/autoimport/autoimporttest.py @@ -1,11 +1,13 @@ import sqlite3 from concurrent.futures import ThreadPoolExecutor from contextlib import closing, contextmanager +from threading import Barrier, Event from textwrap import dedent from unittest.mock import ANY, patch import pytest +from rope.base import exceptions from rope.base.project import Project from rope.base.resources import File, Folder from rope.contrib.autoimport import models @@ -46,16 +48,16 @@ def test_autoimport_connection_parameter_with_in_memory( project: Project, autoimport: AutoImport, ): - connection = AutoImport.create_database_connection(memory=True) - assert is_in_memory_database(connection) + with closing(AutoImport.create_database_connection(memory=True)) as connection: + assert is_in_memory_database(connection) def test_autoimport_connection_parameter_with_project( project: Project, autoimport: AutoImport, ): - connection = AutoImport.create_database_connection(project=project) - assert not is_in_memory_database(connection) + with closing(AutoImport.create_database_connection(project=project)) as connection: + assert not is_in_memory_database(connection) def test_autoimport_create_database_connection_conflicting_parameter( @@ -102,25 +104,314 @@ def foo(): def test_multithreading( - autoimport: AutoImport, project: Project, pkg1: Folder, mod1: File, ): mod1_init = pkg1.get_child("__init__.py") - mod1_init.write(dedent("""\ + mod1_init.write( + dedent("""\ def foo(): pass - """)) - mod1.write(dedent("""\ + """) + ) + mod1.write( + dedent("""\ foo - """)) - autoimport = AutoImport(project, memory=False) - autoimport.generate_cache([mod1_init]) + """) + ) + with closing(AutoImport(project, memory=False)) as autoimport: + autoimport.generate_cache([mod1_init]) - tp = ThreadPoolExecutor(1) - results = tp.submit(autoimport.search, "foo", True).result() - assert [("from pkg1 import foo", "foo")] == results + with ThreadPoolExecutor(1) as tp: + results = tp.submit(autoimport.search, "foo", True).result() + assert [("from pkg1 import foo", "foo")] == results + + +def test_multithread_connections_closed_on_close(project: Project): + with AutoImport(project, memory=True) as ai: + main_conn = ai.connection + worker_conns = [] + + def worker(): + conn = ai.connection + worker_conns.append(conn) + return list(ai.search("foo")) + + with ThreadPoolExecutor(3) as tp: + futures = [tp.submit(worker) for _ in range(3)] + for f in futures: + f.result() + + all_conns = {main_conn} | set(worker_conns) + assert len(all_conns) > 1 + assert all_conns.issubset(ai._connections.values()) + + assert ai._closed + assert len(ai._connections) == 0 + for conn in all_conns: + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") + + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + _ = ai.connection + + +def test_finished_thread_connections_are_reaped(project: Project): + ai = AutoImport(project, memory=True) + worker_connections = [] + + for _ in range(3): + with ThreadPoolExecutor(1) as executor: + worker_connections.append(executor.submit(lambda: ai.connection).result()) + assert len(ai._connections) <= 2 + + for conn in worker_connections[:-1]: + with pytest.raises( + sqlite3.ProgrammingError, match="Cannot operate on a closed database" + ): + conn.execute("SELECT 1") + + ai.close() + for conn in worker_connections: + with pytest.raises( + sqlite3.ProgrammingError, match="Cannot operate on a closed database" + ): + conn.execute("SELECT 1") + + +def test_close_rejects_live_worker_connections(project: Project): + ai = AutoImport(project, memory=True) + started = Event() + finish = Event() + worker_connections = [] + + def worker(): + worker_connections.append(ai.connection) + started.set() + finish.wait() + + with ThreadPoolExecutor(1) as executor: + future = executor.submit(worker) + try: + assert started.wait(timeout=5) + with pytest.raises(exceptions.RopeError, match="wait for them to finish"): + ai.close() + assert not ai._closed + assert worker_connections[0].execute("SELECT 1").fetchone() == (1,) + finally: + finish.set() + future.result(timeout=5) + + ai.close() + assert ai._closed + + +def test_close_thread_connection(project: Project): + with AutoImport(project, memory=True) as ai: + worker_conn = None + + def worker(): + nonlocal worker_conn + worker_conn = ai.connection + assert worker_conn in ai._connections.values() + ai.close_thread_connection() + assert worker_conn not in ai._connections.values() + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + worker_conn.execute("SELECT 1") + new_conn = ai.connection + assert new_conn is not worker_conn + assert new_conn in ai._connections.values() + + with ThreadPoolExecutor(1) as tp: + tp.submit(worker).result() + + +def test_close_idempotent(project: Project): + ai = AutoImport(project, memory=True) + conn = ai.connection + ai.close() + assert ai._closed + ai.close() + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") + + +@pytest.mark.parametrize( + "fail_commit, fail_close", [(True, False), (False, True), (True, True)] +) +def test_close_reports_errors_after_closing_all_connections( + project: Project, fail_commit: bool, fail_close: bool +): + commit_error = sqlite3.OperationalError("commit failed") + close_error = sqlite3.OperationalError("close failed") + + class FailingConnection(sqlite3.Connection): + def commit(self): + if fail_commit: + raise commit_error + super().commit() + + def close(self): + super().close() + if fail_close: + raise close_error + + with AutoImport(project, memory=True) as ai: + with closing(ai.connection): + conn = sqlite3.connect(":memory:", factory=FailingConnection) + ai.connection = conn + with ThreadPoolExecutor(1) as executor: + worker_conn = executor.submit(lambda: ai.connection).result(timeout=5) + + with pytest.raises(sqlite3.OperationalError) as raised: + ai.close() + assert raised.value is (commit_error if fail_commit else close_error) + assert ai._closed + assert not ai._connections + assert ai._observer is None + for connection in (conn, worker_conn): + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + connection.execute("SELECT 1") + ai.close() + + +def test_registration_closes_new_connection_when_reaping_fails(project: Project): + error = sqlite3.OperationalError("commit failed") + + class FailingConnection(sqlite3.Connection): + def commit(self): + raise error + + with AutoImport(project, memory=True) as ai: + failing_conn = sqlite3.connect( + ":memory:", factory=FailingConnection, check_same_thread=False + ) + ready = Barrier(3, timeout=5) + + def worker(connection=None): + if connection is not None: + ai.connection = connection + conn = ai.connection + ready.wait() + return conn + + with ThreadPoolExecutor(2) as executor: + failing_future = executor.submit(worker, failing_conn) + other_future = executor.submit(worker) + ready.wait() + assert failing_future.result(timeout=5) is failing_conn + other_conn = other_future.result(timeout=5) + + created = [] + create_connection = ai.create_database_connection + + def create(**kwargs): + conn = create_connection(**kwargs) + created.append(conn) + return conn + + with patch.object(ai, "create_database_connection", side_effect=create): + with ThreadPoolExecutor(1) as executor: + future = executor.submit(lambda: ai.connection) + with pytest.raises(sqlite3.OperationalError) as raised: + future.result(timeout=5) + assert raised.value is error + assert len(created) == 1 + for conn in [failing_conn, other_conn, *created]: + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + conn.execute("SELECT 1") + assert list(ai._connections.values()) == [ai.connection] + assert ai.connection.execute("SELECT 1").fetchone() == (1,) + + +@pytest.mark.parametrize("registration_wins", [True, False]) +def test_connection_registration_races_with_close( + project: Project, registration_wins: bool +): + paused = Event() + resume = Event() + created = [] + + with AutoImport(project, memory=True) as ai: + register_connection = ai._register_connection + + def register(conn): + created.append(conn) + if registration_wins: + register_connection(conn) + paused.set() + assert resume.wait(timeout=5) + if not registration_wins: + register_connection(conn) + + def worker(): + return ai.connection.execute("SELECT 1").fetchone() + + with patch.object(ai, "_register_connection", side_effect=register): + with ThreadPoolExecutor(1) as executor: + future = executor.submit(worker) + try: + assert paused.wait(timeout=5) + if registration_wins: + with pytest.raises( + exceptions.RopeError, match="wait for them to finish" + ): + ai.close() + assert not ai._closed + assert created[0] in ai._connections.values() + else: + ai.close() + assert ai._closed + assert not ai._connections + finally: + resume.set() + if registration_wins: + assert future.result(timeout=5) == (1,) + else: + with pytest.raises( + exceptions.RopeError, match="AutoImport instance has been closed" + ): + future.result(timeout=5) + assert len(created) == 1 + + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + created[0].execute("SELECT 1") + + +def test_register_connection_after_close(project: Project): + ai = AutoImport(project, memory=True) + ai.close() + conn = AutoImport.create_database_connection(memory=True) + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + ai._register_connection(conn) + assert conn not in ai._connections.values() + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") + + +def test_connection_setter_after_close(project: Project): + ai = AutoImport(project, memory=True) + ai.close() + with closing(AutoImport.create_database_connection(memory=True)) as conn: + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + ai.connection = conn + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + conn.execute("SELECT 1") + + +def test_connection_setter_replaces_existing(project: Project): + with AutoImport(project, memory=True) as ai: + with closing(ai.connection) as old_conn: + assert old_conn in ai._connections.values() + conn = AutoImport.create_database_connection(memory=True) + ai.connection = conn + assert ai.connection is conn + assert conn in ai._connections.values() + assert old_conn not in ai._connections.values() + assert old_conn.execute("SELECT 1").fetchone() == (1,) + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + conn.execute("SELECT 1") def test_connection(project: Project, project2: Project): diff --git a/ropetest/contrib/autoimport/modeltest.py b/ropetest/contrib/autoimport/modeltest.py index 0044434cc..8eef06d35 100644 --- a/ropetest/contrib/autoimport/modeltest.py +++ b/ropetest/contrib/autoimport/modeltest.py @@ -8,7 +8,9 @@ @pytest.fixture def empty_db(): - return sqlite3.connect(":memory:") + conn = sqlite3.connect(":memory:") + yield conn + conn.close() class TestQuery: diff --git a/ropetest/contrib/autoimporttest.py b/ropetest/contrib/autoimporttest.py index d42ec14f7..ebf21af56 100644 --- a/ropetest/contrib/autoimporttest.py +++ b/ropetest/contrib/autoimporttest.py @@ -15,6 +15,7 @@ def setUp(self): self.importer = autoimport.AutoImport(self.project, observe=False) def tearDown(self): + self.importer.close() testutils.remove_project(self.project) super().tearDown() @@ -178,12 +179,12 @@ def test_skipping_directories_not_accessible_because_of_permission_error(self): def test_search_submodule(project, external_fixturepkg): - importer = autoimport.AutoImport(project, observe=False) - importer.update_module("external_fixturepkg") - import_statement = ("from external_fixturepkg import mod1", "mod1") - assert import_statement in importer.search("mod1", exact_match=True) - assert import_statement in importer.search("mo") - assert import_statement in importer.search("mod1") + with autoimport.AutoImport(project, observe=False) as importer: + importer.update_module("external_fixturepkg") + import_statement = ("from external_fixturepkg import mod1", "mod1") + assert import_statement in importer.search("mod1", exact_match=True) + assert import_statement in importer.search("mo") + assert import_statement in importer.search("mod1") class AutoImportObservingTest(unittest.TestCase): @@ -196,6 +197,7 @@ def setUp(self): self.importer = autoimport.AutoImport(self.project, observe=True) def tearDown(self): + self.importer.close() testutils.remove_project(self.project) super().tearDown()