code: def prefix_connection(
      src_connection,
      dst_stream,
      prefix,
      index,
      prefix_output_with_timestamp,
  ):
      """
      Prefixes the given source connection with timestamp, a prefix and an index.
      Each line of the source will be prefix in this format, if index and prefix are not None:
          {time}[{index}]<{prefix}>:{line}
      The dst_stream must be text streams.
  
      :param src_connection: source pipe connection
      :param dst_stream: destination text stream
      :param prefix: prefix string
      :param index: index value
      :param prefix_output_with_timestamp: prefix lines in dst_stream with timestamp
      :return: None
      """
  
      def get_context(rank, prefix):
          localtime = (
              time.asctime(time.localtime(time.time()))
              if prefix_output_with_timestamp
              else ""
          )
          return "{time}[{rank}]<{prefix}>:".format(
              time=localtime, rank=str(rank), prefix=prefix
          )
  
      def write(text):
          if index is not None and prefix is not None:
              context = get_context(index, prefix)
              text = context + text
          dst_stream.write(text)
          dst_stream.flush()
  
      # the incremental encoder allows us to decode chunks of utf8 bytes
      # with utf8 characters spread across the boundary chunks
      decoder = codecs.getincrementaldecoder("utf8")()
  
      line_buffer = ""
      while True:
          # read at most that many bytes, but do not wait until that many bytes become available
          # waits for the first bytes so this is not a busy loop
          buf = os.read(src_connection.fileno(), 1000)
  
          # turn the bytes into string decoding them as utf8
          # we need to use an incremental decoder as characters can span multiple bytes
          # where the last character might not be completely in buf
          # see https://github.com/horovod/horovod/issues/2367
          text = decoder.decode(buf or b"", final=not buf)
  
          # write line_buffer out when we reach an \n or \r
          # the latter is used to update the current line (e.g. progress bar)
          # which we want to flush out (and prefix) as soon as possible
          for line in re.split("([\r\n])", text):
              line_buffer += line
              if line == "\r" or line == "\n":
                  write(line_buffer)
                  line_buffer = ""
  
          # not an empty text indicates EOF but an empty buf
          if not buf:
              break
  
      # flush the line buffer if it is not empty
      if len(line_buffer):
          write(line_buffer)
  
      src_connection.close()
  
file_path: horovod/horovod/runner/common/util/safe_shell_exec.py
imports: import codecs
  import multiprocessing
  import os
  import re
  import signal
  import subprocess
  import sys
  import threading
  import time
  import psutil
  from horovod.runner.util.threads import in_thread, on_event
label_desc: Use image node_id 2 for calling a global function with example usage: prefix_connection(src_connection, dst_stream, prefix, index, prefix_output_with_timestamp) and returns: str