Worker
class Worker
Traits
Constants
| EXIT_SUCCESS |
|
| EXIT_ERROR |
|
| EXIT_MEMORY_LIMIT |
|
| RESTART_SIGNAL_CACHE_KEY |
The cache key for the restart signal. IMPORTANT: Uses Laravel's key for cross-framework queue interoperability. |
| protected HANDLED_SIGNALS |
Signals installed when the worker daemon starts. |
| protected SHUTDOWN_WAIT_SECONDS |
The interval between graceful shutdown checks. |
Properties
| protected string|null | $name | The name of the worker. |
|
| protected Repository|null | $cache | The cache repository implementation. |
|
| protected callable | $isDownForMaintenance | The callback used to determine if the application is in maintenance mode. |
|
| protected array | $runningJobs | The running jobs. |
|
| protected array<string, float> | $timeoutJobIds | The timeout job IDs. |
|
| protected WaitGroup> | $timeoutNotifications | Notifications that may still publish a timed-out attempt's outcome. |
|
| protected int|null | $monitorId | The job monitor's ID. |
|
| protected int|null | $livenessTimerId | The independent event-loop liveness timer. |
|
| protected float | $lastLoopProgressAt | The last time the event loop made progress. |
|
| protected bool | $ownsTimeoutAlarm | Indicates whether this daemon owns the timeout alarm. |
|
| protected float|null | $timeoutAlarmAt | The deadline currently armed with PCNTL. |
|
| protected bool | $recoveringBlockedLoop | Indicates that blocked-loop recovery has committed to terminating the worker. |
|
| protected int|null | $parkedCoroutineId | The coroutine frozen by emergency recovery. |
|
| protected Timer | $timer | The timer that owns the timeout monitor. |
|
| protected int | $jobsProcessed | The number of jobs completed by the worker. |
|
| protected float|null | $lastJobProcessedAt | The timestamp of the last completed job. |
|
| protected WorkerOptions}> | $pendingSignals | Signals awaiting delivery outside the asynchronous PCNTL handler. |
|
| bool | $shouldQuit | Indicates if the worker should exit. |
|
| bool | $lostConnection | Indicates if the worker lost its connection. |
|
| bool | $paused | Indicates if the worker is paused. |
|
| protected array<int, string> | $pausedQueues | The queues the worker last observed to be paused. |
|
| protected string|null | $lastPolledConnection | The connection used for the last pause-state observation. |
|
| protected string|null | $lastPolledQueues | The queue list used for the last pause-state observation. |
|
| static protected callable[] | $popCallbacks | The callbacks used to pop jobs from queues. |
|
| static int|null | $memoryExceededExitCode | The custom exit code to be used when memory is exceeded. |
|
| static int|null | $timedOutExitCode | The custom exit code to be used when a job times out. |
|
| static bool | $killOnTimeout | Indicates if the worker should be killed when a job exceeds its timeout. |
|
| static protected null|callable(int): mixed | $killCallback | The callback used to kill the worker process. |
|
| static bool | $reportJobExceptions | Indicates if the worker should report job exceptions. |
|
| static bool | $stopOnLostConnection | Indicates if the worker should stop when a lost connection is detected. |
|
| static bool | $restartable | Indicates if the worker should check for the restart signal in the cache. |
|
| static bool | $pausable | Indicates if the worker should check for the paused signal in the cache. |
Methods
Determine if the given exception was caused by a lost connection.
Create a new queue worker.
Listen to the given queue in a loop.
Create the waiter used to pop jobs.
Deliver pending signals outside their asynchronous handlers.
Monitor the jobs for timeout.
Allow a full monitor interval plus whole-second alarm granularity.
Recover kill-mode timeouts when PHP code prevents the monitor from running.
Arm an earlier timeout check without scanning jobs on the admission path.
Give the normal timeout handler a bounded chance to recover a blocked loop.
Interrupt an expired body without canceling a completed body's deferred cleanup.
Determine if there are any timeout jobs.
Handle a timed-out job whose execution has ended or is frozen.
Raise the event for an attempt that exceeded its timeout.
Report a timeout-handling error without abandoning other expired attempts.
Report an exception, writing it to the PHP error log if reporting fails.
Determine if the daemon should process on this iteration.
Pause the worker for the current loop.
Determine the exit code to stop the process if necessary.
Process the next job on the queue.
Determine which of the given queues are currently paused.
Raise events for any queues that have been paused or resumed since the last check.
Run the callback with the worker option context active.
Process the given job from the queue.
Prepare cleanup shared by the job's handoffs and its timeout monitor.
Determine whether the job counts interrupted attempts as exceptions.
Fail a job whose payload cannot be consumed.
Register a coroutine job to running jobs.
Handle an exception that occurred while the job was running.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if it has exceeded the maximum allowed exceptions.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if the exception handler determines it should not be retried.
Mark the given job as failed if it should fail on timeouts.
Raise an event indicating the worker is starting.
Raise the before job has been popped.
Raise the after queue job event.
Raise the exception occurred queue job event.
Determine if the queue worker should restart.
Get the last queue restart timestamp, or null.
Enable async signals for the process.
Handle a worker interruption signal.
Handle a worker pause signal.
Handle a worker resume signal.
Pass the signal to the running jobs.
Determine if "async" signals are supported.
Determine if the memory limit has been exceeded.
Get the current memory usage in megabytes.
Get the current monotonic time.
Stop listening and bail out of the script.
Kill the process.
Terminate the current worker process immediately.
Sleep the script for a given number of seconds.
Register a callback to be executed to pick jobs.
Register a callback to be used to kill the worker process.
Get the queue manager instance.
Flush all static state.
Details
in
DetectsLostConnections at line 16
protected bool
causedByLostConnection(Throwable $e)
Determine if the given exception was caused by a lost connection.
at line 297
__construct(Factory $manager, Dispatcher $events, ExceptionHandler $exceptions, callable $isDownForMaintenance, Timer|null $timer = null)
Create a new queue worker.
at line 311
int
daemon(string $connectionName, string $queue, WorkerOptions $options)
Listen to the given queue in a loop.
at line 527
protected Waiter
createPopWaiter()
Create the waiter used to pop jobs.
at line 535
protected void
waitForRunningJobs(WaitConcurrent $concurrent)
Wait for every admitted job coroutine to finish.
at line 557
protected void
drainPendingSignals(Waiter $lifecycleWaiter)
Deliver pending signals outside their asynchronous handlers.
at line 605
protected void
monitorTimeoutJobs(WorkerOptions $options, string $connectionName, string $queue)
Monitor the jobs for timeout.
at line 635
protected int
timeoutGracePeriod(WorkerOptions $options)
Allow a full monitor interval plus whole-second alarm granularity.
at line 643
protected void
startTimeoutAlarm(WorkerOptions $options)
Recover kill-mode timeouts when PHP code prevents the monitor from running.
at line 667
protected void
armTimeoutAlarm(float $deadline)
Arm an earlier timeout check without scanning jobs on the admission path.
at line 681
protected void
handleTimeoutAlarm(WorkerOptions $options)
Give the normal timeout handler a bounded chance to recover a blocked loop.
at line 729
protected void
terminateTimeoutJobs(WorkerOptions $options)
Scan the running jobs and terminate the timeout jobs.
at line 792
protected void
interruptTimeoutJob(string $jobId, array $job)
Interrupt an expired body without canceling a completed body's deferred cleanup.
at line 815
protected void
notifyTimeoutJob(Job $job)
Notify a job before cancellation unwinds its running command.
at line 827
protected bool
hasTimeoutJobs()
Determine if there are any timeout jobs.
at line 835
protected void
handleTimeoutJob(Job $job, WorkerOptions $options, JobProcessingMarker|null $processingMarker = null)
Handle a timed-out job whose execution has ended or is frozen.
at line 874
protected void
raiseTimedOutJobEvent(Job $job, WorkerOptions $options)
Raise the event for an attempt that exceeded its timeout.
at line 892
protected void
reportTimeoutException(Throwable $exception)
Report a timeout-handling error without abandoning other expired attempts.
at line 904
protected void
reportException(Throwable $exception)
Report an exception, writing it to the PHP error log if reporting fails.
at line 924
protected int
timeoutForJob(Job|null $job, WorkerOptions $options)
Get the appropriate timeout for the given job.
at line 932
protected bool
daemonShouldRun(WorkerOptions $options, string $connectionName, string $queue)
Determine if the daemon should process on this iteration.
at line 943
protected array|null
pauseWorker(WorkerOptions $options, int|null $lastRestart, float|int $startTime, int $jobsAdmitted)
Pause the worker for the current loop.
at line 963
protected array|null
stopIfNecessary(WorkerOptions $options, int|null $lastRestart, float|int $startTime, int $jobsAdmitted, mixed $job = null, bool $checkQueueEmpty = true, bool $hasRunningJobs = false)
Determine the exit code to stop the process if necessary.
at line 992
null
runNextJob(string $connectionName, string $queue, WorkerOptions $options)
Process the next job on the queue.
at line 1017
protected Job|null
getNextJob(Queue $connection, string $queue)
Get the next job from the queue connection.
at line 1080
protected array
getPausedQueues(string $connectionName, array $queues)
Determine which of the given queues are currently paused.
at line 1099
protected void
raisePausedQueueEvents(string $connectionName, array $paused)
Raise events for any queues that have been paused or resumed since the last check.
at line 1119
protected null
runJob(Job $job, string $connectionName, WorkerOptions $options)
Process the given job.
at line 1143
protected mixed
withCoroutineContext(WorkerOptions $options, callable $callback)
Run the callback with the worker option context active.
at line 1174
protected void
stopWorkerIfLostConnection(Throwable $e)
Stop the worker if we have lost connection to a database.
at line 1202
void
process(string $connectionName, Job $job, WorkerOptions $options)
Process the given job from the queue.
at line 1395
protected JobProcessingMarker|null
prepareJobProcessingMarker(Job $job)
Prepare cleanup shared by the job's handoffs and its timeout monitor.
at line 1423
protected bool
shouldCountCrashesAsExceptions(Job $job)
Determine whether the job counts interrupted attempts as exceptions.
at line 1434
protected void
handleInvalidPayload(string $connectionName, Job $job, InvalidPayloadException $e)
Fail a job whose payload cannot be consumed.
at line 1443
protected string
registerCoroutineJob(Job $job, WorkerOptions $options, JobProcessingMarker|null $processingMarker = null)
Register a coroutine job to running jobs.
at line 1466
protected void
handleJobException(string $connectionName, Job $job, WorkerOptions $options, Throwable $e)
Handle an exception that occurred while the job was running.
at line 1540
protected void
markJobAsFailedIfAlreadyExceedsMaxAttempts(string $connectionName, Job $job, int $maxTries)
Mark the given job as failed if it has exceeded the maximum allowed attempts.
This will likely be because the job previously exceeded a timeout.
at line 1566
protected void
markJobAsFailedIfAlreadyExceedsMaxExceptions(string $connectionName, Job $job)
Mark the given job as failed if it has exceeded the maximum allowed exceptions.
This will likely be because the worker previously died while processing the job.
at line 1592
protected void
markJobAsFailedIfWillExceedMaxAttempts(string $connectionName, Job $job, int $maxTries, Throwable $e)
Mark the given job as failed if it has exceeded the maximum allowed attempts.
at line 1608
protected void
markJobAsFailedIfWillExceedMaxExceptions(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if it has exceeded the maximum allowed attempts.
at line 1632
protected void
markJobAsFailedIfItShouldntBeRetried(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if the exception handler determines it should not be retried.
at line 1643
protected void
markJobAsFailedIfItShouldFailOnTimeout(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if it should fail on timeouts.
at line 1653
protected void
failJob(Job $job, Throwable $e)
Mark the given job as failed and raise the relevant event.
at line 1661
protected int
calculateBackoff(Job $job, WorkerOptions $options)
Calculate the backoff for the given job.
at line 1675
protected void
raiseWorkerStartingEvent(string $connectionName, string $queue, WorkerOptions $options)
Raise an event indicating the worker is starting.
at line 1685
protected void
raiseBeforeJobPopEvent(string $connectionName, string|null $queue = null)
Raise the before job has been popped.
at line 1695
protected void
raiseAfterJobPopEvent(string $connectionName, Job|null $job)
Raise the after job has been popped.
at line 1708
protected void
raiseBeforeJobEvent(string $connectionName, Job|null $job)
Raise the before queue job event.
at line 1721
protected void
raiseAfterJobEvent(string $connectionName, Job $job, float|null $duration = null)
Raise the after queue job event.
at line 1735
protected void
raiseExceptionOccurredJobEvent(string $connectionName, Job|null $job, Throwable $e)
Raise the exception occurred queue job event.
at line 1749
protected bool
queueShouldRestart(int|null $lastRestart)
Determine if the queue worker should restart.
at line 1761
protected int|null
getTimestampOfLastQueueRestart()
Get the last queue restart timestamp, or null.
at line 1779
protected void
listenForSignals(string $connectionName, string $queue, WorkerOptions $options)
Enable async signals for the process.
at line 1805
protected void
handleInterruptionSignal(int $signal, string $connectionName, string $queue, WorkerOptions $options)
Handle a worker interruption signal.
at line 1814
protected void
handlePauseSignal(string $connectionName, string $queue, WorkerOptions $options)
Handle a worker pause signal.
at line 1828
protected void
handleResumeSignal(string $connectionName, string $queue, WorkerOptions $options)
Handle a worker resume signal.
at line 1842
protected void
notifyJobsOfSignal(int $signal)
Pass the signal to the running jobs.
at line 1852
protected void
notifyJobOfSignal(Job $job, int $signal)
Pass the signal to the given job.
at line 1884
protected bool
supportsAsyncSignals()
Determine if "async" signals are supported.
at line 1892
bool
memoryExceeded(float|string $memoryLimit)
Determine if the memory limit has been exceeded.
at line 1902
protected float
currentMemoryUsage()
Get the current memory usage in megabytes.
at line 1910
protected float
currentTime()
Get the current monotonic time.
at line 1918
int
stop(string $connectionName, string $queue, int $status = 0, WorkerOptions|null $options = null, WorkerStopReason|null $reason = null)
Stop listening and bail out of the script.
at line 1945
never
kill(string $connectionName, string $queue, int $status = 0, WorkerOptions|null $options = null, WorkerStopReason|null $reason = null)
Kill the process.
at line 1985
protected never
terminateProcess(int $status)
Terminate the current worker process immediately.
at line 1997
protected MaxAttemptsExceededException
maxAttemptsExceededException(Job $job)
Create an instance of MaxAttemptsExceededException.
at line 2005
protected TimeoutExceededException
timeoutExceededException(Job|null $job)
Create an instance of TimeoutExceededException.
at line 2013
void
sleep(float|int $seconds)
Sleep the script for a given number of seconds.
at line 2026
Worker
setCache(Repository $cache)
Set the cache repository implementation.
The repository persists on the singleton worker for the worker lifetime and is read by every subsequent pause or restart check.
at line 2042
Worker
setName(string $name)
Set the name of the worker.
The name persists on the singleton worker for the worker lifetime and selects the static pop callback used by every job pop.
at line 2057
static void
popUsing(string $workerName, callable|null $callback)
Register a callback to be executed to pick jobs.
The callback persists in a static property for the worker process lifetime and runs on every job pop for the named worker. Passing null removes the callback.
at line 2075
static void
killUsing(callable|null $callback)
Register a callback to be used to kill the worker process.
The callback persists for the worker lifetime and runs before forced process termination. Passing null clears it.
at line 2084
Factory
getManager()
Get the queue manager instance.
at line 2095
void
setManager(Factory $manager)
Set the queue manager instance.
Swaps the manager reference on the singleton worker; runtime use races across coroutines and changes every concurrent queue lookup.
at line 2104
static void
flushState()
Flush all static state.