cl-jobpond

cl-jobpond runs caller-supplied job bodies on a bounded pool of reusable threads. Every job is supervised: it publishes exactly one terminal result, it can be cancelled while it runs, it can carry a wall-clock deadline, and it reports progress to listeners.

(let ((pool (cl-jobpond:make-job-pool :maximum-concurrency 4)))
  (unwind-protect
       (let ((job (cl-jobpond:job-pool-submit
                   pool
                   (lambda (job)
                     (* 2 (cl-jobpond:job-payload job)))
                   :name "double"
                   :payload 21)))
         (getf (cl-jobpond:job-await job) :result))
    (cl-jobpond:job-pool-close pool)))
;; => 42

The pool owns admission, concurrency, lifecycle, cancellation, and retention. The body owns the work.

Jobs

A job body is a function of one argument, the job itself. Its outcome is typed:

• returning a value publishes :completed with that value as the result
• signalling job-aborted publishes :aborted
• signalling any other error publishes :failed with a bounded condition report

A body that escapes even those paths is caught by the worker backstop and published as :failed. A job that has been admitted always becomes terminal.

job-snapshot returns a coherent plist of state, result, cancellation reason, condition report, timings, and progress. Prefer the snapshot when several fields must agree.

Admission

job-pool-submit admits one job. job-pool-submit-batch admits a list of entry plists atomically: either every entry is queued, or none is.

(cl-jobpond:job-pool-submit-batch
 pool
 (list (list :function #'summarize :payload report-a :name "report-a")
       (list :function #'summarize :payload report-b :name "report-b")))

Three bounds are enforced:

• :maximum-concurrency limits jobs running at the same time
• :maximum-batch-size limits one atomic admission
• :maximum-live-jobs limits queued and running jobs together

The last two refuse work with job-pool-capacity-exceeded.

A pool starts one worker thread per unit of its concurrency bound at construction. Pass :start-threads-p nil for a pool that starts them on its first admission. Close a pool with job-pool-close when it is no longer needed.

job-pool-update-limits replaces the concurrency, batch-size, live-job, and default-runtime limits under the pool lock. Every value is required and validated before any policy changes.

(cl-jobpond:job-pool-update-limits
 pool
 :maximum-concurrency 2
 :maximum-batch-size 4
 :maximum-live-jobs 8
 :maximum-runtime-milliseconds 30000)

Progress

A body calls job-report-progress to publish a progress value, append to a bounded output tail, and advance a step count. job-progress-snapshot reads the result from any thread.

(cl-jobpond:job-report-progress job :detail '(:stage :parsing) :output line)

Every progress report is also a cancellation point: it signals job-aborted when a controller asked for cancellation.

Cancellation

job-cancel is first-writer. The first accepted request records its reason; later requests return nil. A queued job leaves the queue and publishes :aborted immediately. A running job is interrupted on its worker thread by default.

Interrupts are best effort. A body blocked where the host cannot deliver them stops at its next job-check-cancellation or job-report-progress call. job-aborted is a serious-condition, so a body that handles error broadly lets cancellation through.

For a body with its own cancellation polling, specialize job-interrupt-on-cancellation-p to return nil:

(defclass polling-job (cl-jobpond:job) ())

(defmethod cl-jobpond:job-interrupt-on-cancellation-p ((job polling-job))
  nil)

(cl-jobpond:make-job-pool :job-class 'polling-job)

The method runs on the cancelling thread outside lifecycle locks. Return promptly without signalling, and keep the policy stable throughout the run. The body must poll job-cancellation-requested-p or call job-check-cancellation at bounded intervals and unwind promptly, including during pool shutdown. No asynchronous interrupt is sent for this job. Cancellation reasons, subtree cancellation and deadlines use the same state transitions. If the body returns normally after cancellation, publication resolves it to :aborted and discards the returned value.

Waiting for several jobs

job-await waits for one job. job-await-any waits until any of a list of jobs is terminal and returns that job and its snapshot, or NIL after :timeout-seconds. With several terminal jobs it returns the first in list order, so a caller removing each returned job collects results as they finish and can cancel the rest once it has enough.

(loop with pending = jobs
      while pending
      do (let ((job (cl-jobpond:job-await-any pending)))
           (setf pending (remove job pending))
           (when (enough-p (cl-jobpond:job-snapshot job))
             (dolist (rest pending) (cl-jobpond:job-cancel rest :reason :not-needed))
             (return))))

Deadlines

A positive :maximum-runtime-milliseconds gives every job of the pool a wall-clock cap, and an individual entry may override it. A single monitor thread cancels expired jobs with the reason :timeout. Zero disables deadlines.

Terminal results

Terminal jobs stay available for inspection through job-pool-find-job and job-pool-list-jobs, bounded by :terminal-retention-limit. Once the ring is full the oldest terminal job is evicted, and looking it up afterwards signals job-not-found. The evicted job object itself remains valid for anyone still holding it.

Publication is single-writer. A host that wants to own what a terminal job carries supplies :terminal-result-function. It is called with the job, the resolved state, the result being published, and the report, and returns its own result, report, and state:

(lambda (job state result report)
  (values (list :status state :value result :error report)
          report
          state))

The hook runs inside the publication claim and outside the job’s lifecycle lock, so exactly one writer ever runs it. It may take locks or do input and output.

Listeners

job-pool-add-listener registers a function of a channel keyword and an event plist. The pool emits:

• :job-lifecycle events with the statuses :started, :completed, :failed, and :aborted
• :job-progress events for each report

Every event carries the job it is about as :job. A listener that signals is ignored. Listeners run on the thread that produced the event.

Shutdown

Wait for the jobs that matter with job-await first. Closing is idempotent and safe from several threads at once. job-pool-detach then drops the runtime state of a closed pool, and job-pool-refresh reopens it with fresh workers.

Job trees

A job may name its ancestors with :owner-identifiers, outermost first, and the tree it belongs to with :root-identifier. job-pool-descendant-jobs finds a whole subtree in one pass. job-cancel with :cascade-p cancels that subtree and returns the descendants it took. The job itself is cancelled first.

Running a job on a chosen thread

job-run-inline runs a queued job on the calling thread, which keeps a pool from deadlocking when every worker is blocked waiting on another job. A host that must have a particular thread admits the job with :inline-only-p. Such a job counts against the live bound and is findable and cancellable like any other, but only job-run-inline will ever run it.

Host fields

A host with fields of its own subclasses job and gives the pool :job-class; each admission entry then carries :initargs for that class.

Durable schedules (optional)

Load cl-jobpond/schedules. This system supplies a locked portable state machine; the host supplies execution and a persistent process or event loop.

(let ((scheduler (cl-jobpond:make-scheduler :store #'atomic-save)))
  (cl-jobpond:scheduler-add scheduler :id "daily" :version "v3"
                           :payload '(:task "summarize")
                           :at (get-universal-time) :interval 86400
                           :missed-policy ':latest)
  (cl-jobpond:scheduler-wake scheduler)
  (let ((claim (cl-jobpond:scheduler-claim scheduler :version "v3")))
    (when claim
      ;; Admit CLAIM's payload into the existing execution facility, then settle it.
      (cl-jobpond:scheduler-ack scheduler :id (getf claim :id)
                               :token (getf claim :token) :result '(:proof "done")))))

make-scheduler accepts :clock (default universal time), :store, :snapshot and :capacity (default 4096). Times use nonnegative integer clock units. IDs and versions are bounded strings. Registration is immutable and identical registration is idempotent; use a new ID when replacing a cancelled version.

scheduler-next-time returns the earliest active deadline or NIL. Call scheduler-wake :now on a host deadline, and scheduler-event :event :id :payload for an observable event. Stable external event IDs suppress duplicates and reject changed payloads. Occurrence IDs include schedule ID, version and time/event ID. Claim with :id and/or :version to guard stale work.

Missed-run policies are :all (every occurrence), :latest (coalesce to the most recent due occurrence), and :skip (omit past occurrences, emit one exactly at NOW). Catch-up capacity failure rolls back the entire wake operation. No terminal history is evicted implicitly: exhausted retention refuses admission. Capacity bounds each schedule/wakeup collection.

Claim before external admission. scheduler-ack :id :token :outcome :result settles :completed, :failed or :cancelled and retains host proof. Identical ACK is idempotent; stale tokens or conflicting outcomes are refused. On reload, claimed records become :unknown and are never automatically reexecuted. Inspect them, ACK a known outcome, or explicitly authorize scheduler-resolve :action :retry with the existing token. Retry invalidates that token. :action :unknown also records uncertainty in a running host.

scheduler-cancel :id :version prevents future/pending work. Claimed outcomes retain cancellation requests for host settlement. scheduler-close closes new admission and cancels pending work while permitting ACK/inspection.

Acknowledged mailboxes (optional)

Load cl-jobpond/mailboxes. make-mailbox accepts :capacity (64), :history-limit (4096), :clock, :store and :snapshot. Capacity includes queued plus every unacknowledged delivery, including unknown recovered outcomes. Retained history has a separate bound and is never automatically evicted.

(let* ((mailbox (cl-jobpond:make-mailbox))
       (sent (cl-jobpond:mailbox-send mailbox :id "m1" :sender "a" :receiver "b"
                                    :payload '(:text "ready")))
       (delivery (cl-jobpond:mailbox-receive mailbox :receiver "b")))
  (declare (ignore sent))
  (cl-jobpond:mailbox-ack mailbox :id "m1" :receiver "b"
                         :token (getf delivery :token) :result "accepted"))

SEND returns the retained message and a second value indicating an identical duplicate. Changed identity/payload is a conflict, including after acknowledgement. RECEIVE atomically claims the oldest queued message for its receiver, optionally restricted by :id. Receiver identity and the delivery token are required for ACK. This library records identity; hosts authorize it before invoking these operations.

mailbox-find and mailbox-snapshot return isolated records. mailbox-wait takes either :id (wait for terminal acknowledgement/cancellation) or :receiver (wait for a queued message without claiming it), plus a real-seconds :timeout and optional :cancelled-p. It returns record/state, or NIL and :timeout, :cancelled or :closed. Waiters use condition notifications; only injected cancellation predicates require bounded periodic sampling.

Reload converts deliveries to :unknown. mailbox-resolve :id :receiver :token :action :retry authorizes redelivery explicitly and invalidates the old token. :action :unknown records uncertainty without redelivery. mailbox-cancel :id :sender terminates queued messages; delivered messages retain their unacknowledged slot until settlement. mailbox-close closes SEND/RECEIVE and cancels queued work. ACK and identical SEND lookup are available after closure to resolve in-flight work.

mailbox-forget mailbox :ids ids atomically removes exactly the caller-specified acknowledged/cancelled records and returns their count. Supply a proper list of unique stable string IDs. A missing ID (:not-found), unsettled queued, delivered or unknown record (:state), duplicate (:duplicate-identity), or invalid identity (:identity) signals durable-state-error and rejects the whole operation. Other records, delivery tokens and the sequence are preserved; a store refusal leaves live state unchanged. Nothing is pruned automatically. Retain your own durable receipts and prevent re-admission of forgotten identities.

Snapshot persistence

Both optional systems share cl-jobpond/durable-state without additional runtime dependencies. :store is a function of one isolated complete version-one snapshot. It runs under the state lock before visibility and must atomically durably replace storage, or signal an ordinary ERROR without changing storage. It must not reenter the same state. Snapshots contain proper lists, strings, rational numbers, keywords, T and NIL. Data is copied at boundaries; cycles, opaque objects and excessive nesting/node/string sizes are refused. Read persisted snapshots with *read-eval* bound to NIL, and construct a new state with =:snapshot and the same store.

A nonlocal escape from a store callback may follow publication. That live object then refuses access with :uncertain-store; reload the durable snapshot rather than repeating an effect against potentially stale state. A power loss after durable claim likewise recovers as UNKNOWN. Claims/ACKs give explicit at-most-once admission with uncertain outcomes, not a promise of exactly-once external effects.

Transition failures signal durable-state-error with durable-state-error-reason and durable-state-error-identity. Expected reasons include :capacity, :identity-conflict, :stale-version, :stale-token, :outcome-conflict and :closed. Snapshot format/lifecycle validation precedes recovery and persistence.

Completion subscriptions (optional)

Load cl-jobpond/completions. Subscribe to terminal results without a polling thread, using the acknowledged mailbox for bounded storage and delivery claims.

(let ((subscription
        (cl-jobpond:make-completion-subscription
         :store #'atomic-save
         :identity-function (lambda (job snapshot)
                              (declare (ignore snapshot))
                              (my-durable-execution-id job))
         :wakeup (lambda (subscription)
                   (declare (ignore subscription))
                   (signal-my-event-loop)))))
  (unwind-protect
       (progn
         (cl-jobpond:completion-subscription-attach subscription pool)
         ;; Submit jobs after attachment. Later, at the caller's safe boundary:
         (dolist (record (cl-jobpond:completion-subscription-collect subscription :limit 16))
           (persist-delivery-receipt (getf record :id) (getf record :payload))
           (cl-jobpond:completion-subscription-ack
            subscription :id (getf record :id) :token (getf record :token)
            :result '(:receipt "durable-delivery-proof"))))
    (cl-jobpond:completion-subscription-close subscription)))

For an individual job, use completion-subscription-watch, optionally with :id. Watch safely reconciles running and already terminal jobs, even terminal objects that the pool has evicted. Attach reconciles jobs retained by the pool and installs its listener before that reconciliation. Use one attachment per pool, before explicit watches. :identity-function receives the job and coherent raw snapshot; return a durable unique string identity, or NIL to ignore the job. Pool job counters are process-local, so use caller identities for persistence across pool lifetimes.

A collected record has :id, :token and :payload. The payload event contains :format 1, :id, :job-id, :outcome and :snapshot. Outcomes are :success, :failure, :cancelled and :unknown. The full snapshot includes result, condition report, cancellation reason, lifecycle timings and progress. A caller with opaque results supplies :snapshot-function of two arguments (job and raw snapshot), returning the same complete snapshot shape with portable values, such as artifact references. Preserve the raw :identifier and :state; event IDs can differ from that pool-local identifier. Identity and projection callbacks run outside locks, may repeat concurrently, and must be idempotent. The identity callback must be reliable; explicit watch can recover an unidentified job after its callback fails.

Wakeup callbacks run after enqueue, outside locks, on producer threads. They must only signal or schedule short caller work. Repeated/concurrent wakeups and a wakeup already in flight during close are possible. Wakeup failures retain the event and an inspectable completion-subscription-error report. Persistence/projection failures retain bounded watches for completion-subscription-refresh; producer listeners report the failure and request a wakeup. Explicit calls propagate the failure. No callback failure changes the job’s truthful terminal outcome.

Use completion-subscription-forget subscription :ids ids to reclaim settled history after retaining durable caller receipts. It delegates to mailbox-forget without changing pending events or runtime watches. Return NIL from your identity callback for receipted jobs, and exclude them from explicit replay/watch. Pruning removes generic deduplication evidence, so receipt-based exclusion is required before pruning and during restart reconciliation.

:capacity bounds queued plus unacknowledged events and unresolved runtime watches. :history-limit bounds retained deduplication identities; exhausted history refuses new identities rather than dropping evidence. Successful enqueue order is delivery order; concurrent jobs have no promised wall-clock ordering. Each collect claims at most its limit. Caller receipts must precede acknowledgement. A failed collection may have persisted some claims; inspect snapshot messages to settle them.

Persist completion-subscription-snapshot and recreate with :snapshot and the same store. The snapshot exposes :messages for inspection without claiming. Claims restored after a crash have UNKNOWN delivery state, preserving their terminal payload and token. Inspect the caller’s receipts, then acknowledge with the old token or explicitly retry using mailbox-resolve on completion-subscription-mailbox with :receiver "completion" and :action :retry. Runtime watches/listeners are not serialized: persist outstanding job snapshots separately, then reconstruct with completion-subscription-replay and :id. Replay of an interrupted nonterminal job produces an UNKNOWN execution outcome; replay with identical identity/content is detectable and harmless. Conflicting content for a retained identity is refused. Closing detaches producer listeners and releases job references; queued/claimed events can still be collected and acknowledged. It does not close or cancel the underlying mailbox.

Development

./script/bootstrap
./script/check
./script/check --fix

./script/check runs the tests, then checks the source with cheque as cheque.sexp configures it. --fix skips the tests and repairs what cheque can repair. cheque and its tools are development dependencies: checks/qlfile lists them, and the check installs them on first use into ~/.cache/cheque/tools/.

The check runs these test systems:

(asdf:test-system :cl-jobpond)
(asdf:test-system :cl-jobpond/coordination-tests)
(asdf:test-system :cl-jobpond/completions)

License

Licensed under COLL-Attribution. See LICENSE.lisp for the authoritative terms.

Part of the Lambda Symbolics library shelf.