Batch redis atomic pop push benchmark

rpoplpushn-benchmark.rb

#!/usr/bin/env ruby

require 'benchmark'
require 'redis'
require 'wolverine'

ITERATIONS = 200
BATCH_SIZE = 200
PROCESS_COUNT = 8
redis = Redis.new(:host => 'localhost')
Wolverine.config.redis = redis
Wolverine.config.script_path = Pathname.new(File.expand_path("app/wolverine"))

def setup(redis, queue)
  redis.del(queue)
  (PROCESS_COUNT * ITERATIONS * 2).times do |i|
    redis.lpush(queue, BATCH_SIZE.times.to_a)
  end
end

RANGETRIM_QUEUE = "bm:rangetrim:queue"
RPOPLPUSH_QUEUE = "bm:rpoplpush:queue"
LUA_QUEUE = "bm:rngtrmlua:queue"

puts "setup:"
m = Benchmark.measure do
  [RANGETRIM_QUEUE, RPOPLPUSH_QUEUE, LUA_QUEUE].each do |queue|
    setup(redis, queue)
  end
end
puts m
puts

def run
  PROCESS_COUNT.times do
    ITERATIONS.times do
      yield
    end
  end
  Process.waitall
end

def lrange_ltrim_test(redis)
  batch = redis.lrange(RANGETRIM_QUEUE, -BATCH_SIZE, -1)
  redis.ltrim(RANGETRIM_QUEUE, 0, -BATCH_SIZE - 1)
end

RPOPLPUSH_WORK = "bm:rpoplpush:work"

def rpoplpush_test(redis)
  batch = redis.pipelined do
    BATCH_SIZE.times { redis.rpoplpush(RPOPLPUSH_QUEUE, RPOPLPUSH_WORK) }
  end
  redis.del(RPOPLPUSH_WORK)
end

def lua_test(redis)
  batch = Wolverine.util.rpoplpushn([LUA_QUEUE, RPOPLPUSH_WORK], [BATCH_SIZE])
  redis.del(RPOPLPUSH_WORK)
  raise "rpoplpushn return batch of size #{batch.size}" if batch.size < BATCH_SIZE
end

Benchmark.bmbm do |x|
  x.report("rpoplpush") { run{ rpoplpush_test(redis) } }
  x.report("lrange ltrim") { run{ lrange_ltrim_test(redis) } }
  x.report("lua") { run{ lua_test(redis) } }
end

[RANGETRIM_QUEUE, RPOPLPUSH_QUEUE].each do |queue|
  len = redis.llen(queue)
  puts "'#{queue}' queue still has #{len} items left" if len != 0
end

rpoplpushn.lua

local source = KEYS[1]
local dest = KEYS[2]
local batch_size = ARGV[1]

local function reverse(t)
  local newTable = {}
  local j = #t
  for i = 1, #t do
    newTable[i] = t[j]
    j = j - 1
  end
  return newTable
end

local batch = redis.call('lrange', source, -batch_size, -1)
batch = reverse(batch)
if #batch > 0 then
  redis.call('lpush', dest, unpack(batch))
  redis.call('ltrim', source, 0, -#batch - 1)
end
return batch