Merge pull request #7200 from drew2a/feature/refactor_tiny_tribler_se…

…rvice

Refactoring of  the `tiny_tribler_service.py`

scripts/experiments/tunnel_community/hidden_peer_discovery.py

@@ -11,30 +11,23 @@
 from tribler.core.components.key.key_component import KeyComponent
 from tribler.core.components.restapi.restapi_component import RESTComponent
 from tribler.core.components.tunnel.tunnel_component import TunnelsComponent
-from tribler.core.config.tribler_config import TriblerConfig
 from tribler.core.utilities.tiny_tribler_service import TinyTriblerService
-from tribler.core.utilities.utilities import make_async_loop_fragile
 
 EXPERIMENT_RUN_TIME = int(os.environ.get('EXPERIMENT_RUN_TIME', 3600 * 3))
 
 
 class Service(TinyTriblerService, TaskManager):
-    def __init__(self, working_dir, config_path):
-        super().__init__(Service.create_config(working_dir, config_path), working_dir=working_dir,
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs,
                          components=[Ipv8Component(), KeyComponent(), RESTComponent(), TunnelsComponent()])
         TaskManager.__init__(self)
+        self.config.dht.enabled = True
         self.swarm = None
         self.start = time.time()
         self.results = []
         self.register_task('monitor_swarm', self.monitor_swarm, interval=5)
         self.register_task('_graceful_shutdown', self._graceful_shutdown, delay=EXPERIMENT_RUN_TIME)
 
-    @staticmethod
-    def create_config(working_dir, config_path):
-        config = TriblerConfig(state_dir=working_dir, file=config_path)
-        config.dht.enabled = True
-        return config
-
     def _graceful_shutdown(self):
         task = asyncio.create_task(self.on_tribler_shutdown())
         task.add_done_callback(lambda result: TinyTriblerService._graceful_shutdown(self))
@@ -64,22 +57,10 @@ def monitor_swarm(self):
                              int(self.swarm.last_dht_response - self.start) if self.swarm.last_dht_response else 0))
 
 
-def run_experiment(arguments):
-    service = Service(working_dir=Path('/tmp/tribler').absolute(), config_path=Path('./tribler.conf'))
-    loop = asyncio.get_event_loop()
-    if arguments.fragile:
-        make_async_loop_fragile(loop)
-    loop.create_task(service.start_tribler())
-    try:
-        loop.run_forever()
-    finally:
-        loop.run_until_complete(loop.shutdown_asyncgens())
-        loop.close()
-
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description='Run hidden peer discovery experiment')
     parser.add_argument('--fragile', '-f', help='Fail at the first error', action='store_true')
-    args = parser.parse_args()
+    arguments = parser.parse_args()
 
-    run_experiment(args)
+    service = Service(state_dir=Path('/tmp/tribler'))
+    service.run(fragile=arguments.fragile)

scripts/experiments/tunnel_community/speed_test_e2e.py

@@ -1,5 +1,4 @@
 import argparse
-import asyncio
 import os
 from binascii import unhexlify
 from pathlib import Path
@@ -10,7 +9,6 @@
 from scripts.experiments.tunnel_community.speed_test_exit import EXPERIMENT_NUM_CIRCUITS, EXPERIMENT_NUM_HOPS, \
     Service as SpeedTestExitService
 from tribler.core.components.tunnel.tunnel_component import TunnelsComponent
-from tribler.core.utilities.utilities import make_async_loop_fragile
 
 EXPERIMENT_NUM_MB = int(os.environ.get('EXPERIMENT_NUM_MB', 10))
 
@@ -43,23 +41,10 @@ async def on_circuit_ready(self, address):
             self._graceful_shutdown()
 
 
-def run_experiment(arguments):
-    service = Service(working_dir=Path('.Tribler').absolute(), config_path=Path('./tribler.conf'))
-    loop = asyncio.get_event_loop()
-    if arguments.fragile:
-        make_async_loop_fragile(loop)
-
-    loop.create_task(service.start_tribler())
-    try:
-        loop.run_forever()
-    finally:
-        loop.run_until_complete(loop.shutdown_asyncgens())
-        loop.close()
-
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description='Run speed e2e experiment')
     parser.add_argument('--fragile', '-f', help='Fail at the first error', action='store_true')
-    args = parser.parse_args()
+    arguments = parser.parse_args()
 
-    run_experiment(args)
+    service = Service(state_dir=Path('.Tribler'))
+    service.run(fragile=arguments.fragile)

scripts/experiments/tunnel_community/speed_test_exit.py

@@ -11,29 +11,23 @@
 from tribler.core.components.key.key_component import KeyComponent
 from tribler.core.components.restapi.restapi_component import RESTComponent
 from tribler.core.components.tunnel.tunnel_component import TunnelsComponent
-from tribler.core.config.tribler_config import TriblerConfig
 from tribler.core.utilities.tiny_tribler_service import TinyTriblerService
-from tribler.core.utilities.utilities import make_async_loop_fragile
 
 EXPERIMENT_NUM_MB = int(os.environ.get('EXPERIMENT_NUM_MB', 25))
 EXPERIMENT_NUM_CIRCUITS = int(os.environ.get('EXPERIMENT_NUM_CIRCUITS', 10))
 EXPERIMENT_NUM_HOPS = int(os.environ.get('EXPERIMENT_NUM_HOPS', 1))
 
 
 class Service(TinyTriblerService, TaskManager):
-    def __init__(self, working_dir, config_path):
-        super().__init__(Service.create_config(working_dir, config_path), working_dir=working_dir,
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs,
                          components=[Ipv8Component(), KeyComponent(), RESTComponent(), TunnelsComponent()])
         TaskManager.__init__(self)
+        self.config.dht.enabled = True
+
         self.results = []
         self.output_file = 'speed_test_exit.txt'
 
-    @staticmethod
-    def create_config(working_dir, config_path):
-        config = TriblerConfig(state_dir=working_dir, file=config_path)
-        config.dht.enabled = True
-        return config
-
     def _graceful_shutdown(self):
         task = asyncio.create_task(self.on_tribler_shutdown())
         task.add_done_callback(lambda result: TinyTriblerService._graceful_shutdown(self))
@@ -92,23 +86,10 @@ async def run_speed_test(self, direction, circuit, index, size):
         return results
 
 
-def run_experiment(arguments):
-    service = Service(working_dir=Path('.Tribler').absolute(), config_path=Path('./tribler.conf'))
-    loop = asyncio.get_event_loop()
-    if arguments.fragile:
-        make_async_loop_fragile(loop)
-
-    loop.create_task(service.start_tribler())
-    try:
-        loop.run_forever()
-    finally:
-        loop.run_until_complete(loop.shutdown_asyncgens())
-        loop.close()
-
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description='Run speed e2e experiment')
     parser.add_argument('--fragile', '-f', help='Fail at the first error', action='store_true')
-    args = parser.parse_args()
+    arguments = parser.parse_args()
 
-    run_experiment(args)
+    service = Service(state_dir=Path('.Tribler'))
+    service.run(fragile=arguments.fragile)

scripts/seedbox/disseminator.py

@@ -16,7 +16,6 @@
 """
 
 import argparse
-import asyncio
 import logging
 import os
 from json import dumps
@@ -38,9 +37,7 @@
 from tribler.core.components.metadata_store.db.orm_bindings.channel_node import NEW
 from tribler.core.components.metadata_store.metadata_store_component import MetadataStoreComponent
 from tribler.core.components.socks_servers.socks_servers_component import SocksServersComponent
-from tribler.core.config.tribler_config import TriblerConfig
 from tribler.core.utilities.tiny_tribler_service import TinyTriblerService
-from tribler.core.utilities.utilities import make_async_loop_fragile
 
 _description_file_name = 'description.md'
 _thumbnail_file_name = 'thumbnail.png'
@@ -163,15 +160,14 @@ def flush(self):
 
 
 class Service(TinyTriblerService):
-    def __init__(self, source_dir, working_dir, testnet: bool):
-        config = TriblerConfig(state_dir=working_dir)
-        config.general.testnet = testnet
-        super().__init__(config,
-                         working_dir=working_dir,
+    def __init__(self, source_dir, testnet: bool, *args, **kwargs):
+        super().__init__(*args, **kwargs,
                          components=[
                              KnowledgeComponent(), MetadataStoreComponent(), KeyComponent(), Ipv8Component(),
                              SocksServersComponent(), LibtorrentComponent(), GigachannelManagerComponent(),
-                             GigaChannelComponent()])
+                             GigaChannelComponent()
+                         ])
+        self.config.general.testnet = testnet
         self.source_dir = Path(source_dir)
 
     def get_torrents_from_source(self):
@@ -212,29 +208,16 @@ async def on_tribler_started(self):
                                   gigachannel_manager_component.gigachannel_manager)
 
 
-def run_tribler(arguments):
-    working_dir = Path(arguments.tribler_dir).absolute()
-    service = Service(
-        source_dir=Path(arguments.source),
-        working_dir=working_dir,
-        testnet=arguments.testnet
-    )
-
-    loop = asyncio.get_event_loop()
-    if arguments.fragile:
-        make_async_loop_fragile(loop)
-
-    loop.create_task(service.start_tribler())
-    try:
-        loop.run_forever()
-    finally:
-        loop.run_until_complete(loop.shutdown_asyncgens())
-        loop.close()
-
-
 if __name__ == "__main__":
     _arguments = parse_args()
     print(f"Arguments: {_arguments}")
 
     setup_logger(_arguments.verbosity)
-    run_tribler(_arguments)
+
+    service = Service(
+        source_dir=Path(_arguments.source),
+        state_dir=Path(_arguments.tribler_dir),
+        testnet=_arguments.testnet
+    )
+
+    service.run(fragile=_arguments.fragile)

src/tribler/core/utilities/tiny_tribler_service.py

@@ -6,8 +6,10 @@
 
 from tribler.core.components.component import Component
 from tribler.core.components.session import Session
+from tribler.core.config.tribler_config import TriblerConfig
 from tribler.core.utilities.osutils import get_root_state_directory
 from tribler.core.utilities.process_checker import ProcessChecker
+from tribler.core.utilities.utilities import make_async_loop_fragile
 
 
 class TinyTriblerService:
@@ -16,13 +18,12 @@ class TinyTriblerService:
     All overlays are disabled by default.
     """
 
-    def __init__(self, config, components: List[Component], timeout_in_sec=None, working_dir=Path('/tmp/tribler')):
+    def __init__(self, components: List[Component], timeout_in_sec=None, state_dir=Path('/tmp/tribler')):
         self.logger = logging.getLogger(self.__class__.__name__)
 
         self.session = None
         self.process_checker: Optional[ProcessChecker] = None
-        self.working_dir = working_dir
-        self.config = config
+        self.config = TriblerConfig(state_dir=state_dir.absolute())
         self.timeout_in_sec = timeout_in_sec
         self.components = components
 
@@ -32,17 +33,29 @@ async def on_tribler_started(self):
         It is good place to add a custom code.
         """
 
-    async def start_tribler(self):
-        self.logger.info(f'Starting tribler instance in directory: {self.working_dir}')
+    def run(self, fragile: bool = False):
+        async def start_tribler():
+            self.logger.info(f'Starting tribler instance in directory: {self.config.state_dir}')
 
-        self._check_already_running()
-        await self._start_session()
+            self._check_already_running()
+            await self._start_session()
 
-        if self.timeout_in_sec:
-            asyncio.create_task(self._terminate_by_timeout())
+            if self.timeout_in_sec:
+                asyncio.create_task(self._terminate_by_timeout())
 
-        self._enable_graceful_shutdown()
-        await self.on_tribler_started()
+            self._enable_graceful_shutdown()
+            await self.on_tribler_started()
+
+        loop = asyncio.get_event_loop()
+        if fragile:
+            make_async_loop_fragile(loop)
+
+        loop.create_task(start_tribler())
+        try:
+            loop.run_forever()
+        finally:
+            loop.run_until_complete(loop.shutdown_asyncgens())
+            loop.close()
 
     async def _start_session(self):
         self.logger.info(f"Starting Tribler session with config: {self.config}")
@@ -52,7 +65,7 @@ async def _start_session(self):
         self.logger.info("Tribler session started")
 
     def _check_already_running(self):
-        self.logger.info(f'Check if we are already running a Tribler instance in: {self.working_dir}')
+        self.logger.info(f'Check if we are already running a Tribler instance in: {self.config.state_dir}')
 
         root_state_dir = get_root_state_directory()
         self.process_checker = ProcessChecker(root_state_dir)