create the child process set the proc attribute has the child process finished the child process has finished but the transport hasn t been notified yet don t clear the proc reference yet postinit may still run asyncio uses a child watcher copy the status into the popen object on python 3 6 it is required to avoid a resourcewarning wait until the process exit and return the process return code this method is a coroutine if self returncode is not none return self returncode waiter self loop createfuture self exitwaiters appendwaiter return await waiter def tryfinishself assert not self finished if self returncode is none return if allp is not none and p disconnected for p in self pipes values self finished true self callself callconnectionlost none def callconnectionlostself exc try self protocol connectionlostexc finally wake up futures waiting for wait for waiter in self exitwaiters if not waiter cancelled waiter setresultself returncode self exitwaiters none self loop none self proc none self protocol none class writesubprocesspipeprotoprotocols baseprotocol def initself proc fd self proc proc self fd fd self pipe none self disconnected false def connectionmadeself transport self pipe transport def reprself return f self class name fdself fd pipeself pipe r def connectionlostself exc self disconnected true self proc pipeconnectionlostself fd exc self proc none def pausewritingself self proc protocol pausewriting def resumewritingself self proc protocol resumewriting class readsubprocesspipeprotowritesubprocesspipeproto protocols protocol def datareceivedself data self proc pipedatareceivedself fd data create the child process set the _proc attribute has the child process finished the child process has finished but the transport hasn t been notified yet don t clear the _proc reference yet _post_init may still run asyncio uses a child watcher copy the status into the popen object on python 3 6 it is required to avoid a resourcewarning wait until the process exit and return the process return code this method is a coroutine wake up futures waiting for wait
import collections
import subprocess
import warnings

from . import protocols
from . import transports
from .log import logger


class BaseSubprocessTransport(transports.SubprocessTransport):

    def __init__(self, loop, protocol, args, shell,
                 stdin, stdout, stderr, bufsize,
                 waiter=None, extra=None, **kwargs):
        super().__init__(extra)
        self._closed = False
        self._protocol = protocol
        self._loop = loop
        self._proc = None
        self._pid = None
        self._returncode = None
        self._exit_waiters = []
        self._pending_calls = collections.deque()
        self._pipes = {}
        self._finished = False

        if stdin == subprocess.PIPE:
            self._pipes[0] = None
        if stdout == subprocess.PIPE:
            self._pipes[1] = None
        if stderr == subprocess.PIPE:
            self._pipes[2] = None

        
        try:
            self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
                        stderr=stderr, bufsize=bufsize, **kwargs)
        except:
            self.close()
            raise

        self._pid = self._proc.pid
        self._extra['subprocess'] = self._proc

        if self._loop.get_debug():
            if isinstance(args, (bytes, str)):
                program = args
            else:
                program = args[0]
            logger.debug('process %r created: pid %s',
                         program, self._pid)

        self._loop.create_task(self._connect_pipes(waiter))

    def __repr__(self):
        info = [self.__class__.__name__]
        if self._closed:
            info.append('closed')
        if self._pid is not None:
            info.append(f'pid={self._pid}')
        if self._returncode is not None:
            info.append(f'returncode={self._returncode}')
        elif self._pid is not None:
            info.append('running')
        else:
            info.append('not started')

        stdin = self._pipes.get(0)
        if stdin is not None:
            info.append(f'stdin={stdin.pipe}')

        stdout = self._pipes.get(1)
        stderr = self._pipes.get(2)
        if stdout is not None and stderr is stdout:
            info.append(f'stdout=stderr={stdout.pipe}')
        else:
            if stdout is not None:
                info.append(f'stdout={stdout.pipe}')
            if stderr is not None:
                info.append(f'stderr={stderr.pipe}')

        return '<{}>'.format(' '.join(info))

    def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
        raise NotImplementedError

    def set_protocol(self, protocol):
        self._protocol = protocol

    def get_protocol(self):
        return self._protocol

    def is_closing(self):
        return self._closed

    def close(self):
        if self._closed:
            return
        self._closed = True

        for proto in self._pipes.values():
            if proto is None:
                continue
            proto.pipe.close()

        if (self._proc is not None and
                
                self._returncode is None and
                
                
                self._proc.poll() is None):

            if self._loop.get_debug():
                logger.warning('Close running child process: kill %r', self)

            try:
                self._proc.kill()
            except ProcessLookupError:
                pass

            

    def __del__(self, _warn=warnings.warn):
        if not self._closed:
            _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
            self.close()

    def get_pid(self):
        return self._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 _check_proc(self):
        if self._proc is None:
            raise ProcessLookupError()

    def send_signal(self, signal):
        self._check_proc()
        self._proc.send_signal(signal)

    def terminate(self):
        self._check_proc()
        self._proc.terminate()

    def kill(self):
        self._check_proc()
        self._proc.kill()

    async def _connect_pipes(self, waiter):
        try:
            proc = self._proc
            loop = self._loop

            if proc.stdin is not None:
                _, pipe = await loop.connect_write_pipe(
                    lambda: WriteSubprocessPipeProto(self, 0),
                    proc.stdin)
                self._pipes[0] = pipe

            if proc.stdout is not None:
                _, pipe = await loop.connect_read_pipe(
                    lambda: ReadSubprocessPipeProto(self, 1),
                    proc.stdout)
                self._pipes[1] = pipe

            if proc.stderr is not None:
                _, pipe = await loop.connect_read_pipe(
                    lambda: ReadSubprocessPipeProto(self, 2),
                    proc.stderr)
                self._pipes[2] = pipe

            assert self._pending_calls is not None

            loop.call_soon(self._protocol.connection_made, self)
            for callback, data in self._pending_calls:
                loop.call_soon(callback, *data)
            self._pending_calls = None
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            if waiter is not None and not waiter.cancelled():
                waiter.set_exception(exc)
        else:
            if waiter is not None and not waiter.cancelled():
                waiter.set_result(None)

    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 _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
        if self._loop.get_debug():
            logger.info('%r exited with return code %r', self, returncode)
        self._returncode = returncode
        if self._proc.returncode is None:
            
            
            self._proc.returncode = returncode
        self._call(self._protocol.process_exited)

        self._try_finish()

    async def _wait(self):
        
        if self._returncode is not None:
            return self._returncode

        waiter = self._loop.create_future()
        self._exit_waiters.append(waiter)
        return await waiter

    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._call(self._call_connection_lost, None)

    def _call_connection_lost(self, exc):
        try:
            self._protocol.connection_lost(exc)
        finally:
            
            for waiter in self._exit_waiters:
                if not waiter.cancelled():
                    waiter.set_result(self._returncode)
            self._exit_waiters = None
            self._loop = None
            self._proc = None
            self._protocol = None


class WriteSubprocessPipeProto(protocols.BaseProtocol):

    def __init__(self, proc, fd):
        self.proc = proc
        self.fd = fd
        self.pipe = None
        self.disconnected = False

    def connection_made(self, transport):
        self.pipe = transport

    def __repr__(self):
        return f'<{self.__class__.__name__} fd={self.fd} pipe={self.pipe!r}>'

    def connection_lost(self, exc):
        self.disconnected = True
        self.proc._pipe_connection_lost(self.fd, exc)
        self.proc = None

    def pause_writing(self):
        self.proc._protocol.pause_writing()

    def resume_writing(self):
        self.proc._protocol.resume_writing()


class ReadSubprocessPipeProto(WriteSubprocessPipeProto,
                              protocols.Protocol):

    def data_received(self, data):
        self.proc._pipe_data_received(self.fd, data)