java.lang.Object
net.i2p.stat.RateSampleDelivery
class RateSampleDeliveryextends Object
Delivers coalesced rate samples to RateSummaryListeners off the coalesce thread.

This exists because the natural place to notify a summary listener — inline in Rate.coalesce() — puts two things on one thread that must not be there. coalesce() runs on the shared SimpleTimer2 pool, which has a fixed two threads for every periodic event in the router, and it is reached from StatManager.coalesceStats() while that method holds the StatManager monitor. A listener that touches storage does I/O: the graph listeners write a sample into an RRD database, which ultimately writes a FileChannel to a shared scratch file. A slow or stalled disk therefore blocks a pool thread while it holds two monitors, and the whole router's periodic scheduling degrades with it — silently, because a blocked thread neither throws nor logs.

Moving delivery here decouples the two concerns. The counters a rate exposes (getLastTotalValue() and friends) are still updated synchronously under the rate's own monitor, so nothing that reads coalesced statistics changes behaviour; only the listener callback moves. A stalled disk now costs graph history and nothing else.

The queue is bounded, and what happens when it fills is the whole point of this class. Storage can only hold one value per archive step, so two samples for the same step are not two pieces of history — only the newer one is writable. A same-step sample therefore supersedes the one behind it (see submit(RateSummaryListener, double, long, double, long, long)) instead of queueing behind it: that is not a loss, and counting it as one would be a lie. Only when the hard capacity is reached is something genuinely evicted, which is counted by getOverruns().

Samples for a single rate are delivered in order because delivery is single-threaded and the queue is FIFO, which Rate relies on: an RRD rejects a sample whose timestamp does not advance.

The consumer holds no monitor when it calls a listener, and never takes a Rate, RateStat or StatManager lock, so a listener blocked in I/O cannot keep a rate from coalescing.

  • Field Details

    • DEFAULT_CAPACITY

      static final int DEFAULT_CAPACITY
      Pending samples retained before the oldest is evicted. Sized to cover a few coalesce cycles of the whole rate table at once; the graph listeners are the only consumers in practice.
      See Also:

    • SHUTDOWN_JOIN_MS

      private static final long SHUTDOWN_JOIN_MS
      How long shutdown() waits for the consumer before abandoning whatever is still queued.
      See Also:

    • _queue

      private final LinkedBlockingDeque<RateSampleDelivery.Sample> _queue
      The samples waiting to be delivered.

    • _capacity

      private final int _capacity
      Retained capacity, kept for the overrun report.

    • _running

      private final AtomicBoolean _running
      True while the delivery thread should run.

    • _overruns

      private final AtomicLong _overruns
      Overruns reported so far; see reportOverruns() for the consumer-only reader.

    • _superseded

      private final AtomicLong _superseded
      Samples dropped in favour of newer ones.

    • _delivered

      private final AtomicLong _delivered
      Samples delivered to their listeners.

    • _reportedOverruns

      private long _reportedOverruns
      Overruns already written to the log. Touched only by the consumer thread, so it needs no synchronisation.

    • _context

      private final I2PAppContext _context
      The app context.

    • _thread

      private final Thread _thread
      The thread delivering the samples.

  • Constructor Details

    • RateSampleDelivery

      RateSampleDelivery(I2PAppContext context)
      Create a delivery with the default pending-sample capacity.
      Parameters:
      context - router context, used for the logger; may be null

    • RateSampleDelivery

      RateSampleDelivery(I2PAppContext context, int capacity)
      Create a delivery that retains at most the given number of pending samples.
      Parameters:
      context - router context, used for the logger; may be null
      capacity - pending samples to retain before the oldest is evicted

  • Method Details

    • submit

      boolean submit(RateSummaryListener listener, double totalValue, long eventCount, double totalEventTime, long period, long timestampMs)
      Queue one coalesced sample for delivery.

      Never blocks and never throws: this is on the coalesce path, where a stall is the failure being guarded against.

      If the tail entry is the same listener and belongs to the same archive step, it is replaced rather than appended to. Storage holds one value per step, so the older of the two can never be written and the newer is the only one worth delivering. The replacement is counted by getSuperseded() so the decision is visible rather than silent. Only a hard capacity overflow evicts anything, and that is counted by getOverruns().

      Parameters:
      listener - the listener to notify; a null listener is refused rather than queued, because most rates have no listener and an entry that fails on the consumer thread costs a logged exception per rate per period
      totalValue - the period's total value
      eventCount - the period's event count
      totalEventTime - the period's accumulated event time
      period - the rate period in ms
      timestampMs - the sample's archive step stamp, as stamped by Rate.coalesce()
      Returns:
      true if the sample was queued or superseded an existing entry, false if the listener was null, if it was rejected after shutdown(), or if an eviction could not be completed

    • supersedeTail

      private boolean supersedeTail(RateSampleDelivery.Sample s)
      Replace the tail entry if it is the same listener for the same archive step. O(1): one peek, no scan, and identity rather than equality on the listener because two distinct listeners for one rate compare equal but only one of them is the registered instance.
      Parameters:
      s - the incoming sample
      Returns:
      true if the tail was replaced by s

    • getOverruns

      long getOverruns()
      Samples evicted because the queue was at capacity and the next sample could not be held. This is the counter that must stay zero: nonzero means graph history has a hole in it, and the router's own rate counters are unaffected because they are updated before delivery.
      Returns:
      the running overrun count

    • getSuperseded

      long getSuperseded()
      Samples replaced by a newer sample for the same archive step and listener.

      Not a loss: storage keeps one value per step, so the replaced sample could not have been written. Counted because "how many samples were merged away" is otherwise unanswerable from the outside.

      Returns:
      the running supersession count

    • getDelivered

      long getDelivered()
      Samples handed to a listener.
      Returns:
      the running count of samples passed to the listener so far

    • getPending

      int getPending()
      Samples waiting to be delivered.
      Returns:
      the number of samples queued for the consumer to deliver next

    • getOldestPendingAgeMs

      long getOldestPendingAgeMs()
      How long the oldest pending sample has been waiting, from the step stamp Rate.coalesce() gave it.

      This is the live signal that the consumer has stopped making progress, and it keeps rising while a listener is wedged — which is exactly when getOverruns() stops being able to report anything useful.

      Returns:
      the age in ms of the head of the queue, or 0 when nothing is pending

    • isStopped

      boolean isStopped()
      True once shutdown() has been called.
      Returns:
      true after shutdown() has begun tearing the consumer down, false while the delivery thread is still accepting samples

    • shutdown

      void shutdown()
      Stop accepting samples, then wait briefly for the consumer to finish what it already holds.

      The join is what makes teardown ordering meaningful. Without it, a caller that closes the listeners' databases immediately after this returns can race an in-flight delivery and fail a write against a closed backend. The wait is bounded because the consumer may be parked inside a listener's I/O, and a shutdown must not hang on a stalled disk; anything still queued after the timeout is abandoned, which is the correct trade at shutdown.

    • run

      private void run()
      Delivers the queued samples to their listeners.

    • reportOverruns

      private void reportOverruns()
      Write any overrun the consumer has not yet reported, at ERROR.

      Reported here rather than from submit(RateSummaryListener, double, long, double, long, long) because submit runs under the Rate and StatManager monitors on the shared coalesce timer, where a log call that can block is not allowed. The cost is that a wedged consumer delays its own report until it unblocks — acceptable, because getOldestPendingAgeMs() and getOverruns() expose the same condition to the console's ledger continuously and without needing a log line to have been written first. Not rate-limited: an overrun must never be absorbed, so every advance of the counter produces a record.