Ensure shared process plugins shutdown cleanly

[cburgdorf]

What was wrong?

Currently, plugins that run in the shared network process do not shutdown cleanly in all scenarios (see #1284). To be more precise, they only shutdown if they happen to be chained with some CancelToken which is triggered. This means that for all current plugins that run in the networking process (TxPool, LightPeerChainBridge) there isn't a real problem as these will simply shutdown based the fact that e.g the PeerPool shuts down or the LightPeerChain shuts down. However, this is still leaving us with two problems:

1. Plugins that do not have a logical dependency that they can rely on to initiate cancellation are left out
2. Plugins can't stopped individually (think, future command trinity stop-plugin <plugin-name>

How was it fixed?

Understanding the sync vs async scenario

Before we dive into how exactly it was fixed, let's first take a moment to reflect on the three different kind of operational models that we currently have for plugins.

1. Plugins that overtake the whole main process

This category of plugin doesn't need any special tear down treatment as it is basically breaking out of the standard trinity boot sequence early and entirely redefines what the trinity process even means.

2. Plugins that spawn there own process

The teardown for these plugins is similar to how we shut down other non-plugin processes.

py-evm/trinity/extensibility/plugin.py

Lines 180 to 182 in d225f4a

	def stop(self) -> None:
	self.context.event_bus.stop()
	kill_process_gracefully(self._process, self.logger)

Notice that kill_process_gracefully is a blocking operation.

3. Plugins that run in a shared process (currently networking)

Plugins that run in the shared networking process are required to be written in a non-blocking fashion using asyncio (This is a MUST for builtin plugins and at least a strong recommendation for any user defined plugin).

Sidenote: That doesn't mean that plugins that spawn their own process can't be written in the same async fashion (in fact, most should) but since they run in an isolated space they can do pretty much everything they want including spinning up a virtual system to run a PHP server on Windows 95)

Means, since these plugins run asynchronously, they need to be stopped asynchronously as well.

Enter shutdown vs shutdown_nowait

Initially, I tried to streamline the shutdown of the PluginManager to either enforce being:

1. Always synchronous (turning non blocking operations into blocking ones)
2. Always asynchronous (turning blocking operations into non-blocking ones)

This has proven to be problematic for various reasons (I can elaborate if needed). So what I settled with instead is to come up with shutdown and shutdown_nowait (naming is inline with other places of our code base (and python stdlib) that use that pattern ).

Then inside the kill_trinity_gracefully() function, we continue to call the synchronous version (only isolated plugins are managed by this plugin manager instance) and in the unwind routine of the networking process we await the async version of shutdown (only async plugins run in this plugin manager instance.

Refactorings

I refactored a bunch of stuff around plugins to make this possible:

1. Because some plugins need no stop() at all, some need async def stop() and some need def stop(), this method was removed from the BasePlugin and new derived plugins BaseAsyncStopPlugin as well as BaseSyncStopPlugin where defined accordingly.

2. PluginProcessScope is no more. Otherwise it would be possible to e.g. derive from BaseAsyncStopPlugin but set PluginProcessScope.MAIN which is a contradicting configuration.

3. The PluginManager itself was refactored to rely more on composition. It now delegates to scope.is_responsible_for_plugin(...) and scope.create_plugin_context(...) instead of hardcoding if/else rules in methods of the PluginManager directly.

4. A new AsyncContextManager (backport from Python 3.7) was created to allow more flexible teardown of event loops compared to what we previously had with exit_on_signal. The exit_on_signal helper does still exist but it is now built on top of said AsyncContextManager-based API.

Cute Animal Picture

[carver]

Generally seems fine (although I'm only vaguely familiar with the plugin stuff).

Two things keeping me from 👍 --

1. the default loop value in wait_for_exit
2. the TODO left in the code

The rest are suggestions, if you like.

[carver]

A await asyncio.gather() seems pretty doable in this PR. If there's some significant complication I'm not seeing, an issue is preferable to a TODO.

[cburgdorf]

Yep, that was just me slacking 🌴 ☀️

[carver]

minor naming nitpick

I'm not sure the _nowait suffix is being used here in quite the same way as other asyncio library usages. This is potentially a long blocking wait for each plugin to shut down, right? Seems like it's more like a shutdown_blocking to me.

[cburgdorf]

Ah, that seems like the right name for it 👍

[carver]

I think this default value is going to be problematic. It will be evaluated once at import time, which is probably not be desired. So far all method calls explicitly set the loop, anyway.

How about we require the loop to be passed for now. If leaving it out becomes convenient at some point in the future, we can re-add it. (The default lookup would go in the method body, to detect the loop at runtime).

[cburgdorf]

It will be evaluated once at import time

Oh, I assumed that would only be evaluated in the case it is called without that parameter. Didn't know these things are evaluated eagerly in Python (Still a Python noob after all).

How about we require the loop to be passed for now

Yes, let's do that.

[carver]

Maybe add another one like:

@asynccontextmanager
async def wait_for_service_exit(service_to_exit: BaseService) -> None:
    loop = service_to_exit.get_event_loop()
    async with wait_for_exit(loop):
        await service_to_exit.cancel()
        yield
        service_to_exit._executor.shutdown(wait=True)

exit_on_signal might go away or be pushed elsewhere, into exit_service_with_plugin_endpoint(service, endpoint) which just stops the endpoint, and simple_service_exit(service) which does nothing at all in wait_for_service_exit.

[cburgdorf]

I went down that way but slightly adjusted naming scheme:

async def exit_with_service_and_endpoint(service_to_exit: BaseService, endpoint: Endpoint) -> None:
    async with exit_signal_with_service(service_to_exit):
        endpoint.stop()


async def exit_with_service(service_to_exit: BaseService) -> None:
    async with exit_signal_with_service(service_to_exit):
        pass


@asynccontextmanager
async def exit_signal_with_service(service_to_exit: BaseService) -> AsyncGenerator[None, None]:
    loop = service_to_exit.get_event_loop()
    async with exit_signal(loop):
        await service_to_exit.cancel()
        yield
        service_to_exit._executor.shutdown(wait=True)


@asynccontextmanager
async def exit_signal(loop: asyncio.AbstractEventLoop) -> AsyncGenerator[None, None]:
    sigint_received = asyncio.Event()
    for sig in [signal.SIGINT, signal.SIGTERM]:
        # TODO also support Windows
        loop.add_signal_handler(sig, sigint_received.set)

    await sigint_received.wait()
    try:
        yield
    finally:
        loop.stop()

I think this reads more natural:

async with exit_signal(loop):
    # do something

Same with:

asyncio.ensure_future(exit_with_service_and_endpoint(service, endpoint))

[carver]

With a new wait_for_service_exit this could be:

    async with wait_for_service_exit(service):
        await plugin_manager.shutdown()
        endpoint.stop()

[cburgdorf]

I pushed this out into a private method rather than using an inline comprehension for two reasons:

1. I wanted to do self._logger.info("Stopping plugin: %s", plugin.name) to have the same logs as with sync stops
2. We need an isinstance check to avoid a cast

[cburgdorf]

@carver I think I addressed all your points

[carver]

There isn't a result for sync plugins in self._started_plugins, so this will emit the wrong logs below, when a sync plugin comes before an async one.

Maybe add a private method that returns all async plugins, so you can do:

async_plugins = self._get_async_plugins()
stop_results = await asyncio.gather(*self._stop_plugins(async_plugins), return_exceptions=True)
for plugin, result in zip(async_plugins, stop_results):
  ...

[cburgdorf]

Oh, indeed! 🙉👍

[cburgdorf]

@carver doing this inline here to not clutter the class with methods that are only used by this method too much. For a moment I thought we could reuse _stop_plugins() for both async / sync but since the sync stop() needs to be surrounded with a try/catch and is considered a success as soon as stop() returns, that isn't really reusable for the async version. I guess its cleaner to handle these separately.

[carver]

LGTM!

[pipermerriam]

I failed to submit this single comment from sometime last week.

[pipermerriam]

I generally consider this an anti-pattern since return value checking is error prone. Don't feel strongly.... but I kind of do.... If there's a good reason why try/except doesn't fit well with this I'm fine with it.