Process Pool Packet Parsing

packet_parsing_process_pool.rb

require 'packetgen'

class Worker
  attr_accessor :reader
  attr_accessor :writer
  def initialize
    @reader, @writer = IO.pipe
    @output_reader, @output_writer = IO.pipe
    @process = fork do
      trap "SIGINT" do
        @reader.close
        @writer.close
        exit 
      end
      while true do
        begin
          binary = @reader.gets
          packet = PacketGen.parse(binary)
          next unless packet.headers.first.ethertype
          packet = packet.headers[0]
          @output_writer.puts packet.src << " -> " << packet.dst
          #puts packet.src << " -> " << packet.dst
        rescue 
          # yolo
        end
      end
    end
  end

  def output
    @output_reader
  end

  def kill!
    Process.kill('KILL', @process)
  end
end

class WorkerPool 
  attr_accessor :workers

  def initialize(workers: 0)
    @index = 0
    @workers = []
    workers.times do
      add_worker
    end
  end

  def add_worker
    @workers << Worker.new
  end

  def broadcast(mesg)
    @workers.each do |worker|
      worker.writer.puts mesg
    end
  end

  def randomly_allocate(mesg)
    @workers.sample.writer.puts mesg
  end

  def cleanup!
    @workers.each do |worker|
      worker.kill!
    end
    @workers = []
  end

  def read_outputs
    @workers.each do |worker|
      worker.output do |line|
        puts output
      end
      worker.kill!
    end
  end
end

pool = WorkerPool.new(workers: 8)

trap "SIGINT" do
  pool.cleanup!
  exit 0
end

begin
  fork do
    trap "SIGINT" do
      @reader.close
      @writer.close
      exit 
    end
    loop do
      pool.workers.each do |worker|
        begin
          buffer = ""
          buffer << worker.output.read_nonblock(1) while buffer[-1] != "\n"
          print buffer
        rescue
          next
        end
      end
    end
  end
  PacketGen.capture do |packet|
    pool.randomly_allocate(packet)
  end
  # CTRL+C to exit
ensure
  pool.cleanup!
end