java.lang.Object
net.i2p.internal.QueuedI2CPMessageReader.InternalI2CPDispatcher
All Implemented Interfaces:
Runnable
Enclosing class:
QueuedI2CPMessageReader
private static class QueuedI2CPMessageReader.InternalI2CPDispatcherextends Object implements Runnable
Shared dispatcher pool that multiplexes all internal I2CP message readers.

The ready queue holds readers that have work: one arrives at registration, is handed back after a pass that filled its budget, and is picked up again by the idle tick if a message turns up for it later. A reader that drained is left out, so on an idle system the queue empties, IDLE_POLL_MS actually parks a worker instead of timing out against a queue the workers keep refilling themselves, and surplus workers can retire. The idle cost is therefore a single worker however many clients are connected.

Reoffering after a full pass goes to the back of the queue, so a client with a backlog cannot monopolise the pool. Readers picked up by the idle tick are offered in no particular order, which only matters when several become ready in the same tick.

The pool never falls to zero workers while a reader is registered. Nothing wakes a pool that has gone to sleep, so a reader stranded without one would sit on a queue nobody drains until it filled, at which point the router starts failing to send to that client. register(QueuedI2CPMessageReader) and workerStopped(boolean) both run under this monitor, so a registration can never land between a worker's decision to exit and its departure.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    (package private) static final long
    IDLE_POLL_MS
    Idle poll interval, which is also the latency floor for a queued message, and the spacing between retirement of surplus workers.
    private long
    lastRetireMs
    When the last retirement happened, to shed one worker per idle tick.
    private int
    liveThreads
    Live workers, plus workers that have decided to exit but not yet decremented.
    (package private) static final int
    MAX_WORKERS
    Ceiling on concurrent workers, started on demand rather than up front.
    private final Set<QueuedI2CPMessageReader>
    readers
    The readers currently registered.
    private final LinkedBlockingQueue<QueuedI2CPMessageReader>
    ready
    Readers waiting to be served.
    (package private) static final long
    SLOW_PASS_WARN_MS
    A pass slower than this means a client's message handler is blocking the pool.
    private int
    threadCount
    Thread name counter; prefix plus digits stays within the 12-char convention.

  • Constructor Summary

    Constructors
    Modifier
    Constructor
    Description
    private
    InternalI2CPDispatcher()
    Dispatches the messages read from the client.

  • Method Summary

    Modifier and Type
    Method
    Description
    (package private) void
    grow()
    Add a worker when the ready backlog is deeper than the pool can serve serially.
    private boolean
    hasReaders()
    Whether any reader is registered, and so whether the pool must stay up.
    private boolean
    isSurplus()
    Whether this worker can retire, keeping at least one for the registered readers.
    private boolean
    mayRetire()
    Reserve a retirement slot for the calling worker, shedding at most one worker per idle tick.
    private void
    offerPendingReaders()
    Offer every registered reader that has messages waiting.
    (package private) void
    register(QueuedI2CPMessageReader reader)
    Add a reader and make sure something will serve it.
    void
    run()
     
    private void
    startWorker()
    Start one worker.
    (package private) void
    unregister(QueuedI2CPMessageReader reader)
    Stop serving a reader.
    private static void
    warnIfSlowPass(QueuedI2CPMessageReader reader, long startedMs)
    Warn when one pass over a reader took long enough to look like a blocked handler.
    private void
    workerStopped(boolean alreadyCounted)
    Account for a worker leaving, whatever the reason, and uphold the invariant that a registered reader always has a worker.

    Methods inherited from class Object

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

  • Field Details

    • readers

      private final Set<QueuedI2CPMessageReader> readers
      The readers currently registered.

    • ready

      private final LinkedBlockingQueue<QueuedI2CPMessageReader> ready
      Readers waiting to be served. Requeued readers go to the back, so no client monopolises it.

    • liveThreads

      private int liveThreads
      Live workers, plus workers that have decided to exit but not yet decremented.

    • lastRetireMs

      private long lastRetireMs
      When the last retirement happened, to shed one worker per idle tick.

    • threadCount

      private int threadCount
      Thread name counter; prefix plus digits stays within the 12-char convention.

    • IDLE_POLL_MS

      static final long IDLE_POLL_MS
      Idle poll interval, which is also the latency floor for a queued message, and the spacing between retirement of surplus workers.
      See Also:

    • MAX_WORKERS

      static final int MAX_WORKERS
      Ceiling on concurrent workers, started on demand rather than up front.

      Only reached while several clients are backed up at once. At that point one worker per client would be unbounded, and 8 concurrent workers is already far more than the queue drain rate needs.

      See Also:

    • SLOW_PASS_WARN_MS

      static final long SLOW_PASS_WARN_MS
      A pass slower than this means a client's message handler is blocking the pool.
      See Also:

  • Constructor Details

    • InternalI2CPDispatcher

      private InternalI2CPDispatcher()
      Dispatches the messages read from the client.

  • Method Details

    • register

      void register(QueuedI2CPMessageReader reader)
      Add a reader and make sure something will serve it.
      Parameters:
      reader - the reader to serve

    • unregister

      void unregister(QueuedI2CPMessageReader reader)
      Stop serving a reader. No monitor: readers is concurrent, and a worker re-checks membership before requeueing, so an entry left in the ready queue is skipped rather than served.
      Parameters:
      reader - the reader to drop

    • grow

      void grow()
      Add a worker when the ready backlog is deeper than the pool can serve serially.

    • startWorker

      private void startWorker()
      Start one worker. Caller must hold the monitor.

    • hasReaders

      private boolean hasReaders()
      Whether any reader is registered, and so whether the pool must stay up.
      Returns:
      true if a reader is registered

    • isSurplus

      private boolean isSurplus()
      Whether this worker can retire, keeping at least one for the registered readers.
      Returns:
      true if more workers are live than the readers need

    • mayRetire

      private boolean mayRetire()
      Reserve a retirement slot for the calling worker, shedding at most one worker per idle tick.

      Every parked worker notices the same idle moment, so a plain surplus check would let all of them decide to leave before any of them is counted, collapsing the pool to zero and restarting it. Spacing the retirements by one tick steps the pool down to the single worker the design promises instead. The count is decremented here, under the monitor, so the caller must not decrement it again.

      Returns:
      true if the caller is surplus and must exit

    • offerPendingReaders

      private void offerPendingReaders()
      Offer every registered reader that has messages waiting.

      A reader whose queue was drained is not requeued by the worker that served it, so this is how a message arriving afterwards is still noticed. Called once per idle tick, which bounds the cost to one pass over the registered readers per IDLE_POLL_MS - the latency floor a queued message already had.

    • workerStopped

      private void workerStopped(boolean alreadyCounted)
      Account for a worker leaving, whatever the reason, and uphold the invariant that a registered reader always has a worker. The check and the restart happen in one critical section so a concurrent registration cannot be missed.
      Parameters:
      alreadyCounted - true when the worker reserved its retirement slot in mayRetire() and so must not be counted again

    • run

      public void run()
      Specified by:
      run in interface Runnable

    • warnIfSlowPass

      private static void warnIfSlowPass(QueuedI2CPMessageReader reader, long startedMs)
      Warn when one pass over a reader took long enough to look like a blocked handler. The stall that matters here is invisible from the router side: the client simply stops draining its queue until the queue is full.
      Parameters:
      reader - the reader that was served
      startedMs - wall-clock ms the pass began