converter.rb

require 'celluloid'
require './lib/discog_downloader/product'

module DiscogDownloader
  class Converter
    include Celluloid

    def initialize(queue)
      @queue = queue
    end

    def convert(node)
      @queue << Product.new(:id => node['id']) 
    end
  end
end

node_parser.rb

require 'nokogiri'
require 'celluloid'
module DiscogDownloader
  class Parser
    include Celluloid
    attr_reader :reader, :next_pool

    def initialize(io, next_pool)
      @reader     = Nokogiri::XML::Reader.from_io(io)
      @next_pool  = next_pool
    end

    def parse!
      reader.each do |node|
        if node.name == "release" && node.node_type == Nokogiri::XML::Reader::TYPE_ELEMENT
          next_pool.async.parse(node.outer_xml)
        end
      end
      # done. wat do?
    end
  end
end

parser.rb

require 'nokogiri'
require 'celluloid'
module DiscogDownloader
  class Parser
    include Celluloid
    attr_reader :reader, :next_pool

    def initialize(io, next_pool)
      @reader     = Nokogiri::XML::Reader.from_io(io)
      @next_pool  = next_pool
    end

    def parse!
      reader.each do |node|
        if node.name == "release" && node.node_type == Nokogiri::XML::Reader::TYPE_ELEMENT
          next_pool.async.parse(node.outer_xml)
        end
      end
      # done. wat do?
    end
  end
end

test.rb

require './lib/discog_downloader/node_parser'
require './lib/discog_downloader/parser'
require './lib/discog_downloader/converter'
require 'thread'
require 'benchmark'

queue = Queue.new

converter_pool = DiscogDownloader::Converter.pool(:args => [queue])
node_parser_pool = DiscogDownloader::NodeParser.pool(:args => [converter_pool])

parser = DiscogDownloader::Parser.new(ARGF, node_parser_pool)

parser.async.parse!

# wait for everyone to finish?

it seems like your NodeParser and Parser are the same -- was that a copy/paste error? if this is all "fire and forget", the end of the line Celluloid actor (it looks like Converter in this case) must complete the final task. is that to save them in a database after the xml is converted to raw attributes?