RAII-manage enqueue semaphore permits - #22
Conversation
This change replaces the explicit acquire/try_acquire_for plus release API with an RAII "permit" mechanism that enforces correct releases. The pairing between acquire and release was not really hermetic: * check_confirm_timeout() removed timed-out messages from m_unconfirmed and reported ConfirmTimeout, but didn't release the semaphore for them. This could leak capacity and enough timeouts might exhaust max_unconfirmed eventually. * send() acquired the permit before calling accumulator->add(), which throws InvalidInputException for a message over max_frame_size. That exception isn't one send() catches, so it also leaked a permit. * handle_publish_confirm()/handle_publish_error() released unconditionally, even when the publishing id wasn't found in m_unconfirmed (a duplicate or late confirm for an already-timed-out message). This could over-release. Semaphore::acquire()/try_acquire_for() now return a Permit that releases on drop. AccumulatedMessage holds one for as long as it's outstanding (queued, in flight, or unconfirmed) so the permit's lifetime is tied directly to the message's lifetime instead of depending on every code path that can end that lifetime (confirm, error, timeout, oversized rejection, or shutdown) to remember to release it itself. For context: the official Java client's equivalent (unconfirmedMessagesSemaphore) doesn't have this problem because every path that removes an entry from unconfirmedMessages funnels through the same release-on-removal logic. RAII enforces this in the type rather than in the calling code.
|
Thanks for this PR! The semaphore class was intended to be mostly a simple implementation for a std::counting_semaphore since initially we didn't target c++20 for the library. However, there was some more exotic stuff added to it for the library over time, so I think deviating from the standard API is warranted in this case, as it makes usage less error prone. We're not really attached to the java client either, it was just a good starting point to structure the API of the library, the implementation doesn't have to adhere to its exact semantics (that, and it probably evolved at lot since hareflow was implemented). Good catch on the edge cases! |
|
Thanks! I took a quick look at counting_semaphore and I think it might not buy very much anyways since the semaphore here has extra functionality like |
2 participants
I noticed that the semaphore on the producer side could miscount available credit in a few edge-cases like messages that are too large (semaphore acquired and then an exception is raised) or if there is a confirm timeout (semaphore not released).
I'm more familiar with Rust so naturally I want to use Drop-glue to fix this 😄. The idea is to have the semaphore hand out
Permits. On drop/destroy of anAccumulatedMessage, thePermitreleases its hold on the semaphore.This is a departure from the official Java client, the Java client releases its semaphore in calling code. I think it's more idiomatic to solve this with RAII in C++ since the type can do the book-keeping itself.