# The selectors module is in the stdlib in Python 3.4 but not in 3.3.
# Do this first, so the other submodules can use "from . import selectors".
+# Prefer asyncio/selectors.py over the stdlib one, as ours may be newer.
try:
- import selectors # Will also be exported.
-except ImportError:
from . import selectors
+except ImportError:
+ import selectors # Will also be exported.
+
+if sys.platform == 'win32':
+ # Similar thing for _overlapped.
+ try:
+ from . import _overlapped
+ except ImportError:
+ import _overlapped # Will also be exported.
# This relies on each of the submodules having an __all__ variable.
from .futures import *
return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra)
def _make_write_pipe_transport(self, sock, protocol, waiter=None,
- extra=None):
- return _ProactorWritePipeTransport(self, sock, protocol, waiter, extra)
+ extra=None, check_for_hangup=True):
+ if check_for_hangup:
+ # We want connection_lost() to be called when other end closes
+ return _ProactorDuplexPipeTransport(self,
+ sock, protocol, waiter, extra)
+ else:
+ # If other end closes we may not notice for a long time
+ return _ProactorWritePipeTransport(self, sock, protocol, waiter,
+ extra)
def close(self):
if self._proactor is not None:
"""Selector eventloop for Unix with signal handling."""
-import collections
import errno
import fcntl
import os
import sys
+from . import base_subprocess
from . import constants
from . import events
from . import protocols
self._loop = None
-class _UnixWriteSubprocessPipeProto(protocols.BaseProtocol):
- pipe = None
+class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
- def __init__(self, proc, fd):
- self.proc = proc
- self.fd = fd
- self.connected = False
- self.disconnected = False
- proc._pipes[fd] = self
-
- def connection_made(self, transport):
- self.connected = True
- self.pipe = transport
- self.proc._try_connected()
-
- def connection_lost(self, exc):
- self.disconnected = True
- self.proc._pipe_connection_lost(self.fd, exc)
-
-
-class _UnixReadSubprocessPipeProto(_UnixWriteSubprocessPipeProto,
- protocols.Protocol):
-
- def data_received(self, data):
- self.proc._pipe_data_received(self.fd, data)
-
- def eof_received(self):
- pass
-
-
-class _UnixSubprocessTransport(transports.SubprocessTransport):
-
- def __init__(self, loop, protocol, args, shell,
- stdin, stdout, stderr, bufsize,
- extra=None, **kwargs):
- super().__init__(extra)
- self._protocol = protocol
- self._loop = loop
-
- self._pipes = {}
+ def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
stdin_w = None
if stdin == subprocess.PIPE:
- self._pipes[STDIN] = None
# Use a socket pair for stdin, since not all platforms
# support selecting read events on the write end of a
# socket (which we use in order to detect closing of the
# other end). Notably this is needed on AIX, and works
# just fine on other platforms.
stdin, stdin_w = self._loop._socketpair()
- if stdout == subprocess.PIPE:
- self._pipes[STDOUT] = None
- if stderr == subprocess.PIPE:
- self._pipes[STDERR] = None
- self._pending_calls = collections.deque()
- self._finished = False
- self._returncode = None
-
self._proc = subprocess.Popen(
args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
universal_newlines=False, bufsize=bufsize, **kwargs)
if stdin_w is not None:
stdin.close()
self._proc.stdin = open(stdin_w.detach(), 'rb', buffering=bufsize)
- self._extra['subprocess'] = self._proc
-
- def close(self):
- for proto in self._pipes.values():
- proto.pipe.close()
- if self._returncode is None:
- self.terminate()
-
- def get_pid(self):
- return self._proc.pid
-
- def get_returncode(self):
- return self._returncode
-
- def get_pipe_transport(self, fd):
- if fd in self._pipes:
- return self._pipes[fd].pipe
- else:
- return None
-
- def send_signal(self, signal):
- self._proc.send_signal(signal)
-
- def terminate(self):
- self._proc.terminate()
-
- def kill(self):
- self._proc.kill()
-
- @tasks.coroutine
- def _post_init(self):
- proc = self._proc
- loop = self._loop
- if proc.stdin is not None:
- transp, proto = yield from loop.connect_write_pipe(
- lambda: _UnixWriteSubprocessPipeProto(self, STDIN),
- proc.stdin)
- if proc.stdout is not None:
- transp, proto = yield from loop.connect_read_pipe(
- lambda: _UnixReadSubprocessPipeProto(self, STDOUT),
- proc.stdout)
- if proc.stderr is not None:
- transp, proto = yield from loop.connect_read_pipe(
- lambda: _UnixReadSubprocessPipeProto(self, STDERR),
- proc.stderr)
- if not self._pipes:
- self._try_connected()
-
- def _call(self, cb, *data):
- if self._pending_calls is not None:
- self._pending_calls.append((cb, data))
- else:
- self._loop.call_soon(cb, *data)
-
- def _try_connected(self):
- assert self._pending_calls is not None
- if all(p is not None and p.connected for p in self._pipes.values()):
- self._loop.call_soon(self._protocol.connection_made, self)
- for callback, data in self._pending_calls:
- self._loop.call_soon(callback, *data)
- self._pending_calls = None
-
- def _pipe_connection_lost(self, fd, exc):
- self._call(self._protocol.pipe_connection_lost, fd, exc)
- self._try_finish()
-
- def _pipe_data_received(self, fd, data):
- self._call(self._protocol.pipe_data_received, fd, data)
-
- def _process_exited(self, returncode):
- assert returncode is not None, returncode
- assert self._returncode is None, self._returncode
- self._returncode = returncode
- self._loop._subprocess_closed(self)
- self._call(self._protocol.process_exited)
- self._try_finish()
-
- def _try_finish(self):
- assert not self._finished
- if self._returncode is None:
- return
- if all(p is not None and p.disconnected
- for p in self._pipes.values()):
- self._finished = True
- self._loop.call_soon(self._call_connection_lost, None)
-
- def _call_connection_lost(self, exc):
- try:
- self._protocol.connection_lost(exc)
- finally:
- self._proc = None
- self._protocol = None
- self._loop = None
import errno
import socket
+import subprocess
import weakref
import struct
import _winapi
+from . import base_subprocess
from . import futures
from . import proactor_events
from . import selector_events
from . import tasks
from . import windows_utils
from .log import logger
-
-try:
- import _overlapped
-except ImportError:
- from . import _overlapped
+from . import _overlapped
__all__ = ['SelectorEventLoop', 'ProactorEventLoop', 'IocpProactor']
def _stop_serving(self, server):
server.close()
+ @tasks.coroutine
+ def _make_subprocess_transport(self, protocol, args, shell,
+ stdin, stdout, stderr, bufsize,
+ extra=None, **kwargs):
+ transp = _WindowsSubprocessTransport(self, protocol, args, shell,
+ stdin, stdout, stderr, bufsize,
+ extra=None, **kwargs)
+ yield from transp._post_init()
+ return transp
+
+ def _subprocess_closed(self, transport):
+ pass
+
class IocpProactor:
"""Proactor implementation using IOCP."""
if self._iocp is not None:
_winapi.CloseHandle(self._iocp)
self._iocp = None
+
+
+class _WindowsSubprocessTransport(base_subprocess.BaseSubprocessTransport):
+
+ def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
+ self._proc = windows_utils.Popen(
+ args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
+ bufsize=bufsize, **kwargs)
+ def callback(f):
+ returncode = self._proc.poll()
+ self._process_exited(returncode)
+ f = self._loop._proactor.wait_for_handle(int(self._proc._handle))
+ f.add_done_callback(callback)
BUFSIZE = 8192
PIPE = subprocess.PIPE
+STDOUT = subprocess.STDOUT
_mmap_counter = itertools.count()
#
The stdin, stdout, stderr are None or instances of PipeHandle.
"""
def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds):
+ assert not kwds.get('universal_newlines')
+ assert kwds.get('bufsize', 0) == 0
stdin_rfd = stdout_wfd = stderr_wfd = None
stdin_wh = stdout_rh = stderr_rh = None
if stdin == PIPE:
- stdin_rh, stdin_wh = pipe(overlapped=(False, True))
+ stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True)
stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY)
+ else:
+ stdin_rfd = stdin
if stdout == PIPE:
stdout_rh, stdout_wh = pipe(overlapped=(True, False))
stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0)
+ else:
+ stdout_wfd = stdout
if stderr == PIPE:
stderr_rh, stderr_wh = pipe(overlapped=(True, False))
stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0)
+ elif stderr == STDOUT:
+ stderr_wfd = stdout_wfd
+ else:
+ stderr_wfd = stderr
try:
- super().__init__(args, bufsize=0, universal_newlines=False,
- stdin=stdin_rfd, stdout=stdout_wfd,
+ super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd,
stderr=stderr_wfd, **kwds)
except:
for h in (stdin_wh, stdout_rh, stderr_rh):
- _winapi.CloseHandle(h)
+ if h is not None:
+ _winapi.CloseHandle(h)
raise
else:
if stdin_wh is not None:
r.close()
w.close()
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
+
+class SubprocessTestsMixin:
+
+ def check_terminated(self, returncode):
+ if sys.platform == 'win32':
+ self.assertIsInstance(returncode, int)
+ self.assertNotEqual(0, returncode)
+ else:
+ self.assertEqual(-signal.SIGTERM, returncode)
+
+ def check_killed(self, returncode):
+ if sys.platform == 'win32':
+ self.assertIsInstance(returncode, int)
+ self.assertNotEqual(0, returncode)
+ else:
+ self.assertEqual(-signal.SIGKILL, returncode)
+
def test_subprocess_exec(self):
proto = None
transp = None
self.loop.run_until_complete(proto.got_data[1].wait())
transp.close()
self.loop.run_until_complete(proto.completed)
- self.assertEqual(-signal.SIGTERM, proto.returncode)
+ self.check_terminated(proto.returncode)
self.assertEqual(b'Python The Winner', proto.data[1])
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_interactive(self):
proto = None
transp = None
transp.close()
self.loop.run_until_complete(proto.completed)
- self.assertEqual(-signal.SIGTERM, proto.returncode)
+ self.check_terminated(proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_shell(self):
proto = None
transp = None
nonlocal proto, transp
transp, proto = yield from self.loop.subprocess_shell(
functools.partial(MySubprocessProtocol, self.loop),
- 'echo "Python"')
+ 'echo Python')
self.assertIsInstance(proto, MySubprocessProtocol)
self.loop.run_until_complete(connect())
self.loop.run_until_complete(proto.completed)
self.assertEqual(0, proto.returncode)
self.assertTrue(all(f.done() for f in proto.disconnects.values()))
- self.assertEqual({1: b'Python\n', 2: b''}, proto.data)
+ self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
+ self.assertEqual(proto.data[2], b'')
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_exitcode(self):
proto = None
self.loop.run_until_complete(proto.completed)
self.assertEqual(7, proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_close_after_finish(self):
proto = None
transp = None
self.assertEqual(7, proto.returncode)
self.assertIsNone(transp.close())
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_kill(self):
proto = None
transp = None
transp.kill()
self.loop.run_until_complete(proto.completed)
- self.assertEqual(-signal.SIGKILL, proto.returncode)
+ self.check_killed(proto.returncode)
+
+ def test_subprocess_terminate(self):
+ proto = None
+ transp = None
+
+ prog = os.path.join(os.path.dirname(__file__), 'echo.py')
+
+ @tasks.coroutine
+ def connect():
+ nonlocal proto, transp
+ transp, proto = yield from self.loop.subprocess_exec(
+ functools.partial(MySubprocessProtocol, self.loop),
+ sys.executable, prog)
+ self.assertIsInstance(proto, MySubprocessProtocol)
+
+ self.loop.run_until_complete(connect())
+ self.loop.run_until_complete(proto.connected)
+
+ transp.terminate()
+ self.loop.run_until_complete(proto.completed)
+ self.check_terminated(proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
+ @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
def test_subprocess_send_signal(self):
proto = None
transp = None
self.loop.run_until_complete(proto.completed)
self.assertEqual(-signal.SIGHUP, proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_stderr(self):
proto = None
transp = None
self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
self.assertEqual(0, proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_stderr_redirect_to_stdout(self):
proto = None
transp = None
transp.close()
self.assertEqual(0, proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_close_client_stream(self):
proto = None
transp = None
self.loop.run_until_complete(proto.disconnects[1])
stdin.write(b'xxx')
self.loop.run_until_complete(proto.got_data[2].wait())
- self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
-
+ if sys.platform != 'win32':
+ self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
+ else:
+ # After closing the read-end of a pipe, writing to the
+ # write-end using os.write() fails with errno==EINVAL and
+ # GetLastError()==ERROR_INVALID_NAME on Windows!?! (Using
+ # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
+ self.assertEqual(b'ERR:OSError', proto.data[2])
transp.close()
self.loop.run_until_complete(proto.completed)
- self.assertEqual(-signal.SIGTERM, proto.returncode)
+ self.check_terminated(proto.returncode)
- @unittest.skipIf(sys.platform == 'win32',
- "Don't support subprocess for Windows yet")
def test_subprocess_wait_no_same_group(self):
proto = None
transp = None
def create_event_loop(self):
return windows_events.SelectorEventLoop()
- class ProactorEventLoopTests(EventLoopTestsMixin, unittest.TestCase):
+
+ class ProactorEventLoopTests(EventLoopTestsMixin,
+ SubprocessTestsMixin,
+ unittest.TestCase):
def create_event_loop(self):
return windows_events.ProactorEventLoop()
from asyncio import unix_events
if hasattr(selectors, 'KqueueSelector'):
- class KqueueEventLoopTests(EventLoopTestsMixin, unittest.TestCase):
+ class KqueueEventLoopTests(EventLoopTestsMixin,
+ SubprocessTestsMixin,
+ unittest.TestCase):
def create_event_loop(self):
return unix_events.SelectorEventLoop(
selectors.KqueueSelector())
if hasattr(selectors, 'EpollSelector'):
- class EPollEventLoopTests(EventLoopTestsMixin, unittest.TestCase):
+ class EPollEventLoopTests(EventLoopTestsMixin,
+ SubprocessTestsMixin,
+ unittest.TestCase):
def create_event_loop(self):
return unix_events.SelectorEventLoop(selectors.EpollSelector())
if hasattr(selectors, 'PollSelector'):
- class PollEventLoopTests(EventLoopTestsMixin, unittest.TestCase):
+ class PollEventLoopTests(EventLoopTestsMixin,
+ SubprocessTestsMixin,
+ unittest.TestCase):
def create_event_loop(self):
return unix_events.SelectorEventLoop(selectors.PollSelector())
# Should always exist.
- class SelectEventLoopTests(EventLoopTestsMixin, unittest.TestCase):
+ class SelectEventLoopTests(EventLoopTestsMixin,
+ SubprocessTestsMixin,
+ unittest.TestCase):
def create_event_loop(self):
return unix_events.SelectorEventLoop(selectors.SelectSelector())
import _winapi
from asyncio import windows_utils
-
-try:
- import _overlapped
-except ImportError:
- from asyncio import _overlapped
+from asyncio import _overlapped
class WinsocketpairTests(unittest.TestCase):
projects / python / commitdiff