2929import org .slf4j .Logger ;
3030import org .slf4j .LoggerFactory ;
3131
32- import java .util .concurrent .LinkedBlockingDeque ;
33- import java .util .concurrent .TimeUnit ;
3432import java .util .concurrent .atomic .AtomicLong ;
33+ import java .util .concurrent .locks .LockSupport ;
3534
3635/**
37- * Bounded inbox + lazy-started daemon thread that delivers ack-watermark
38- * advances to a user-supplied {@link SenderProgressHandler} off the I/O thread.
36+ * Single-slot watermark mailbox + lazy-started daemon thread that delivers
37+ * ack-watermark advances to a user-supplied {@link SenderProgressHandler} off
38+ * the I/O thread.
3939 *
4040 * <h2>Why a separate thread</h2>
4141 * The I/O thread must never block on user code. The same rationale as the
4242 * {@link SenderErrorDispatcher} sibling: a slow handler (e.g. journal write)
4343 * cannot stall send progress. The I/O thread {@link #offer offers} the new
44- * watermark and continues; the daemon dispatcher takes from the queue and
44+ * watermark and continues; the daemon dispatcher takes the latest value and
4545 * invokes the handler.
4646 *
47- * <h2>Why drop-on-overflow is safe here</h2>
48- * Watermarks are <em>monotonically increasing</em>. If the inbox is full and an
49- * offer is dropped, the next successful offer carries a higher (or equal) FSN
50- * — so any handler watching for "ackedFsn >= target" still catches up. This
51- * is the key reason a bounded queue is acceptable for what is otherwise a
52- * high-volume signal: drops compress, they don't lose information.
47+ * <h2>Why a single-slot mailbox is sufficient</h2>
48+ * Watermarks are <em>monotonically increasing</em>. If the handler is busy
49+ * when several offers arrive in succession, only the highest watermark is
50+ * worth delivering — earlier advances are subsumed by it. The mailbox is
51+ * therefore a single {@code volatile long} that the producer overwrites
52+ * unconditionally; the consumer reads it and delivers anything strictly
53+ * greater than its previous delivery.
5354 *
54- * <p>Implementation detail: a {@link LinkedBlockingDeque} is used so on full
55- * inbox we drop the <em>oldest</em> entry rather than the newest. The freshest
56- * watermark is the one the handler cares about, and a sustained burst should
57- * not bury it behind stale values.
55+ * <p>This collapses a high-volume signal to a constant memory footprint and
56+ * keeps the I/O thread on a zero-allocation hot path: no autoboxing, no
57+ * per-offer node allocation, no {@code java.util.*} structures.
5858 *
5959 * <h2>Lifecycle</h2>
6060 * The dispatcher thread is started lazily on the first successful
@@ -69,24 +69,30 @@ public final class SenderProgressDispatcher implements QuietCloseable {
6969
7070 public static final int DEFAULT_CAPACITY = 256 ;
7171 private static final long DRAIN_DEADLINE_NANOS = 100_000_000L ; // 100 ms
72+ // Park interval used when the mailbox is idle. Bounded so a missed unpark
73+ // (e.g. from a future code path) cannot wedge the dispatcher.
74+ private static final long IDLE_PARK_NANOS = 100_000_000L ; // 100 ms
7275 private static final Logger LOG = LoggerFactory .getLogger (SenderProgressDispatcher .class );
73- // Sentinel pushed during close() to nudge the dispatcher out of poll().
74- // Identity-compared in the loop body; never delivered to the handler.
75- // Box the sentinel so it has reference identity distinct from any real
76- // value the I/O thread might offer.
77- private static final Long POISON = Long .MIN_VALUE ;
78- private final AtomicLong dropped = new AtomicLong ();
76+ // Sentinel for "no value ever offered". Real FSNs are non-negative, so
77+ // Long.MIN_VALUE is unambiguous and the dispatcher's "deliver if greater
78+ // than lastDelivered" check naturally skips it.
79+ private static final long EMPTY = Long .MIN_VALUE ;
7980 // volatile so the producer of progress events can swap the handler
8081 // post-connect. Each delivery reads the current reference; concurrent
8182 // updates may interleave a final-old / first-new delivery, which is
8283 // acceptable for the watermark contract (monotonic + idempotent).
8384 private volatile SenderProgressHandler handler ;
84- private final LinkedBlockingDeque <Long > inbox ;
8585 private final Object lock = new Object ();
8686 private final String threadName ;
8787 private final AtomicLong totalDelivered = new AtomicLong ();
88+ private final AtomicLong totalOffered = new AtomicLong ();
8889 private volatile boolean closed ;
8990 private Thread dispatcherThread ;
91+ // Single-slot inbox. Producer (I/O thread) overwrites unconditionally;
92+ // consumer reads via volatile and skips when the value has already been
93+ // delivered. Monotonic FSNs make overwrite safe -- the latest value
94+ // subsumes any prior undelivered value.
95+ private volatile long latestFsn = EMPTY ;
9096
9197 public SenderProgressDispatcher (SenderProgressHandler handler , int capacity ) {
9298 this (handler , capacity , "qdb-sf-progress-dispatcher" );
@@ -96,11 +102,13 @@ public SenderProgressDispatcher(SenderProgressHandler handler, int capacity, Str
96102 if (handler == null ) {
97103 throw new IllegalArgumentException ("handler must be non-null" );
98104 }
105+ // Capacity is retained on the constructor for API compatibility but
106+ // the mailbox is single-slot by design (see class doc). The bound is
107+ // still validated so misconfiguration surfaces at construction.
99108 if (capacity < 1 ) {
100109 throw new IllegalArgumentException ("capacity must be >= 1, was " + capacity );
101110 }
102111 this .handler = handler ;
103- this .inbox = new LinkedBlockingDeque <>(capacity );
104112 this .threadName = threadName ;
105113 }
106114
@@ -111,9 +119,11 @@ public void close() {
111119 return ;
112120 }
113121 closed = true ;
114- inbox .offer (POISON );
115122 Thread t = dispatcherThread ;
116123 if (t != null ) {
124+ // Wake the dispatcher in case it is parked. After observing
125+ // closed=true with no pending value, the loop returns.
126+ LockSupport .unpark (t );
117127 long deadline = System .nanoTime () + DRAIN_DEADLINE_NANOS ;
118128 long remainingMillis ;
119129 while ((remainingMillis = (deadline - System .nanoTime ()) / 1_000_000L ) > 0 ) {
@@ -126,7 +136,7 @@ public void close() {
126136 }
127137 if (t .isAlive ()) {
128138 LOG .warn ("progress-dispatcher thread did not exit within drain deadline; "
129- + "abandoning {} queued advances" , inbox . size () );
139+ + "abandoning latest pending advance" );
130140 t .interrupt ();
131141 }
132142 dispatcherThread = null ;
@@ -135,13 +145,17 @@ public void close() {
135145 }
136146
137147 /**
138- * Total advances dropped since startup due to inbox-overflow. Non-zero
139- * means the user's handler is slower than the ack rate — typically not a
140- * correctness issue (later advances catch up), but useful as an ops signal
141- * when handler latency matters.
148+ * Watermark advances overwritten before the handler observed them, since
149+ * startup. Computed as {@code totalOffered - totalDelivered}; a small
150+ * non-zero value typically reflects a single in-flight advance, while a
151+ * persistently growing value signals a handler slower than the ack rate.
152+ * Useful as an ops signal when handler latency matters.
142153 */
143154 public long getDroppedNotifications () {
144- return dropped .get ();
155+ long offered = totalOffered .get ();
156+ long delivered = totalDelivered .get ();
157+ long diff = offered - delivered ;
158+ return diff > 0L ? diff : 0L ;
145159 }
146160
147161 /**
@@ -164,72 +178,57 @@ public void setHandler(SenderProgressHandler handler) {
164178
165179 /**
166180 * Non-blocking enqueue of an ack-watermark advance. Returns {@code true}
167- * if the value will be delivered (eventually, on the dispatcher thread).
168- * Returns {@code false} if the dispatcher was closed.
181+ * if the value will be visible to the dispatcher (eventually delivered,
182+ * possibly subsumed by a later advance). Returns {@code false} if the
183+ * dispatcher was closed.
169184 *
170- * <p>On a full inbox this method evicts the oldest queued value to make
171- * room for the new one — the freshest watermark is what observers want.
172- * The dropped-old count is bumped for ops visibility.
173- *
174- * <p>Lazy-starts the dispatcher thread on the first successful offer.
185+ * <p>Watermarks are monotonic, so the slot is overwritten unconditionally:
186+ * a later advance subsumes any earlier one not yet observed by the
187+ * handler. Lazy-starts the dispatcher thread on the first offer.
175188 */
176189 public boolean offer (long ackedFsn ) {
177190 if (closed ) {
178191 return false ;
179192 }
180- // Try the fast path first: a non-full inbox accepts immediately.
181- if (inbox .offerLast (ackedFsn )) {
182- if (dispatcherThread == null ) {
183- startDispatcherIfNeeded ();
184- }
185- return true ;
186- }
187- // Inbox full: evict the oldest entry and try again. We do this rather
188- // than dropping the newest so the handler always sees the most recent
189- // watermark even under sustained backpressure.
190- inbox .pollFirst ();
191- dropped .incrementAndGet ();
192- boolean accepted = inbox .offerLast (ackedFsn );
193- if (accepted && dispatcherThread == null ) {
193+ totalOffered .incrementAndGet ();
194+ latestFsn = ackedFsn ;
195+ Thread t = dispatcherThread ;
196+ if (t == null ) {
194197 startDispatcherIfNeeded ();
198+ t = dispatcherThread ;
195199 }
196- return accepted ;
200+ if (t != null ) {
201+ LockSupport .unpark (t );
202+ }
203+ return true ;
197204 }
198205
199206 private void dispatchLoop () {
200- // Local high-water filter: skip any queued value <= the last one we
201- // already delivered, in case multiple advances queued while we were
202- // running the handler. The contract guarantees the user only sees
203- // strictly-increasing values.
204- long lastDelivered = Long .MIN_VALUE ;
205- while (!closed || !inbox .isEmpty ()) {
206- Long v ;
207- try {
208- v = inbox .poll (100 , TimeUnit .MILLISECONDS );
209- } catch (InterruptedException e ) {
210- if (closed ) {
211- return ;
207+ // Strictly-monotonic delivery: skip the slot when its value is not
208+ // greater than the last value handed to the handler.
209+ long lastDelivered = EMPTY ;
210+ while (true ) {
211+ long v = latestFsn ;
212+ if (v > lastDelivered ) {
213+ lastDelivered = v ;
214+ // Increment before invoking the handler so observers using a
215+ // CountDownLatch in the handler can read the updated counter
216+ // once their latch fires.
217+ totalDelivered .incrementAndGet ();
218+ SenderProgressHandler h = handler ;
219+ try {
220+ h .onAcked (v );
221+ } catch (Throwable t ) {
222+ LOG .error ("SenderProgressHandler threw on fsn={}: {}" , v , t .getMessage (), t );
212223 }
213- Thread .currentThread ().interrupt ();
214- continue ;
215- }
216- if (v == null || v .equals (POISON )) {
217- continue ;
218- }
219- long fsn = v ;
220- if (fsn <= lastDelivered ) {
221224 continue ;
222225 }
223- lastDelivered = fsn ;
224- // Increment before invoking the handler so observers using a
225- // CountDownLatch in the handler can read the updated counter
226- // once their latch fires.
227- totalDelivered .incrementAndGet ();
228- try {
229- handler .onAcked (fsn );
230- } catch (Throwable t ) {
231- LOG .error ("SenderProgressHandler threw on fsn={}: {}" , fsn , t .getMessage (), t );
226+ if (closed ) {
227+ return ;
232228 }
229+ // No new value; park until the producer unparks us or the idle
230+ // tick fires. Spurious wakeups loop back to the top and re-check.
231+ LockSupport .parkNanos (this , IDLE_PARK_NANOS );
233232 }
234233 }
235234
0 commit comments