def forward_stream()

in horovod/runner/common/util/safe_shell_exec.py [0:0]

• 31 lines of code
• 11 McCabe index (conditional complexity)

def forward_stream(src_stream, dst_stream, prefix, index):
    def prepend_context(line, rank, prefix):
        localtime = time.asctime(time.localtime(time.time()))
        return '{time}[{rank}]<{prefix}>:{line}'.format(
            time=localtime,
            rank=str(rank),
            prefix=prefix,
            line=line
        )

    def write(text):
        if index is not None:
            text = prepend_context(text, index, prefix)
        dst_stream.write(text)
        dst_stream.flush()

    line_buffer = ''
    while True:
        text = os.read(src_stream.fileno(), 1000)
        if text is None:
            break

        if not isinstance(text, str):
            text = text.decode('utf-8')

        if not text:
            break

        for line in re.split('([\r\n])', text):
            line_buffer += line
            if line == '\r' or line == '\n':
                write(line_buffer)
                line_buffer = ''

    # flush the line buffer if it is not empty
    if len(line_buffer):
        write(line_buffer)

    src_stream.close()