"Assertion `loop->watchers[w->fd] == w' failed." with multiprocessing and OS pipes

[momocow]

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.

[1st1]

Thanks so much! I'll take a look in a few weeks when it's time for uvloop 0.15. If anyone wants to further debug & work on fix please go ahead!

[fantix]

This seems to be a pretty serious issue - when the buffer is full writing to a pipe, this is 100% reproducible:

import asyncio
from asyncio.subprocess import PIPE

import uvloop


async def main():
    proc = await asyncio.create_subprocess_shell('sleep 3600', stdin=PIPE)
    while True:
        proc.stdin.write(b'x' * 32768)
        await proc.stdin.drain()


if __name__ == '__main__':
    uvloop.install()
    asyncio.run(main())

The error is the same:

Assertion failed: (loop->watchers[w->fd] == w), function uv__io_stop, file src/unix/core.c, line 932.
Abort trap: 6

Before we have a proper fix from libuv (I'll create a PR in libuv when time, refs libuv/libuv#2058), we can easily fix uvloop by partially reverting d8fe153 and adding this patch:

diff --git a/uvloop/handles/pipe.pyx b/uvloop/handles/pipe.pyx
index 581554f..7a2c8ec 100644
--- a/uvloop/handles/pipe.pyx
+++ b/uvloop/handles/pipe.pyx
@@ -12,6 +12,7 @@ cdef __pipe_init_uv_handle(UVStream handle, Loop loop):
     err = uv.uv_pipe_init(handle._loop.uvloop,
                           <uv.uv_pipe_t*>handle._handle,
                           0)
+    # UV_HANDLE_READABLE allows calling uv_read_start() on this pipe
+    # even if it is O_WRONLY, see also #317, libuv/libuv#2058
+    handle._handle.flags |= 0x00004000
     if err < 0:
         handle._abort_init()
         raise convert_error(err)
diff --git a/uvloop/includes/uv.pxd b/uvloop/includes/uv.pxd
index 5f034b3..9efdab1 100644
--- a/uvloop/includes/uv.pxd
+++ b/uvloop/includes/uv.pxd
@@ -82,6 +82,7 @@ cdef extern from "uv.h" nogil:
     ctypedef struct uv_handle_t:
         void* data
         uv_loop_t* loop
+        unsigned int flags
         # ...

     ctypedef struct uv_idle_t:

But UV_HANDLE_READABLE is an internal flag in uv-common.h, so using the magic number here is a bit dirty - @1st1 shall I create a PR for this patch in uvloop with proper tests, or do you prefer to wait for a new version of libuv?

[mikepurvis]

I also hit this error— I have a fairly generic asyncio application which downloads a bunch of tarballs in parallel with httpx.stream and pipes the content into tar subprocesses which are unpacking them onto the filesystem:

        async with self.stream_repo_tarball() as tarball_stream:
            tar_proc = await asyncio.create_subprocess_exec('tar', '--extract', '--verbose',
                    '--gzip', '--strip-components=1', cwd=path, stdin=asyncio.subprocess.PIPE)

            async for chunk in tarball_stream.aiter_bytes():
                tar_proc.stdin.write(chunk)
                await tar_proc.stdin.drain()
            tar_proc.stdin.close()
            await tar_proc.wait()

This works under asyncio, but uvloop crashes pretty reliably with:

python3: src/unix/core.c:932: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.

If it's relevant, this is in the context of a Sanic webserver, so I believe there is a second worker thread running the loop. For now I am disabling uvloop and then it is fine again:

asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())

[1st1]

@fantix have we fixed this one in master?

[enumag]

just ran into this as well... what's the current status? is it fixed in master already? if so, when can we expect a release?

[rgeronimi]

Critical issue as well