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.

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    (package private) static interface 
    PacketDispatcher.PacketProcessor
    Process one dispatched (connection, packet) pair.

  • Constructor Summary

    Constructors
    Constructor
    Description
    PacketDispatcher(int workerCount, int queueCapacity, PacketDispatcher.PacketProcessor processor)
    Create and start the shard workers.
    PacketDispatcher(int workerCount, int queueCapacity, PacketDispatcher.PacketProcessor processor, I2PAppContext context)
    Create and start the shard workers.

  • Method Summary

    Modifier and Type
    Method
    Description
    (package private) void
    dispatch(long sendStreamId, Connection con, Packet packet)
    Route (con, packet) to the shard owning the connection and return promptly.
    (package private) int
    getWorkerCount()
    Current worker count (shard array length).
    (package private) void
    resize(int n)
    Rebuild the shard array at a new size.
    (package private) 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.
    (package private) static int
    shardFor(long sendStreamId, int nThreads)
    Deterministic, in-range shard index for a connection's inbound stream id.
    (package private) void
    shutdown()
    Stop the workers and free queued-but-unprocessed packets.

    Methods inherited from class Object

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

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

      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

    • 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