|
# Copyright 2014 Ryan Biesemeyer |
|
# License: MIT |
|
# |
|
# Sometimes your redis needs a little help keeping up with expiries. |
|
# |
|
# First, it's helpful to know how [expiries][http://redis.io/commands/expire] |
|
# are handled in redis internals: |
|
# |
|
# > Redis keys are expired in two ways: a passive way, and an active way. |
|
# > |
|
# > A key is actively expired simply when some client tries to access it, and |
|
# > the key is found to be timed out. |
|
# > |
|
# > Of course this is not enough as there are expired keys that will never be |
|
# > accessed again. This keys should be expired anyway, so periodically Redis |
|
# > test a few keys at random among keys with an expire set. All the keys that |
|
# > are already expired are deleted from the keyspace. |
|
# > |
|
# > Specifically this is what Redis does 10 times per second: |
|
# > |
|
# > Test 100 random keys from the set of keys with an associated expire. |
|
# > Delete all the keys found expired. |
|
# > If more than 25 keys were expired, start again from step 1. |
|
# > |
|
# > This is a trivial probabilistic algorithm, basically the assumption is that |
|
# > our sample is representative of the whole key space, and we continue to |
|
# > expire until the percentage of keys that are likely to be expired is under |
|
# > 25% |
|
# > |
|
# > This means that at any given moment the maximum amount of keys already |
|
# > expired that are using memory is at max equal to max amount of write |
|
# > operations per second divided by 4. |
|
# |
|
# Effectively, there are scenarios where there are just too many expiring keys |
|
# in a dataset for redis to be able to keep up with them via its passive method, |
|
# so we need to take matters into our own hands. |
|
# |
|
# @example |
|
# ~~~ ruby |
|
# redis = Redis.new(connection_params) |
|
# reh = RedisExpiryHelper.new(redis) |
|
# reh.run |
|
# ~~~ |
|
# |
|
class RedisExpiryHelper |
|
# @param redis [Redis,Redis::Namespace] |
|
def initialize(redis) |
|
if defined?(Redis::Namespace) && redis.kind_of?(Redis::Namespace) |
|
redis = redis.instance_exec { @redis } |
|
end |
|
@redis = redis |
|
|
|
@counts = Hash.new(0) |
|
end |
|
attr_reader :counts |
|
attr_reader :redis |
|
|
|
# Requests thousands of keys from redis and their ttls, pipelined. |
|
# Effectively kicks off the `active` clause in expiry reaping. |
|
# @param options [Hash{Symbol=>Object}] |
|
# @option options [IO] :io ($stderr) |
|
# @option options [Integer] :chunk (1000) |
|
# @option options [Integer] :throttle (1.0) |
|
def run(options = {}) |
|
io = options.fetch(:io, $stderr) |
|
chunk = options.fetch(:chunk, 1000) |
|
throttle = options.fetch(:throttle, 1.0) |
|
|
|
keep_processing do |
|
counts[:total] += 1 |
|
|
|
if (counts[:total] % 100).zero? |
|
io && io.puts(counts.inspect) |
|
sleep(throttle) unless throttle <= 0 |
|
end |
|
|
|
# First, get a chunk of random keys |
|
random_keys = redis.pipelined do |
|
chunk.times { redis.randomkey } |
|
end |
|
|
|
# Then access all of the keys to invoke `active` reaping |
|
redis.pipelined do |
|
random_keys.each do |random_key| |
|
redis.ttl(random_key) |
|
end |
|
end.each do |ttl| |
|
counts[ttl > 0] += 1 |
|
end |
|
end |
|
end |
|
|
|
private |
|
|
|
def keep_processing |
|
keep_processing = true |
|
backup_trap = trap('INT') { keep_processing = false } |
|
|
|
while(keep_processing) do |
|
yield |
|
end |
|
ensure |
|
if backup_trap.kind_of?(Proc) |
|
trap('INT', &backup_trap) |
|
else |
|
trap('INT', backup_trap) |
|
end |
|
end |
|
end |
