Use ThreadSelectorEventLoop on Windows with ProactorEventLoop

[davidbrochart]

Changes

This PR allows anyio.wait_socket_readable(sock) to work on Windows when using the asyncio backend with a proactor event loop, by using Tornado's SelectorThread, as discussed with @minrk in zeromq/pyzmq#1827.

Checklist

If this is a user-facing code change, like a bugfix or a new feature, please ensure that
you've fulfilled the following conditions (where applicable):

• You've added tests (in tests/) added which would fail without your patch
• You've updated the documentation (in docs/, in case of behavior changes or new
  features)
• You've added a new changelog entry (in docs/versionhistory.rst).

[agronholm]

To me this seems like overkill. Why would we have to shoehorn these into the async event loop?

[agronholm]

What I'm saying is that a simpler solution would be better, where we just spawn a thread on demand that runs select(...) on all the sockets that need to wait for read/write. Or am I missing something?

[davidbrochart]

But that would mean one thread per socket, right? While with this solution there is only one thread per event loop.

[agronholm]

But that would mean one thread per socket, right? While with this solution there is only one thread per event loop.

No, I never said that. There would be just one thread which would select() on multiple sockets.

[davidbrochart]

OK, well there might be a better way than this PR, which is trying to bring Tornado's implementation almost untouched.

[agronholm]

I think Tornado had different needs, hence their more elaborate solution.

[minrk]

Why would we have to shoehorn these into the async event loop?

I don't really understand what you mean by shoehorning into the loop. The selector is not being added to the loop itself, it is a single Thread running select which implements the add_reader interface of the EventLoop API in a threadsafe way, to call select in a thread and run callbacks in the event loop when select returns. It is an object with a reference to the event loop and a weakref registry to store one of these threads per event loop (only once add_reader is called).

However you want to do it, I think it is appropriate to implement it optimistically as if add_reader can be assumed, because asyncio maintainers agree that it should be, even on Proactor, even though nobody has had the time to implement it yet (I still think it was a mistake to allow Proactor to become the default without this fundamental API). That way you can remove the workaround when you drop support for Pythons that didn't support it without changing ~anything. That is I think the main thing the tornado approach accomplishes because you only need to replace loop.add_reader with get_selector(loop).add_reader, which does nothing but return loop.add_reader everywhere but Windows + Proactor.

where we just spawn a thread on demand that runs select(...) on all the sockets that need to wait for read/write. Or am I missing something?

It is indeed very simple if you have a single thread per socket, but if you want to use a shared thread instead of thread-per-socket, you need to handle the fact that add_reader and friends may be called when select is already blocking in a background thread, which means needing:

1. a threadsafe way to update the sockets for select, and
2. a mechanism to wake the thread when the socket list is updated

which is where most of the logic in the tornado implementation resides. You could reimplement it, but it seems to me like it's going to need to be quite similar to tornado's implementation if it's going to work. But maybe you have an idea I'm missing about how to handle waking the select thread in a simpler way?

[davidbrochart]

I don't understand that test_wait_socket_readable fails on trio now.

[davidbrochart]

And now I don't understand that the test won't be skipped for Windows/Trio :)

[agronholm]

And now I don't understand that the test won't be skipped for Windows/Trio :)

Have you tried using pytest.mark.skipif to skip the test?

[davidbrochart]

It seems that pytest.mark.skipif doesn't support fixtures, and we need the anyio_backend_name fixture. I don't see any other way than using pytest.skip inside the test.

[agronholm]

Why a timeout this strict? It's bound to cause flakiness on slow/busy systems.
Also, wouldn't fail_after eliminate the need to set a flag?

Suggested change

	with move_on_after(0.1):
	with fail_after(5):

On another note, I think this should be on the same line as with conn:, given that it's the only place where the timeout can work.

[agronholm]

I don't think this needs to be daemon, given how it will either succeed or fail instantly. Given the continuous effort on CPython's side to get rid of daemon threads, I'm also looking for alternatives.

Also, the thread needs to be joined uncoditionally, even if the test fails. With this code, it won't be joined if the test fails.

[agronholm]

It seems that pytest.mark.skipif doesn't support fixtures, and we need the anyio_backend_name fixture. I don't see any other way than using pytest.skip inside the test.

Do you have a Windows machine to run this locally on? Setting a breakpoint on the first line would probably shed some light on this.

[davidbrochart]

Do you have a Windows machine to run this locally on?

No, but with platform.system() == "Linux" I see that the test is skipped on my Linux machine for trio, so I really have no clue.

[agronholm]

Do you have a Windows machine to run this locally on?

No, but with platform.system() == "Linux" I see that the test is skipped on my Linux machine for trio, so I really have no clue.

I have a Windows machine here, maybe I could take a look.

[agronholm]

Something weird is going on here. If I only run the test on Trio, it's skipped as expected. I would say that the exception is some sort of fallout from the asyncio version of the test.

[agronholm]

Specifically, it has to do with the thread that's spawned when wait_socket_readable() is called, not so much the thread that's spawned in the test code directly – that looks totally fine.

[agronholm]

I checked that the ResourceWarning about the unclosed socket had nothing to do with any of the three sockets created during the test.

[agronholm]

For the record, the test passes just fine on Trio+Windows if you skip the problematic asyncio version.

[davidbrochart]

Thanks @agronholm, I had forgotten to close the loop when removing AddThreadSelectorEventLoop.
I think the failure in PyPy is unrelated?

[graingert]

I think it would be better to try to use add_reader, and catch the NotImplementedError - raising if we're not on sys.platform == "win32"?

Or maybe we don't care if sys.platform == "win32" or not at this point? if add_reader is not implemented on the loop we can still work around it with the selector in a thread

[minrk]

Yeah, I think if you catch the error you can register the selector independent of the platform in case someone else decides to make another incomplete EventLoop implementation like Proactor. The selector thread should work everywhere.

[graingert]

probably will need a different name

[minrk]

perhaps something like:

Suggested change

	add_reader(sock, event.set)
	try:
	add_reader(sock, event.set)
	except NotImplementedError:
	# Proactor on Windows does not yet implement add/remove reader
	selector = _get_selector_windows(loop)
	selector.add_reader(sock, event.set)
	remove_reader = selector.remove_reader
	else:
	remove_reader = loop.remove_reader

[graingert]

This causes an issue for dask/distributed where they can't use tornado on the proactor because they send a shutdown message on a socket then close the event loop in atexit

the socket add_reader/add_writer then stops working while it's trying to send the shutdown message

anyio already has a thread pool implementation that shuts down when run_forever completes I think that should be used here

[agronholm]

To be precise, it ties itself into the life cycle of the root task.

[graingert]

Also the thread here is closed explicitly using the async generator hook and patching EventLoop.close so I think it's ok to leak the demon thread atexit if nobody explicitly closes the loop

[graingert]

I don't think we need this, because we know the loop has started when we create the Selector thread. We should probably also stash the task somewhere