Merge pull request #7150 from drew2a/fix/run_threaded

Extract the `run_threaded` function to the `pony_utils.py`

scripts/seedbox/disseminator.py

@@ -159,7 +159,7 @@ def commit(self):
     def flush(self):
         _logger.debug('Flush')
 
-        self.community.mds._db.flush()  # pylint: disable=protected-access
+        self.community.mds.db.flush()
 
 
 class Service(TinyTriblerService):

src/tribler/core/components/gigachannel_manager/gigachannel_manager.py

@@ -14,6 +14,7 @@
 from tribler.core.components.metadata_store.db.serialization import CHANNEL_TORRENT
 from tribler.core.components.metadata_store.db.store import MetadataStore
 from tribler.core.utilities.notifier import Notifier
+from tribler.core.utilities.pony_utils import run_threaded
 from tribler.core.utilities.simpledefs import DLSTATUS_SEEDING, NTFY
 from tribler.core.utilities.unicode import hexlify
 
@@ -283,7 +284,7 @@ def _process_download():
             mds.process_channel_dir(channel_dirname, channel.public_key, channel.id_, external_thread=True)
 
         try:
-            await mds.run_threaded(_process_download)
+            await run_threaded(mds.db, _process_download)
         except Exception as e:  # pylint: disable=broad-except  # pragma: no cover
             self._logger.error("Error when processing channel dir download: %s", e)
 

src/tribler/core/components/metadata_store/db/store.py

@@ -1,13 +1,10 @@
 import logging
 import re
-import threading
-from asyncio import get_event_loop
 from datetime import datetime, timedelta
 from time import sleep, time
 from typing import Optional, Union
 
 from lz4.frame import LZ4FrameDecompressor
-
 from pony import orm
 from pony.orm import db_session, desc, left_join, raw_sql, select
 from pony.orm.dbproviders.sqlite import keep_exception
@@ -50,12 +47,11 @@
 from tribler.core.exceptions import InvalidSignatureException
 from tribler.core.utilities.notifier import Notifier
 from tribler.core.utilities.path_util import Path
-from tribler.core.utilities.pony_utils import get_max, get_or_create
+from tribler.core.utilities.pony_utils import get_max, get_or_create, run_threaded
 from tribler.core.utilities.search_utils import torrent_rank
 from tribler.core.utilities.unicode import hexlify
 from tribler.core.utilities.utilities import MEMORY_DB
 
-
 BETA_DB_VERSIONS = [0, 1, 2, 3, 4, 5]
 CURRENT_DB_VERSION = 14
 
@@ -164,12 +160,12 @@ def __init__(
         # We have to dynamically define/init ORM-managed entities here to be able to support
         # multiple sessions in Tribler. ORM-managed classes are bound to the database instance
         # at definition.
-        self._db = orm.Database()
+        self.db = orm.Database()
 
         # This attribute is internally called by Pony on startup, though pylint cannot detect it
         # with the static analysis.
         # pylint: disable=unused-variable
-        @self._db.on_connect(provider='sqlite')
+        @self.db.on_connect(provider='sqlite')
         def on_connect(_, connection):
             cursor = connection.cursor()
             cursor.execute("PRAGMA journal_mode = WAL")
@@ -189,31 +185,31 @@ def on_connect(_, connection):
 
             # pylint: enable=unused-variable
 
-        self.MiscData = misc.define_binding(self._db)
+        self.MiscData = misc.define_binding(self.db)
 
-        self.TrackerState = tracker_state.define_binding(self._db)
-        self.TorrentState = torrent_state.define_binding(self._db)
+        self.TrackerState = tracker_state.define_binding(self.db)
+        self.TorrentState = torrent_state.define_binding(self.db)
 
-        self.ChannelNode = channel_node.define_binding(self._db, logger=self._logger, key=my_key)
+        self.ChannelNode = channel_node.define_binding(self.db, logger=self._logger, key=my_key)
 
-        self.MetadataNode = metadata_node.define_binding(self._db)
-        self.CollectionNode = collection_node.define_binding(self._db)
+        self.MetadataNode = metadata_node.define_binding(self.db)
+        self.CollectionNode = collection_node.define_binding(self.db)
         self.TorrentMetadata = torrent_metadata.define_binding(
-            self._db,
+            self.db,
             notifier=notifier,
             tag_processor_version=tag_processor_version
         )
-        self.ChannelMetadata = channel_metadata.define_binding(self._db)
+        self.ChannelMetadata = channel_metadata.define_binding(self.db)
 
-        self.JsonNode = json_node.define_binding(self._db, db_version)
-        self.ChannelDescription = channel_description.define_binding(self._db)
+        self.JsonNode = json_node.define_binding(self.db, db_version)
+        self.ChannelDescription = channel_description.define_binding(self.db)
 
-        self.BinaryNode = binary_node.define_binding(self._db, db_version)
-        self.ChannelThumbnail = channel_thumbnail.define_binding(self._db)
+        self.BinaryNode = binary_node.define_binding(self.db, db_version)
+        self.ChannelThumbnail = channel_thumbnail.define_binding(self.db)
 
-        self.ChannelVote = channel_vote.define_binding(self._db)
-        self.ChannelPeer = channel_peer.define_binding(self._db)
-        self.Vsids = vsids.define_binding(self._db)
+        self.ChannelVote = channel_vote.define_binding(self.db)
+        self.ChannelPeer = channel_peer.define_binding(self.db)
+        self.Vsids = vsids.define_binding(self.db)
 
         self.ChannelMetadata._channels_dir = channels_dir  # pylint: disable=protected-access
 
@@ -224,13 +220,13 @@ def on_connect(_, connection):
             create_db = not db_filename.is_file()
             db_path_string = str(db_filename)
 
-        self._db.bind(provider='sqlite', filename=db_path_string, create_db=create_db, timeout=120.0)
-        self._db.generate_mapping(
+        self.db.bind(provider='sqlite', filename=db_path_string, create_db=create_db, timeout=120.0)
+        self.db.generate_mapping(
             create_tables=create_db, check_tables=check_tables
         )  # Must be run out of session scope
         if create_db:
             with db_session(ddl=True):
-                self._db.execute(sql_create_fts_table)
+                self.db.execute(sql_create_fts_table)
                 self.create_fts_triggers()
                 self.create_torrentstate_triggers()
                 self.create_partial_indexes()
@@ -245,15 +241,6 @@ def on_connect(_, connection):
                 default_vsids = self.Vsids.create_default_vsids()
             self.ChannelMetadata.votes_scaling = default_vsids.max_val
 
-    async def run_threaded(self, func, *args, **kwargs):
-        def wrapper():
-            try:
-                return func(*args, **kwargs)
-            finally:
-                self.disconnect_thread()
-
-        return await get_event_loop().run_in_executor(None, wrapper)
-
     def set_value(self, key: str, value: str):
         key_value = get_or_create(self.MiscData, name=key)
         key_value.value = value
@@ -263,14 +250,14 @@ def get_value(self, key: str, default: Optional[str] = None) -> Optional[str]:
         return data.value if data else default
 
     def drop_indexes(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute("select name from sqlite_master where type='index' and name like 'idx_%'")
         for [index_name] in cursor.fetchall():
             cursor.execute(f"drop index {index_name}")
 
     def get_objects_to_create(self):
-        connection = self._db.get_connection()
-        schema = self._db.schema
+        connection = self.db.get_connection()
+        schema = self.db.schema
         provider = schema.provider
         created_tables = set()
         result = []
@@ -284,28 +271,28 @@ def get_db_file_size(self):
         return 0 if self.db_path is MEMORY_DB else Path(self.db_path).size()
 
     def drop_fts_triggers(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute("select name from sqlite_master where type='trigger' and name like 'fts_%'")
         for [trigger_name] in cursor.fetchall():
             cursor.execute(f"drop trigger {trigger_name}")
 
     def create_fts_triggers(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute(sql_add_fts_trigger_insert)
         cursor.execute(sql_add_fts_trigger_delete)
         cursor.execute(sql_add_fts_trigger_update)
 
     def fill_fts_index(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute("insert into FtsIndex(rowid, title) select rowid, title from ChannelNode")
 
     def create_torrentstate_triggers(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute(sql_add_torrentstate_trigger_after_insert)
         cursor.execute(sql_add_torrentstate_trigger_after_update)
 
     def create_partial_indexes(self):
-        cursor = self._db.get_connection().cursor()
+        cursor = self.db.get_connection().cursor()
         cursor.execute(sql_create_partial_index_channelnode_subscribed)
         cursor.execute(sql_create_partial_index_channelnode_metadata_type)
 
@@ -332,13 +319,7 @@ def vote_bump(self, public_key, id_, voter_pk):
 
     def shutdown(self):
         self._shutting_down = True
-        self._db.disconnect()
-
-    def disconnect_thread(self):
-        # Ugly workaround for closing threadpool connections
-        # Remark: maybe subclass ThreadPoolExecutor to handle this automatically?
-        if not isinstance(threading.current_thread(), threading._MainThread):  # pylint: disable=W0212
-            self._db.disconnect()
+        self.db.disconnect()
 
     @staticmethod
     def get_list_of_channel_blobs_to_process(dirname, start_timestamp):
@@ -467,7 +448,7 @@ def process_mdblob_file(self, filepath, **kwargs):
 
     async def process_compressed_mdblob_threaded(self, compressed_data, **kwargs):
         try:
-            return await self.run_threaded(self.process_compressed_mdblob, compressed_data, **kwargs)
+            return await run_threaded(self.db, self.process_compressed_mdblob, compressed_data, **kwargs)
         except Exception as e:  # pylint: disable=broad-except  # pragma: no cover
             self._logger.warning("DB transaction error when tried to process compressed mdblob: %s", str(e))
             return None
@@ -787,7 +768,7 @@ def get_entries_query(
         return pony_query
 
     async def get_entries_threaded(self, **kwargs):
-        return await self.run_threaded(self.get_entries, **kwargs)
+        return await run_threaded(self.db, self.get_entries, **kwargs)
 
     @db_session
     def get_entries(self, first=1, last=None, **kwargs):
@@ -838,7 +819,7 @@ def get_auto_complete_terms(self, text, max_terms, limit=10):
         suggestion_re = re.compile(suggestion_pattern, re.UNICODE)
 
         with db_session:
-            titles = self._db.select("""
+            titles = self.db.select("""
                 cn.title
                 FROM ChannelNode cn
                 INNER JOIN FtsIndex ON cn.rowid = FtsIndex.rowid

src/tribler/core/components/metadata_store/db/tests/test_store.py

@@ -6,12 +6,10 @@
 from datetime import datetime
 from unittest.mock import patch
 
+import pytest
 from ipv8.keyvault.crypto import default_eccrypto
-
 from pony.orm import db_session
 
-import pytest
-
 from tribler.core.components.metadata_store.db.orm_bindings.channel_metadata import (
     CHANNEL_DIR_NAME_LENGTH,
     entries_to_chunk,
@@ -29,8 +27,10 @@
 from tribler.core.components.metadata_store.tests.test_channel_download import CHANNEL_METADATA_UPDATED
 from tribler.core.tests.tools.common import TESTS_DATA_DIR
 from tribler.core.utilities.path_util import Path
+from tribler.core.utilities.pony_utils import run_threaded
 from tribler.core.utilities.utilities import random_infohash
 
+
 # pylint: disable=protected-access,unused-argument
 
 
@@ -269,13 +269,12 @@ def test_process_forbidden_payload(metadata_store):
 def test_process_payload(metadata_store):
     sender_key = default_eccrypto.generate_key("curve25519")
     for md_class in (
-        metadata_store.ChannelMetadata,
-        metadata_store.TorrentMetadata,
-        metadata_store.CollectionNode,
-        metadata_store.ChannelDescription,
-        metadata_store.ChannelThumbnail,
+            metadata_store.ChannelMetadata,
+            metadata_store.TorrentMetadata,
+            metadata_store.CollectionNode,
+            metadata_store.ChannelDescription,
+            metadata_store.ChannelThumbnail,
     ):
-
         node, node_payload, node_deleted_payload = get_payloads(md_class, sender_key)
         node_dict = node.to_dict()
         node.delete()
@@ -333,8 +332,8 @@ def test_process_payload_with_known_channel_public_key(metadata_store):
 
     # Check accepting a payload with matching public key
     assert (
-        metadata_store.process_payload(payload, channel_public_key=key1.pub().key_to_bin()[10:])[0].obj_state
-        == ObjState.NEW_OBJECT
+            metadata_store.process_payload(payload, channel_public_key=key1.pub().key_to_bin()[10:])[0].obj_state
+            == ObjState.NEW_OBJECT
     )
     assert metadata_store.TorrentMetadata.get()
 
@@ -465,8 +464,8 @@ def f1(a, b, *, c, d):
             return threading.get_ident()
         raise ThreadedTestException('test exception')
 
-    result = await metadata_store.run_threaded(f1, 1, 2, c=3, d=4)
+    result = await run_threaded(metadata_store.db, f1, 1, 2, c=3, d=4)
     assert result != thread_id
 
     with pytest.raises(ThreadedTestException, match='^test exception$'):
-        await metadata_store.run_threaded(f1, 1, 2, c=5, d=6)
+        await run_threaded(metadata_store.db, f1, 1, 2, c=5, d=6)

src/tribler/core/components/metadata_store/remote_query_community/remote_query_community.py

@@ -23,6 +23,7 @@
 from tribler.core.components.metadata_store.utils import RequestTimeoutException
 from tribler.core.components.knowledge.community.knowledge_validator import is_valid_resource
 from tribler.core.components.knowledge.db.knowledge_db import ResourceType
+from tribler.core.utilities.pony_utils import run_threaded
 from tribler.core.utilities.unicode import hexlify
 
 BINARY_FIELDS = ("infohash", "channel_pk")
@@ -213,12 +214,13 @@ async def process_rpc_query(self, sanitized_parameters: Dict[str, Any]) -> List:
         :raises ValueError: if no JSON could be decoded.
         :raises pony.orm.dbapiprovider.OperationalError: if an illegal query was performed.
         """
-        # tags should be extracted because `get_entries_threaded` doesn't expect them as a parameter
-        tags = sanitized_parameters.pop('tags', None)
+        if self.knowledge_db:
+            # tags should be extracted because `get_entries_threaded` doesn't expect them as a parameter
+            tags = sanitized_parameters.pop('tags', None)
 
-        infohash_set = await self.mds.run_threaded(self.search_for_tags, tags)
-        if infohash_set:
-            sanitized_parameters['infohash_set'] = {bytes.fromhex(s) for s in infohash_set}
+            infohash_set = await run_threaded(self.knowledge_db.instance, self.search_for_tags, tags)
+            if infohash_set:
+                sanitized_parameters['infohash_set'] = {bytes.fromhex(s) for s in infohash_set}
 
         return await self.mds.get_entries_threaded(**sanitized_parameters)
 

src/tribler/core/components/metadata_store/restapi/search_endpoint.py

@@ -14,6 +14,7 @@
 from tribler.core.components.metadata_store.restapi.metadata_schema import MetadataParameters, MetadataSchema
 from tribler.core.components.restapi.rest.rest_endpoint import HTTP_BAD_REQUEST, RESTResponse
 from tribler.core.components.knowledge.db.knowledge_db import ResourceType
+from tribler.core.utilities.pony_utils import run_threaded
 from tribler.core.utilities.utilities import froze_it
 
 SNIPPETS_TO_SHOW = 3          # The number of snippets we return from the search results
@@ -151,7 +152,7 @@ def search_db():
                     if infohash_set:
                         sanitized['infohash_set'] = {bytes.fromhex(s) for s in infohash_set}
 
-            search_results, total, max_rowid = await mds.run_threaded(search_db)
+            search_results, total, max_rowid = await run_threaded(mds.db, search_db)
         except Exception as e:  # pylint: disable=broad-except;  # pragma: no cover
             self._logger.exception("Error while performing DB search: %s: %s", type(e).__name__, e)
             return RESTResponse(status=HTTP_BAD_REQUEST)

src/tribler/core/components/metadata_store/tests/test_channel_metadata.py

@@ -5,14 +5,11 @@
 from pathlib import Path
 from unittest.mock import Mock, patch
 
+import pytest
 from ipv8.keyvault.crypto import default_eccrypto
-
 from lz4.frame import LZ4FrameDecompressor
-
 from pony.orm import ObjectNotFound, db_session
 
-import pytest
-
 from tribler.core.components.libtorrent.torrentdef import TorrentDef
 from tribler.core.components.metadata_store.db.orm_bindings.channel_metadata import (
     CHANNEL_DIR_NAME_LENGTH,
@@ -31,6 +28,7 @@
 from tribler.core.utilities.simpledefs import CHANNEL_STATE
 from tribler.core.utilities.utilities import random_infohash
 
+
 # pylint: disable=protected-access
 
 
@@ -80,7 +78,7 @@ def mds_with_some_torrents_fixture(metadata_store):
     #   torrent6 aaa zzz
 
     def save():
-        metadata_store._db.flush()  # pylint: disable=W0212
+        metadata_store.db.flush()
 
     def new_channel(**kwargs):
         params = dict(subscribed=True, share=True, status=NEW, infohash=random_infohash())