java.lang.Object
net.i2p.router.JobQueueScaler
All Implemented Interfaces:
Runnable
class JobQueueScalerextends Object implements Runnable
Dynamic job runner scaling controller for the JobQueue.

Monitors queue metrics (ready job count, lag) and automatically scales the number of JobQueueRunner threads up or down based on load conditions.

Scaling Algorithm:

  • Scale UP when: readyJobs > activeRunners * scaleUpJobsRatio OR maxLag > scaleUpLagThreshold
  • Scale DOWN when: readyJobs < activeRunners AND avgLag < scaleDownLagThreshold (sustained)
  • Bounds: min runners = max(4, cores), max runners = min(2 * router.maxJobRunners, RAM-based limit)

Resource Saturation Gates: Scale-up is suppressed (not just limited) when the bottleneck is resource saturation rather than thread count. Emergency signals (critical lag, wedged jobs, dropped jobs) override the suppression.

  • CPU saturation: router.cpuLoad > 85%
  • GC storm: router.gcPauseTime > 50 ms/min
  • Network congestion: tunnel.buildSuccessRate < 50%
  • Bandwidth saturation: bwLimiter.outboundDelayedTime > 500ms

Feedback Mechanism:

  • Captures pre-scale metrics before adding runners
  • Compares post-scale metrics after 2 check intervals
  • If lag, maxLag, message delay, or ready jobs INCREASED after scaling, rolls back
  • After 5 failed scale attempts, circuit breaker opens — stops scaling up
  • Tracks memory usage to prevent OOM conditions

  • Field Details

    • _log

      private final Log _log
      Logger for scaling decisions.

    • _jobQueue

      private final JobQueue _jobQueue
      The queue whose runners are scaled.

    • _context

      private final RouterContext _context
      The router context.

    • _isRunning

      private volatile boolean _isRunning
      True while the scaler job is scheduled.

    • _isAlive

      private volatile boolean _isAlive
      False once the scaler has shut down.

    • _consecutiveScaleUpChecks

      private int _consecutiveScaleUpChecks
      Checks in a row that wanted to scale up.

    • _consecutiveScaleDownChecks

      private int _consecutiveScaleDownChecks
      Checks in a row that wanted to scale down.

    • _lastScaleTime

      private long _lastScaleTime
      When the last scale event happened.

    • _configuredMaxRunners

      private int _configuredMaxRunners
      Runner limit from the config.

    • _currentMaxRunners

      private int _currentMaxRunners
      Runner limit in effect right now.

    • _preScaleSnapshot

      private JobQueueScaler.PreScaleSnapshot _preScaleSnapshot
      State captured before the last scale, for feedback.

    • _checksSinceLastScale

      private int _checksSinceLastScale
      Checks run since the last scale event.

    • _periodicCheckCount

      private int _periodicCheckCount
      Total checks run, never reset by scale events.

    • _consecutiveFailedScaleUps

      private int _consecutiveFailedScaleUps
      Scale-ups in a row that did not help.

    • _isInExtendedCooldown

      private boolean _isInExtendedCooldown
      True while the failure cooldown runs.

    • _scalingUpDisabled

      private boolean _scalingUpDisabled
      Circuit breaker set after repeated failures.

    • _cachedScaleJobsRatioProp

      private String _cachedScaleJobsRatioProp
      Config string the cached ratio was parsed from, kept to avoid re-parsing on the hot path.

    • _cachedScaleJobsRatio

      private double _cachedScaleJobsRatio
      Backlog ratio needed to scale up.

    • _previousReadyJobs

      private int _previousReadyJobs
      Ready jobs seen at the previous check.

    • _previousMessageDelay

      private long _previousMessageDelay
      Message delay seen at the previous check.

    • _messageDelayHistory

      private long[] _messageDelayHistory
      Recent message delay samples for trend tracking.

    • _msgDelayHistoryIndex

      private int _msgDelayHistoryIndex
      Write index into the delay history.

    • _msgDelayHistoryCount

      private int _msgDelayHistoryCount
      Samples held in the delay history.

    • _externalFloor

      private final AtomicInteger _externalFloor
      Runner floor requested by subsystems.

    • _externalRequests

      private final ConcurrentHashMap<String,int[]> _externalRequests
      Runner requests by subsystem name, with expiry times.

    • _loadEpisodeStart

      private long _loadEpisodeStart
      When the current overload episode began.

    • _lastRecoveryTime

      private long _lastRecoveryTime
      When the queue last recovered from an overload.

    • FEEDBACK_CHECKS_AFTER_SCALE

      private static final int FEEDBACK_CHECKS_AFTER_SCALE
      Checks after a scale before judging its result.
      See Also:

    • MAX_CONSECUTIVE_FAILED_SCALES

      private static final int MAX_CONSECUTIVE_FAILED_SCALES
      Failed scale-ups before the circuit breaker opens.
      See Also:

    • EXTENDED_COOLDOWN_MULTIPLIER

      private static final long EXTENDED_COOLDOWN_MULTIPLIER
      Cooldown multiplier after a failed scale.
      See Also:

    • LAG_INCREASE_THRESHOLD

      private static final double LAG_INCREASE_THRESHOLD
      Lag growth tolerated after a scale before rollback.
      See Also:

    • READY_JOBS_INCREASE_THRESHOLD

      private static final double READY_JOBS_INCREASE_THRESHOLD
      Backlog growth tolerated after a scale before rollback.
      See Also:

    • MSG_DELAY_HISTORY_SIZE

      private static final int MSG_DELAY_HISTORY_SIZE
      Delay samples kept, a 12 second window at 2s checks.
      See Also:

    • BASELINE_MULTIPLIER

      private static final double BASELINE_MULTIPLIER
      Delay over the baseline median that triggers action.
      See Also:

    • QUEUE_GROWTH_RATE_THRESHOLD

      private static final int QUEUE_GROWTH_RATE_THRESHOLD
      Jobs added per check that count as backlog growth.
      See Also:

    • MB

      private static final long MB
      One megabyte, for the memory limits.
      See Also:

    • RUNNER_MEMORY_ESTIMATE

      private static final long RUNNER_MEMORY_ESTIMATE
      A runner thread is estimated at 2 MB of heap.
      See Also:

    • MAX_MEMORY_PERCENTAGE

      private static final double MAX_MEMORY_PERCENTAGE
      Share of the heap allowed for runner threads.
      See Also:

    • DEFAULT_SCALE_CHECK_INTERVAL

      private static final long DEFAULT_SCALE_CHECK_INTERVAL
      The default interval between checks is 2 seconds.
      See Also:

    • DEFAULT_SCALE_COOLDOWN

      private static final long DEFAULT_SCALE_COOLDOWN
      The default wait between scale events is 10 seconds.
      See Also:

    • DEFAULT_SCALE_UP_LAG_THRESHOLD

      private static final int DEFAULT_SCALE_UP_LAG_THRESHOLD
      The default lag that justifies scaling up is 10 ms.
      See Also:

    • DEFAULT_SCALE_UP_MESSAGE_DELAY_THRESHOLD

      private static final int DEFAULT_SCALE_UP_MESSAGE_DELAY_THRESHOLD
      The default message delay that justifies scaling up is 200 ms.
      See Also:

    • DEFAULT_SCALE_DOWN_LAG_THRESHOLD

      private static final int DEFAULT_SCALE_DOWN_LAG_THRESHOLD
      The default lag low enough to scale down is 5 ms.
      See Also:

    • DEFAULT_SCALE_UP_JOBS_RATIO

      private static final double DEFAULT_SCALE_UP_JOBS_RATIO
      Default backlog ratio needed to scale up.
      See Also:

    • DEFAULT_SCALE_UP_STEP

      private static final int DEFAULT_SCALE_UP_STEP
      Runners added per scale-up.
      See Also:

    • DEFAULT_SCALE_DOWN_STEP

      private static final int DEFAULT_SCALE_DOWN_STEP
      Runners removed per scale-down, faster than we grow.
      See Also:

    • SUSTAINED_CHECKS_REQUIRED

      private static final int SUSTAINED_CHECKS_REQUIRED
      Checks that must want a scale-up before acting.
      See Also:

    • SUSTAINED_CHECKS_REQUIRED_DOWN

      private static final int SUSTAINED_CHECKS_REQUIRED_DOWN
      Checks that must want a scale-down before acting.
      See Also:

    • PROP_DYNAMIC_SCALING

      private static final String PROP_DYNAMIC_SCALING
      Config key turning dynamic scaling on or off.
      See Also:

    • PROP_SCALE_UP_LAG

      private static final String PROP_SCALE_UP_LAG
      Config key for the lag that justifies scaling up.
      See Also:

    • PROP_SCALE_DOWN_LAG

      private static final String PROP_SCALE_DOWN_LAG
      Config key for the lag low enough to scale down.
      See Also:

    • PROP_SCALE_JOBS_RATIO

      private static final String PROP_SCALE_JOBS_RATIO
      Config key for the backlog ratio needed to scale up.
      See Also:

    • PROP_SCALE_CHECK_INTERVAL

      private static final String PROP_SCALE_CHECK_INTERVAL
      Config key for the interval between checks.
      See Also:

    • PROP_SCALE_COOLDOWN

      private static final String PROP_SCALE_COOLDOWN
      Config key for the wait between scale events.
      See Also:

    • PROP_SCALE_UP_MSG_DELAY

      private static final String PROP_SCALE_UP_MSG_DELAY
      Config key for the message delay that justifies scaling up.
      See Also:

    • PROP_MIN_RUNNERS

      private static final String PROP_MIN_RUNNERS
      Config key for the minimum runner count.
      See Also:

    • PROP_FEEDBACK_ENABLED

      private static final String PROP_FEEDBACK_ENABLED
      Config key turning the post-scale feedback check on or off.
      See Also:

    • CPU_SATURATION_THRESHOLD

      private static final int CPU_SATURATION_THRESHOLD
      CPU percent above which scale-up is skipped.
      See Also:

    • GC_STORM_THRESHOLD

      private static final long GC_STORM_THRESHOLD
      GC pause ms per minute above which scale-up is skipped.
      See Also:

  • Constructor Details

    • JobQueueScaler

      public JobQueueScaler(RouterContext context, JobQueue jobQueue)
      Create a new JobQueueScaler.
      Parameters:
      context - the router context
      jobQueue - the job queue to scale

  • Method Details

    • calculateMaxRunnersBasedOnRAM

      private int calculateMaxRunnersBasedOnRAM(int configuredMax)
      Calculate max runners based on available RAM. Prevents OOM by limiting runners to use max 10% of heap.
      Parameters:
      configuredMax - the configured maximum from properties
      Returns:
      the lower of configuredMax*2 or RAM-based limit

    • recalculateMaxRunners

      private void recalculateMaxRunners()
      Recalculate max runners based on current memory conditions. Called periodically to adapt to changing memory pressure.

    • getUsedMemory

      private long getUsedMemory()
      Current used memory in bytes.
      Returns:
      the used memory

    • getMemoryUsagePercent

      private double getMemoryUsagePercent()
      Memory usage percentage.
      Returns:
      the memory usage percent

    • computeMedianBaseline

      private long computeMedianBaseline()
      Compute the median of the message delay history ring buffer.
      Returns:
      the median of the recorded message delays in ms, or 0 when no samples have been recorded yet

    • readBuildSuccessRate

      private double readBuildSuccessRate()
      Read the 60s rolling average of tunnel build success rate (0-100 scale). NaN if no data available.
      Returns:
      the average tunnel build success percentage for the last minute, or NaN when the rate or its 1-minute rate has no data

    • startup

      public void startup()
      Start the scaler thread.

    • shutdown

      public void shutdown()
      Shutdown the scaler gracefully.

    • isDynamicScalingEnabled

      private boolean isDynamicScalingEnabled()
      Check if dynamic scaling is enabled.
      Returns:
      whether dynamic scaling enabled

    • isFeedbackEnabled

      private boolean isFeedbackEnabled()
      Check if feedback mechanism is enabled.
      Returns:
      whether feedback enabled

    • getMinRunnersDynamic

      int getMinRunnersDynamic()
      Minimum number of runners (floor), dynamic - reads property each time. On router startup, returns more runners to handle startup load.
      Returns:
      the configured minimum, floored at 1, raised to 8 in the first 3 minutes

    • getCheckInterval

      private long getCheckInterval()
      Check interval in milliseconds.
      Returns:
      the check interval

    • getCooldownPeriod

      private long getCooldownPeriod()
      Cooldown period between scale events.
      Returns:
      the cooldown period

    • getScaleUpLagThreshold

      private int getScaleUpLagThreshold()
      Lag threshold for scaling up.
      Returns:
      the scale up lag threshold

    • getScaleUpMessageDelayThreshold

      private int getScaleUpMessageDelayThreshold()
      Message delay threshold for scaling up.
      Returns:
      the scale up message delay threshold

    • getScaleDownLagThreshold

      private int getScaleDownLagThreshold()
      Lag threshold for scaling down.
      Returns:
      the scale down lag threshold

    • getScaleUpJobsRatio

      private double getScaleUpJobsRatio()
      Jobs-to-runners ratio threshold for scaling up.
      Returns:
      the scale up jobs ratio

    • run

      public void run()
      Main scaling loop.
      Specified by:
      run in interface Runnable

    • checkAndScale

      private void checkAndScale()
      Check current conditions and scale if necessary.

    • checkScaleDown

      private void checkScaleDown(int activeRunners, int readyJobs, long maxLag, long avgLag, int minRunners, boolean inCooldown)
      Check if we should scale down.
      Parameters:
      activeRunners - the current number of JobQueueRunner threads
      readyJobs - jobs ready to run now, used for the job-to-runner ratio
      maxLag - ms the oldest ready job has waited, logged with the action
      avgLag - ms average lag of the ready jobs, compared with the scale-down lag threshold
      minRunners - the floor the runner count may not be reduced below
      inCooldown - true when a scale action happened recently enough to suppress another one

    • evaluateScalingFeedback

      private void evaluateScalingFeedback(int currentRunners, int currentReadyJobs, long currentMaxLag, long currentAvgLag, long currentMessageDelay, boolean currentDroppedJobs)
      Evaluate whether the last scale-up actually helped. Checks multiple signals: avg lag, max lag, ready jobs, message delay, and dropped jobs. If any signal worsened beyond threshold, rolls back.
      Parameters:
      currentRunners - runner count now, caps how many added runners roll back
      currentReadyJobs - ready jobs now, compared with the pre-scale snapshot
      currentMaxLag - ms, worst ready-job lag now, compared with the snapshot
      currentAvgLag - ms average lag now, compared with the pre-scale average
      currentMessageDelay - ms message delay now, checked against pre-scale lag
      currentDroppedJobs - true when the queue has dropped jobs since the snapshot, which forces a rollback

    • scaleUp

      private void scaleUp(int count, int readyJobs, long maxLag, long avgLag)
      Scale up by adding runners, and snapshot the current queue metrics so the feedback check can judge whether the change helped.
      Parameters:
      count - runners to add, already capped at the maximum by the caller
      readyJobs - ready job count now, logged and stored in the snapshot
      maxLag - ms of worst ready-job lag now, stored in the snapshot
      avgLag - ms of average lag now, the snapshot's lag baseline

    • scaleDown

      private void scaleDown(int count, int readyJobs, long maxLag)
      Scale down by removing runners. The queue retires idle runners only, so fewer than requested may be removed.
      Parameters:
      count - runners to remove
      readyJobs - ready job count now, logged with the action
      maxLag - ms of worst ready-job lag now, logged with the action

    • requestRunners

      public void requestRunners(int count, String reason, long autoReleaseMs)
      Request additional runners for an anticipated burst. Called by subsystems (e.g., Tuner) that know demand is imminent. The floor prevents the scaler from reducing below active + requested until auto-decay or explicit release.
      Parameters:
      count - number of additional runners to reserve
      reason - label for tracking and logging
      autoReleaseMs - if > 0, auto-decay after this many ms

    • releaseRunners

      public void releaseRunners(int count, String reason)
      Release previously requested runners. Idempotent — releasing more than requested just floors the per-reason count at 0.
      Parameters:
      count - number of runners to release
      reason - must match the reason used in requestRunners

    • recalcExternalFloor

      private void recalcExternalFloor()
      Recalculate the external floor from all outstanding requests.

    • getEffectiveMinRunners

      private int getEffectiveMinRunners()
      Effective minimum runners including external requests.
      Returns:
      base min runners + external floor

    • isAlive

      public boolean isAlive()
      Check if the scaler is currently running.
      Returns:
      true between startup() and shutdown(), and false when dynamic scaling is disabled, so startup() may leave it false

    • getCurrentMaxRunners

      public int getCurrentMaxRunners()
      Current maximum runner limit (may be up to 2x configured or RAM-limited).
      Returns:
      the current max runners

    • updateMaxRunners

      public void updateMaxRunners(int configuredMax)
      Update the maximum runner limit (called when configuration changes).
      Parameters:
      configuredMax - the new limit from configuration, before the RAM cap is applied