Script showing issue mixing uvloop threading and multiprocessing servers

uvloop-fail-mixing-multi-processing-and-threads.py

"""Script showing issue mixing uvloop threading and multiprocessing servers

Running the script hangs.

Changing the line:

    for threaded in [True, False]:

to have just True or just False makes the script run fine.
"""
import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

import sys
import multiprocessing
import queue
import threading

class EchoServerProtocol(asyncio.Protocol):

    def connection_made(self, transport):
        transport.write(b'z')

class EchoClientProtocol(asyncio.Protocol):

    def __init__(self, loop):
        self.loop = loop

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        self.transport.close()

    def connection_lost(self, exc):
        self.loop.stop()

def run_server(quin, qout):
    def server_thread():
        global server_loop
        loop = server_loop = asyncio.new_event_loop()
        coro = loop.create_server(EchoServerProtocol, '127.0.0.1', 0)
        server = loop.run_until_complete(coro)
        addr = server.sockets[0].getsockname()
        qout.put(addr)
        loop.run_forever()
        server.close()
        loop.run_until_complete(server.wait_closed())
        try:
            loop.close()
        except Exception as exc:
            print(exc)
        qout.put('stopped')

    thread = threading.Thread(target=server_thread, daemon=True)
    thread.start()

    qin.get()
    server_loop.call_soon_threadsafe(server_loop.stop)
    thread.join(1)

for i in range(5):

    for threaded in [True, False]:

        print(i, threaded)

        if threaded:
            qin, qout = queue.Queue(), queue.Queue()
            threading.Thread(
                target=run_server, args=(qin, qout), daemon=True).start()
        else:
            qin, qout = multiprocessing.Queue(), multiprocessing.Queue()
            multiprocessing.Process(
                target=run_server, args=(qin, qout), daemon=True).start()

        addr = qout.get()
        loop = asyncio.new_event_loop()
        loop.create_task(
            loop.create_connection(lambda : EchoClientProtocol(loop),
                                   host=addr[0], port=addr[1]))
        loop.run_forever()
        loop.close()
        qin.put('stop')
        qout.get()