BaseSelectorEventLoop.sock_{recv,sendall}() don't remove their callbacks when canceled

@fantix

BPO	30064
Nosy	@fantix, @ned-deily, @asvetlov, @ambv, @1st1, @achimnol, @miss-islington, @aeros
PRs	bpo-30064: Refactor sock_* asyncio API #10419 bpo-30064: Fix asyncio loop.sock_* race condition issue #20369 [3.9] bpo-30064: Fix asyncio loop.sock_* race condition issue (GH-20369) #20460 bpo-30064: Fix unstable asyncio "racing" socket tests #20485 [3.9] bpo-30064: Fix unstable asyncio "racing" socket tests (GH-20485) #20487 bpo-30064: Properly skip unstable loop.sock_connect() racing test #20494 [3.9] bpo-30064: Properly skip unstable loop.sock_connect() racing test (GH-20494) #20503 bpo-30064: Fix slow asyncio sock test #20868 [3.9] bpo-30064: Fix slow asyncio sock test (GH-20868) #20869
Files	cancelled-sock-recv-race.patch

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = 'https://github.com/asvetlov'
closed_at = None
created_at = <Date 2017-04-13.08:55:09.854>
labels = ['type-bug', '3.9', '3.10', 'expert-asyncio']
title = "BaseSelectorEventLoop.sock_{recv,sendall}() don't remove their callbacks when canceled"
updated_at = <Date 2020-06-15.16:37:16.856>
user = 'https://bugs.python.org/abacabadabacaba'

bugs.python.org fields:

activity = <Date 2020-06-15.16:37:16.856>
actor = 'vstinner'
assignee = 'asvetlov'
closed = False
closed_date = None
closer = None
components = ['asyncio']
creation = <Date 2017-04-13.08:55:09.854>
creator = 'abacabadabacaba'
dependencies = []
files = ['49146']
hgrepos = []
issue_num = 30064
keywords = ['patch']
message_count = 41.0
messages = ['291593', '326294', '329746', '355272', '355273', '357989', '358011', '358098', '358108', '362637', '362671', '362690', '368553', '368701', '368702', '370123', '370131', '370143', '370147', '370159', '370161', '370205', '370215', '370218', '370219', '370234', '370268', '370269', '371240', '371247', '371467', '371478', '371479', '371485', '371489', '371490', '371491', '371492', '371495', '371498', '371499']
nosy_count = 9.0
nosy_names = ['fantix', 'ned.deily', 'asvetlov', 'abacabadabacaba', 'lukasz.langa', 'yselivanov', 'achimnol', 'miss-islington', 'aeros']
pr_nums = ['10419', '20369', '20460', '20485', '20487', '20494', '20503', '20868', '20869']
priority = 'high'
resolution = None
stage = 'commit review'
status = 'open'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue30064'
versions = ['Python 3.9', 'Python 3.10']

Code:

    import asyncio as a
    import socket as s

    @a.coroutine
    def coro():
        s1, s2 = s.socketpair()
        s1.setblocking(False)
        s2.setblocking(False)
        try:
            yield from a.wait_for(loop.sock_recv(s2, 1), 1)
        except a.TimeoutError:
            pass
        yield from loop.sock_sendall(s1, b'\x00')
        yield
        s1.close()
        s2.close()

    loop = a.get_event_loop()
    loop.run_until_complete(coro())

Result:

    Exception in callback BaseSelectorEventLoop._sock_recv(<Future cancelled>, True, <socket.socke...2049, proto=0>, 1)
    handle: <Handle BaseSelectorEventLoop._sock_recv(<Future cancelled>, True, <socket.socke...2049, proto=0>, 1)>
    Traceback (most recent call last):
      File "/usr/lib/python3.6/asyncio/events.py", line 127, in _run
        self._callback(*self._args)
      File "/usr/lib/python3.6/asyncio/selector_events.py", line 378, in _sock_recv
        self.remove_reader(fd)
      File "/usr/lib/python3.6/asyncio/selector_events.py", line 342, in remove_reader
        return self._remove_reader(fd)
      File "/usr/lib/python3.6/asyncio/selector_events.py", line 279, in _remove_reader
        key = self._selector.get_key(fd)
      File "/usr/lib/python3.6/selectors.py", line 189, in get_key
        return mapping[fileobj]
      File "/usr/lib/python3.6/selectors.py", line 70, in __getitem__
        fd = self._selector._fileobj_lookup(fileobj)
      File "/usr/lib/python3.6/selectors.py", line 224, in _fileobj_lookup
        return _fileobj_to_fd(fileobj)
      File "/usr/lib/python3.6/selectors.py", line 41, in _fileobj_to_fd
        raise ValueError("Invalid file descriptor: {}".format(fd))
    ValueError: Invalid file descriptor: -1

See also https://bugs.python.org/issue34795

New changeset 7438792 by Andrew Svetlov in branch 'master':
bpo-30064: Refactor sock_* asyncio API (bpo-10419)
7438792

I think this broke asyncio a bit. These two uvloop sock_* API tests no longer pass on asyncio 3.8:

Andrew, since this was your patch, please take a look.

I'll elevate the status so that we don't forget before 3.8.1 is too close

I'll elevate the status so that we don't forget before 3.8.1 is too close

Andrew, Yury: ping. 3.8.1 cutoff is approaching.

Not sure if failed uvloop tests are correct.
The scenario is like the following:

Suppose we have an unblocking socket connected to peer.
Create a task for reading data:
task = asyncio.create_task(loop.sock_read(sock, 1))
Note, the task is not started yet.
Just after it, cancel the task without waiting for the actual cancellation
task.cancel()
Finally, read from the socket again:
data = await loop.sock_read(sock, 1)

If I put context switch (await asyncio.sleep(0)) between 3) and 4) *or* replace direct sock_read() call in 4) with creation a task (data = await asyncio.create_task(loop.sock_read(sock, 1))) the cancellation of former read is performed and test passes.

I very doubt if any sane code is organizing like this test: start delayed reading, cancel it and read again.

The worse, neither previous not current sock_read() implementation doesn't prevent the concurrent reading which basically delivers data in an unpredictable order. Honestly, I'm not sure if we need to provide the concurrency guarantee for such low-level functions. The code built on top of these low-level primitives should handle the cuncurrent access problem IMHO.

Note: this is going to miss Python 3.8.1 as I'm releasing 3.8.1rc1 right now. Please address this before 3.8.2 (due in February).

No problem at all, and thanks for clarifying.

If enlarging the timeout works in some of your cases, that really inspires me on a potential cause of the issue in the test. To simulate the race condition specifically for loop.sock_sendall(), I needed a connected socket with a full send buffer so that the next sock_sendall() call would block. To achieve this, I wrote something like this:

It might be that it cost too much time in these loops (especially the 2nd one because the logs showed that it was blocked on calling sock_sendall()). But that's just a guess yet, let me add some more debug logs and test locally.

That seems reasonable to me and would make the most sense at a glance. If this is confirmed to be the source of the issue, we should be able to just reduce the size of the socket's send buffer via SO_SNDBUF before sending to it so that it fills up much more quickly.

OK I think I solved the problem by improving the two loops using smarter buffer sizes:

            # fill the buffer until sending 5 chars would block
            size = 8192
            while size > 5:
                with self.assertRaises(BlockingIOError):
                    while True:
                        sock.send(b' ' * size)
                size = int(size / 2)

and

            # receive everything that is not a space
            async def recv_all():
                rv = b''
                while True:
                    buf = await self.loop.sock_recv(server, 8192)
                    if not buf:
                        return rv
                    rv += buf.strip()

This test is now running ~25x faster, and the load test crashed my laptop a few times before it hits a timeout.

Last question before PR pls: given the fact that this test is to cover a fixed case when loop.sock_*() was hanging forever, should I keep the wait_for(..., timeout=10)?

and yes, updated the SO_SNDBUF too

Last question before PR pls: given the fact that this test is to cover a fixed case when loop.sock_*() was hanging forever, should I keep the wait_for(..., timeout=10)?

For now, I think we can keep the asyncio.wait_for() w/ 10 the sec timeout. I'll have to run the tests locally though w/ the patch to double check (once the PR is opened), and it may still end up needing to be increased if the test fails again in the future. It should be okay though.

New changeset 8f04a84 by Fantix King in branch 'master':
bpo-30064: Fix slow asyncio sock test (GH-20868)
8f04a84

New changeset 83cd968 by Miss Islington (bot) in branch '3.9':
bpo-30064: Fix slow asyncio sock test (GH-20868)
83cd968

@fantix

Thanks again for working on this @fantix, and for the continued vigilance on the issue after the test failures occurred in the buildbots. I think this recent PR-20868 will do the trick. At the very least, it will make the failure much more rare than before, which is definitely an improvement.

Navigation Menu

Repository navigation

BaseSelectorEventLoop.sock_{recv,sendall}() don't remove their callbacks when canceled #74250

27 remaining items

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

BaseSelectorEventLoop.sock_{recv,sendall}() don't remove their callbacks when canceled #74250

Description

Activity

abacabadabacaba commented on Apr 13, 2017

1st1 commented on Sep 24, 2018

asvetlov commented on Nov 12, 2018

1st1 commented on Oct 23, 2019

1st1 commented on Oct 23, 2019

ned-deily commented on Dec 7, 2019

asvetlov commented on Dec 8, 2019

ambv commented on Dec 9, 2019

27 remaining items

aeros commented on Jun 14, 2020

fantix commented on Jun 14, 2020

fantix commented on Jun 14, 2020

aeros commented on Jun 14, 2020

miss-islington commented on Jun 14, 2020

miss-islington commented on Jun 14, 2020

aeros commented on Jun 14, 2020

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions