refactor(reactor): remove module-level global reactor

Author: glevco

extras/custom_checks.sh

@@ -101,13 +101,26 @@ function check_do_not_import_from_hathor_in_entrypoints() {
     return 0
 }
 
+function check_do_not_import_twisted_reactor_directly() {
+    EXCLUDES="--exclude=reactor.py --exclude=conftest.py"
+    PATTERN='\<.*from .*twisted.internet import .*reactor\>'
+
+    if grep -R $EXCLUDES "$PATTERN" "${SOURCE_DIRS[@]}"; then
+        echo 'do not use `from twisted.internet import reactor` directly.'
+        echo 'instead, use `hathor.reactor.get_global_reactor()`.'
+        return 1
+    fi
+    return 0
+}
+
 # List of functions to be executed
 checks=(
 	check_version_match
 	check_do_not_use_builtin_random_in_tests
 	check_deprecated_typing
 	check_do_not_import_tests_in_hathor
 	check_do_not_import_from_hathor_in_entrypoints
+	check_do_not_import_twisted_reactor_directly
 )
 
 # Initialize a variable to track if any check fails

hathor/builder/builder.py

@@ -34,6 +34,7 @@
 from hathor.p2p.manager import ConnectionsManager
 from hathor.p2p.peer_id import PeerId
 from hathor.pubsub import PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.storage import RocksDBStorage
 from hathor.stratum import StratumFactory
 from hathor.transaction.storage import (
@@ -42,7 +43,7 @@
     TransactionRocksDBStorage,
     TransactionStorage,
 )
-from hathor.util import Random, Reactor, get_environment_info, not_none
+from hathor.util import Random, get_environment_info, not_none
 from hathor.verification.verification_service import VerificationService, VertexVerifiers
 from hathor.wallet import BaseWallet, Wallet
 
@@ -311,7 +312,7 @@ def _get_or_create_pubsub(self) -> PubSubManager:
         return self._pubsub
 
     def _create_stratum_server(self, manager: HathorManager) -> StratumFactory:
-        stratum_factory = StratumFactory(manager=manager)
+        stratum_factory = StratumFactory(manager=manager, reactor=self._get_reactor())
         manager.stratum_factory = stratum_factory
         manager.metrics.stratum_factory = stratum_factory
         return stratum_factory

hathor/builder/cli_builder.py

@@ -35,8 +35,9 @@
 from hathor.p2p.peer_id import PeerId
 from hathor.p2p.utils import discover_hostname, get_genesis_short_hash
 from hathor.pubsub import PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.stratum import StratumFactory
-from hathor.util import Random, Reactor, not_none
+from hathor.util import Random, not_none
 from hathor.verification.verification_service import VerificationService
 from hathor.verification.vertex_verifiers import VertexVerifiers
 from hathor.wallet import BaseWallet, HDWallet, Wallet
@@ -271,7 +272,7 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
         p2p_manager.set_manager(self.manager)
 
         if self._args.stratum:
-            stratum_factory = StratumFactory(self.manager)
+            stratum_factory = StratumFactory(self.manager, reactor=reactor)
             self.manager.stratum_factory = stratum_factory
             self.manager.metrics.stratum_factory = stratum_factory
 

hathor/builder/resources_builder.py

@@ -224,15 +224,15 @@ def create_resources(self) -> server.Site:
             root.putChild(b'_debug', debug_resource)
             resources.extend([
                 (b'log', DebugLogResource(), debug_resource),
-                (b'raise', DebugRaiseResource(), debug_resource),
-                (b'reject', DebugRejectResource(), debug_resource),
+                (b'raise', DebugRaiseResource(self.manager.reactor), debug_resource),
+                (b'reject', DebugRejectResource(self.manager.reactor), debug_resource),
                 (b'print', DebugPrintResource(), debug_resource),
             ])
         if self._args.enable_crash_api:
             crash_resource = Resource()
             root.putChild(b'_crash', crash_resource)
             resources.extend([
-                (b'exit', DebugCrashResource(), crash_resource),
+                (b'exit', DebugCrashResource(self.manager.reactor), crash_resource),
                 (b'mess_around', DebugMessAroundResource(self.manager), crash_resource),
             ])
 

hathor/cli/events_simulator/events_simulator.py

@@ -44,15 +44,16 @@ def execute(args: Namespace) -> None:
     os.environ['HATHOR_CONFIG_YAML'] = UNITTESTS_SETTINGS_FILEPATH
     from hathor.cli.events_simulator.event_forwarding_websocket_factory import EventForwardingWebsocketFactory
     from hathor.cli.events_simulator.scenario import Scenario
+    from hathor.reactor import get_global_reactor
     from hathor.simulator import Simulator
-    from hathor.util import reactor
 
     try:
         scenario = Scenario[args.scenario]
     except KeyError as e:
         possible_scenarios = [scenario.name for scenario in Scenario]
         raise ValueError(f'Invalid scenario "{args.scenario}". Choose one of {possible_scenarios}') from e
 
+    reactor = get_global_reactor()
     log = logger.new()
     simulator = Simulator(args.seed)
     simulator.start()

hathor/cli/run_node.py

@@ -138,7 +138,8 @@ def prepare(self, *, register_resources: bool = True) -> None:
         self.check_unsafe_arguments()
         self.check_python_version()
 
-        from hathor.util import reactor
+        from hathor.reactor import get_global_reactor
+        reactor = get_global_reactor()
         self.reactor = reactor
 
         from hathor.builder import CliBuilder, ResourcesBuilder

hathor/cli/stratum_mining.py

@@ -30,8 +30,8 @@ def create_parser() -> ArgumentParser:
 
 def execute(args: Namespace) -> None:
     from hathor.crypto.util import decode_address
+    from hathor.reactor import get_global_reactor
     from hathor.stratum import StratumClient
-    from hathor.util import reactor
     from hathor.wallet.exceptions import InvalidAddress
 
     address = None
@@ -43,7 +43,8 @@ def execute(args: Namespace) -> None:
             print('The given address is invalid')
             sys.exit(-1)
 
-    miner = StratumClient(proc_count=args.nproc, address=address)
+    reactor = get_global_reactor()
+    miner = StratumClient(proc_count=args.nproc, address=address, reactor=reactor)
     miner.start()
     point = TCP4ClientEndpoint(reactor, args.host, args.port)
     connectProtocol(point, miner)

hathor/debug_resources.py

@@ -24,7 +24,7 @@
 from hathor.cli.openapi_files.register import register_resource
 from hathor.exception import HathorError
 from hathor.manager import HathorManager
-from hathor.util import reactor
+from hathor.reactor import ReactorProtocol
 from hathor.utils.zope import asserted_cast
 
 logger = get_logger()
@@ -54,6 +54,10 @@ class DebugRaiseResource(Resource):
     }
     default_msg = 'exception raised for debugging purposes'
 
+    def __init__(self, reactor: ReactorProtocol) -> None:
+        super().__init__()
+        self._reactor = reactor
+
     def run(self, exc_cls: type[BaseException], msg: str) -> None:
         raise exc_cls(msg)
 
@@ -63,7 +67,7 @@ def render_GET(self, request: Request) -> bytes:
         assert exc_cls_name in self.exc_class_map
         exc_cls = self.exc_class_map[exc_cls_name]
         msg = get_arg_default(raw_args, 'msg', self.default_msg)
-        threaded_reactor = asserted_cast(IReactorFromThreads, reactor)
+        threaded_reactor = asserted_cast(IReactorFromThreads, self._reactor)
         threaded_reactor.callFromThread(self.run, exc_cls, msg)
         return b'OK: no side-effects\n'
 
@@ -188,7 +192,7 @@ def render_GET(self, request: Request) -> bytes:
         mess = get_arg_default(get_args(request), 'with', self.default_mess)
         assert mess in self.mess_map
         mess_func = self.mess_map[mess]
-        threaded_reactor = asserted_cast(IReactorFromThreads, reactor)
+        threaded_reactor = asserted_cast(IReactorFromThreads, self.manager.reactor)
         threaded_reactor.callFromThread(mess_func)
         return b'OK: database yanked, full-node will break\n'
 
@@ -208,12 +212,16 @@ class DebugCrashResource(Resource):
         }
     }
 
+    def __init__(self, reactor: ReactorProtocol) -> None:
+        super().__init__()
+        self._reactor = reactor
+
     def run(self, code: int) -> None:
         # XXX: sys.exit will raise a SystemExit exception that get's trapped by twisted
         #      os._exit will bypass that by exiting directly, note that no cleanup methods will be called
         os._exit(code)
 
     def render_GET(self, request: Request) -> bytes:
         code = get_arg_default(get_args(request), 'code', -1)
-        reactor.callLater(1.0, self.run, code)
+        self._reactor.callLater(1.0, self.run, code)
         return b'OK: full-node will exit and probably break database\n'

hathor/event/event_manager.py

@@ -23,8 +23,9 @@
 from hathor.event.storage import EventStorage
 from hathor.event.websocket import EventWebsocketFactory
 from hathor.pubsub import EventArguments, HathorEvents, PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.transaction import BaseTransaction
-from hathor.util import Reactor, not_none, progress
+from hathor.util import not_none, progress
 from hathor.utils.iter import batch_iterator
 
 logger = get_logger()

hathor/event/websocket/factory.py

@@ -21,7 +21,8 @@
 from hathor.event.storage import EventStorage
 from hathor.event.websocket.protocol import EventWebsocketProtocol
 from hathor.event.websocket.response import EventResponse, InvalidRequestType
-from hathor.util import Reactor, not_none
+from hathor.reactor import ReactorProtocol as Reactor
+from hathor.util import not_none
 
 logger = get_logger()
 

hathor/manager.py

@@ -50,14 +50,15 @@
 from hathor.p2p.protocol import HathorProtocol
 from hathor.profiler import get_cpu_profiler
 from hathor.pubsub import HathorEvents, PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.stratum import StratumFactory
 from hathor.transaction import BaseTransaction, Block, MergeMinedBlock, Transaction, TxVersion, sum_weights
 from hathor.transaction.exceptions import TxValidationError
 from hathor.transaction.storage import TransactionStorage
 from hathor.transaction.storage.exceptions import TransactionDoesNotExist
 from hathor.transaction.storage.tx_allow_scope import TxAllowScope
 from hathor.types import Address, VertexId
-from hathor.util import EnvironmentInfo, LogDuration, Random, Reactor, calculate_min_significant_weight, not_none
+from hathor.util import EnvironmentInfo, LogDuration, Random, calculate_min_significant_weight, not_none
 from hathor.verification.verification_service import VerificationService
 from hathor.wallet import BaseWallet
 

hathor/metrics.py

@@ -22,11 +22,11 @@
 from hathor.conf import HathorSettings
 from hathor.p2p.manager import ConnectionsManager, PeerConnectionsMetrics
 from hathor.pubsub import EventArguments, HathorEvents, PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.transaction.base_transaction import sum_weights
 from hathor.transaction.block import Block
 from hathor.transaction.storage import TransactionRocksDBStorage, TransactionStorage
 from hathor.transaction.storage.cache_storage import TransactionCacheStorage
-from hathor.util import Reactor
 
 if TYPE_CHECKING:
     from hathor.stratum import StratumFactory  # noqa: F401
@@ -63,7 +63,7 @@ class Metrics:
     connections: ConnectionsManager
     tx_storage: TransactionStorage
     # Twisted reactor that handles the time and callLater
-    reactor: Optional[Reactor] = None
+    reactor: Reactor
 
     # Transactions count in the network
     transactions: int = 0
@@ -127,10 +127,6 @@ def __post_init__(self) -> None:
         # Stores caculated block weights saved in tx storage
         self.weight_block_deque: deque[WeightValue] = deque(maxlen=self.weight_block_deque_len)
 
-        if self.reactor is None:
-            from hathor.util import reactor as twisted_reactor
-            self.reactor = twisted_reactor
-
         # A timer to periodically collect data
         self._lc_collect_data = LoopingCall(self._collect_data)
         self._lc_collect_data.clock = self.reactor

hathor/p2p/manager.py

@@ -34,8 +34,9 @@
 from hathor.p2p.sync_version import SyncVersion
 from hathor.p2p.utils import description_to_connection_string, parse_whitelist
 from hathor.pubsub import HathorEvents, PubSubManager
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.transaction import BaseTransaction
-from hathor.util import Random, Reactor
+from hathor.util import Random
 
 if TYPE_CHECKING:
     from twisted.internet.interfaces import IDelayedCall

hathor/p2p/netfilter/matches_remote.py

@@ -20,7 +20,7 @@
 from twisted.internet.task import LoopingCall
 
 from hathor.p2p.netfilter.matches import NetfilterMatch, NetfilterMatchIPAddress
-from hathor.util import Reactor
+from hathor.reactor import ReactorProtocol as Reactor
 
 if TYPE_CHECKING:
     from hathor.p2p.netfilter.context import NetfilterContext

hathor/p2p/rate_limiter.py

@@ -14,7 +14,7 @@
 
 from typing import NamedTuple, Optional
 
-from hathor.util import Reactor
+from hathor.reactor import ReactorProtocol as Reactor
 
 
 class RateLimiterLimit(NamedTuple):
@@ -32,12 +32,9 @@ class RateLimiter:
     # Stores the last hit for each key
     hits: dict[str, RateLimiterLimit]
 
-    def __init__(self, reactor: Optional[Reactor] = None):
+    def __init__(self, reactor: Reactor):
         self.keys = {}
         self.hits = {}
-        if reactor is None:
-            from hathor.util import reactor as twisted_reactor
-            reactor = twisted_reactor
         self.reactor = reactor
 
     def set_limit(self, key: str, max_hits: int, window_seconds: float) -> None:

hathor/p2p/sync_factory.py

@@ -13,16 +13,16 @@
 # limitations under the License.
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from hathor.p2p.sync_agent import SyncAgent
-from hathor.util import Reactor
+from hathor.reactor import ReactorProtocol as Reactor
 
 if TYPE_CHECKING:
     from hathor.p2p.protocol import HathorProtocol
 
 
 class SyncAgentFactory(ABC):
     @abstractmethod
-    def create_sync_agent(self, protocol: 'HathorProtocol', reactor: Optional[Reactor] = None) -> SyncAgent:
+    def create_sync_agent(self, protocol: 'HathorProtocol', reactor: Reactor) -> SyncAgent:
         pass

hathor/p2p/sync_v1/agent.py

@@ -26,10 +26,11 @@
 from hathor.p2p.messages import GetNextPayload, GetTipsPayload, NextPayload, ProtocolMessages, TipsPayload
 from hathor.p2p.sync_agent import SyncAgent
 from hathor.p2p.sync_v1.downloader import Downloader
+from hathor.reactor import ReactorProtocol as Reactor
 from hathor.transaction import BaseTransaction
 from hathor.transaction.base_transaction import tx_or_block_from_bytes
 from hathor.transaction.storage.exceptions import TransactionDoesNotExist
-from hathor.util import Reactor, json_dumps, json_loads
+from hathor.util import json_dumps, json_loads
 
 logger = get_logger()
 
@@ -59,7 +60,7 @@ class NodeSyncTimestamp(SyncAgent):
 
     MAX_HASHES: int = 40
 
-    def __init__(self, protocol: 'HathorProtocol', downloader: Downloader, reactor: Optional[Reactor] = None) -> None:
+    def __init__(self, protocol: 'HathorProtocol', downloader: Downloader, reactor: Reactor) -> None:
         """
         :param protocol: Protocol of the connection.
         :type protocol: HathorProtocol
@@ -72,9 +73,6 @@ def __init__(self, protocol: 'HathorProtocol', downloader: Downloader, reactor:
         self.manager = protocol.node
         self.downloader = downloader
 
-        if reactor is None:
-            from hathor.util import reactor as twisted_reactor
-            reactor = twisted_reactor
         self.reactor: Reactor = reactor
 
         # Rate limit for this connection.

hathor/p2p/sync_v1/factory.py

@@ -19,7 +19,7 @@
 from hathor.p2p.sync_factory import SyncAgentFactory
 from hathor.p2p.sync_v1.agent import NodeSyncTimestamp
 from hathor.p2p.sync_v1.downloader import Downloader
-from hathor.util import Reactor
+from hathor.reactor import ReactorProtocol as Reactor
 
 if TYPE_CHECKING:
     from hathor.p2p.protocol import HathorProtocol
@@ -36,5 +36,5 @@ def get_downloader(self) -> Downloader:
             self._downloader = Downloader(self.connections.manager)
         return self._downloader
 
-    def create_sync_agent(self, protocol: 'HathorProtocol', reactor: Optional[Reactor] = None) -> SyncAgent:
+    def create_sync_agent(self, protocol: 'HathorProtocol', reactor: Reactor) -> SyncAgent:
         return NodeSyncTimestamp(protocol, downloader=self.get_downloader(), reactor=reactor)