From: Willy Tarreau Date: Tue, 25 Aug 2026 15:53:59 +0000 (+0200) Subject: MEDIUM: ring: attach readers under the tail lock instead of isolating X-Git-Tag: v3.5-dev6~72 X-Git-Url: https://git.kaiwu.me/?a=commitdiff_plain;h=8a4b50b0743105011e2ea8dbc4a0992466067b04;p=haproxy.git MEDIUM: ring: attach readers under the tail lock instead of isolating ring_dispatch_messages() isolates the thread to grab the initial reader offset, as introduced by commit 31b93b40b ("MEDIUM: ring: protect the initialization of the initial reader offset"), whose message already noted that the tail lock could be used instead. Doing it that way removes a trap for the callers: isolating from a function that may be called with a lock held deadlocks against any thread spinning on that lock, which is what happened with the sink forward applet. Only a writer holding the tail lock may move the head, so taking that same lock is sufficient to perform the initialization while we attach. Contrary to the isolation, this does not exclude a writer that released the tail lock and is still copying its payload, so the readers count byte may transiently be seen as RING_WRITING_SIZE if we're unlucky, but the existing retry loop on values above RING_MAX_READERS already handles that case. This also makes attaching a reader a lot cheaper even if that has never been a problem. Note that this only aims at making the code more robust and tolerant to accidental abuse, it's not a reason for calling ring_dispatch_messages() under a lock anyway! --- diff --git a/src/ring.c b/src/ring.c index ebe3b042c..ae0ccdfbe 100644 --- a/src/ring.c +++ b/src/ring.c @@ -587,15 +587,30 @@ int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t * and keep it while being scheduled. Thus instead let's take it the * first time we enter here so that we have a chance to pass many * existing messages before grabbing a reference to a location. This - * value cannot be produced after initialization. The first offset - * needs to be taken under isolation as it must not move while we're - * trying to catch it. + * value cannot be produced after initialization. The head must not move + * while we're trying to catch it, and since only a writer holding the + * tail lock may move it, taking that same lock is sufficient here. We + * must not use thread_isolate() for this, as some callers may hold a + * lock (e.g. the sink forward applet holds the sft lock), and isolating + * while holding a lock deadlocks against any thread spinning on it, + * since a spinning thread never becomes harmless. */ if (unlikely(*ofs_ptr == ~0)) { - thread_isolate(); + size_t *tail_ptr = &ring->storage->tail; + + /* grab the tail lock, i.e. atomically set the lock bit on the + * tail offset only if it's not already set, exactly like a + * writer does before recycling the oldest messages. + */ + tail_ofs = HA_ATOMIC_LOAD(tail_ptr); + while (1) { + tail_ofs &= ~RING_TAIL_LOCK; + if (HA_ATOMIC_CAS(tail_ptr, &tail_ofs, tail_ofs | RING_TAIL_LOCK)) + break; + __ha_cpu_relax(); + } head_ofs = HA_ATOMIC_LOAD(&ring->storage->head); - tail_ofs = ring_tail(ring); if (flags & RING_WF_SEEK_NEW) { /* going to the end means looking at tail-1 */ @@ -604,13 +619,18 @@ int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t head_ofs -= ring_size; } - /* reserve our slot here (inc readers count) */ + /* reserve our slot here (inc readers count). The byte may still + * be seen as RING_WRITING_SIZE by a writer which released the + * tail lock and is still copying its payload, hence the retry + * on values above RING_MAX_READERS. + */ do { readers = _HA_ATOMIC_LOAD(ring_area + head_ofs); } while ((readers > RING_MAX_READERS || !_HA_ATOMIC_CAS(ring_area + head_ofs, &readers, readers + 1)) && __ha_cpu_relax()); - thread_release(); + /* release the tail lock */ + HA_ATOMIC_STORE(tail_ptr, tail_ofs); /* store this precious offset in our context, and we're done */ *ofs_ptr = head_ofs;