Demo Popen() methods, for discussion:

    def _read_output(self, fileobj):
        handle = msvcrt.get_osfhandle(fileobj.fileno())
        output = self._fileobj2output[fileobj]
        
        while True:
            try:
                size = _winapi.PeekNamedPipe(handle)[0] or 1
                data = _winapi.ReadFile(handle, size)[0]
            except BrokenPipeError:
                break
            except OSError as e:
                if e.winerror == _winapi.ERROR_OPERATION_ABORTED:
                    # Should this be mapped to InterruptedError
                    # (EINTR) in PC/errmap.h?
                    break
                raise
            output.append(data)

        fileobj.close()


    def _communicate(self, input, endtime, orig_timeout):
        if not self._communication_started:
            self._fileobj2thread = {}
            self._fileobj2output = {}
            if self.stdout:
                self._fileobj2output[self.stdout] = []
            if self.stderr:
                self._fileobj2output[self.stderr] = []

        stdout = self._fileobj2output.get(self.stdout)
        stderr = self._fileobj2output.get(self.stderr)

        thread_list = []
        for fileobj in (self.stdin, self.stdout, self.stderr):
            if fileobj is None:
                continue
            if fileobj in self._fileobj2thread:
                thread = self._fileobj2thread[fileobj]
            else:
                if fileobj == self.stdin:
                    target, args = self._stdin_write, (input,)
                else:
                    target, args = self._read_output, (fileobj,)
                thread = threading.Thread(target=target, args=args,
                            daemon=True)
                thread.start()
                self._fileobj2thread[fileobj] = thread
            thread_list.append(thread)

        for thread in thread_list:
            thread.join(self._remaining_time(endtime))
            if thread.is_alive():
                self._check_timeout(endtime, orig_timeout,
                    stdout, stderr, skip_check_and_raise=True)

        # Join partial reads.
        if stdout is not None:
            stdout = b''.join(stdout)
        if stderr is not None:
            stderr = b''.join(stderr)

        if self.text_mode:
            if stdout is not None:
                stdout = self._translate_newlines(stdout,
                    self.stdout.encoding, self.stdout.errors)
            if stderr is not None:
                stderr = self._translate_newlines(stderr,
                    self.stderr.encoding, self.stderr.errors)

        return (stdout, stderr)


    def _cancel_io(self):
        if not hasattr(self, '_fileobj2thread'):
            return
        for fileobj in (self.stdin, self.stdout, self.stderr):
            thread = self._fileobj2thread.get(fileobj)
            if thread is None or not thread.is_alive():
                continue
            try:
                handle = _winapi.OpenThread(
                    _winapi.TERMINATE_THREAD, False, thread.ident)
            except OSError:
                pass
            else:
                try:
                    try:
                        _winapi.CancelSynchronousIo(handle)
                    except OSError:
                        pass
                finally:
                    _winapi.CloseHandle(handle)


    def __exit__(self, exc_type, value, traceback):
        if _mswindows:
            self._cancel_io()

        # rest unchanged...