java.lang.Object
net.i2p.client.streaming.impl.ConnectionManager
class ConnectionManagerextends Object
Coordinate all of the connections for a single local destination.

  • Field Details

    • _context

      private final I2PAppContext _context
      The app context.

    • _log

      private final Log _log
      Logger for the streaming session.

    • _session

      private final I2PSession _session
      The session carrying the packets.

    • _messageHandler

      private final MessageHandler _messageHandler
      Handles the session's messages.

    • _packetHandler

      private final PacketHandler _packetHandler
      Processes inbound packets.

    • _connectionHandler

      private final ConnectionHandler _connectionHandler
      Runs the connection state machines.

    • _outboundQueue

      private final PacketQueue _outboundQueue
      Queue of outbound packets.

    • _schedulerChooser

      private final SchedulerChooser _schedulerChooser
      Chooses the outbound packet scheduler.

    • _conPacketHandler

      private final ConnectionPacketHandler _conPacketHandler
      Handles packets for one connection.

    • _tcbShare

      private final TCBShare _tcbShare
      Shares the congestion window between connections.

    • _connectionFilter

      private final IncomingConnectionFilter _connectionFilter
      Decides whether to accept inbound connections.

    • _connectionByInboundId

      private final ConcurrentHashMap<Long, Connection> _connectionByInboundId
      Inbound stream ID (Long) to Connection map.

    • _connectionByOutboundId

      private final ConcurrentHashMap<Long, Connection> _connectionByOutboundId
      Outbound stream ID (Long) to Connection map — reverse index for O(1) lookup.

    • _pendingPings

      private final ConcurrentHashMap<Long, ConnectionManager.PingRequest> _pendingPings
      Ping ID (Long) to PingRequest.

    • _throttlersInitialized

      private volatile boolean _throttlersInitialized
      Whether the throttlers have been created.

    • _defaultOptions

      private final ConnectionOptions _defaultOptions
      Options used for new connections.

    • _numWaiting

      private final AtomicInteger _numWaiting
      Threads waiting for an inbound connection slot.

    • _soTimeout

      private long _soTimeout
      Socket read timeout in ms.

    • _minuteThrottler

      private volatile ConnThrottler _minuteThrottler
      Limits inbound connections per minute.

    • _hourThrottler

      private volatile ConnThrottler _hourThrottler
      Limits inbound connections per hour.

    • _dayThrottler

      private volatile ConnThrottler _dayThrottler
      Limits inbound connections per day.

    • _timer

      private final RetransmissionTimer _timer
      The retransmission timer for this manager.

    • _recentlyClosed

      private final Map<Long,Object> _recentlyClosed
      Recently closed connections, to reject quick reopens.

    • _cache

      private final ByteCache _cache
      Pooled buffers for packet payloads.

    • DUMMY

      private static final Object DUMMY
      Placeholder value in the recently-closed map.

    • _cooldownLock

      private final Object _cooldownLock
      Lock for interruptible cooldown wait in connect(Destination, ConnectionOptions, I2PSession) — notified on shutdown.

    • _destFailures

      private final ConcurrentHashMap<Hash,Long> _destFailures
      Per-destination cooldown to avoid hammering unreachable peers. Key is destination Hash, value is timestamp of last failed connect.

    • _cooldownWarned

      private final ConcurrentHashMap<Hash,Long> _cooldownWarned
      When we last logged the "Delaying connect" cooldown WARN per destination. Complements _destFailures: a dest that keeps failing re-enters cooldown (and thus arms the WARN) on every attempt, so without this the message would fire once per connect() against an unreachable peer. The key is pruned together with _destFailures using the same cutoff.

    • _streamPools

      private final ConcurrentHashMap<Hash, ConcurrentLinkedDeque<ConnectionManager.PooledConnection>> _streamPools
      Stream connection pool — reuses established streams to the same destination. Key: destination Hash, Value: deque of pooled connections.

    • RATES

      private static final long[] RATES
      Rate stat periods for the throttler stats.

    • RTX_RATIO_RATES

      static final long[] RTX_RATIO_RATES
      Sample periods for the stream-close retransmission ratios. The five minute period is mandatory, not cosmetic: Tuner.getAdditionalStat5Min() reads exactly FIVE_MINUTES from both stats to measure routing congestion, and a RateStat without that period makes addRateData() silently drop those samples, leaving the tuner with NaN forever.

    • EVENT_RATES

      static final long[] EVENT_RATES
      Sample periods for the send-event counters whose ratio the streaming ceiling reads. The five-minute period is required so the event-count ratio can be compared against RTX_RATIO_RATES on the same window — without it the two signals live on different clocks and every comparison mixes a 60 s event figure with a 5 min byte figure.

    • currentBlacklist

      private static volatile String currentBlacklist
      Cache of the property to detect changes.

    • _globalBlacklist

      private static final Set<Hash> _globalBlacklist
      Destinations blacklisted across all managers.

    • _tempBanUntil

      private final ConcurrentHashMap<Hash,Long> _tempBanUntil
      Temporary bans keyed by destination hash -> bannedUntil epoch ms. Autoban hammering dests for 24h (or i2p.streaming.tempBanMinutes). Enforced in shouldRejectConnection(Packet) before the port/budget counters so a banned dest no longer consumes its per-dest budget or per-peer throttlers. Expired entries are removed by BanExpiry.

    • _tempBanReason

      private final ConcurrentHashMap<Hash,String> _tempBanReason
      Trigger reason for an active temp-ban, kept alongside _tempBanUntil so the enforcement path can tell a client *why* it was banned (e.g. the per-minute limit it tripped). Written on ban, removed on expiry by BanExpiry.

    • _refusalCounter

      private final ObjectCounter<Hash> _refusalCounter
      Rolling refusals per destination, incremented each time the MAXIMUM streams gate refuses a SYN from that dest. Empties when a dest is banned or when BanExpiry clears the window.

    • _streamsByDest

      private final ConcurrentHashMap<Hash, AtomicInteger> _streamsByDest
      Live concurrent-stream count keyed by remote destination. Each remote dest gets its own stream budget (captured from the same effective max as the old global gate), so a flood from one dest can no longer starve legitimate clients sharing the listener. Slots are only ever taken by reserveStreamSlot(ConcurrentHashMap, Hash, int) (an atomic check-and-reserve, so two SYNs racing the gate cannot both observe "room left") and released through _streamReservations so a teardown cannot double-count. Shrunk toward the live table by ConnectionManager.BanExpiry only after the same excess is seen twice. @since 0.9.71+.

    • _streamReservations

      private final ConcurrentHashMap<Long,Hash> _streamReservations
      Reservation tokens: receive stream ID → the dest whose budget the stream was admitted under. Bound inside assignReceiveStreamId(Connection, Hash) in the same critical section that publishes the ID, and consumed exactly once by releaseReservation(long) when that ID leaves the manager. The token, not the connection's remote peer, is the source of truth on teardown, so a connection torn down before its peer was ever learned (or one whose dest hash is expensive to recompute) still returns its slot to the right budget. @since 0.9.71+.

    • _streamShrinkCandidates

      private final ConcurrentHashMap<Hash,Integer> _streamShrinkCandidates
      Reconciliation memory: what ConnectionManager.BanExpiry last observed for a dest whose ledger count exceeded the live connection table. The excess is only rebased downward when it repeats on a consecutive sweep, which rules out an in-flight reservation (accepted, ID not yet published).

    • _admissionLock

      private final ReentrantReadWriteLock _admissionLock
      Admission barrier: readers hold it across a whole reserve-to-bind span (tryReserveStream through assignReceiveStreamId) so a reservation can never straddle disconnectAllHard()'s ledger clear. Without the barrier a SYN admitted mid-clear would bind a token against an erased count and its teardown would then decrement a DIFFERENT stream's fresh count. The write side wraps only the hard-disconnect loop and the clears, so the rare teardown pays for quiescing the (microsecond) admission spans.

    • _banExpiry

      private volatile ConnectionManager.BanExpiry _banExpiry
      The sweeper owned by this manager. Held so shutdown() can stop it: a fired-and-cancelled TimedEvent would otherwise re-arm itself from inside ConnectionManager.BanExpiry.timeReached(), outliving the manager it reads.

    • _recentSyns

      private final ConcurrentHashMap<Hash,long[]> _recentSyns
      Per-destination burst state for the sub-second SYN gate: value is a two-element array {burstStartMs, count}, updated at most once per validated SYN via compare-and-swap (replace). A dest that quiesces naturally falls out of the next window. No per-dest entries accumulate beyond one array, so this is bounded and cheap on the hot path. Reset per-dest by the SYN-rate sweeper. @since 0.9.71+.

    • _synBurstStrikes

      private final ConcurrentHashMap<Hash, ConnectionManager.SynStrike> _synBurstStrikes
      Each dest's most recent SYN-burst strikes: the start of the burst window that crossed the threshold, the time that strike was recorded, and how many distinct windows have tripped inside the current strike window. Forgiveness ages from the strike time — window start can precede the strike by up to a whole burst window, and aging from it would forgive a later window's trip early. A trip from a DIFFERENT burst window within STRIKE_WINDOW_MS of the last strike adds one, and reaching PROP_TEMP_BAN_STRIKES of them autobans; repeat trips inside the window that already struck are ignored, so a single unlucky page-load burst never bans no matter how far over threshold it goes. After the strike window the entry is swept by BanExpiry.

    • _synRateMs

      private volatile long _synRateMs
      Cached sub-second burst config, refreshed by the BanExpiry sweeper rather than re-read from the property store on every validated SYN (hot path).

    • _synBurst

      private volatile int _synBurst
      SYNs in a burst that trigger a temp ban.

    • _tempBanStrikes

      private volatile int _tempBanStrikes
      Cached strikes-to-ban count, refreshed by the BanExpiry sweeper. 0 or 1 is raised to 2 at the decision point, so no setting can let a single burst window ban on its own.

    • _tempBanMs

      private volatile long _tempBanMs
      Cached autoban duration (ms) and refusal threshold, refreshed by the BanExpiry sweeper so the hot path (isTempBanned / refusal latch) avoids per-SYN property-store reads. Snapshot-at-mark-time still holds: a ban uses the duration current when it was placed.

    • _tempBanRefusals

      private volatile long _tempBanRefusals
      Refusals that trigger a temp ban.

    • _autobanEnabled

      private volatile boolean _autobanEnabled
      Cached autoban switch (PROP_AUTOBAN), refreshed by the BanExpiry sweeper like the other hot-path config. When false, banPeer() is a no-op, the SYN burst gates do not count or strike, and no SYN is dropped for a would-ban trip. Bans placed while it was enabled still expire naturally.

    • PROP_BLACKLIST

      public static final String PROP_BLACKLIST
      Blacklist property for streaming.
      See Also:

    • PROP_TEMP_BAN_MINUTES

      public static final String PROP_TEMP_BAN_MINUTES
      Autoban property: a dest is temporarily banned until this many minutes after it first trips a flood threshold. Default 5 minutes. Short enough that a legitimate client's retry window (30s connect timeout) overlaps with the ban expiry, so retries succeed on the next attempt. The previous 24-hour default was far too aggressive: a burst of page-load SYNs from a browser (local or remote) could ban a peer for an entire day. Tunable via i2p.streaming.tempBanMinutes. 0 disables autoban. @since 0.9.71+.
      See Also:

    • DEFAULT_TEMP_BAN_MINUTES

      private static final long DEFAULT_TEMP_BAN_MINUTES
      The default temp ban lasts 5 minutes.
      See Also:

    • PROP_AUTOBAN

      public static final String PROP_AUTOBAN
      Autoban property: whether autoban enforcement is enabled. 0 disables (no new bans are recorded and the SYN flood gates do not strike or drop); nonzero enables. Default on. The BanExpiry sweeper runs regardless, so per-dest gate state stays bounded and the cached autoban config keeps refreshing while enforcement is off. @since 0.9.71+.
      See Also:

    • DEFAULT_AUTOBAN

      private static final int DEFAULT_AUTOBAN
      Negative means no permanent autoban.
      See Also:

    • PROP_TEMP_BAN_REFUSALS

      public static final String PROP_TEMP_BAN_REFUSALS
      Autoban property: refusals from a single dest within a one-minute window (as counted by _refusalCounter) at which the dest is auto-banned. Tunable via i2p.streaming.tempBanRefusals. @since 0.9.71+.
      See Also:

    • DEFAULT_TEMP_BAN_REFUSALS

      private static final long DEFAULT_TEMP_BAN_REFUSALS
      Default refused connections that trigger a temp ban.
      See Also:

    • PROP_TEMP_BAN_RATE_MS

      public static final String PROP_TEMP_BAN_RATE_MS
      Autoban property: a dest sending more than tempBanSynBurst SYNs within a tempBanSynRate-ms rolling window is auto-banned. This is the sub-second rate gate: a legit client (e.g. a BitTorrent announce on a timer) never bursts tens of SYNs within a few hundred ms, but the observed tracker flood does (~25 SYNs in 259ms). Catches the burst before it consumes the stream budget.
      See Also:

    • DEFAULT_TEMP_BAN_RATE_MS

      private static final long DEFAULT_TEMP_BAN_RATE_MS
      The default SYN rate window is 1 second.
      See Also:

    • PROP_TEMP_BAN_SYN_BURST

      public static final String PROP_TEMP_BAN_SYN_BURST
      Autoban property: sub-second burst threshold (SYNs within rate window). Default 40 SYNs per 1s = 40 req/s instantaneous. A browser page load can fire 15-20 parallel connections, each with SYN retransmits; empty-response retry dual-race adds up to 4 SYNs/s per dest. The threshold must sit well above legitimate bursts so only a serious abuser trips it. First trip is a strike (no ban); a second trip within STRIKE_WINDOW_MS autobans.
      See Also:

    • DEFAULT_TEMP_BAN_SYN_BURST

      private static final int DEFAULT_TEMP_BAN_SYN_BURST
      Default SYNs in a burst that trigger a temp ban.
      See Also:

    • PROP_TEMP_BAN_STRIKES

      public static final String PROP_TEMP_BAN_STRIKES
      Autoban property: how many DISTINCT burst windows inside STRIKE_WINDOW_MS a dest must trip before the SYN-burst gate autobans it. Default 3.

      A single burst window is not abuse: a browser page load, a client with a high concurrent-stream ceiling, or a peer resuming after a pause can all legitimately cross the sub-second threshold. Two was low enough that a merely busy peer could lose five minutes of connectivity, so the default demands sustained tripping of the gate. Raise it further on a high-legitimate-traffic router, or set PROP_AUTOBAN 0 to turn off autoban entirely (which also disables the refusal-based ban).

      Values below 2 are raised to 2: a single window must never ban alone. Tunable via i2p.streaming.tempBanStrikes.

      See Also:

    • DEFAULT_TEMP_BAN_STRIKES

      private static final int DEFAULT_TEMP_BAN_STRIKES
      Temp bans before a permanent ban.
      See Also:

    • STRIKE_WINDOW_MS

      static final long STRIKE_WINDOW_MS
      Strike window for the multi-strike SYN-burst gate: strikes more than this many ms apart are forgiven, so the required count has to be met inside one rolling window. A dest that trips once and then quiesces for the window is forgiven (single unlucky page-load burst).
      See Also:

    • MAX_PONG_PAYLOAD

      private static final int MAX_PONG_PAYLOAD
      Largest pong payload accepted, in bytes.
      See Also:

    • PROP_ENABLE_PONG_DELAY

      private static final String PROP_ENABLE_PONG_DELAY
      Config key enabling the PONG response delay.
      See Also:

    • DEFAULT_ENABLE_PONG_DELAY

      private static final boolean DEFAULT_ENABLE_PONG_DELAY
      Pong delay is off by default.
      See Also:

    • MAX_PONG_DELAY

      private static final int MAX_PONG_DELAY
      Longest artificial pong delay, in ms.
      See Also:

    • PROP_MAX_PONG_DELAY

      static final String PROP_MAX_PONG_DELAY
      Property: maximum pong delay in ms.
      See Also:

    • LIMIT_HTTP_RESPONSE

      private static final String LIMIT_HTTP_RESPONSE
      The http 429 response sent when over the limit. See also: RFC 6585 (additional HTTP status codes) See also: rate limit responses
      See Also:

    • MAX_TIME

      private static final int MAX_TIME
      Largest ban length accepted, effectively forever.
      See Also:

  • Constructor Details

    • ConnectionManager

      public ConnectionManager(I2PAppContext context, I2PSession session, ConnectionOptions defaultOptions, IncomingConnectionFilter connectionFilter)
      Manage all conns for this session.
      Parameters:
      context - the I2P app context
      session - the primary session, packets may come in on subsessions also
      defaultOptions - the default connection options
      connectionFilter - the incoming connection filter

  • Method Details

    • shouldLogCooldownWarn

      static boolean shouldLogCooldownWarn(long lastWarnTime, long now, long cooldownMs)
      Whether to log another cooldown-delay WARN for a destination. Rate-limits to one warning per cooldown window per destination so a peer that keeps failing to connect cannot turn every retry into a log line. A value of 0 (never warned, or pruned) always logs.
      Parameters:
      lastWarnTime - last timestamp the WARN was emitted for this dest, 0 if never
      now - router clock time in milliseconds
      cooldownMs - length of the cooldown window that arms the WARN
      Returns:
      true if the WARN should be emitted

    • awaitCooldown

      static long awaitCooldown(Object lock, long delay, Clock clock)
      Wait out the remainder of a destination's connect cooldown. Re-arms after every wakeup until the deadline passes, so a notify - the one shutdown sends, or a spurious one - cannot cut the cooldown short and let a connect through to a peer we just failed to reach. The deadline is fixed at entry, so an early wakeup costs one more short wait, never a longer cooldown. An interrupt does end the wait, since that is a deliberate cancellation rather than a stray signal.
      Parameters:
      lock - the monitor to wait on, notified on shutdown
      delay - milliseconds of cooldown left to sit out, non-positive to skip
      clock - the clock the deadline is measured against
      Returns:
      milliseconds actually waited

    • getPoolMaxIdleMs

      private long getPoolMaxIdleMs()
      Max idle time before a pooled stream is closed and evicted. Default 20s — balances reuse vs staleness. Tunable via i2p.streaming.poolMaxIdleMs.
      Returns:
      the pool max idle ms

    • getPoolMaxPerDestination

      private int getPoolMaxPerDestination()
      Max pooled streams per destination. Default 8 — limits memory per peer. Tunable via i2p.streaming.poolMaxPerDestination.
      Returns:
      the pool max per destination

    • isPoolEnabled

      private boolean isPoolEnabled()
      Whether stream pooling is enabled. Disabled by default — pooled connections retain stale remote state that causes zero-data-flow after reuse. Enable only with proper connection state reset. Tunable via i2p.streaming.streamPool.enabled (default: false).
      Returns:
      whether pool enabled

    • getDestCooldownMs

      private long getDestCooldownMs()
      Cooldown between connection attempts to the same failed destination. Tunable via i2p.streaming.destinationCooldownMs (default: 5000).
      Returns:
      the dest cooldown ms

    • trimDestFailures

      private void trimDestFailures()
      Remove expired entries from _destFailures and _cooldownWarned. Called on every connect failure to prevent unbounded growth. Uses the cooldown window as the eviction threshold.

    • registerRtxRatioStats

      static void registerRtxRatioStats(StatManager sm)
      Register the stream-close retransmission ratio stats, including the five-minute period the router's congestion tuner samples. Separated from the constructor so tests can register against a fresh StatManager without building a whole manager.
      Parameters:
      sm - stat manager to register with, non-null

    • banPeer

      boolean banPeer(Hash h, String why, long now)
      Ban a dest for the configured duration. Idempotent; an existing longer ban is left in place so repeated abuse can't shrink it. No-op when autoban is disabled (PROP_AUTOBAN 0) or the duration is 0.
      Parameters:
      h - dest hash to ban
      why - human-readable trigger, e.g. "exceeded max 50 conns/minute"
      now - current clock time
      Returns:
      true if the dest is banned after this call (newly banned, extended, or already banned); false if bans are disabled

    • canReserveStream

      static boolean canReserveStream(int held, int max)
      Pure admission decision for the per-dest stream budget: one more stream fits while the dest's live count is strictly below the ceiling. A non-positive ceiling disables the gate entirely (no cap enforced).
      Parameters:
      held - streams currently reserved by the dest
      max - the per-dest concurrent stream ceiling
      Returns:
      true if a further stream would stay under the ceiling

    • tooManyStreamsForDest

      static boolean tooManyStreamsForDest(int streamCount, int max)
      Pure decision for the per-destination stream budget: a dest is over budget once its live concurrent-stream count reaches the per-dest ceiling. A non-positive ceiling disables the gate entirely.
      Parameters:
      streamCount - live streams currently held by the dest
      max - the per-dest concurrent stream ceiling
      Returns:
      true if the dest is at or over its own budget

    • reserveStreamSlot

      static boolean reserveStreamSlot(ConcurrentHashMap<Hash, AtomicInteger> ledger, Hash h, int max)
      Atomically take one stream slot from a dest's budget: the ceiling check and the increment happen inside a single per-key compute(), so two SYNs racing the gate can no longer both observe "room left" and admit past the ceiling (the check-then-add this replaced).

      Refusal leaves the counter untouched, so a rejected SYN never costs the dest anything. An unknown dest (null hash) is admitted without accounting, matching the old gate's "no hash, no budget" behavior.

      Parameters:
      ledger - the per-dest stream ledger, non-null
      h - remote dest hash, may be null
      max - the per-dest concurrent stream ceiling; ≤ 0 disables the gate
      Returns:
      true if the slot was taken (the caller now owns a slot and must either bind it to a reservation token or release it)

    • releaseStreamSlot

      static void releaseStreamSlot(ConcurrentHashMap<Hash, AtomicInteger> ledger, Hash h)
      Return one stream slot to a dest's budget. The count is clamped at zero and the entry is dropped once drained, so a mismatched teardown can never drive a budget negative (which would hand out free slots) and an idle dest leaves no entry behind.
      Parameters:
      ledger - the per-dest stream ledger, non-null
      h - remote dest hash, null-safe

    • streamSlotCount

      static int streamSlotCount(ConcurrentHashMap<Hash, AtomicInteger> ledger, Hash h)
      Read the stream count a dest is actually holding in the ledger, or 0 if it holds none.
      Parameters:
      ledger - the per-dest stream ledger, non-null
      h - remote dest hash, non-null
      Returns:
      the reserved stream count for the dest, never negative

    • consumeReservation

      static Hash consumeReservation(ConcurrentHashMap<Long,Hash> tokens, ConcurrentHashMap<Hash, AtomicInteger> ledger, long receiveStreamId)
      Consume the reservation token bound to a receive stream ID, releasing the dest budget slot it owns. A token is removed first and released after, so concurrent teardowns of the same ID cannot both return the slot; a missing token is a no-op rather than a guess about which dest to debit.
      Parameters:
      tokens - receive stream ID → dest hash, non-null
      ledger - the per-dest stream ledger, non-null
      receiveStreamId - the connection's receive stream ID
      Returns:
      the dest hash whose slot was released, or null if no token was bound

    • releaseWaiting

      static int releaseWaiting(AtomicInteger waiting)
      Decrement a connect-attempt waiter count without ever going negative.

      Waiters are released from a finally around the wait loop, so an abandoned or failed attempt always drains its own increment exactly once and a double release can only stop at zero.

      Parameters:
      waiting - the waiter counter, non-null
      Returns:
      the value after the release, or the unchanged value if it was already zero

    • reconcileStreamCount

      static int reconcileStreamCount(int held, int observed, Integer candidate, int boundTokens)
      Pure reconciliation step for one dest: how much of a ledger count that exceeds the live connection table should survive this sweep.

      The excess is only dropped when the SAME excess was recorded by the previous sweep (the candidate), which means it persisted across a whole sweep interval and therefore cannot be an in-flight reservation admitted moments before the first sweep. Counts at or below the table are never raised: the ledger only ever shrinks toward observed reality, so a sweep can free leaked slots but can never hand out new ones.

      Even a repeated candidate cannot shrink below the bound-token floor. A release followed by a re-reserve can return the count to the same value while the observed snapshot still shows the old, smaller table (ABA): shrinking to observed there would erase brand-new reservations. Tokens are bound one-per-admission, so max(observed, boundTokens) is never below the live floor and a genuine leak (no token behind the excess) still shrinks to observed.

      Parameters:
      held - the count currently in the ledger
      observed - the count of live connections for the dest
      candidate - the excess last sweep recorded for this dest, or null
      boundTokens - reservation tokens bound for this dest right now
      Returns:
      the count to keep in the ledger this sweep

    • reconcileStreamSlots

      static void reconcileStreamSlots(ConcurrentHashMap<Hash, AtomicInteger> ledger, Map<Hash,Integer> observed, ConcurrentHashMap<Hash,Integer> candidates, Map<Hash,Integer> boundTokens)
      Reconcile the per-dest stream ledger against the live connection table. See reconcileStreamCount(int, int, Integer, int) for the policy; this walks the ledger, records or clears per-dest candidates, and prunes candidates for dests that have left the ledger.

      A rebase is applied with a compare-and-set on the count we read, so a reserve or release landing mid-sweep makes us skip that dest rather than overwrite the newer value.

      Parameters:
      ledger - the per-dest stream ledger, non-null
      observed - live connection count per dest, non-null
      candidates - sweep-over-sweep observation memory, non-null
      boundTokens - reservation tokens bound per dest, non-null

    • tryReserveStream

      private boolean tryReserveStream(Hash h, int max)
      Take a stream slot from this dest's budget. See reserveStreamSlot(ConcurrentHashMap, Hash, int).
      Parameters:
      h - remote dest hash, non-null on every real path
      max - the per-dest concurrent stream ceiling
      Returns:
      true if the slot was taken

    • releaseStream

      private void releaseStream(Hash h)
      Return a stream slot taken outside the token protocol (a SYN rejected after admission, or a connection built before a token was bound).
      Parameters:
      h - remote dest hash, null-safe

    • releaseReservation

      private Hash releaseReservation(long receiveStreamId)
      Consume the reservation bound to a receive stream ID, releasing its slot.
      Parameters:
      receiveStreamId - the connection's receive stream ID
      Returns:
      the dest hash released, or null if no token was bound

    • releaseConnectionReservation

      private Hash releaseConnectionReservation(Connection con)
      Release whatever this connection still holds against a dest budget: first its reservation token, and only if no token was bound, its remote peer (a connection built outside the reservation protocol, so the slot must still be drained or the budget leaks forever).

      The release is claimed on the connection first: any path may reach here for a connection whose slot was already returned (a hard-disconnect sweep racing an async teardown, or a teardown after the sweeper released it), and the peer fallback would then decrement a DIFFERENT stream's count. The claim is re-armed by assignReceiveStreamId(Connection, Hash) when a pooled connection starts a new generation.

      Parameters:
      con - the connection leaving the manager, non-null
      Returns:
      the dest hash released, or null if nothing was held

    • tempBanReason

      private String tempBanReason(Hash h)
      The stored trigger description for an active temp-ban, if any.
      Parameters:
      h - destination hash to look up
      Returns:
      the recorded reason, or null if none stored

    • banIsLonger

      static boolean banIsLonger(Long existing, Long candidate)
      Whether a new ban end time should replace an existing one: only when it is strictly longer. Prevents a late-arriving shorter ban from shrinking an active ban under repeated abuse.
      Parameters:
      existing - current bannedUntil (aged), null if none
      candidate - proposed new bannedUntil
      Returns:
      true if candidate exceeds existing

    • isTempBanned

      boolean isTempBanned(Hash h, long now)
      Whether the dest is currently temp-banned.
      Parameters:
      h - dest hash to check
      now - current clock time
      Returns:
      true if temp-banned and not yet expired

    • banActive

      static boolean banActive(Long bannedUntil, long now)
      Whether a bannedUntil time is still in the future.
      Parameters:
      bannedUntil - epoch ms; null treated as not banned
      now - current clock time
      Returns:
      true if bannedUntil is non-null and greater than now

    • refusalThresholdMet

      static boolean refusalThresholdMet(long refusals, long threshold)
      Whether a dest has tripped the autoban refusal threshold.
      Parameters:
      refusals - count of refusals seen for the dest in the current window
      threshold - configured refusals-to-ban threshold
      Returns:
      true when refusals exceed threshold

    • synBurstTripped

      static boolean synBurstTripped(Long burstStartMs, int count, long now, long windowMs, int burst)
      Pure decision for the sub-second SYN-burst gate: whether a dest has sent more than burst SYNs within the last windowMs ms.
      Parameters:
      burstStartMs - epoch ms of the first SYN in the current burst window (the ''oldest'' still counted), null if none
      count - number of SYNs attributed to the open window
      now - current clock time
      windowMs - rolling window length
      burst - SYNs-per-window that constitutes an abusive burst
      Returns:
      true if the dest tripped the burst gate

    • checkSynBurst

      private long checkSynBurst(Hash h, long now, boolean autoban)
      Whether a SYN from a dest tripped the sub-second burst gate. Side effects: records the SYN in the per-dest rolling window. Allocation-free on the hot path: an existing window is bumped in place rather than replaced, and the window/burst limits are read from cached volatile fields refreshed by the BanExpiry sweeper. Call only for a validated SYN source.

      The autoban policy is a parameter, snapshotted once per SYN by the caller, so one SYN can never be counted under one policy and then recorded (or not) under a different one that changed mid-evaluation.

      Parameters:
      h - dest hash to record against
      now - current clock time
      autoban - the autoban policy snapshotted when this SYN was evaluated
      Returns:
      the window start of the burst window that tripped (used to key the strike to a distinct burst), or -1 if this SYN did not trip the gate (below threshold, window just re-armed, gate disabled, or autoban enforcement off)

    • strikeForgiven

      private static boolean strikeForgiven(ConnectionManager.SynStrike last, long now, long strikeWindowMs)
      Whether last has aged out of the forgiveness window (or was stamped in the future by a clock adjustment, which is treated the same way so a skewed clock can never ban).

      Single source of truth: the decision function uses it to choose BAN vs RECORD, and the recorder uses it to decide whether a RECORD extends the running strike count or restarts it at 1. Sharing it keeps those two from drifting, which would otherwise let a dest accumulate strikes across unrelated hours of traffic.

      Parameters:
      last - recorded strike to age-check, non-null
      now - wall-clock ms to measure the strike's age against
      strikeWindowMs - strike forgiveness window (pass STRIKE_WINDOW_MS)
      Returns:
      true if the strike is old enough (or future-stamped) to be forgiven, false while it still counts toward a ban

    • synBurstStrikeAction

      static ConnectionManager.SynBurstAction synBurstStrikeAction(ConnectionManager.SynStrike last, long windowStart, long now, long strikeWindowMs, int requiredStrikes)
      Pure decision for the multi-strike SYN-burst gate. Strikes mean DISTINCT burst windows: repeat trips inside the window that recorded the most recent strike are IGNOREd, so a single unlucky page-load burst can never cost a 5-minute outage however far over threshold it goes. Only when a dest has tripped requiredStrikes distinct windows inside STRIKE_WINDOW_MS of each other is this trip a BAN, i.e. the demonstrable pattern of sustained abuse. After the window the count is forgiven and a new one RECORDs instead.

      Forgiveness ages from the most recent strike time, so the window is rolling and the clock starts when the strike was recorded, not when the (millisecond-scale) burst window began.

      Parameters:
      last - recorded strike for this dest, or null if this dest has none
      windowStart - window start of the burst window that just tripped
      now - current clock time
      strikeWindowMs - strike forgiveness window (pass STRIKE_WINDOW_MS)
      requiredStrikes - distinct burst windows needed to ban; values below 2 are raised to 2 so a single window can never ban on its own
      Returns:
      the action to take for this trip

    • applySynBurstStrike

      static ConnectionManager.SynBurstAction applySynBurstStrike(ConcurrentHashMap<Hash, ConnectionManager.SynStrike> strikes, Hash h, long windowStart, long now, long strikeWindowMs, int requiredStrikes)
      Atomically evaluate and apply a burst-gate trip against h's recorded strike. The read-modify-write runs under ConcurrentHashMap.compute(K, BiFunction) so concurrent trips from one dest can neither both miss a just-recorded first strike (double-counting a single burst as two strikes) nor lose it to a racing put (letting a sustained flood escape the ban).
      Parameters:
      strikes - per-dest strike map
      h - remote dest hash, non-null
      windowStart - window start of the burst window that just tripped
      now - current clock time
      strikeWindowMs - strike forgiveness window (pass STRIKE_WINDOW_MS)
      requiredStrikes - distinct burst windows needed to autoban
      Returns:
      the action taken

    • noteSynBurstStrike

      private boolean noteSynBurstStrike(Hash h, long windowStart, long now)
      Record a SYN-burst strike for h and decide whether this trip autobans (see synBurstStrikeAction(ConnectionManager.SynStrike, long, long, long, int)). First trip in a window (or a trip after the strike window): store the strike, return false. Repeat trips in an already-struck window: no-op, return false. Second distinct window inside the strike window: leave the original strike in place (the ban supersedes further strikes), return true.

      Autoban is re-checked live here (the callers already gate checkSynBurst(Hash, long, boolean) on a per-SYN snapshot): if enforcement is switched off between the trip and this record, no strike may be stored, or the map would keep evidence recorded under a policy the operator has since disabled. Callers' snapshot decides BAN for the trip in flight; this check only gates the record.

      Parameters:
      h - remote dest hash, non-null
      windowStart - window start of the burst window that just tripped
      now - current clock time
      Returns:
      true if this trip should autoban

    • checkInboundSynFlood

      boolean checkInboundSynFlood(Hash h, long now)
      Package-visible flood gate for the retransmit-SYN path in ConnectionHandler.receiveNewSyn(Packet). A retransmitted SYN carries the stream IDs of a connection that already exists in the manager, so it never reaches receiveConnection(Packet) (and thus never passes the checkSynBurst(Hash, long, boolean) gate at the top of that method). An attacker exploits that by planting a handful of half-open connections and then blasting retransmitted SYNs against them. Without this gate every hit makes ConnectionHandler.resendSynAck mint and enqueue a fresh SYN-ACK, consuming CPU and egress, and the connection never establishes.

      This routes the retransmit through the same per-destination sub-second burst window as fresh SYNs, so a dest that crosses the burst threshold in PROP_TEMP_BAN_STRIKES DISTINCT windows within the strike window — the demonstrable pattern of sustained abuse — is autobanned and dropped. A single burst only strikes, and with autoban disabled (PROP_AUTOBAN 0) nothing is recorded or dropped here. Once banned, subsequent calls return true (drop) immediately.

      Parameters:
      h - remote dest hash, non-null
      now - current clock time
      Returns:
      true if this SYN should be dropped (dest already temp-banned, or this SYN is a second distinct burst strike and just banned it)

    • getTempBanMinutes

      private long getTempBanMinutes()
      Returns the temp ban length from the config.
      Returns:
      the ban length in minutes

    • getEffectiveMaxStreams

      private int getEffectiveMaxStreams()
      The concurrent-stream cap to enforce right now. Normally this is the value captured into _defaultOptions at init (the user-configured i2p.streaming.maxConcurrentStreams); if the router's Tuner has armed an override via I2PSocketManagerFull.setMaxStreamsOverride(int), the effective cap is the lower of the two so a user ceiling is never exceeded a Tuner. A volatile read; no config lookup per call.
      Returns:
      the cap; ≤ 0 means no cap is enforced

    • getLogMaxStreams

      private int getLogMaxStreams()
      The cap to report in refusal log messages: the Tuner's current override when it has armed one, otherwise the user-configured ceiling.

      This deliberately differs from getEffectiveMaxStreams(), which min's the override against the user ceiling and is the real enforcement limit at the per-destination stream budget gate (reserveStreamSlot(ConcurrentHashMap, Hash, int)). When the Tuner relaxes a healthy host back up toward its own max (e.g. 512) that is higher than a user's configured ceiling (e.g. 256), the enforcement limit stays 256 while the operator reading the log wants to see the Tuner's current target (512).

      Returns:
      the tuner override when armed, else the configured cap

    • getMaxPingTimeout

      private long getMaxPingTimeout()
      Maximum ping timeout. Tunable via i2p.streaming.maxPingTimeout (default: 300000).
      Returns:
      the max ping timeout

    • getDropOverLimit

      private int getDropOverLimit()
      Once over throttle limits, respond this many times before just dropping. Tunable via i2p.streaming.dropOverLimit (default: 3).
      Returns:
      the drop over limit

    • getConnectionByInboundId

      Connection getConnectionByInboundId(long id)
      Look up a connection by its inbound stream ID.
      Parameters:
      id - the inbound stream ID
      Returns:
      the connection, or null if not found

    • getConnectionByOutboundId

      Connection getConnectionByOutboundId(long id)
      Look up a connection by its outbound stream ID. Not guaranteed to be unique, but in case we receive more than one packet on an inbound connection that we haven't ack'ed yet...
      Parameters:
      id - the outbound stream ID
      Returns:
      the connection, or null if not found

    • wasRecentlyClosed

      public boolean wasRecentlyClosed(long inboundID)
      Was this conn recently closed?
      Parameters:
      inboundID - the inbound stream ID to look up in the recently-closed LRU
      Returns:
      true if the connection was closed recently and is still being tracked

    • setSoTimeout

      public void setSoTimeout(long x)
      Socket accept() timeout.
      Parameters:
      x - accept timeout in milliseconds, passed through to the listener's accept()

    • getSoTimeout

      public long getSoTimeout()
      Socket accept() timeout.
      Returns:
      accept timeout in ms.

    • setAllowIncomingConnections

      public void setAllowIncomingConnections(boolean allow)
      Enable or disable acceptance of incoming connections. When first enabled, initializes throttlers if needed.
      Parameters:
      allow - true to accept incoming connections

    • updateOptions

      public void updateOptions()
      Update the throttler options.

    • getAllowIncomingConnections

      public boolean getAllowIncomingConnections()
      Whether incoming connections are accepted.
      Returns:
      if we should accept connections

    • receiveConnection

      public Connection receiveConnection(Packet synPacket)
      Create a new connection based on the SYN packet we received.

      Ownership of the packet's pooled payload is decided here: acceptSyn(Packet) never touches it, so every exit from that stage - all of the rejections, and any throw - returns the buffer. The receivePacket() call below takes ownership, and nothing after it may release the payload, its own verification-failure path included.

      Parameters:
      synPacket - SYN packet to process
      Returns:
      created Connection with the packet's data already delivered to it, or null if the syn's streamId was already taken, or if the connection was rejected

    • acceptSyn

      private Connection acceptSyn(Packet synPacket)
      Everything about a new inbound connection up to the point where its payload is handed over: the SYN burst gates, the admission reservation, the options and MTU negotiation, and the connection's construction. Leaves the packet's payload alone - the caller owns it and returns it on every exit from here.
      Parameters:
      synPacket - the SYN packet to process
      Returns:
      the connection with its receive slot bound, or null if the SYN was dropped or refused

    • sendRejectResponse

      private void sendRejectResponse(Packet synPacket, Destination from, Hash fromHash, int retryAfter)
      Send the configured response for a rejected SYN (RESET, HTTP-style Retry-After payload, or custom payload), subject to the peer's throttler status. Only called when the SYN was already judged rejectable; extracted from receiveConnection(Packet) to bound its complexity.
      Parameters:
      synPacket - the incoming SYN being rejected
      from - the peer destination
      fromHash - from.calculateHash() (already computed by the caller)
      retryAfter - seconds for the Retry-After header; MAX_TIME means never retry (silent drop)

    • limitResponse

      static String limitResponse(String limitAction, int retryAfter, boolean sslPort)
      The payload to send in a limit-response SYN/CLOSE, or null for a RESET.

      Pure decision helper (extracted from receiveConnection): the "reset" limit action, an empty action, and port 443 (the synth HTTPS over I2P port, which must not be tricked into a text payload) all produce a plain RESET. "http" produces an HTTP Retry-After body; anything else is echoed verbatim after normalizing \r/\code{n} escape sequences. HTTPS (443) always stacks the port check too, per RFC-compliant servers that disregard limit responses on that port.

      Parameters:
      limitAction - the configured i2p.streaming.limitAction value, may be null
      retryAfter - seconds for the Retry-After header, or 0 to omit it
      sslPort - true if the SYN arrived on local port 443
      Returns:
      the response payload, or null for a RESET

    • synNacksMatch

      static boolean synNacksMatch(long[] nacks, Hash localHash)
      Whether a SYN's 8 NACK words match the local destination hash. A NACK that does not match the local hash means the sender is wrong about who we are; extracted so the word-split comparison is unit-testable.
      Parameters:
      nacks - the packet's NACK values, or null
      localHash - the local destination hash to compare against
      Returns:
      true if the NACKs are absent or all 8 words match

    • receivePing

      public boolean receivePing(Connection con, Packet ping)
      Process a ping by checking for throttling, etc., then sending a pong.
      Parameters:
      con - null if unknown
      ping - Ping packet to process, must have From and Sig fields, with signature already verified, only if answerPings() returned true
      Returns:
      true if we sent a pong

    • assignReceiveStreamId

      private void assignReceiveStreamId(Connection con, Hash destHash)
      Pick a new random stream ID for the con and assign it, taking care to avoid duplicates, and put it in the connection table.

      Binds this stream's budget reservation here, inside the same critical section that publishes the ID, so the token becomes the source of truth for the slot from the moment the ID is visible to teardown paths. An earlier id the con may have held is deliberately NOT re-released: a pooled connection hands back its previous slot when it leaves the manager, and releasing here as well would double-count.

      Parameters:
      con - the connection to assign an ID to
      destHash - the remote dest whose budget admitted this stream; the reservation token for the new ID points here

    • assignPingId

      private long assignPingId(ConnectionManager.PingRequest req)
      Pick a new random stream ID for a ping and assign it, taking care to avoid duplicates, and return it.
      Parameters:
      req - the ping request to register under the new ID
      Returns:
      the fresh receive stream ID the ping is pending under

    • assignRejectId

      private long assignRejectId()
      Pick a new random stream ID that we are rejecting, taking care to avoid duplicates, and return it.
      Returns:
      the fresh stream ID recorded in the recently-closed set

    • getDefaultStreamDelayMax

      private long getDefaultStreamDelayMax()
      Default stream delay maximum when no connect timeout is set. Tunable via i2p.streaming.defaultStreamDelayMax (default: 10000).
      Returns:
      the default stream delay max

    • connect

      public Connection connect(Destination peer, ConnectionOptions opts, I2PSession session)
      Build a new connection to the given peer. This blocks if there is no connection delay, otherwise it returns immediately.
      Parameters:
      peer - Destination to contact, non-null
      opts - connection options, never null
      session - generally the session from the constructor, but could be a subsession
      Returns:
      new connection, or null if we have exceeded our limit

    • finalizeConnection

      private Connection finalizeConnection(Connection con, Destination peer, ConnectionOptions opts, long connectStart, boolean fromPool)
      Common post-creation setup for both pooled and new connections. Handles cooldown, waitForConnect, stats recording.
      Parameters:
      con - the new or pooled connection being set up, non-null
      peer - the remote destination, used for the per-dest cooldown and failure bookkeeping
      opts - options supplying the connect delay and timeout passed to waitForConnect
      connectStart - wall-clock ms when the connect attempt began, for the elapsed-time stats
      fromPool - if true, skip cooldown delay (connection already established)
      Returns:
      the same con, after waitForConnect; a failed connect is reported through getConnectionError(), not a null return

    • shouldRejectConnection

      private ConnectionManager.Reason shouldRejectConnection(Packet syn)
      Check if a connection should be rejected based on blacklists, access lists, throttling, and the connection filter.
      Parameters:
      syn - the incoming SYN packet
      Returns:
      a Reason with seconds for Retry-After header; MAX_TIME for drop, 0 if unknown; or null if not rejected

    • getMessageHandler

      public MessageHandler getMessageHandler()
      Message handler for this manager.
      Returns:
      the message handler

    • getPacketHandler

      public PacketHandler getPacketHandler()
      Packet handler for this manager.
      Returns:
      the packet handler

    • getSession

      public I2PSession getSession()
      This is the primary session only.
      Returns:
      the primary session passed to the constructor, not a subsession

    • updateOptsFromShare

      public void updateOptsFromShare(Connection con)
      Update opts from share.
      Parameters:
      con - the connection whose options are being refreshed

    • updateShareOpts

      public void updateShareOpts(Connection con)
      Update share opts.
      Parameters:
      con - the connection whose options are being refreshed

    • getConnectionHandler

      public ConnectionHandler getConnectionHandler()
      Connection handler for this manager.
      Returns:
      the handler running the connection state machines

    • getPacketQueue

      public PacketQueue getPacketQueue()
      Outbound packet queue.
      Returns:
      the queue of packets waiting to be sent

    • answerPings

      public boolean answerPings()
      Do we respond to pings that aren't on an existing connection?
      Returns:
      true if we answer pings

    • disconnectAllHard

      public void disconnectAllHard()
      Drop every connection, sending a RESET instead of a CLOSE handshake. This will not close the ServerSocket. This will not kill the timer threads.

      CAN continue to use the manager.

    • shutdown

      public void shutdown()
      Kill all connections and the timers. Don't bother sending close packets. As of 0.9.17, this will close the ServerSocket, killing one thread in accept().

      CANNOT continue to use the manager or restart.

    • acquireFromPool

      private Connection acquireFromPool(Destination peer)
      Try to acquire a pooled connection for the given destination. Returns a valid connection or null if none available.
      Parameters:
      peer - the destination whose pool bucket is polled, non-null
      Returns:
      an established, healthy pooled connection, or null if the bucket is empty or every entry was stale or unhealthy

    • returnToPool

      private void returnToPool(Connection con)
      Return a connection to the pool if healthy. Called from removeConnection(Connection).
      Parameters:
      con - the connection being retired; skipped unless it is established, healthy and has clean streams, null is ignored

    • cleanupPool

      void cleanupPool()
      Periodic pool cleanup — evicts stale connections, then drops the buckets they leave behind. Without the second step the _streamPools key set grows with every destination the manager has ever pooled a stream to: an emptied deque is still a map entry, and a destination that is never requested again is never drained by acquireFromPool(Destination).

    • pruneEmptyPoolBuckets

      static int pruneEmptyPoolBuckets(Map<Hash, ? extends Collection<?>> pools)
      Remove destination buckets whose connection deque has gone empty, so a destination that is no longer contacted does not keep a (now-empty) pool entry alive forever. returnToPool(Connection) recreates a bucket on demand via computeIfAbsent, so dropping an empty one is safe. Weakly-consistent iteration, so it never blocks a concurrent pool acquire/return. Generic over the value type so the prune is testable without constructing a PooledConnection.
      Parameters:
      pools - the per-destination pool map to prune in place
      Returns:
      the number of empty buckets removed

    • registerOutboundId

      void registerOutboundId(Connection con)
      Register outbound id.
      Parameters:
      con - the connection whose send stream ID is bound to it, ignored when <= 0

    • removeConnection

      public void removeConnection(Connection con)
      Remove a connection from the manager.
      Parameters:
      con - Connection to drop.

    • publishLifetimeStats

      void publishLifetimeStats(Connection con)
      Publish a closing connection's lifetime counters. Called once per stream from Connection.disconnectComplete(), which is the single funnel every teardown goes through — deliberately not gated on whether the dispatch table still held the connection. A hard-killed or already-unlinked stream would otherwise silently lose every lifetime stat, and with them the retransmission ratios the Tuner reads for its loss decisions.

      RTT and send-window are the exception: they are only published once a real sample exists, since a connection killed during connect holds class defaults rather than observations.

      Parameters:
      con - Connection that is done; never null

    • listConnections

      public Set<Connection> listConnections()
      Connections currently managed.
      Returns:
      set of Connection objects

    • ping

      public boolean ping(Destination peer, long timeoutMs)
      Ping the destination and wait for a pong.
      Parameters:
      peer - the resolved destination to ping
      timeoutMs - greater than zero
      Returns:
      true if pong received

    • ping

      public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs)
      Ping the destination and wait for a pong.
      Parameters:
      peer - the resolved destination to ping
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      Returns:
      true if pong received

    • ping

      public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs, boolean blocking)
      Ping the destination, optionally waiting for a pong.
      Parameters:
      peer - the resolved destination to ping
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      blocking - true to block until pong
      Returns:
      true if blocking and pong received

    • ping

      public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs, boolean blocking, ConnectionManager.PingNotifier notifier)
      Ping the destination, optionally waiting for a pong.
      Parameters:
      peer - the resolved destination to ping
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      blocking - true to block until pong
      notifier - may be null
      Returns:
      true if blocking and pong received

    • ping

      public byte[] ping(Destination peer, int fromPort, int toPort, long timeoutMs, byte[] payload)
      Ping the destination with a payload and wait for the pong.
      Parameters:
      peer - the resolved destination to ping
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      payload - non-null, include in packet, up to 32 bytes may be returned in pong not copied, do not modify
      Returns:
      the payload received in the pong, zero-length if none, null on failure or timeout

    • ping

      public boolean ping(Destination peer, I2PSession session, int fromPort, int toPort, long timeoutMs)
      Ping the destination through a specific subsession, waiting for the pong.
      Parameters:
      peer - the resolved destination to ping
      session - the subsession to use (must not be null)
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      Returns:
      true if pong received

    • ping

      public boolean ping(Destination peer, I2PSession session, int fromPort, int toPort, long timeoutMs, boolean blocking)
      Ping the destination through a specific subsession, optionally blocking.
      Parameters:
      peer - the resolved destination to ping
      session - the subsession to use (must not be null)
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      blocking - true to block until pong
      Returns:
      true if blocking and pong received

    • ping

      public byte[] ping(Destination peer, I2PSession session, int fromPort, int toPort, long timeoutMs, byte[] payload)
      Ping the destination with a payload through a specific subsession.
      Parameters:
      peer - the resolved destination to ping
      session - the subsession to use (must not be null)
      fromPort - the source port
      toPort - the destination port
      timeoutMs - greater than zero
      payload - non-null, include in packet, up to 32 bytes may be returned in pong
      Returns:
      the payload received in the pong, zero-length if none, null on failure or timeout

    • observedStreamCounts

      private Map<Hash,Integer> observedStreamCounts()
      Live connection count per dest, preferring the reservation token (the dest that was actually charged at admission) and falling back to the connection's remote peer for anything bound outside the protocol. Built per sweep. A dest with no live connection is simply absent.
      Returns:
      observed stream count per dest, never null

    • tokenStreamCounts

      private Map<Hash,Integer> tokenStreamCounts()
      Bound reservation tokens per dest: the ABA floor for reconcileStreamCount(int, int, Integer, int). A token exists for exactly the admissions that own a slot (it is bound after the connection is published and released together with the count), so a stale shrink candidate can never claim slots that are genuinely held.
      Returns:
      tokens per dest; empty when nothing is bound, never null

    • receivePong

      void receivePong(long pingId, ByteArray payload)
      Process a received pong response.
      Parameters:
      pingId - the ping stream ID to match
      payload - the pong payload, may be null

    • toString

      public String toString()
      Connection summary for log messages.
      Overrides:
      toString in class Object