Hi, I encountered the libuv assertion error while running multiprocessing and communicating with child processes via OS pipes.

I've prepared a minimal reproducible case.

I'm using kchmck/aiopipe to create pipes, which actually wraps file descriptors from os.pipe() into asyncio StreamReaders/StreamWrites via loop.connect_read_pipe() and loop.connect_write_pipe().

Here, I have decapsulated wrapper classes from aiopipe into functions in the aiopipe_decap.py for easier code tracing.

# aiopipe_decap.py

import asyncio
import os
from asyncio import (BaseTransport, StreamReader, StreamReaderProtocol,
                     StreamWriter, get_running_loop)
from contextlib import asynccontextmanager, contextmanager


async def _open_reader(fd):
    rx = StreamReader()
    transport, _ = await get_running_loop().connect_read_pipe(
        lambda: StreamReaderProtocol(rx),
        os.fdopen(fd))
    return transport, rx


async def _open_writer(fd):
    rx = StreamReader()
    transport, proto = await get_running_loop().connect_write_pipe(
        lambda: StreamReaderProtocol(rx),
        os.fdopen(fd, "w"))
    tx = StreamWriter(transport, proto, rx, get_running_loop())
    return transport, tx


@asynccontextmanager
async def open_stream(fd, mode):
    transport, stream = await _open_reader(fd) if mode == "r" \
        else await _open_writer(fd)

    try:
        yield stream
    finally:
        try:
            transport.close()
        except OSError:
            # The transport/protocol sometimes closes the fd before this is reached.
            pass

        # Allow event loop callbacks to run and handle closed transport.
        await asyncio.sleep(0)

@contextmanager
def detach(fd):
    os.set_inheritable(fd, True)
    try:
        yield
    finally:
        os.close(fd)

And here is the reproducible snippet, which creates a pipe connection between a child process and the master process. Once the child starts up, it will start a loop to send messages back to the master.

You can use the following environment variables to control some of the behaviors in this snippet.

• U: enable uvloop if this variable is set (default: False)
• R: number of messages to send (default: 1)
• M: content of messages (default: "a")

# uv.py

import asyncio
import os
from multiprocessing import Process

from aiopipe_decap import *
import uvloop


async def child_task(fd, message, repeat):
    async with open_stream(fd, "w") as tx:
        for i in range(repeat):
            tx.write(message)
            await tx.drain()
        tx.write_eof()


def child_main(fd, message, repeat):
    asyncio.run(child_task(fd, message, repeat))


async def main(*, message=b"a", repeat=1):
    rfd, tfd = os.pipe()

    with detach(tfd):
        proc = Process(target=child_main, args=(tfd, message, repeat))
        proc.start()

    count = 0
    async with open_stream(rfd, "r") as rx:
        while True:
            msg = await rx.read(1)
            if not msg:
                break
            count += 1
            assert msg == message
    assert count == repeat


if __name__ == "__main__":
    if os.getenv("U", ""):
        uvloop.install()
    rp = int(os.getenv("R", "1"))
    msg = os.getenv("M", "a").encode()
    asyncio.run(main(message=msg, repeat=rp))

Here's my result on my NAS (in Ubuntu 16.04 container). I found the assertion error related to the number of repeated times, the more the number is, the more chance to trigger the error. In the log shows that 9216 is a magic number but I doubt it's depending on different environments.

(venv) $ U=1 R=9210 python uv.py
(venv) $ U=1 R=9211 python uv.py
(venv) $ U=1 R=9212 python uv.py
(venv) $ U=1 R=9213 python uv.py
(venv) $ U=1 R=9214 python uv.py
(venv) $ U=1 R=9215 python uv.py
(venv) $ U=1 R=9216 python uv.py
(venv) $ U=1 R=9217 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 85, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 77, in main
    assert count == repeat
AssertionError
(venv) $ U=1 R=9218 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 85, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 77, in main
    assert count == repeat
AssertionError
(venv) $ U=1 R=9219 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 85, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 77, in main
    assert count == repeat
AssertionError
(venv) $ U=1 R=9220 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 85, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 77, in main
    assert count == repeat
AssertionError

It works like a charm with vanilla asyncio.

(venv) $ R=9210 python uv.py
(venv) $ R=9211 python uv.py
(venv) $ R=9212 python uv.py
(venv) $ R=9213 python uv.py
(venv) $ R=9214 python uv.py
(venv) $ R=9215 python uv.py
(venv) $ R=9216 python uv.py
(venv) $ R=9217 python uv.py
(venv) $ R=9218 python uv.py
(venv) $ R=9219 python uv.py
(venv) $ R=9220 python uv.py

With even larger number of repeated times.

• R=100K

(venv) $ R=100000 python uv.py
(venv) $ U=1 R=100000 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 44, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 36, in main
    assert count == repeat
AssertionError

• R=1M

(venv) $ R=1000000 python uv.py
(venv) $ U=1 R=1000000 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 44, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 36, in main
    assert count == repeat
AssertionError

• uvloop version: 0.14.0
• Python version: Python 3.7.3
• Platform: Linux-4.2.8-x86_64-with-debian-stretch-sid'
• Can you reproduce the bug with PYTHONASYNCIODEBUG in env?: Yes

(venv) $ PYTHONASYNCIODEBUG=1 U=1 R=100000 python uv.py
python: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
Traceback (most recent call last):
  File "uv.py", line 44, in <module>
    asyncio.run(main(message=msg, repeat=rp))
  File "/usr/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "uv.py", line 36, in main
    assert count == repeat
AssertionError

• Does uvloop behave differently from vanilla asyncio? How?: Vanilla asyncio works like a charm while uvloop raise the assertion error.