def pprocess_rows
      thread_number = [@inst.sources.size,
                       @inst.options[:n_threads]].min
      queue = Queue.new
      @inst.sources.each { |p| queue << p }
      threads = thread_number.times.map do
        Thread.new do
          begin
            while p = queue.pop(true)
              p.each { |row| process row }
            end
          rescue ThreadError
          end
        end
      end
      threads.each(&:join)
    end
def pprocess_rows
      thread_number = [@inst.sources.size,
                       @inst.options[:n_threads]].min
      queue = Queue.new
      @inst.sources.each { |p| queue << p }
      threads = thread_number.times.map do
        Thread.new do
          begin
            while p = queue.pop(true)
              p.each { |row| process row }
            end
          rescue ThreadError
          end
        end
      end
      threads.each(&:join)
    end
process rows in parallel