Calling publisher coroutine from another thread

[dwsutherland]

While reviewing #3492 I noticed the publisher was being called differently.
I had a closer look after re-basing a branch and realised the publisher is running in a different thread, and awaiting the publish coroutine from the main thread:

cylc-flow/cylc/flow/scheduler.py

Lines 1608 to 1609 in e904325

	await self.publisher.publish(
	self.data_store_mgr.get_publish_deltas())

This isn't desirable (although may not be an issue):
https://docs.python.org/3/library/asyncio-dev.html#concurrency-and-multithreading

So, to be safe, we have two options here:

1. Use a threadsafe call to this coroutine from the main thread (i.e. loop.call_soon_threadsafe()).
2. As the publisher.py has no blocking calls or loops (unlike the REQ/RES server.py), we could run the publisher in the main thread and keep the awaited publish the same.

In my opinion, and assuming publishing is not that intensive/blocking; we should go with option 2, as it simplifies the workflow scheduler somewhat (if extra thread isn't needed, why use it).

Thoughts? (ping @oliver-sanders)

Pull requests welcome!

[oliver-sanders]

The publisher doesn't need to be in lock-step, call_soon_threadsafe should be enough?

[oliver-sanders]

After working on #3654 I'm keen that we stop spawning/managing threads in the add-hoc way we currently do (I've hit all sorts of lovely issues).

As of #3654 the Scheduler is "event loop agnostic", it gets executed in a single loop provided by the executor.

How about re-working the server/publisher code to be async and, if possible, agnostic to the async implementation (plus or minus a couple of asyncio.gather calls). Then just use future = loop.run_in_executor(...) to start the server and await future to stop it. This would cut out the need for Barrier synchronisation etc.

[dwsutherland]

How about re-working the server/publisher code to be async and, if possible, agnostic to the async implementation (plus or minus a couple of asyncio.gather calls). Then just use future = loop.run_in_executor(...) to start the server and await future to stop it. This would cut out the need for Barrier synchronisation etc.

Yeah, we can do this, but we needed to change the socket.recv loop to the no-wait version otherwise it will cause the whole program to hang.

[dwsutherland]

And actually, I think the publisher is already in a state that it can run in the same thread...
It was setup to run in the different thread because it used asyncio and the scheduler didn't (which isn't the case now), ergo option 2..

The ZMQ authenticator still runs in it's own thread (which is fine)