Merge pull request #1157 from pipermerriam/piper/service-context-api

Use shared ProcessPoolExecutor across all services

Author: pipermerriam

p2p/executor.py

@@ -135,9 +135,18 @@ async def handshake(remote: Node,
     except (ConnectionRefusedError, OSError) as e:
         raise UnreachablePeer() from e
     peer = peer_class(
-        remote=remote, privkey=privkey, reader=reader, writer=writer,
-        aes_secret=aes_secret, mac_secret=mac_secret, egress_mac=egress_mac,
-        ingress_mac=ingress_mac, headerdb=headerdb, network_id=network_id)
+        remote=remote,
+        privkey=privkey,
+        reader=reader,
+        writer=writer,
+        aes_secret=aes_secret,
+        mac_secret=mac_secret,
+        egress_mac=egress_mac,
+        ingress_mac=ingress_mac,
+        headerdb=headerdb,
+        network_id=network_id,
+        token=token,
+    )
     await peer.do_p2p_handshake()
     await peer.do_sub_proto_handshake()
     return peer
@@ -167,8 +176,9 @@ def __init__(self,
                  headerdb: 'BaseAsyncHeaderDB',
                  network_id: int,
                  inbound: bool = False,
+                 token: CancelToken = None,
                  ) -> None:
-        super().__init__()
+        super().__init__(token)
         self.remote = remote
         self.privkey = privkey
         self.reader = reader
@@ -1027,8 +1037,15 @@ def _test() -> None:
     network_id = RopstenChain.network_id
     loop = asyncio.get_event_loop()
     nodes = [Node.from_uri(args.enode)]
+
     peer_pool = PeerPool(
-        peer_class, headerdb, network_id, ecies.generate_privkey(), ROPSTEN_VM_CONFIGURATION)
+        peer_class,
+        headerdb,
+        network_id,
+        ecies.generate_privkey(),
+        ROPSTEN_VM_CONFIGURATION,
+    )
+
     asyncio.ensure_future(connect_to_peers_loop(peer_pool, nodes))
 
     async def request_stuff() -> None:

p2p/peer.py

@@ -15,6 +15,7 @@
 from cancel_token import CancelToken, OperationCancelled
 
 from p2p.cancellable import CancellableMixin
+from p2p.utils import get_asyncio_executor
 
 
 class ServiceEvents:
@@ -34,26 +35,44 @@ class BaseService(ABC, CancellableMixin):
     _wait_until_finished_timeout = 5
 
     # the custom event loop to run in, or None if the default loop should be used
-    loop: asyncio.AbstractEventLoop = None
+    _loop: asyncio.AbstractEventLoop = None
 
-    def __init__(self, token: CancelToken=None, loop: asyncio.AbstractEventLoop = None) -> None:
-        if self.logger is None:
-            self.logger = cast(
-                TraceLogger, logging.getLogger(self.__module__ + '.' + self.__class__.__name__))
+    _logger: TraceLogger = None
 
-        self._run_lock = asyncio.Lock()
+    def __init__(self,
+                 token: CancelToken=None,
+                 loop: asyncio.AbstractEventLoop = None) -> None:
         self.events = ServiceEvents()
+        self._run_lock = asyncio.Lock()
         self._child_services = []
         self._finished_callbacks = []
 
-        self.loop = loop
+        self._loop = loop
+
         base_token = CancelToken(type(self).__name__, loop=loop)
 
         if token is None:
             self.cancel_token = base_token
         else:
             self.cancel_token = base_token.chain(token)
 
+        self._executor = get_asyncio_executor()
+
+    @property
+    def logger(self) -> TraceLogger:
+        if self._logger is None:
+            self._logger = cast(
+                TraceLogger,
+                logging.getLogger(self.__module__ + '.' + self.__class__.__name__)
+            )
+        return self._logger
+
+    def get_event_loop(self) -> asyncio.AbstractEventLoop:
+        if self._loop is None:
+            return asyncio.get_event_loop()
+        else:
+            return self._loop
+
     async def run(
             self,
             finished_callback: Optional[Callable[['BaseService'], None]] = None) -> None:
@@ -102,6 +121,10 @@ def run_child_service(self, child_service: 'BaseService') -> 'asyncio.Future[Any
         self._child_services.append(child_service)
         return asyncio.ensure_future(child_service.run())
 
+    async def _run_in_executor(self, callback: Callable[..., Any], *args: Any) -> Any:
+        loop = self.get_event_loop()
+        return await self.wait(loop.run_in_executor(self._executor, callback, *args))
+
     async def cleanup(self) -> None:
         """
         Run the ``_cleanup()`` coroutine and set the ``cleaned_up`` event after the service as
@@ -146,7 +169,7 @@ async def threadsafe_cancel(self) -> None:
 
         :param poll_period: how many seconds to wait in between each check for service cleanup
         """
-        asyncio.run_coroutine_threadsafe(self.cancel(), loop=self.loop)
+        asyncio.run_coroutine_threadsafe(self.cancel(), loop=self.get_event_loop())
         await asyncio.wait_for(
             self.events.cleaned_up.wait(),
             timeout=self._wait_until_finished_timeout,

p2p/service.py

@@ -1,4 +1,7 @@
 import datetime
+from concurrent.futures import Executor, ProcessPoolExecutor
+import logging
+import os
 from typing import Tuple
 
 import rlp
@@ -32,3 +35,44 @@ def time_since(start_time: datetime.datetime) -> Tuple[int, int, int, int]:
     hours, remainder = divmod(delta.seconds, 3600)
     minutes, seconds = divmod(remainder, 60)
     return delta.days, hours, minutes, seconds
+
+
+CPU_EMPTY_VALUES = {None, 0}
+
+
+_executor: Executor = None
+
+
+def get_asyncio_executor(cpu_count: int=None) -> Executor:
+    """
+    Returns a global `ProcessPoolExecutor` instance.
+
+    NOTE: We use the ProcessPoolExecutor to offload CPU intensive tasks to
+    separate processes to ensure we don't block the main networking process.
+    This pattern will only work correctly if used within a single process.  If
+    multiple processes use this executor API we'll end up with more workers
+    than there are CPU cores at which point the networking process will be
+    competing with all the worker processes for CPU resources.  At the point
+    where we need this in more than one process we will need to come up with a
+    different solution
+    """
+    global _executor
+
+    if _executor is None:
+        # Use CPU_COUNT - 1 processes to make sure we always leave one CPU idle
+        # so that it can run asyncio's event loop.
+        if cpu_count is None:
+            os_cpu_count = os.cpu_count()
+            if os_cpu_count in CPU_EMPTY_VALUES:
+                # Need this because os.cpu_count() returns None when the # of
+                # CPUs is indeterminable.
+                logger = logging.getLogger('p2p')
+                logger.warning(
+                    "Could not determine number of CPUs, defaulting to 1 instead of %s",
+                    os_cpu_count,
+                )
+                cpu_count = 1
+            else:
+                cpu_count = max(1, os_cpu_count - 1)
+        _executor = ProcessPoolExecutor(cpu_count)
+    return _executor

p2p/utils.py

@@ -47,7 +47,10 @@ def make_random_trie(random):
 
 
 @given(random=strategies.randoms())
-@settings(max_examples=10)
+@settings(
+    max_examples=10,
+    deadline=4000,
+)
 @example(random=RandomWithSeed(EXAMPLE_37968))
 @example(random=RandomWithSeed(EXAMPLE_809368))
 def test_trie_sync(random, event_loop):

tests/trinity/core/p2p-proto/test_state_sync.py

@@ -275,7 +275,7 @@ def exit_because_ambigious_filesystem(logger: logging.Logger) -> None:
 
 
 async def exit_on_signal(service_to_exit: BaseService) -> None:
-    loop = asyncio.get_event_loop()
+    loop = service_to_exit.get_event_loop()
     sigint_received = asyncio.Event()
     for sig in [signal.SIGINT, signal.SIGTERM]:
         # TODO also support Windows
@@ -318,9 +318,9 @@ def display_launch_logs(chain_config: ChainConfig) -> None:
 
 
 def run_service_until_quit(service: BaseService) -> None:
-    loop = asyncio.get_event_loop()
-    asyncio.ensure_future(exit_on_signal(service))
-    asyncio.ensure_future(service.run())
+    loop = service.get_event_loop()
+    asyncio.ensure_future(exit_on_signal(service), loop=loop)
+    asyncio.ensure_future(service.run(), loop=loop)
     loop.run_forever()
     loop.close()
 

trinity/main.py

@@ -30,7 +30,7 @@
 
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.eth.commands import (
-    Transactions
+    Transactions,
 )
 
 

trinity/plugins/builtin/tx_pool/pool.py

@@ -20,7 +20,7 @@
 )
 
 from trinity.rpc.main import (
-    RPCServer
+    RPCServer,
 )
 
 MAXIMUM_REQUEST_BYTES = 10000
@@ -144,7 +144,7 @@ async def _run(self) -> None:
         self.server = await asyncio.start_unix_server(
             connection_handler(self.rpc.execute, self.cancel_token),
             str(self.ipc_path),
-            loop=self.loop,
+            loop=self.get_event_loop(),
             limit=MAXIMUM_REQUEST_BYTES,
         )
         self.logger.info('IPC started at: %s', self.ipc_path.resolve())

trinity/rpc/ipc.py

@@ -3,7 +3,6 @@
 from typing import (
     Any,
     AsyncGenerator,
-    Callable,
     Tuple,
     Union,
     cast,
@@ -25,7 +24,6 @@
 from p2p.p2p_proto import DisconnectReason
 from p2p.peer import BasePeer, PeerPool, PeerSubscriber
 from p2p.service import BaseService
-from p2p.executor import get_asyncio_executor
 
 from trinity.db.header import AsyncHeaderDB
 from trinity.p2p.handlers import PeerRequestHandler
@@ -65,7 +63,6 @@ def __init__(self,
         self._syncing = False
         self._sync_complete = asyncio.Event()
         self._sync_requests: asyncio.Queue[HeaderRequestingPeer] = asyncio.Queue()
-        self._executor = get_asyncio_executor()
 
     @property
     def msg_queue_maxsize(self) -> int:
@@ -105,8 +102,10 @@ async def _run(self) -> None:
         asyncio.ensure_future(self._handle_msg_loop())
         with self.subscribe(self.peer_pool):
             while True:
-                peer_or_finished = await self.wait_first(
-                    self._sync_requests.get(), self._sync_complete.wait())  # type: Any
+                peer_or_finished: Any = await self.wait_first(
+                    self._sync_requests.get(),
+                    self._sync_complete.wait()
+                )
 
                 # In the case of a fast sync, we return once the sync is completed, and our caller
                 # must then run the StateDownloader.
@@ -122,10 +121,6 @@ async def _cleanup(self) -> None:
         # run in the background notice the cancel token has been triggered and return.
         await asyncio.sleep(0)
 
-    async def _run_in_executor(self, callback: Callable[..., Any], *args: Any) -> Any:
-        loop = asyncio.get_event_loop()
-        return await self.wait(loop.run_in_executor(self._executor, callback, *args))
-
     async def sync(self, peer: HeaderRequestingPeer) -> None:
         if self._syncing:
             self.logger.debug(

trinity/sync/common/chain.py

@@ -46,7 +46,6 @@
 
 from p2p.exceptions import NoEligiblePeers, NoIdlePeers
 from p2p.peer import BasePeer, PeerPool, PeerSubscriber
-from p2p.executor import get_asyncio_executor
 
 from trinity.db.base import AsyncBaseDB
 from trinity.db.chain import AsyncChainDB
@@ -87,7 +86,6 @@ def __init__(self,
         self._handler = PeerRequestHandler(self.chaindb, self.logger, self.cancel_token)
         self.request_tracker = TrieNodeRequestTracker(self._reply_timeout, self.logger)
         self._peer_missing_nodes: Dict[ETHPeer, Set[Hash32]] = collections.defaultdict(set)
-        self._executor = get_asyncio_executor()
 
     # Throughout the whole state sync our chain head is fixed, so it makes sense to ignore
     # messages related to new blocks/transactions, but we must handle requests for data from
@@ -182,8 +180,7 @@ async def _handle_msg(
             self.logger.debug("Got %d NodeData entries from %s", len(msg), peer)
             _, requested_node_keys = self.request_tracker.active_requests.pop(peer)
 
-            loop = asyncio.get_event_loop()
-            node_keys = await loop.run_in_executor(self._executor, list, map(keccak, msg))
+            node_keys = await self._run_in_executor(list, map(keccak, msg))
 
             missing = set(requested_node_keys).difference(node_keys)
             self._peer_missing_nodes[peer].update(missing)