Can we run multiple workers for a delayed queue

[junwchina]

Hello,

The delayed queue is a good feature. But It seems there is a issue when a delayed queue consumed by multiple workers. The implementation on WorkerImpl#pop will return a null when it is processing the same job with the another worker. That will be a big performance issue if there are many collisions like this.

[gresrun]

@junwchina Do you have a proposal for how to address this issue?

[junwchina]

@gresrun Yes, I have a thought. The basic idea is all ready jobs should be pushed to regular queue which is implemented by list type.

Let's say each queue will have two internal queues, one is a delayed queue in zset, the another is a regular queue in list. In the main loop, I will handle the queue's delayed queue firstly, and push all ready delayed jobs into it's regular queue. No matter there is any ready delayed jobs, I will handle the queue's regular queue then. Since the regular queue is a list type, we can use lpop or rpop to get a job to run. These operators are atomic, so we don't need to care the multiple thread issue. It ensures every thread should can get a job to run if it's not empty.

Also, there is a rpoplpush operator in redis which is atomic too, maybe we can use it to implement lpoplpush. The current implementation will get collision too on multiple threads environments.

That's all, please let me know how do you think.

[argvk]

hey @junwchina, moving that to the main loop sounds nice, but this issue can crop up again when running jesque on multiple machines.
Another way would be to perform zrange and zrem within a multi/exec, that way it acts like an atomic zpop.
Something like this:

> multi
OK
> zrangebyscore resque:queue:fooSchedule -1 1450151702000 withscores
QUEUED
> zremrangebyscore resque:queue:fooSchedule -1 1450151702000
QUEUED
> EXEC
1) 1) "{\"class\":\"fooScheduleJob\",\"args\":[54625],\"vars\":null}"
   2) "1450151700000"
2) (integer) 1

> multi
OK
> zrangebyscore resque:queue:fooSchedule -1 1450151701000 withscores
QUEUED
> zremrangebyscore resque:queue:fooSchedule -1 1450151701000
QUEUED
> EXEC
1) (empty list or set)
2) (integer) 0

related: https://groups.google.com/forum/#!msg/redis-db/ur9U8o-Sko0/tgefLK3zrzQJ

cc @gresrun

[gresrun]

@argvk @junwchina @zhangliuping
This definitely looks like an issue that needs to be addressed. I'm currently in the process of relocating from Florida to California so, I am currently not able to take a look at this. Pull requests are most welcome!

[junwchina]

@argvk Yes, Multi/Exec is a way to fix this issue. Can you clarify why you think my solution will have issue on multiple machines?

I prefer to use two internal queues. The reason is:

1. zremrangebyscore's time complex is O(log(N)), but lpop or rpop is only O(1). I think it's better to save as few as possible jobs in delayed queue. So in the main loop, I would like to get all ready jobs from delayed queue by one operator, and push all of them into regular queue.
2. Since all running jobs are fetching from the internal regular queue. So we can enqueue all kinds of jobs into one queue.
3. is coming.. :)

@argvk @gresrun Please let me know how do you think.

[argvk]

@junwchina I meant to say jesque instances running on multiple machines, might have concurrency issues, due to multiple main loops running across them all. Totally agree with your points, the MULTI/EXEC is in addition to moving it to the main loop.

thoughts ?
👍

[junwchina]

@argvk I think that should not be a problem on multiple machines. We just need to ensure every redis operator must be atomic. That would be possible based on current redis version.

[argvk]

@junwchina OK, I'm trying to work on this and I having a little difficulty understanding this part

I would like to get all ready jobs from delayed queue by one operator,

AFAIK, zremrangebyscore only returns the numbers of elements removed. So, is there something else I can use to return and remove the elements atomically ?

[junwchina]

@argvk Just like you said in the previous message. You can use zrangebyscore and zremrangebyscore together within Multi/Exec.

[abhinavdwivedi]

Any implementation which will require 2 extra redis calls will not scale well under load. Even the query to check type is overloading the redis in my case. This is because redis performance in VM environments is not very good. I am getting around 40k-50k ops per second on a VM as opposed to dedicated hardware where i get around 200K

[gresrun]

Here's a novel idea: replace the majority of WorkerImpl.pop() with an EVAL of a lua script that returns the next job.

local queueKey = KEYS[1]
local inFlightKey = KEYS[2]
local freqKey = KEYS[3]
local now = ARGV[1]

local payload = nil

local not_empty = function(x)
  return (type(x) == 'table') and (not x.err) and (#x ~= 0)
end

local ok, queueType = next(redis.call('TYPE', queueKey))
if queueType == 'zset' then
    local i, lPayload = next(redis.call('ZRANGEBYSCORE', queueKey, '-inf', now, 'WITHSCORES'))
    if lPayload then
        payload = lPayload
        local frequency = redis.call('HGET', freqKey, payload)
        if frequency then
            redis.call('ZINCRBY', queueKey, frequency, payload)
        else
            redis.call('ZREM', queueKey, payload)
        end
    end
elseif queueType == 'list' then
    payload = redis.call('LPOP', queueKey)
    if payload then
        redis.call('LPUSH', inFlightKey, payload)
    end
end

return payload

Since the evaluation of scripts are atomic, this is guaranteed to be safe for multiple workers. Thoughts?

cc @junwchina @argvk @abhinavdwivedi

[gresrun]

Take a look at 5721a43 (on a new branch lua_pop) for a proof-of-concept. As a side-note, the Lua-based pop() is over twice as fast in my micro-benchmark!

cc @junwchina @argvk @abhinavdwivedi