diff --git a/CHANGELOG.md b/CHANGELOG.md index ba328bd9a..58fdb09d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # **Upcoming release** +- #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 35d497337..b8a1f9d64 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -853,7 +853,12 @@ 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. 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..d8ef55a73 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,41 @@ 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: + with contextlib.suppress(sqlite3.ProgrammingError, sqlite3.OperationalError): + conn.commit() + with contextlib.suppress(sqlite3.ProgrammingError, sqlite3.OperationalError): + conn.close() + + def _close_finished_thread_connections(self) -> None: + """Close connections whose owning threads have exited. + + The caller must hold ``_connections_lock``. + """ + for thread, conn in list(self._connections.items()): + if not thread.is_alive(): + del self._connections[thread] + self._commit_and_close(conn) + + 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") + self._close_finished_thread_connections() + self._connections[thread] = conn @property def connection(self) -> sqlite3.Connection: @@ -198,16 +234,31 @@ 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 + 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 +508,55 @@ 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. + """ + thread = current_thread() + with self._connections_lock: + if self._closed: + return + self._close_finished_thread_connections() + 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 + + for conn in connections: + self._commit_and_close(conn) + + 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..77bfde41f 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 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,167 @@ 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") + + +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 + + +def test_connection_setter_replaces_existing(project: Project): + with AutoImport(project, memory=True) as ai: + old_conn = ai.connection + assert old_conn in ai._connections.values() + with closing(AutoImport.create_database_connection(memory=True)) as conn: + ai.connection = conn + assert ai.connection is conn + assert conn in ai._connections.values() + assert old_conn not in ai._connections.values() 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()