java.lang.Object
net.i2p.router.tunnel.TrivialPreprocessor
net.i2p.router.tunnel.BatchedPreprocessor
All Implemented Interfaces:
TunnelGateway.QueuePreprocessor
Direct Known Subclasses:
BatchedRouterPreprocessor
class BatchedPreprocessorextends TrivialPreprocessor
Batching preprocessor that will briefly delay the sending of a message if it doesn't fill up a full tunnel message, in which case it queues up an additional flush task. This is a very simple threshold algorithm - as soon as there is enough data for a full tunnel message, it is sent. If after the delay there still isn't enough data, what is available is sent and padded.

As explained in the tunnel document, the preprocessor has a lot of potential flexibility in delay, padding, or even reordering. We keep things relatively simple for now.

However much of the efficiency results from the clients selecting the correct MTU in the streaming lib such that the maximum-size streaming lib message fits in an integral number of tunnel messages. See ConnectionOptions in the streaming lib for details.

Aside from obvious goals of minimizing delay and padding, we also want to minimize the number of tunnel messages a message occupies, to minimize the impact of a router dropping a tunnel message. So there is some benefit in starting a message in a new tunnel message, especially if it will fit perfectly if we do that (a 964 or 1956 byte message, for example).

An idea for the future...

If we are in the middle of a tunnel msg and starting a new i2np msg, and this one won't fit, let's look to see if we have something that would fit instead by reordering: if (allocated > 0 && msg.getFragment == 0) { for (j = i+1, j < pending.size(); j++) { if it will fit and it is fragment 0 { msg = pending.remove(j) pending.add(0, msg) } } }.

  • Field Details

    • _pendingSince

      private long _pendingSince
      When the first unsent message was queued, 0 if none.

    • _pendingDelay

      private long _pendingDelay
      How long the current batch may still wait, in ms.

    • _name

      private final String _name
      Name shown in the logs for this preprocessor.

    • _fragmentationBatch

      private final RateStatMeanBatch _fragmentationBatch
      Fragment number of each flushed message, accumulated over the flush and reported as one mean sample. Both stats here were sampled per fragment, which is two stat lookups and two monitors per rate period for every fragment leaving the gateway. Both are means, so one sample per flush reports the same quantity.

    • _writeDelayBatch

      private final RateStatMeanBatch _writeDelayBatch
      Gateway lifetime of each flushed message, weight = its byte length.

    • DEBUG

      private static final boolean DEBUG
      Switch for the verbose batching logs.
      See Also:

    • MIN_FLUSH_DELAY

      static final long MIN_FLUSH_DELAY
      Minimum delay even under full pressure (ms) -- avoids busy-wait retry storms.
      See Also:

    • FULL_SIZE

      private static final int FULL_SIZE
      Full preprocessed message size in bytes (1003).
      See Also:

    • DEFAULT_DELAY

      static long DEFAULT_DELAY
      Not final or private so the test code can adjust.

    • FORCE_BATCH_FLUSH

      private static final int FORCE_BATCH_FLUSH
      If we have this many messages queued that are too small, flush them anyway Even small messages take up about 200 bytes or so.
      See Also:

    • FULL_ENOUGH_SIZE

      private static final int FULL_ENOUGH_SIZE
      If we have this much allocated, flush anyway. Tune this to trade off padding vs. fragmentation. The lower the value, the more we are willing to send off a tunnel msg that isn't full so the next message can start in a new tunnel msg to minimize fragmentation.

      This should be at most FULL_SIZE - (39 + a few), since you want to at least fit in the instructions and a few bytes.

      See Also:

  • Constructor Details

    • BatchedPreprocessor

      public BatchedPreprocessor(RouterContext ctx, String name)
      Create a new preprocessor for batched tunnel messages.
      Parameters:
      ctx - the router context
      name - the preprocessor name

  • Method Details

    • getSendDelay

      protected long getSendDelay()
      Wait up to this long before sending (flushing) a small tunnel message Warning - overridden in BatchedRouterPreprocessor.
      Returns:
      the send delay in ms

    • getDelayAmount

      public long getDelayAmount()
      How long to wait before flushing a queued message.
      Specified by:
      getDelayAmount in interface TunnelGateway.QueuePreprocessor
      Overrides:
      getDelayAmount in class TrivialPreprocessor
      Returns:
      delay in milliseconds

    • getDelayAmount

      private long getDelayAmount(boolean shouldStat)
      Chooses how long the current batch may wait.
      Parameters:
      shouldStat - true to record the delay in the statistics
      Returns:
      the remaining delay, in ms

    • computePendingDelay

      private long computePendingDelay(long allocated, int pendingSize)
      Scale the flush delay inversely with buffer pressure so that near-full buffers flush almost immediately (low latency) while near-empty buffers wait longer (better aggregation).

      Pressure is the max of buffer fullness (allocated/FULL_SIZE) and queue pressure (pending.size/FORCE_BATCH_FLUSH), bounded [0,1].

      delay = MIN_FLUSH_DELAY + (getSendDelay() - MIN_FLUSH_DELAY) * (1 - pressure).

      Parameters:
      allocated - bytes allocated in the current tunnel message
      pendingSize - number of messages waiting to be batched
      Returns:
      the flush delay in ms, from MIN_FLUSH_DELAY up to the configured send delay, lower the higher the buffer pressure

    • preprocessQueue

      public boolean preprocessQueue(List<PendingGatewayMessage> pending, TunnelGateway.Sender sender, TunnelGateway.Receiver rec)
      Description copied from class: TrivialPreprocessor
      Return true if there were messages remaining, and we should queue up a delayed flush to clear them

      NOTE: Unused here, see BatchedPreprocessor override, super is not called.

      Specified by:
      preprocessQueue in interface TunnelGateway.QueuePreprocessor
      Overrides:
      preprocessQueue in class TrivialPreprocessor
      Parameters:
      pending - list of Pending objects for messages either unsent or partly sent. This list should be update with any values removed (the preprocessor owns the lock) Messages are not removed from the list until actually sent. The status of unsent and partially-sent messages is stored in the Pending structure.
      sender - the sender to hand the encrypted data to
      rec - the receiver to accept encrypted data
      Returns:
      true if we should delay before preprocessing again

    • preprocessAndSend

      private boolean preprocessAndSend(List<PendingGatewayMessage> pending, TunnelGateway.Sender sender, TunnelGateway.Receiver rec)
      Body of preprocessQueue(List, TunnelGateway.Sender, TunnelGateway.Receiver), split out so the per-flush stat batches are flushed on every return path.
      Parameters:
      pending - the list of pending messages
      sender - the gateway callback that encrypts and forwards the preprocessed data
      rec - the receiver for preprocessed data
      Returns:
      true if messages remain queued for a later flush

    • display

      private void display(long allocated, List<PendingGatewayMessage> pending, String title)
      Logs the state of the queue, only if Log.INFO. Format: title: allocated: X pending: X (delay: X) [0]:offset/length/lifetime [1]:etc.
      Parameters:
      allocated - bytes already allocated to the batch
      pending - the messages still queued
      title - a label for what triggered the display

    • send

      protected void send(List<PendingGatewayMessage> pending, int startAt, int sendThrough, TunnelGateway.Sender sender, TunnelGateway.Receiver rec)
      Preprocess the messages from the pending list, grouping items startAt through sendThrough (though only part of the last one may be fully sent), delivering them through the sender/receiver.
      Parameters:
      pending - the list of pending messages
      startAt - first index in pending to send (inclusive)
      sendThrough - last index in pending to send (inclusive)
      sender - the gateway callback that encrypts and forwards the preprocessed data
      rec - the receiver for preprocessed data

    • writeFragments

      private int writeFragments(List<PendingGatewayMessage> pending, int startAt, int sendThrough, byte[] target, int offset)
      Write the fragments out of the pending list onto the target, updating each of the Pending message's offsets accordingly.
      Parameters:
      pending - the list of pending messages
      startAt - first index in pending to write (inclusive)
      sendThrough - last index in pending to write (inclusive)
      target - the preprocessed buffer to write the fragments into
      offset - where in target to start writing
      Returns:
      new offset into the target for further bytes to be written