<commit_before># Parallel copy-on-write streaming (PCOWS)

require 'tempfile'

class PCOWS

  RUNNINGEXT = 'part' # file extension
  
  def initialize(num_threads,name=File.basename(__FILE__),timeout=180,quiet=false)
    num_threads = cpu_count() if not num_threads # FIXME: set to cpu_num by default
    # $stderr.print "Using ",num_threads,"threads \n"
    @num_threads = num_threads
    @pid_list = []
    @name = name
    @timeout = timeout
    if multi_threaded
      @tmpdir =  Dir::mktmpdir(@name+'_')
    end
    @last_output = 0 # counter
    @output_locked = nil
    @quiet = quiet
  end

  # Feed the worker 'func and state' to COWS. Note that func is a
  # lambda closure so it can pick up surrounding scope at invocation
  # in addition to the data captured in 'state'.
  
  def submit_worker(func,state)
    pid = nil
    if multi_threaded
      count = @pid_list.size+1
      fn = mktmpfilename(count)
      pid = fork do
        # ---- This is running a new copy-on-write process
        tempfn = fn+'.'+RUNNINGEXT
        STDOUT.reopen(File.open(tempfn, 'w+'))
        func.call(state).each { | line | print line }
        STDOUT.flush
        STDOUT.close
        FileUtils::mv(tempfn,fn)
        exit 0
      end
    else
      # ---- Single threaded: call in main process and output immediately
      func.call(state).each { | line | print line }
    end
    @pid_list << [ pid,count,fn ]
    return true
  end

  # Make sure no more than num_threads are running at the same time -
  # this is achieved by checking the PID table and the running files
  # in the tmpdir

  def wait_for_worker_slot()
    return if single_threaded
    Timeout.timeout(@timeout) do
    
      while true
        # ---- count running pids
        running = @pid_list.reduce(0) do | sum, info |
          (pid,count,fn) = info
          if pid_or_file_running?(pid,fn)
            sum+1
          else
            sum
          end
        end
        return if running < @num_threads
        $stderr.print "Waiting for slot (timeout=#{@timeout})\n" if not @quiet
        sleep 0.1
        
      end
    end
  end

  # ---- In this section the output gets collected and passed on to a
  #      printer thread. This function makes sure the printing is
  #      ordered and that no printers are running at the same
  #      time. The printer thread should be doing as little processing
  #      as possible.
  #
  #      In this implementation type==:by_line will call func for
  #      each line. Otherwise it is called once with the filename.
  def process_output(func=nil,type=:by_line, blocking=false)
    return if single_threaded
    output = lambda { |fn|
      if type == :by_line
        File.new(fn).each_line { |buf|
          print buf
        }
      else
        func.call(fn)
      end
    }
    if @output_locked
      # ---- is the other thread still running?
      (pid,count,fn) = @output_locked
      return if File.exist?(fn)  # yes, thread still processing
      @last_output += 1          # get next one in line
      @output_locked = nil
    end
    # Walk the pid list to find the next one
    if info = @pid_list[@last_output]
      (pid,count,fn) = info
      if File.exist?(fn)
        # Yes! We have the next output, create outputter
        $stderr.print "Processing #{fn}\n" if not @quiet
        if not blocking
          pid = fork do
            output.call(fn)
            $stderr.print "Removing #{fn}\n" if not @quiet
            File.unlink(fn)
            exit(0)
          end
          @output_locked = info
        else<commit_msg>This may have been subtle<commit_after># Parallel copy-on-write streaming (PCOWS)

require 'tempfile'

class PCOWS

  RUNNINGEXT = 'part' # file extension
  
  def initialize(num_threads,name=File.basename(__FILE__),timeout=180,quiet=false)
    num_threads = cpu_count() if not num_threads # FIXME: set to cpu_num by default
    # $stderr.print "Using ",num_threads,"threads \n"
    @num_threads = num_threads
    @pid_list = []
    @name = name
    @timeout = timeout
    if multi_threaded
      @tmpdir =  Dir::mktmpdir(@name+'_')
    end
    @last_output = 0 # counter
    @output_locked = nil
    @quiet = quiet
  end

  # Feed the worker 'func and state' to COWS. Note that func is a
  # lambda closure so it can pick up surrounding scope at invocation
  # in addition to the data captured in 'state'.
  
  def submit_worker(func,state)
    pid = nil
    if multi_threaded
      count = @pid_list.size+1
      fn = mktmpfilename(count)
      pid = fork do
        # ---- This is running a new copy-on-write process
        tempfn = fn+'.'+RUNNINGEXT
        STDOUT.reopen(File.open(tempfn, 'w+'))
        func.call(state).each { | line | print line }
        STDOUT.flush
        STDOUT.close
        FileUtils::mv(tempfn,fn)
        exit 0
      end
    else
      # ---- Single threaded: call in main process and output immediately
      func.call(state).each { | line | print line }
    end
    @pid_list << [ pid,count,fn ]
    return true
  end

  # Make sure no more than num_threads are running at the same time -
  # this is achieved by checking the PID table and the running files
  # in the tmpdir

  def wait_for_worker_slot()
    return if single_threaded
    Timeout.timeout(@timeout) do
    
      while true
        # ---- count running pids
        running = @pid_list.reduce(0) do | sum, info |
          (pid,count,fn) = info
          if pid_or_file_running?(pid,fn)
            sum+1
          else
            sum
          end
        end
        return if running < @num_threads
        $stderr.print "Waiting for slot (timeout=#{@timeout})\n" if not @quiet
        sleep 0.1
        
      end
    end
  end

  # ---- In this section the output gets collected and passed on to a
  #      printer thread. This function makes sure the printing is
  #      ordered and that no printers are running at the same
  #      time. The printer thread should be doing as little processing
  #      as possible.
  #
  #      In this implementation type==:by_line will call func for
  #      each line. Otherwise it is called once with the filename.
  def process_output(func=nil,type=:by_line, blocking=false)
    return if single_threaded
    output = lambda { |fn|
      if type == :by_line
        File.new(fn).each_line { |buf|
          print buf
        }
      else
        func.call(fn)
      end
    }
    if @output_locked
      # ---- is the other thread still running?
      (pid,count,fn) = @output_locked
      return if File.exist?(fn)  # yes, thread still processing
      @last_output += 1          # get next one in line
      @output_locked = nil
    end
    # Walk the pid list to find the next one
    if info = @pid_list[@last_output]
      (pid,count,fn) = info
      if File.exist?(fn)
        # Yes! We have the next output, create outputter
        $stderr.print "Processing #{fn}\n" if not @quiet
        if blocking==false
          pid = fork do
            output.call(fn)
            $stderr.print "Removing #{fn}\n" if not @quiet
            File.unlink(fn)
            exit(0)
          end
          @output_locked = info
        else
{'fraction_non_alphanumeric': 0.0883298392732355, 'fraction_numerical': 0.0033542976939203353, 'mean_word_length': 2.097835497835498, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2600833', 'n_tokens_mistral': 2182, 'n_tokens_neox': 2133, 'n_words': 726}