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 Summary

    Fields
    Modifier and Type
    Field
    Description
    (package private) static final int
    DEFAULT_CAPACITY
    Pending samples retained before the oldest is evicted.

  • Constructor Summary

    Constructors
    Constructor
    Description
    RateSampleDelivery(I2PAppContext context)
    Create a delivery with the default pending-sample capacity.
    RateSampleDelivery(I2PAppContext context, int capacity)
    Create a delivery that retains at most the given number of pending samples.

  • Method Summary

    All Methods
    Modifier and Type
    Method
    Description
    (package private) long
    getDelivered()
    Samples handed to a listener.
    (package private) long
    getOldestPendingAgeMs()
    How long the oldest pending sample has been waiting, from the step stamp Rate.coalesce() gave it.
    (package private) long
    getOverruns()
    Samples evicted because the queue was at capacity and the next sample could not be held.
    (package private) int
    getPending()
    Samples waiting to be delivered.
    (package private) long
    getSuperseded()
    Samples replaced by a newer sample for the same archive step and listener.
    (package private) boolean
    isStopped()
    True once shutdown() has been called.
    (package private) void
    shutdown()
    Stop accepting samples, then wait briefly for the consumer to finish what it already holds.
    (package private) boolean
    submit(RateSummaryListener listener, double totalValue, long eventCount, double totalEventTime, long period, long timestampMs)
    Queue one coalesced sample for delivery.

    Methods inherited from class Object

    clone, equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • 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:

  • 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

    • 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.