chore: reduce pipelining latency by reusing existing shard fibers

[romange]

To prove the benefits, run ./dfly_bench --pipeline=50 -n 20000 --ratio 0:1 --qps=0 --key_maximum=1
Before: the average pipelining latency was 10ms
After: the average pipelining latency is 5ms.
Avg latency: pipelined_latency_usec / total_pipelined_squashed_commands

[dranikpg]

Damn, yes, that's wasteful. nit: we have AwaitRunningOnShardQueue (though without predicate filter)

[RafalBielickiIM]

This one actually is something I wanted to report.

Also I saw that running many evalsha in pipeline causes huge spikes occasionally - p999 are getting close to 500ms.
Disabling LUA atomicity & setting multi_exec_mode=3 reduces the issue but not completely.

Seems like this would fix the issue? Or should I create a ticket?

We are running latest 1.21.1.

[romange]

@RafalBielickiIM could you post an sample lua script here? do you use hashtags in your database?

[romange]

I kept it to keep debugging why there is a memory blow up with RunBlockingInParallel

[dranikpg]

Yeah but you still have to rebuild it, so copy paste replace would've done it 😛

[RafalBielickiIM]

@RafalBielickiIM could you post an sample lua script here? do you use hashtags in your database?
Those are flags in DF we are using

• --unixsocket=/hostdf/dragonfly_0.sock
• --unixsocketperm=660
• --lua_resp2_legacy_float=true
• --default_lua_flags=disable-atomicity
• --multi_exec_mode=3
• --lua_auto_async=true
• --hz=400
• --enable_direct_fd=true
• --proactor_threads=2
• --conn_io_threads=2
• --interpreter_per_thread=50

We have 2 CPU + 8GB requests on those pods. (LIMIT is 3CPU & 12GB)

Looking at traces we see spikes when running 10+ EVALSHA in pipeline.
We have many LUA scripts running on those instances, frequently in pipelines.
While in other services where simpler ops are performed DF is way better, here it play out slower and way less stable than Redis (and this is with atomicity disabled & multiexec mode = 3).

Example LUAs run in there

-- KEYS = {}
-- ARGV = {}

local key = KEYS[1]
local key_prev = KEYS[2]
local key_1 = KEYS[3]
local key_1_next = KEYS[4]
local key_2 = KEYS[5]
local key_2_next = KEYS[6]

local some_id_key = KEYS[7]
local some_id_key_next = KEYS[8]

local some_key =  KEYS[9]
local some_key_prev = KEYS[10]

local some_id = ARGV[1]
local value = ARGV[2]
local exp = tonumber(ARGV[3])

local window_weight = tonumber(ARGV[4])

local size_1 = redis.call('SCARD', key_1) or 0
local size_2 = redis.call('SCARD', key_2) or 0
local size_1_next = redis.call('SCARD', key_1_next) or 0
local size_2_next = redis.call('SCARD', key_2_next) or 0

local some_value_size = redis.call('SCARD', some_id_key) or 0
local some_value_next_size = redis.call('SCARD', some_id_key_next) or 0

local this_counts = redis.call('INCR', key)
local prev_counts = tonumber(redis.call('GET', key_prev) or 0)

local this_some_count = redis.call('INCR', some_key)
local prev_some_count = tonumber(redis.call('GET', some_key_prev) or 0)

local counts = this_counts + prev_counts * window_weight
local some_count = this_some_count + prev_some_count * window_weight

if size_1 < 100 then
    redis.call('SADD', key_1, some_id)
end

if size_2 < 100 then
    redis.call('SADD', key_2, value)
end

if some_value_size < 100 then
    redis.call('SADD', some_id_key, value)
end

if some_value_next_size < 100 then
    redis.call('SADD', some_id_key_next, value)
end

if size_1_next < 100 then
    redis.call('SADD', key_1_next, some_id)
end

if size_2_next < 100 then
    redis.call('SADD', key_2_next, value)
end

if this_counts == 1 then
    local half_ttl = math.floor(exp / 2) + 10
    redis.call('EXPIRE', key_1_next, exp)
    redis.call('EXPIRE', key_2_next, exp)
    redis.call('EXPIRE', key_1, half_ttl)
    redis.call('EXPIRE', key_2, half_ttl)
    redis.call('EXPIRE', key, exp)
end

return {counts, size_1, size_2, some_value_size, some_count}

OR

local d = redis.call('mget', unpack(KEYS))

local last_count = tonumber(d[1]) or 0
local current_count = (tonumber(d[2]) or 0)
local current_time = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local current_rate = last_count * ((window - (current_time % window)) / window) + current_count + 1
if current_rate > limit then
    return math.floor(limit - current_rate)
end
redis.call('incr', KEYS[2])
redis.call('expire', KEYS[2], 2*window)
return math.floor(limit - current_rate)

@romange Maybe you have a suggestion for different config?
Overall in this case lower thread count equals with better AVG, but increasing thread count is getting us more stable p99 & p999.

[dranikpg]

👍🏻

[romange]

@RafalBielickiIM it probably belongs to Discussions section and not to this PR. The short answer is that this PR won't help you with your use-case., sorry.

[dranikpg]

@RafalBielickiIM

1. multiexec mode is for MULTI/EXEC, pipelines are not affected by this setting
2. I see you found --lua_auto_async=true. We currently have it disabled for non-atomic modes even when requested with this flag. Insider tip: you can use redis.apcall if you want to execute a bulk of write-only commands. It only makes sense to use it you have at least ~5 commands per thread. This will trigger the optimization that is modified in this PR.

You can try swapping apcall with call, my total time drops significantly. Please note that apcall is all experimental with DF. The more clients you have, the less will be the impact of this optimization currently

import redis
import time

s = """
redis.pcall('GET', 'NEX')
for i = 1, 20 do
    redis.apcall('PEXPIRE', 'key' .. i, 0)
end
"""

r = redis.Redis()
sha = r.script_load(s)

for _ in range(10):
    p = r.pipeline(transaction=False)
    for _ in range(50):
        p.evalsha(sha, 0)
    s = time.time()
    p.execute()
    print(1000 * (time.time() - s))

[romange]

But I think the latency is caused by the fact the script touches multiple shards so it has tail latency which can not be avoided.
The only way to fight it is to group all the variables under the same slot using the emulated cluster mode but I am guessing you can not do it and probably use Dragonfly to avoid cluster mode.

[dranikpg]

But I think the latency is caused by the fact the script touches multiple shards so it has tail latency which can not be avoided

So the last write commands can be squashed with apcall. But with 10 tail writes and 2-3 shards it's effect is questionable

[romange]

And for the rate limiting script consider using https://www.dragonflydb.io/docs/command-reference/strings/cl.throttle#can-clthrottle-be-used-for-non-api-related-rate-limiting though it's mostly stateless unlike your approach