Skip to content

Commit d5aeccf

Browse files
vstinnervstinner
committed
asyncio, Tulip issue 205: Fix a race condition in BaseSelectorEventLoop.sock_connect()
There is a race condition in create_connection() used with wait_for() to have a timeout. sock_connect() registers the file descriptor of the socket to be notified of write event (if connect() raises BlockingIOError). When create_connection() is cancelled with a TimeoutError, sock_connect() coroutine gets the exception, but it doesn't unregister the file descriptor for write event. create_connection() gets the TimeoutError and closes the socket. If you call again create_connection(), the new socket will likely gets the same file descriptor, which is still registered in the selector. When sock_connect() calls add_writer(), it tries to modify the entry instead of creating a new one. This issue was originally reported in the Trollius project, but the bug comes from Tulip in fact (Trollius is based on Tulip): https://bitbucket.org/enovance/trollius/issue/15/after-timeouterror-on-wait_for This change fixes the race condition. It also makes sock_connect() more reliable (and portable) is sock.connect() raises an InterruptedError.
1 parent 41f3c3f commit d5aeccf

File tree

2 files changed

+83
-35
lines changed

2 files changed

+83
-35
lines changed

‎Lib/asyncio/selector_events.py‎

Lines changed: 31 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88

99
import collections
1010
import errno
11+
import functools
1112
import socket
1213
try:
1314
import ssl
@@ -345,26 +346,43 @@ def sock_connect(self, sock, address):
345346
except ValueError as err:
346347
fut.set_exception(err)
347348
else:
348-
self._sock_connect(fut, False, sock, address)
349+
self._sock_connect(fut, sock, address)
349350
return fut
350351

351-
def _sock_connect(self, fut, registered, sock, address):
352+
def _sock_connect(self, fut, sock, address):
352353
fd = sock.fileno()
353-
if registered:
354-
self.remove_writer(fd)
354+
try:
355+
while True:
356+
try:
357+
sock.connect(address)
358+
except InterruptedError:
359+
continue
360+
else:
361+
break
362+
except BlockingIOError:
363+
fut.add_done_callback(functools.partial(self._sock_connect_done,
364+
sock))
365+
self.add_writer(fd, self._sock_connect_cb, fut, sock, address)
366+
except Exception as exc:
367+
fut.set_exception(exc)
368+
else:
369+
fut.set_result(None)
370+
371+
def _sock_connect_done(self, sock, fut):
372+
self.remove_writer(sock.fileno())
373+
374+
def _sock_connect_cb(self, fut, sock, address):
355375
if fut.cancelled():
356376
return
377+
357378
try:
358-
if not registered:
359-
# First time around.
360-
sock.connect(address)
361-
else:
362-
err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
363-
if err != 0:
364-
# Jump to the except clause below.
365-
raise OSError(err, 'Connect call failed %s' % (address,))
379+
err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
380+
if err != 0:
381+
# Jump to any except clause below.
382+
raise OSError(err, 'Connect call failed %s' % (address,))
366383
except (BlockingIOError, InterruptedError):
367-
self.add_writer(fd, self._sock_connect, fut, True, sock, address)
384+
# socket is still registered, the callback will be retried later
385+
pass
368386
except Exception as exc:
369387
fut.set_exception(exc)
370388
else:

‎Lib/test/test_asyncio/test_selector_events.py‎

Lines changed: 52 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,9 @@ def list_to_buffer(l=()):
4040
class BaseSelectorEventLoopTests(test_utils.TestCase):
4141

4242
def setUp(self):
43-
selector = mock.Mock()
44-
self.loop = TestBaseSelectorEventLoop(selector)
43+
self.selector = mock.Mock()
44+
self.selector.select.return_value = []
45+
self.loop = TestBaseSelectorEventLoop(self.selector)
4546
self.set_event_loop(self.loop, cleanup=False)
4647

4748
def test_make_socket_transport(self):
@@ -303,63 +304,92 @@ def test_sock_connect(self):
303304
f = self.loop.sock_connect(sock, ('127.0.0.1', 8080))
304305
self.assertIsInstance(f, asyncio.Future)
305306
self.assertEqual(
306-
(f, False, sock, ('127.0.0.1', 8080)),
307+
(f, sock, ('127.0.0.1', 8080)),
307308
self.loop._sock_connect.call_args[0])
308309

310+
def test_sock_connect_timeout(self):
311+
# Tulip issue #205: sock_connect() must unregister the socket on
312+
# timeout error
313+
314+
# prepare mocks
315+
self.loop.add_writer = mock.Mock()
316+
self.loop.remove_writer = mock.Mock()
317+
sock = test_utils.mock_nonblocking_socket()
318+
sock.connect.side_effect = BlockingIOError
319+
320+
# first call to sock_connect() registers the socket
321+
fut = self.loop.sock_connect(sock, ('127.0.0.1', 80))
322+
self.assertTrue(sock.connect.called)
323+
self.assertTrue(self.loop.add_writer.called)
324+
self.assertEqual(len(fut._callbacks), 1)
325+
326+
# on timeout, the socket must be unregistered
327+
sock.connect.reset_mock()
328+
fut.set_exception(asyncio.TimeoutError)
329+
with self.assertRaises(asyncio.TimeoutError):
330+
self.loop.run_until_complete(fut)
331+
self.assertTrue(self.loop.remove_writer.called)
332+
309333
def test__sock_connect(self):
310334
f = asyncio.Future(loop=self.loop)
311335

312336
sock = mock.Mock()
313337
sock.fileno.return_value = 10
314338

315-
self.loop._sock_connect(f, False, sock, ('127.0.0.1', 8080))
339+
self.loop._sock_connect(f, sock, ('127.0.0.1', 8080))
316340
self.assertTrue(f.done())
317341
self.assertIsNone(f.result())
318342
self.assertTrue(sock.connect.called)
319343

320-
def test__sock_connect_canceled_fut(self):
344+
def test__sock_connect_cb_cancelled_fut(self):
321345
sock = mock.Mock()
346+
self.loop.remove_writer = mock.Mock()
322347

323348
f = asyncio.Future(loop=self.loop)
324349
f.cancel()
325350

326-
self.loop._sock_connect(f, False, sock, ('127.0.0.1', 8080))
327-
self.assertFalse(sock.connect.called)
351+
self.loop._sock_connect_cb(f, sock, ('127.0.0.1', 8080))
352+
self.assertFalse(sock.getsockopt.called)
353+
354+
def test__sock_connect_writer(self):
355+
# check that the fd is registered and then unregistered
356+
self.loop._process_events = mock.Mock()
357+
self.loop.add_writer = mock.Mock()
358+
self.loop.remove_writer = mock.Mock()
328359

329-
def test__sock_connect_unregister(self):
330360
sock = mock.Mock()
331361
sock.fileno.return_value = 10
362+
sock.connect.side_effect = BlockingIOError
363+
sock.getsockopt.return_value = 0
364+
address = ('127.0.0.1', 8080)
332365

333366
f = asyncio.Future(loop=self.loop)
334-
f.cancel()
367+
self.loop._sock_connect(f, sock, address)
368+
self.assertTrue(self.loop.add_writer.called)
369+
self.assertEqual(10, self.loop.add_writer.call_args[0][0])
335370

336-
self.loop.remove_writer = mock.Mock()
337-
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
371+
self.loop._sock_connect_cb(f, sock, address)
372+
# need to run the event loop to execute _sock_connect_done() callback
373+
self.loop.run_until_complete(f)
338374
self.assertEqual((10,), self.loop.remove_writer.call_args[0])
339375

340-
def test__sock_connect_tryagain(self):
376+
def test__sock_connect_cb_tryagain(self):
341377
f = asyncio.Future(loop=self.loop)
342378
sock = mock.Mock()
343379
sock.fileno.return_value = 10
344380
sock.getsockopt.return_value = errno.EAGAIN
345381

346-
self.loop.add_writer = mock.Mock()
347-
self.loop.remove_writer = mock.Mock()
348-
349-
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
350-
self.assertEqual(
351-
(10, self.loop._sock_connect, f,
352-
True, sock, ('127.0.0.1', 8080)),
353-
self.loop.add_writer.call_args[0])
382+
# check that the exception is handled
383+
self.loop._sock_connect_cb(f, sock, ('127.0.0.1', 8080))
354384

355-
def test__sock_connect_exception(self):
385+
def test__sock_connect_cb_exception(self):
356386
f = asyncio.Future(loop=self.loop)
357387
sock = mock.Mock()
358388
sock.fileno.return_value = 10
359389
sock.getsockopt.return_value = errno.ENOTCONN
360390

361391
self.loop.remove_writer = mock.Mock()
362-
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
392+
self.loop._sock_connect_cb(f, sock, ('127.0.0.1', 8080))
363393
self.assertIsInstance(f.exception(), OSError)
364394

365395
def test_sock_accept(self):

0 commit comments

Comments
 (0)