java.lang.Object
net.i2p.router.transport.udp.OutboundMessageFragments
class OutboundMessageFragmentsextends Object
Coordinate the outbound fragments and select the next one to be built. This pool contains messages we are actively trying to send, essentially doing a round robin across each message to send one fragment, as implemented in getNextVolley(). This also honors per-peer throttling, taking note of each peer's allocations. If a message has each of its fragments sent more than a certain number of times, it is failed out. In addition, this instance also receives notification of message ACKs from the InboundMessageFragments, signaling that we can stop sending a message.

  • Field Details

    • _context

      private final RouterContext _context
      The router context.

    • _log

      private final Log _log
      The log for the fragments.

    • _transport

      private final UDPTransport _transport
      The transport owning this job.

    • _activePeers

      private final CopyOnWriteArrayList<PeerState> _activePeers
      List of peers currently sending outbound messages. Thread-safe for iteration and modification with reduced array copying.

    • _peerIndex

      private int _peerIndex
      The index of the next peer to serve.

    • _peersToRemove

      private final CopyOnWriteArrayList<PeerState> _peersToRemove
      The peers to remove from the pool.

    • _waitLock

      private final Object _waitLock
      The lock held while waiting for messages.

    • _alive

      private volatile boolean _alive
      True while the job runs.

    • _builder2

      private final PacketBuilder2 _builder2
      The builder of the SSU packets.

    • MAX_VOLLEYS

      static final int MAX_VOLLEYS
      Maximum number of times to send a packet before failing it.
      See Also:

    • MAX_WAIT_MS

      static final int MAX_WAIT_MS
      Ceiling on any wait taken here, so one peer reporting a long delay cannot stall the pool.

      Doubled on a host SystemVersion.isSlow(), where the extra millisecond buys nothing but headroom. Package-visible so tests assert against the value in force rather than a literal: hardcoding it made OutboundMessageFragmentsDecisionTest pass only on a host with native BigInteger loaded, and fail everywhere else.

    • MIN_WAIT_MS

      static final int MIN_WAIT_MS
      Floor on any wait taken here.

      A round that allocated nothing has to cost a bounded wait. It used to cost none: see foldSendDelay(int, int).

      See Also:

    • _statEmitCounter

      private int _statEmitCounter
      Counter for periodic aggregate stat emission.

    • MEMORY_STAT_INTERVAL

      private static final int MEMORY_STAT_INTERVAL
      How many add() calls between "udp.memory.activePeers" samples. @since 0.9.71+.
      See Also:

    • _memoryStatCounter

      private int _memoryStatCounter
      Counter gating the memory stat. Single sender thread, no synchronization needed.

    • _consumed

      private BitSet _consumed
      Reusable consumed-fragment marker for preparePackets(List, PeerState). Reused across volleys (cleared per call) instead of allocating a fresh BitSet every volley. BitSet.set(int) grows it as needed, so growth is amortized to the largest volley ever seen.

      Only reachable from getNextVolley() on the single PacketPusher thread, so no synchronization is needed.

  • Constructor Details

    • OutboundMessageFragments

      public OutboundMessageFragments(RouterContext ctx, UDPTransport transport)
      Registers the transport's send-rate and memory statistics.
      Parameters:
      ctx - the router context, whose stat manager registers this transport's send rate stats
      transport - the UDP transport that owns the peer states and the packet pusher

  • Method Details

    • startup

      public void startup()
      Start the message pool.

    • shutdown

      public void shutdown()
      Shut down the message pool.

    • dropPeer

      void dropPeer(PeerState peer)
      Remove a peer from the active outbound list and drop its pending messages.
      Parameters:
      peer - the peer state to forget, whose queued outbound messages are discarded

    • add

      public void add(OutNetMessage msg)
      Add a new message to the active pool.
      Parameters:
      msg - the message to queue; it is ignored when its target has no RouterInfo

    • add

      public void add(OutboundMessageState state, PeerState peer)
      Short circuit the OutNetMessage, letting us send the establish complete message reliably. If you have multiple messages, use the list variant, so the messages may be bundled efficiently.
      Parameters:
      state - the message state to queue immediately
      peer - the peer to send it to; null is a programming error

    • add

      public void add(List<OutboundMessageState> states, PeerState peer)
      Short circuit the OutNetMessage, letting us send multiple messages reliably and efficiently.
      Parameters:
      states - the message states to queue, in the order they should be sent
      peer - the peer to send them to; null is a programming error

    • add

      public void add(PeerState peer, int size)
      Add the peer to the list of peers wanting to transmit something. This wakes up the packet pusher if it is sleeping.

      Avoid synchronization where possible. There are small chances of races. There are larger chances of adding the PeerState "behind" where the iterator is now... but these issues are the same as before concurrentification.

      Parameters:
      peer - the peer that now has something queued to transmit
      size - the minimum size we can send, or 0 to always notify

    • shouldEmitMemoryStat

      private boolean shouldEmitMemoryStat()
      Should the "udp.memory.activePeers" gauge be sampled on this add()? True once every MEMORY_STAT_INTERVAL messages.
      Returns:
      true if the gauge should be updated now

    • getNextVolley

      public List<UDPPacket> getNextVolley()
      Fetch all the packets for a message volley.

      Blocks while no peer can send, then gives up after at most MAX_WAIT_MS rather than waiting indefinitely, so this returns even when nothing became sendable.

      NOT thread-safe. Called by the PacketPusher thread only.

      If this is ever changed to retry internally instead of returning null, it must also break on thread interruption. Shutdown interrupts the pusher, and once that interrupt has been consumed every later wait returns immediately, which would turn the retry into a spin - the bug foldSendDelay(int, int) documents.

      Returns:
      the packets to send, or null when there is nothing to send. Null does not mean shutdown: a round can allocate states whose fragments are all already acked, leaving preparePackets(List, PeerState) with nothing to push, and that case is transient. Callers must treat null as "retry later" and must not spin.

    • nudge

      void nudge()
      Wakes up the packet pusher thread.

    • preparePackets

      private List<UDPPacket> preparePackets(List<OutboundMessageState> states, PeerState peer)
      Build the packets for one volley.
      Parameters:
      states - the states a peer offered, never null here
      peer - the peer they belong to
      Returns:
      the packets to send, or null when there is nothing to send: null states or peer, or every offered state pushed no fragment because all of its fragments are already acked. The last case is transient, which is why getNextVolley() can return null without being shut down.

    • foldSendDelay

      static int foldSendDelay(int runningMin, int reportedDelay)
      Fold one peer's reported delay into the smallest delay seen so far this round.

      A peer reports 0 to mean it wants a prompt retry: its retransmit timer has elapsed, or it is holding a message past its lifetime and should be failed out. That is a reason to retry soon, not a reason not to wait.

      Folding 0 in literally is what made this loop burn a core. The round minimum started at Integer.MAX_VALUE and was only ever lowered, so a single peer reporting 0 pinned it there for everyone. The end-of-round test is nextSendDelay > 0, which then failed and took the branch that re-scanned the same peers with no blocking at all, once per iteration, for as long as that peer stayed in that state.

      Parameters:
      runningMin - smallest delay seen so far this round
      reportedDelay - what this peer reported
      Returns:
      the updated minimum, never below MIN_WAIT_MS and never zero

    • roundWaitMs

      static int roundWaitMs(int roundMin)
      How long to block after a full round allocated nothing.

      Always at least MIN_WAIT_MS, so a round that produced nothing cannot turn into a spin, and never more than MAX_WAIT_MS, so one peer reporting a long delay cannot stall the pool.

      Parameters:
      roundMin - result of folding every peer's delay this round
      Returns:
      milliseconds to block

    • waitForMessages

      private void waitForMessages()
      Waits until there is a message to send.

    • waitForMessages

      private void waitForMessages(int timeout)
      Waits until there is a message to send.
      Parameters:
      timeout - the most time to wait, in milliseconds