Merge with Python 3.4 (asyncio)

- Close #22063: socket operations (socket,recv, sock_sendall, sock_connect,
  sock_accept) now raise an exception in debug mode if sockets are in blocking
  mode.
- asyncio: Use the new os.set_blocking() function of Python 3.5 if available
This commit is contained in:
Victor Stinner 2014-07-29 23:09:56 +02:00
commit 66565649b5
5 changed files with 48 additions and 4 deletions

View File

@ -385,12 +385,18 @@ class BaseProactorEventLoop(base_events.BaseEventLoop):
self._selector = None self._selector = None
def sock_recv(self, sock, n): def sock_recv(self, sock, n):
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
return self._proactor.recv(sock, n) return self._proactor.recv(sock, n)
def sock_sendall(self, sock, data): def sock_sendall(self, sock, data):
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
return self._proactor.send(sock, data) return self._proactor.send(sock, data)
def sock_connect(self, sock, address): def sock_connect(self, sock, address):
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try: try:
base_events._check_resolved_address(sock, address) base_events._check_resolved_address(sock, address)
except ValueError as err: except ValueError as err:
@ -401,6 +407,8 @@ class BaseProactorEventLoop(base_events.BaseEventLoop):
return self._proactor.connect(sock, address) return self._proactor.connect(sock, address)
def sock_accept(self, sock): def sock_accept(self, sock):
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
return self._proactor.accept(sock) return self._proactor.accept(sock)
def _socketpair(self): def _socketpair(self):

View File

@ -256,6 +256,8 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
This method is a coroutine. This method is a coroutine.
""" """
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = futures.Future(loop=self) fut = futures.Future(loop=self)
self._sock_recv(fut, False, sock, n) self._sock_recv(fut, False, sock, n)
return fut return fut
@ -292,6 +294,8 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
This method is a coroutine. This method is a coroutine.
""" """
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = futures.Future(loop=self) fut = futures.Future(loop=self)
if data: if data:
self._sock_sendall(fut, False, sock, data) self._sock_sendall(fut, False, sock, data)
@ -333,6 +337,8 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
This method is a coroutine. This method is a coroutine.
""" """
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = futures.Future(loop=self) fut = futures.Future(loop=self)
try: try:
base_events._check_resolved_address(sock, address) base_events._check_resolved_address(sock, address)
@ -374,6 +380,8 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
This method is a coroutine. This method is a coroutine.
""" """
if self.get_debug() and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = futures.Future(loop=self) fut = futures.Future(loop=self)
self._sock_accept(fut, False, sock) self._sock_accept(fut, False, sock)
return fut return fut

View File

@ -258,6 +258,16 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
return server return server
if hasattr(os, 'set_blocking'):
def _set_nonblocking(fd):
os.set_blocking(fd, False)
else:
def _set_nonblocking(fd):
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
flags = flags | os.O_NONBLOCK
fcntl.fcntl(fd, fcntl.F_SETFL, flags)
class _UnixReadPipeTransport(transports.ReadTransport): class _UnixReadPipeTransport(transports.ReadTransport):
max_size = 256 * 1024 # max bytes we read in one event loop iteration max_size = 256 * 1024 # max bytes we read in one event loop iteration
@ -273,7 +283,7 @@ class _UnixReadPipeTransport(transports.ReadTransport):
stat.S_ISSOCK(mode) or stat.S_ISSOCK(mode) or
stat.S_ISCHR(mode)): stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is for pipes/sockets only.") raise ValueError("Pipe transport is for pipes/sockets only.")
os.set_blocking(self._fileno, False) _set_nonblocking(self._fileno)
self._protocol = protocol self._protocol = protocol
self._closing = False self._closing = False
self._loop.add_reader(self._fileno, self._read_ready) self._loop.add_reader(self._fileno, self._read_ready)
@ -366,7 +376,7 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
stat.S_ISCHR(mode)): stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is only for " raise ValueError("Pipe transport is only for "
"pipes, sockets and character devices") "pipes, sockets and character devices")
os.set_blocking(self._fileno, False) _set_nonblocking(self._fileno)
self._protocol = protocol self._protocol = protocol
self._buffer = [] self._buffer = []
self._conn_lost = 0 self._conn_lost = 0

View File

@ -383,6 +383,24 @@ class EventLoopTestsMixin:
self.assertEqual(read, data) self.assertEqual(read, data)
def _basetest_sock_client_ops(self, httpd, sock): def _basetest_sock_client_ops(self, httpd, sock):
# in debug mode, socket operations must fail
# if the socket is not in blocking mode
self.loop.set_debug(True)
sock.setblocking(True)
with self.assertRaises(ValueError):
self.loop.run_until_complete(
self.loop.sock_connect(sock, httpd.address))
with self.assertRaises(ValueError):
self.loop.run_until_complete(
self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n'))
with self.assertRaises(ValueError):
self.loop.run_until_complete(
self.loop.sock_recv(sock, 1024))
with self.assertRaises(ValueError):
self.loop.run_until_complete(
self.loop.sock_accept(sock))
# test in non-blocking mode
sock.setblocking(False) sock.setblocking(False)
self.loop.run_until_complete( self.loop.run_until_complete(
self.loop.sock_connect(sock, httpd.address)) self.loop.sock_connect(sock, httpd.address))

View File

@ -306,7 +306,7 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.pipe = mock.Mock(spec_set=io.RawIOBase) self.pipe = mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value = 5 self.pipe.fileno.return_value = 5
blocking_patcher = mock.patch('os.set_blocking') blocking_patcher = mock.patch('asyncio.unix_events._set_nonblocking')
blocking_patcher.start() blocking_patcher.start()
self.addCleanup(blocking_patcher.stop) self.addCleanup(blocking_patcher.stop)
@ -469,7 +469,7 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
self.pipe = mock.Mock(spec_set=io.RawIOBase) self.pipe = mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value = 5 self.pipe.fileno.return_value = 5
blocking_patcher = mock.patch('os.set_blocking') blocking_patcher = mock.patch('asyncio.unix_events._set_nonblocking')
blocking_patcher.start() blocking_patcher.start()
self.addCleanup(blocking_patcher.stop) self.addCleanup(blocking_patcher.stop)