Use Mutex lock released on delivery instead of a sleep wait with timeout and a loop

[mensfeld]

Use Mutex lock released on delivery instead of a sleep wait with timeout and a loop

[mensfeld]

I thought so but wanted to be sure ;)

My only concern is that I was able to shave off one array allocation per message recently and here we will introduce a mutex + condition variable. Those are not expensive objects to create but I want to make sure it does not have a huge impact.

My initial assessment shows, that due to less context switching (sleep 10ms anyhow switches) the GC cost is still lower than context switching but would be nice to have some better numbers. In the end once the response is available, we can immediately nullify the objects references, meaning we would not hold them for the whole time handle is in use but only until report can be generated.

[mensfeld]

Sidenote to myself if I do not pull it off soon: we can use mutex to lock on wait within time boundaries instead of sleep and recheck. This should allow us not to wake up those threads unless total timeout or release occurs. The release should also occur on failure of delivery, hence we should pass the lock as a reference similar to label. While it will create an conditional variable, for any synchronous flow such approach will be more optimal than sleep + recheck. Especially since sleep is either 10ms and wakes up, rechecks and always adds 10m to measurable latency.

@thijsc what do you think about such approach? Did you consider it when working on the producer? Do you find such solution viable? Lowering sleep to 1ms or lower can cause CPU cycles for nothing.

[mensfeld]

actually not cond variable but mutex but concept stays the same.

[mensfeld]

@nijikon I am considering providing this as an option OR building a waterdrop synchronization on top of all of it by utilizing the label API. I think we need to have a discussion whether we should do it here or in waterdrop.

1. We could do it here if it would replace the current sleep based mechanism but this would break the API (unless we would leave the wait_timeout variable).

2. We could however also use the label-like (or label per se) API on top of waterdrop to provide the auto-locked and unlocked state on top of things. This however would mean we would break the waterdrop API as we would have to introduce a new error class for timeouts.

3. We could have this configurable in rdkafka-ruby, that is users not waiting on messages and just relying on callback could just stay with sleep, which would save them 2 objects (mutex + condition variable) while others could use mutex + cond variable for better sync performance.

That said, I think it would be much better to have a fully "sleepless" flow for any delivery handles, hence option 1 with "wait_time" being a deprecation option AS LONG as the memory footprint would not be impacted.

[mensfeld]

My POC showed that when using a fast cluster on fast network or local it could reduce the wait time by 42% and had a 16% better CPU efficiency.

[thijsc]

This sounds good to me. I never consider the approach. Fun fact: I was actually not aware condition variables exist in Ruby. Sounds like a great fit.

[mensfeld]

So the way I see it without breaking the current APIs is as follows:

1. We keep the wait_time as a kwarg not to break the API but as of now it will not do anything. We could issue a logger warning in the future versions and slowly migrate.
2. We use a mutex + condition variable on the dispatch inside the delivery handle and lock it until report availability
3. On report availability we unlock + broadcast the condition variable (broadcast and not signal as wait can be in theory on multiple threads)
4. Code that elevates this can be found here: https://github.com/karafka/karafka/blob/master/lib/karafka/processing/timed_queue.rb
5. We need to make sure we compensate spurious wakeups of the condition wait
6. I am not sure if we need to have separate mutex on each produce or of we could use some sort of a per producer / per thread one since anyhow it needs to be passed through handler to the delivery success or failure callback
7. We need to make sure that unlock happens on all delivery events - that is, on success, failure as well as on purge (critical). I do not remember how this flow operates from rdkafka so this requires investigation.

[mensfeld]

Ideally if we could get by with a shared mutex for locking + conditional variable, giving the recent save on array allocation during dispatch, we would be almost "even" in terms of object allocations needed.

[mensfeld]

ah and we need to rebind the timeout error to match one used at the moment.