knife_drawer.rb

require 'thread'

module KnifeDrawer

  DEFAULT_NUM_FORKS = 4

  def self.do_with_forks(tasks_hash, num_forks = DEFAULT_NUM_FORKS)
    input_queue = Queue.new
    output_queue = Queue.new

    tasks_hash.each do |key, block|
      input_queue.push([key, block])
    end

    num_forks.times.inject(Array.new) { |threads|
      threads << Thread.new(input_queue) { |queue|

        while task = input_queue.pop(true) rescue nil
          key, block = *task
          rd,wr = IO.pipe

          if pid = fork
            wr.close
            payload = rd.read
            rd.close
            Process.wait(pid)
            output_queue.push([key, Marshal.load(payload)])

          else # in the fork
            rd.close
            wr.write Marshal.dump(block.call)
            wr.close
            exit
          end

        end # end while

      } # end thread block
    }.each { |thread| thread.join }

    # make output queue into a hash
    output = Hash.new
    while key_value = output_queue.pop(true) rescue nil
      key, value = *key_value
      output[key] = value
    end

    output
  end

end