code: def read_newlines_or_carriage_returns(
      fd: io.RawIOBase,
  ) -> Iterator[str]:
      r"""
      Read lines, delineated by either '\n' or '\r.
  
      Unlike the default io.BufferedReader used in subprocess.Popen(bufsize=-1), we read until we
      encounter either '\n' or \r', and treat that as one line.
  
      Specifically, io.BufferedReader doesn't handle tqdm progress bar outputs very well; it treats
      all of the '\r' outputs as one enormous line.
  
      Args:
          fd: an unbuffered stdout or stderr from a subprocess.Popen.
  
      Yields:
          A series of str, one per line.  Each line always ends with a '\n'.  Each line will be
          broken to length io.DEFAULT_BUFFER_SIZE, even if the underlying io didn't have a linebreak.
      """
      # Ship lines of length of DEFAULT_BUFFER_SIZE, including the terminating newline.
      limit = io.DEFAULT_BUFFER_SIZE - 1
      nread = 0
      chunks: List[bytes] = []
  
      def oneline() -> str:
          nonlocal nread
          nonlocal chunks
          out = b"".join(chunks).decode("utf8")
          chunks = []
          nread = 0
          return out
  
      while True:
          buf = fd.read(limit - nread)
          if not buf:
              # EOF.
              break
  
          # Extract all the lines from this buffer.
          while buf:
              m = lineend.search(buf)
              if m is None:
                  # No line break here; just append to chunks.
                  chunks.append(buf)
                  nread += len(buf)
                  break
  
              # Line break found!
              start, end = m.span()
              chunks.append(buf[:start])
              # Even if we matched a '\r', emit a '\n'.
              chunks.append(b"\n")
              yield oneline()
              # keep checking the rest of buf
              buf = buf[end:]
  
          # Detect if we reached our buffer limit.
          if nread >= limit:
              # Pretend we got a line anyway.
              chunks.append(b"\n")
              yield oneline()
  
      # One last line, maybe.
      if chunks:
          chunks.append(b"\n")
          yield oneline()
  
file_path: determined/harness/determined/launch/wrap_rank.py
imports: import argparse
  import contextlib
  import io
  import os
  import re
  import subprocess
  import sys
  import threading
  from typing import BinaryIO, Iterator, List
  import determined
  from determined import constants
label_desc: Use image node_id 1 for calling a global function with example usage: read_newlines_or_carriage_returns(fd) and returns: out