java.lang.Object
net.i2p.client.streaming.impl.PacketDispatcher
class PacketDispatcherextends Object
Dispatch inbound stream packets to a small pool of single-threaded shard workers, so a slow or hostile connection no longer stalls the session notifier thread for every other connection on the same destination.

Ordering: shardFor(long, int) maps a connection's inbound stream id to exactly one worker, and each worker drains its queue FIFO, so all packets of a given connection are handled by one thread in arrival order. Only packets for different connections run in parallel.

Back-pressure: the per-shard queue is bounded. When it is full the producer (PacketHandler on the session notifier thread) waits briefly for space instead of dropping or reordering; this reproduces the blocking behaviour of the old fully synchronous path, but only for the saturated connection's shard.

Live resize: the router's Tuner can change the worker count at runtime via resizeAll(int) (applies the Tuner's global default to every live manager). A resize is guarded by the dispatcher's write lock so no dispatch is in flight while the shard array is rebuilt; each old shard is drained to idle before its worker is retired, so every connection stays on exactly one thread with no reordering and no dropped packets.

SYN / unknown-stream establishment is intentionally not offloaded; it continues on the notifier thread and the ConnectionHandler. Shutdown replaces the synchronous path once more: shutdown() stops the workers and frees any queued-but-unprocessed packets without leaving the notifier thread blocked.

  • Field Details

    • _live

      private static final Set<PacketDispatcher> _live
      All live dispatchers, so the Tuner can resize every manager at once.

    • _lock

      private final ReentrantReadWriteLock _lock
      Guards the shard array. Dispatch holds the read lock (one reader in practice — a single notifier thread), so it is uncontended on the hot path. Resize and shutdown hold the write lock, excluding every dispatch while the array is rebuilt, an old worker retired, or the dispatcher stopped.

    • _queueCapacity

      private final int _queueCapacity
      Most packets in the dispatcher queue.

    • _processor

      private final PacketDispatcher.PacketProcessor _processor
      The processor handling the packets.

    • _context

      private final I2PAppContext _context
      The app context.

    • _log

      private final Log _log
      The log for the dispatcher.

    • _workers

      private volatile PacketDispatcher.Worker[] _workers
      The threads processing the packets.

    • _running

      private volatile boolean _running
      True while the dispatcher runs.

  • Constructor Details

    • PacketDispatcher

      PacketDispatcher(int workerCount, int queueCapacity, PacketDispatcher.PacketProcessor processor)
      Create and start the shard workers. Package-visible for unit tests.
      Parameters:
      workerCount - number of shards, at least 1 (higher values clamped)
      queueCapacity - queue size per shard, at least 1 (higher clamped)
      processor - not null

    • PacketDispatcher

      PacketDispatcher(int workerCount, int queueCapacity, PacketDispatcher.PacketProcessor processor, I2PAppContext context)
      Create and start the shard workers.
      Parameters:
      workerCount - number of shards, at least 1 (higher values clamped)
      queueCapacity - queue size per shard, at least 1 (higher clamped)
      processor - not null
      context - used to record the receive-backlog / queue-depth rates, may be null in tests (stats disabled)

  • Method Details

    • dispatch

      void dispatch(long sendStreamId, Connection con, Packet packet) throws InterruptedException
      Route (con, packet) to the shard owning the connection and return promptly. Wait for space only while that shard's queue is full. Records the target shard's queue depth and any producer-blind back-pressure.

      The 1ms back-pressure wait runs OUTSIDE the read lock: a saturated shard must not pin the lock, or shutdown() / resize(int) (which need the write lock to flip _running / rebuild the shard array) would deadlock exactly when a worker is stuck behind a full queue. The shard is therefore re-resolved each iteration, so an entry offered across a resize lands on the shard's current worker — a resize drains each old shard to idle before swapping, so ordering is still preserved. A worker retired in the gap between picking it and handing the entry over refuses the handoff rather than keeping it in a queue nothing will drain again, and the shard is resolved once more.

      Parameters:
      sendStreamId - the connection's inbound stream id (shard key)
      con - the resolved connection, passed through to the processor
      packet - the packet to process
      Throws:
      InterruptedException - if interrupted while waiting for queue space

    • resize

      void resize(int n)
      Rebuild the shard array at a new size. Takes the write lock, so no dispatch is in flight. Each old shard is drained to idle (queue empty, no entry mid-process) before its worker is retired and the new workers take over. A no-op when the size is unchanged.
      Parameters:
      n - the new worker count, at least 1 (lower values clamped)

    • resizeAll

      static void resizeAll(int n)
      Resize every live dispatcher to the Tuner's current default, so existing managers (not just ones created afterwards) follow the tunable value.
      Parameters:
      n - the new worker count, at least 1 (lower values clamped)

    • shutdown

      void shutdown()
      Stop the workers and free queued-but-unprocessed packets. Idempotent; after shutdown, dispatch(long, Connection, Packet) becomes a no-op.

    • getWorkerCount

      int getWorkerCount()
      Current worker count (shard array length). Package-visible for tests.
      Returns:
      the current worker count, >= 1

    • newWorkers

      private PacketDispatcher.Worker[] newWorkers(int n)
      Build and start the shard array of the given size.
      Parameters:
      n - the number of worker shards to create and start
      Returns:
      the started workers, in shard order

    • shardFor

      static int shardFor(long sendStreamId, int nThreads)
      Deterministic, in-range shard index for a connection's inbound stream id.

      The id is masked to 31 bits so the result is always non-negative even if the id space wraps. Inbound stream ids are unique per live connection within a manager, giving a stable per-connection (and thus order-preserving) mapping.

      Parameters:
      sendStreamId - inbound stream id, > 0 for known connections
      nThreads - number of shards, >= 1
      Returns:
      0..nThreads-1