java.lang.Object
net.i2p.router.JobQueue
public class JobQueueextends Object
Prioritizes and executes router jobs with preference for earlier scheduled tasks. Manages job queues, timing, and thread pool execution for router-internal operations only.

For use by the router only. Not to be used by applications or plugins.

  • Field Details

    • _log

      private final Log _log
      Logger for the job queue.

    • _context

      private final RouterContext _context
      The router context.

    • _queueRunners

      private final Map<Integer, JobQueueRunner> _queueRunners
      Integer (runnerId) to JobQueueRunner for created runners.

    • _runnerId

      private static final AtomicInteger _runnerId
      Counter to identify a job runner.

    • _readyJobs

      private final BlockingQueue<Job> _readyJobs
      List of jobs that are ready to run ASAP.

    • _highPriorityJobs

      private final BlockingQueue<Job> _highPriorityJobs
      List of high priority jobs that should run before others.

    • _timedJobs

      private final ConcurrentSkipListSet<Job> _timedJobs
      SortedSet of jobs that are scheduled for running in the future, earliest first. Typed as the concrete skip list because the pumper reads first() as the minimum start time rather than re-deriving one per pass.

    • _timedJobsReady

      private final BlockingQueue<Job> _timedJobsReady
      Queue of timed jobs that are ready to run (moved from _timedJobs when ready).

    • _readyIndex

      private final Set<Job> _readyIndex
      Membership index over the three ready queues, so duplicate detection is O(1). Maintained exclusively through offerReady(Job, BlockingQueue), removeReady(Job, BlockingQueue), takeReady(BlockingQueue), addTimed(Job) and removeTimed(Job) so that no queue mutation can leave it stale.

      Membership is released at dispatch, not at completion: JobImpl.requeue(long) is called from runJob() on the running job itself, so holding the entry until completion would swallow every self-requeueing job.

    • _queuedTestJobs

      private final AtomicInteger _queuedTestJobs
      Number of TestJob instances in the three ready queues or _timedJobs, i.e. what getTestJobCount() reports. Maintained by the same helpers, so the pumper can sample it at 100 Hz without walking three collections under _jobLock.

    • _jobStats

      private final ConcurrentHashMap<String,JobStats> _jobStats
      Job name to JobStat for that job.

    • _pumper

      private final JobQueue.QueuePumper _pumper
      Thread pumping the ready jobs to the runners.

    • _allowParallelOperation

      private volatile boolean _allowParallelOperation
      Will we allow the # job runners to grow beyond 1?

    • _alive

      private volatile boolean _alive
      Have we been killed or are we alive?

    • _jobLock

      private final Object _jobLock
      Lock for the job queues.

    • _nextPumperRun

      private volatile long _nextPumperRun
      When the pumper wakes next.

    • runners

      static int runners
      How many when we go parallel.

    • PROP_MAX_RUNNERS

      static final String PROP_MAX_RUNNERS
      Router.config parameter to override the max runners.
      See Also:

    • DEFAULT_LAG_WARNING

      private static final long DEFAULT_LAG_WARNING
      If a job is this lagged, spit out a warning, but keep going.
      See Also:

    • _lagWarning

      private long _lagWarning
      Job delay that logs a warning.

    • DEFAULT_LAG_FATAL

      private static final long DEFAULT_LAG_FATAL
      If a job is this lagged, the router is hosed, so spit out a warning (don't shut it down).
      See Also:

    • _lagFatal

      private long _lagFatal
      Job delay that restarts the router.

    • DEFAULT_RUN_WARNING

      private static final long DEFAULT_RUN_WARNING
      If a job takes this long to run, spit out a warning, but keep going.
      See Also:

    • _runWarning

      private long _runWarning
      Job run time that logs a warning.

    • DEFAULT_RUN_FATAL

      private static final long DEFAULT_RUN_FATAL
      If a job takes this long to run, the router is hosed, so spit out a warning (don't shut it down).
      See Also:

    • _runFatal

      private long _runFatal
      Job run time that logs an error.

    • DEFAULT_WARMUP_TIME

      private static final long DEFAULT_WARMUP_TIME
      Don't enforce fatal limits until the router has been up for this long.
      See Also:

    • _warmupTime

      private long _warmupTime
      Startup time the lag checks ignore.

    • DEFAULT_MAX_WAITING_JOBS

      private static final int DEFAULT_MAX_WAITING_JOBS
      Max ready and waiting jobs before we start dropping 'em - scale with runner count.

    • _maxWaitingJobs

      private final int _maxWaitingJobs
      Resolved once from PROP_MAX_WAITING_JOBS at construction. shouldDrop(Job, int) read it per addJob() that could exceed the cap. A non-positive value disables dropping.

    • MIN_LAG_TO_DROP

      private static final long MIN_LAG_TO_DROP
      Minimum lag (ms) before the drop policy activates. Must be high enough to avoid drops during normal processing jitter; low enough to shed load before queue saturation causes cascading failure. 500ms means the drop gate is meaningful: with 48+ queued jobs AND a half-second of observed lag, the scaler has failed to keep up.
      See Also:

    • PROP_MAX_WAITING_JOBS

      private static final String PROP_MAX_WAITING_JOBS
      Config key limiting how many jobs may wait for a job thread.
      See Also:

    • _runnerLock

      private final Object _runnerLock
      Queue runners wait on this whenever they're not doing anything, and this gets notified *once* whenever there are ready jobs.

    • _scaler

      private final JobQueueScaler _scaler
      Dynamic scaling controller for adjusting runner count based on load.

    • RATES

      private static final long[] RATES
      Rate stat periods for the job statistics.

    • _droppedJobsCount

      private final AtomicInteger _droppedJobsCount
      Track dropped jobs for UI display.

    • POISON_ID

      private static final int POISON_ID
      Job id that stops the pumper.
      See Also:

  • Constructor Details

    • JobQueue

      public JobQueue(RouterContext context)
      Does not start the pumper. Caller MUST call startup.
      Parameters:
      context - the router context supplying properties, logs and stat manager

  • Method Details

    • offerReady

      private void offerReady(Job job, BlockingQueue<Job> queue)
      Claim the job in _readyIndex before offer() publishes it, so a concurrent addJob(Job) sees it as already queued in that window.
      Parameters:
      job - the job being published
      queue - the ready queue to publish it to

    • removeReady

      private boolean removeReady(Job job, BlockingQueue<Job> queue)
      Remove from a ready queue, releasing the index and TestJob claims only if the job was actually present.
      Parameters:
      job - the job to remove
      queue - the ready queue to remove it from
      Returns:
      true if the job was in the queue and has been removed

    • takeReady

      private Job takeReady(BlockingQueue<Job> queue)
      Dispatch from a ready queue, releasing the index claim here rather than at completion.
      Parameters:
      queue - the ready queue to dispatch from
      Returns:
      the next job, or null if the queue was empty

    • addTimed

      private void addTimed(Job job)
      Add to the future-scheduled skip list, keeping _queuedTestJobs in step.
      Parameters:
      job - the job scheduled for a future time

    • removeTimed

      private boolean removeTimed(Job job)
      Remove from the future-scheduled skip list.
      Parameters:
      job - the job to unschedule
      Returns:
      true if the job was scheduled and has been removed

    • addJob

      public void addJob(Job job)
      Enqueue the specified job for normal processing.
      Parameters:
      job - job to add to the queue

    • addJobToTop

      public void addJobToTop(Job job)
      Enqueue the specified job at the top of the queue, ahead of all normal-priority jobs. The job is exempt from the drop policy and its scheduled start time is overridden to run immediately. Any existing copy in the timed or ready queues is removed first, so promotion can never cause the job to run twice.
      Parameters:
      job - the job to promote to the front of the queue

    • removeJob

      public void removeJob(Job job)
      Remove a job from the job queue.
      Parameters:
      job - the job to remove from the queue

    • getReadyCount

      public int getReadyCount()
      Number of jobs ready to be executed.
      Returns:
      count of ready jobs in normal, high priority, and timed-ready queues

    • getAndResetDroppedCount

      public int getAndResetDroppedCount()
      Dropped jobs count (resets on read).
      Returns:
      number of jobs dropped since last call

    • getMaxLag

      public long getMaxLag()
      Maximum lag time for jobs waiting in the queue. This is the delay between when the earliest job was supposed to start and the current time. Uses peek() on each FIFO queue — the head element is the oldest and therefore has the highest lag.
      Returns:
      maximum lag in milliseconds, or 0 if all queues are empty

    • getMaxActiveJobDuration

      public long getMaxActiveJobDuration()
      Maximum duration of currently running jobs. This measures how long active jobs have been executing, which is important when all runners are busy but queue is empty.
      Returns:
      max duration in milliseconds of any currently running job, or 0 if no jobs running

    • getAvgLag

      public long getAvgLag()
      Average lag time for jobs waiting in the queue. This is the average delay between when jobs were supposed to start and the current time across all ready jobs.

      Uses weakly-consistent iteration on each BlockingQueue — concurrent additions/removals may be missed or double-counted, but the result is a sufficient approximation for diagnostics and scoring.

      Returns:
      average lag in milliseconds, or 0 if queue is empty

    • shouldDrop

      private boolean shouldDrop(Job job, int numReady)
      Checks whether a late job should be dropped.
      Parameters:
      job - the job being considered
      numReady - the jobs already waiting to run
      Returns:
      true to drop the job instead of running it

    • allowParallelOperation

      public void allowParallelOperation()
      Enable parallel job execution and start additional queue runner threads. After calling this, multiple jobs may execute concurrently.

    • startup

      public void startup()
      Initialize and start the job queue pumper thread. Does not start the job runner threads.

    • shutdown

      void shutdown()
      Shutdown the job queue, stopping all runners and clearing all jobs.

    • isAlive

      public boolean isAlive()
      Check if the job queue is currently alive and processing jobs. Public so subsystems that queue long-lived work (notably TestJob's batch dispatch) can refuse to hand it new jobs once the queue is gone — addJob() silently discards then, which would leave their internal state waiting on a pump that never runs.
      Returns:
      true if the queue is alive and running

    • getLastJobBegin

      public long getLastJobBegin()
      Timestamp of when the last job began execution.
      Returns:
      timestamp in milliseconds when the last job started, or -1 if no jobs have run

    • getLastJobEnd

      public long getLastJobEnd()
      Timestamp of when the last job finished execution.
      Returns:
      timestamp in milliseconds when the last job ended, or -1 if no jobs have run

    • getLastJob

      public Job getLastJob()
      Last job that was executed.
      Returns:
      the last Job that was executed, or null if no jobs have run

    • getNext

      Job getNext()
      Next job in queue.
      Returns:
      the next runnable job in priority order, or null once the queue has been shut down

    • runQueue

      public void runQueue(int numThreads)
      Start the job queue with the specified number of runner threads. Does nothing if parallel operation is not enabled and runners already exist.
      Parameters:
      numThreads - the number of runner threads to start

    • removeRunner

      void removeRunner(int id)
      Remove a queue runner from the registry. Package-private for use by JobQueueRunner.
      Parameters:
      id - the runner ID to remove

    • getActiveRunnerCount

      public int getActiveRunnerCount()
      Current number of active job runners. Package-private for use by JobQueueScaler.
      Returns:
      the number of active runners

    • getMaxRunnerCount

      public int getMaxRunnerCount()
      Current maximum number of job runners allowed. Returns the RAM-adjusted limit if scaler is active and RAM is constrained, otherwise returns the hard limit (2× configured).
      Returns:
      the current effective maximum runner limit

    • requestJobRunners

      public void requestJobRunners(int count, String reason, long autoReleaseMs)
      Request additional job runners for an anticipated burst. Delegates to JobQueueScaler for tracking and auto-decay.
      Parameters:
      count - number of additional runners to reserve
      reason - label for logging and tracking
      autoReleaseMs - if > 0, auto-decay after this many ms

    • releaseJobRunners

      public void releaseJobRunners(int count, String reason)
      Release previously requested job runners.
      Parameters:
      count - number of runners to release
      reason - must match the reason used in requestJobRunners

    • addRunners

      void addRunners(int count)
      Add additional job runners to the pool. Package-private for use by JobQueueScaler.
      Parameters:
      count - the number of runners to add

    • removeIdleRunners

      int removeIdleRunners(int maxToRemove)
      Remove idle job runners from the pool. Package-private for use by JobQueueScaler. Only removes runners that are not currently processing a job.
      Parameters:
      maxToRemove - the maximum number of runners to remove
      Returns:
      the number of runners actually removed

    • updateJobTimings

      private void updateJobTimings(long delta)
      Shifts the scheduled job times by a clock change.
      Parameters:
      delta - the clock change, in ms

    • updateStats

      void updateStats(Job job, long doStart, long origStartAfter, long duration)
      Update stats.
      Parameters:
      job - the job that just finished a run, used for its name and timing
      doStart - the millisecond timestamp when the job actually began
      origStartAfter - the previously recorded scheduled start time
      duration - the wall-clock milliseconds the run consumed, never negative

    • getJobs

      public int getJobs(Collection<Job> readyJobs, Collection<Job> timedJobs, Collection<Job> activeJobs, Collection<Job> justFinishedJobs)
      Collect statistics about jobs currently in the queue. This includes jobs that are ready, timed, active, and just finished.
      Parameters:
      readyJobs - collection to populate with ready jobs
      timedJobs - collection to populate with timed/scheduled jobs
      activeJobs - collection to populate with currently running jobs
      justFinishedJobs - collection to populate with recently finished jobs
      Returns:
      the number of queue runners currently active

    • getJobStats

      public Collection<JobStats> getJobStats()
      All job statistics collected by the queue.
      Returns:
      unmodifiable collection of JobStats for all job types

    • getTestJobCount

      public int getTestJobCount()
      Count the number of TestJob instances currently queued in the job queue. This includes both ready jobs and timed jobs waiting to be executed.

      O(1): maintained by the queue-mutation helpers, because the pumper samples it on every pass and walking three collections under _jobLock was measurable work for a stat that only feeds a graph.

      Counts a pumper-promoted job in _timedJobsReady as queued. Those jobs are due to run now, so omitting them understated the queue; this matches the jobQueue.testJobCount description and the admission check in addJob(Job), which treats them as ready.

      Returns:
      the total number of TestJob instances in the queue

    • getReadyTestJobCount

      public int getReadyTestJobCount()
      Count TestJob instances that are ready to run or currently running (excluding future-scheduled timed jobs). Used by the scaler to determine whether TestJobs are dominating runner resources right now.
      Returns:
      the number of ready/active TestJob instances