Utility to demultiplex stdout and stderr when reading frames from the
    socket.
557576
def demux_adaptor(stream_id, data):
    
    if stream_id == STDOUT:
        return (data, None)
    elif stream_id == STDERR:
        return (None, data)
    else:
        raise ValueError('{0} is not a valid stream'.format(stream_id))
922736
def debug(msg):
    '''
    
    '''
    if DEBUG:
        sys.stderr.write("DEBUG: " + msg + "\n")
        sys.stderr.flush()
1356915
def DviPsStrFunction(target = None, source= None, env=None):
    """"""
    if env.GetOption("no_exec"):
        result = env.subst('$PSCOM',0,target,source)
    else:
        result = ''
    return result
287107
def stderr_avail(self):
        """"""
        data = self.interpreter.stderr_write.empty_queue()
        if data:
            self.write(data, error=True)
            self.flush(error=True)
151225
def emit_message(self, lines):
        

        for line in lines:
            print(line, file=sys.stderr)
502976
def stderr(self):
        
        streams = self._payload.get('streams', None)
        return streams[1] if streams is not None and len(streams) >= 2 else ''
741414
def stderr_handle(self):
        """"""
        if self.stderr_file is not None:
            # Needed to prevent any error that could appear.
            # See issue 6267
            try:
                handle = codecs.open(self.stderr_file, 'w', encoding='utf-8')
            except Exception:
                handle = None
        else:
            handle = None

        return handle
933925
def _decompress_into_buffer(self, out_buffer):
        
        zresult = lib.ZSTD_decompressStream(self._decompressor._dctx,
                                            out_buffer, self._in_buffer)

        if self._in_buffer.pos == self._in_buffer.size:
            self._in_buffer.src = ffi.NULL
            self._in_buffer.pos = 0
            self._in_buffer.size = 0
            self._source_buffer = None

            if not hasattr(self._source, 'read'):
                self._finished_input = True

        if lib.ZSTD_isError(zresult):
            raise ZstdError('zstd decompress error: %s' %
                            _zstd_error(zresult))

        # Emit data if there is data AND either:
        # a) output buffer is full (read amount is satisfied)
        # b) we're at end of a frame and not in frame spanning mode
        return (out_buffer.pos and
                (out_buffer.pos == out_buffer.size or
                 zresult == 0 and not self._read_across_frames))
28958
def get_simple_output(self, stderr=STDOUT):
        
        args = shlex.split(self.cmd)
        proc = Popen(args, stdout=PIPE, stderr=stderr)
        return proc.communicate()[0].decode("utf8")
783249
private void peek(String command, Socket socket, DataInputStream din) throws IOException {

        DataOutputStream dout = new DataOutputStream(socket.getOutputStream());
        byte[] obs;
        boolean done;

        lock.lock();
        try {
            obs = getObservation(false);
            done = envState.done;
        } finally {
            lock.unlock();
        }

        dout.writeInt(obs.length);
        dout.write(obs);

        dout.writeInt(1);
        dout.writeByte(done ? 1 : 0);

        dout.flush();
    }
1270016
def _capture_stderr(callable: Callable[..., CaptureResult], *args, **kwargs) -> CaptureResult:
        
        stream = StringIO()
        with redirect_stderr(stream):
            return_value = callable(*args, **kwargs)
        assert return_value.stderr is None, "stderr appears to have already been captured"
        return CaptureResult(return_value=return_value.return_value, stdout=return_value.stdout,
                             stderr=stream.getvalue(), exception=return_value.exception)
1111435
def read_from_buffer(cls, buf, identifier_str=None):
        """"""
        try:
            return cls._read_from_buffer(buf, identifier_str)
        except Exception as e:
            cls._load_error(e, identifier_str)
270615
def handle_stderr(stream)
      @err_array = []
      # synchronize access to @err_array
      @semaphore = Mutex.new
      Thread.new do
        until (line = stream.gets).nil?
          line.strip!
          @semaphore.synchronize { @err_array << line if line.size > 3 }
        end
      end
    end
298854
def tee2(process, filter):
    
    while True:
        line = process.stderr.readline()
        if line:
            if sys.version_info[0] >= 3:
                line = decode(line)
            stripped_line = line.rstrip()
            if filter(stripped_line):
                sys.stderr.write(line)
        elif process.returncode is not None:
            process.stderr.close()
            break
1369850
def _handle_stdout_event(self, fd, events):
        """"""

        assert fd == self.fd_stdout

        if events & self.ioloop.READ:
            # got data ready to read
            data = ''

            # Now basically we have two cases: either the client supports
            # HTTP/1.1 in which case we can stream the answer in chunked mode
            # in HTTP/1.0 we need to send a content-length and thus buffer the complete output
            if self.request.supports_http_1_1():
                if not self.headers_sent:
                    self.sent_chunks = True
                    self.headers.update({'Date': get_date_header(), 'Transfer-Encoding': 'chunked'})
                    data = 'HTTP/1.1 200 OK\r\n' + '\r\n'.join([ k + ': ' + v for k, v in self.headers.items()]) + '\r\n\r\n'

                    if self.output_prelude:
                        data += hex(len(self.output_prelude))[2:] + "\r\n" # cut off 0x
                        data += self.output_prelude + "\r\n"

                    self.headers_sent = True

                payload = os.read(fd, 8192)
                if events & self.ioloop.ERROR: # there might be data remaining in the buffer if we got HUP, get it all
                    remainder = True
                    while remainder != '': # until EOF
                        remainder = os.read(fd, 8192)
                        payload += remainder

                data += hex(len(payload))[2:] + "\r\n" # cut off 0x
                data += payload + "\r\n"

            else:
                if not self.headers_sent:
                    # Use the over-eager blocking read that will get everything until we hit EOF
                    # this might actually be somewhat dangerous as noted in the subprocess documentation
                    # and lead to a deadlock. This is only a legacy mode for HTTP/1.0 clients anyway,
                    # so we might want to remove it entirely anyways
                    payload = self.process.stdout.read()
                    self.headers.update({'Date': get_date_header(), 'Content-Length': str(len(payload))})
                    data = 'HTTP/1.0 200 OK\r\n' + '\r\n'.join([ k + ': ' + v for k, v in self.headers.items()]) + '\r\n\r\n'
                    self.headers_sent = True
                    data += self.output_prelude + payload
                else:
                    # this is actually somewhat illegal as it messes with content-length but 
                    # it shouldn't happen anyways, as the read above should have read anything
                    # python docs say this can happen on ttys...
                    logger.error("This should not happen")
                    data = self.process.stdout.read()

            if len(data) == 8200:
                self.number_of_8k_chunks_sent += 1
            else:
                if self.number_of_8k_chunks_sent > 0:
                    logger.debug('Sent %d * 8192 bytes', self.number_of_8k_chunks_sent)
                    self.number_of_8k_chunks_sent = 0

                logger.debug('Sending stdout to client %d bytes: %r', len(data), data[:20])
            self.request.write(data)

        # now we can also have an error. This is because tornado maps HUP onto error
        # therefore, no elif here!
        if events & self.ioloop.ERROR:
            logger.debug('Error on stdout')
            # ensure file is closed
            if not self.process.stdout.closed:
                self.process.stdout.close()
            # remove handler
            self.ioloop.remove_handler(self.fd_stdout)
            # if all fds are closed, we can finish
            return self._graceful_finish()
1094249
def _read_stderr(self):
        """"""
        # We need to read stderr_file as bytes to be able to
        # detect its encoding with chardet
        f = open(self.stderr_file, 'rb')

        try:
            stderr_text = f.read()

            # This is needed to avoid showing an empty error message
            # when the kernel takes too much time to start.
            # See issue 8581
            if not stderr_text:
                return ''

            # This is needed since the stderr file could be encoded
            # in something different to utf-8.
            # See issue 4191
            encoding = get_coding(stderr_text)
            stderr_text = to_text_string(stderr_text, encoding)
            return stderr_text
        finally:
            f.close()
731870
def _pipe_stdio(cls, sock, stdin_isatty, stdout_isatty, stderr_isatty, handle_stdin):
    """"""
    stdio_writers = (
      (ChunkType.STDOUT, stdout_isatty),
      (ChunkType.STDERR, stderr_isatty)
    )
    types, ttys = zip(*(stdio_writers))

    @contextmanager
    def maybe_handle_stdin(want):
      if want:
        # TODO: Launching this thread pre-fork to handle @rule input currently results
        # in an unhandled SIGILL in `src/python/pants/engine/scheduler.py, line 313 in pre_fork`.
        # More work to be done here in https://github.com/pantsbuild/pants/issues/6005
        with NailgunStreamStdinReader.open(sock, stdin_isatty) as fd:
          yield fd
      else:
        with open('/dev/null', 'rb') as fh:
          yield fh.fileno()

    with maybe_handle_stdin(handle_stdin) as stdin_fd,\
         NailgunStreamWriter.open_multi(sock, types, ttys) as ((stdout_fd, stderr_fd), writer),\
         stdio_as(stdout_fd=stdout_fd, stderr_fd=stderr_fd, stdin_fd=stdin_fd):
      # N.B. This will be passed to and called by the `DaemonExiter` prior to sending an
      # exit chunk, to avoid any socket shutdown vs write races.
      stdout, stderr = sys.stdout, sys.stderr
      def finalizer():
        try:
          stdout.flush()
          stderr.flush()
        finally:
          time.sleep(.001)  # HACK: Sleep 1ms in the main thread to free the GIL.
          writer.stop()
          writer.join()
          stdout.close()
          stderr.close()
      yield finalizer