A tiny task manager built on powpow: a ThreadPool for execution plus a private scheduler Loop (own thread) for delayed one-shot and repeating tasks.
- Immediate work goes straight to the pool: jobs run on worker threads, cb/onError fire serialized on the pool dispatch thread (never on the caller's thread — lock shared state there). submit returns a JobId; cancelJob removes a still-queued job synchronously (running jobs cannot be preempted).
- Everything for the scheduler thread (timer registration, cancels, removals, halts, loop stop) crosses as a SchedOp: one proc plus one unmanaged arg. A heartbeat callback drains the pending ops on the scheduler thread — no managed closure is ever posted across threads, so the per-thread cycle tables stay consistent.
- Delayed/repeating tasks may carry a name: cancelTask drops the timer, removeTask additionally frees the name for reuse and guarantees no further fires. Unnamed tasks are tracked by id only.
- Wall-clock scheduling on std/times (local time): scheduleAt runs once at a DateTime (past times stay tracked as taskInactive and never fire); scheduleDaily/scheduleWeekly re-arm per occurrence. taskStatus reports taskArmed/taskCancelled/taskInactive/taskUnknown per id or name.
- Framework-agnostic: this module imports only pkg/powpow and the standard library. No C libraries, no event loop to drive.
Threading contract: Nim ORC tracks cycle candidates per thread, so a managed cell released last on a thread other than the one that shared it corrupts that table. Hence: job/cb closures must not capture ref objects — directly or nested (e.g. seq[SomeRef]) — across threads. Capture values, strings, seqs of values, locks, atomics and raw pointers instead. This matches powpow's submitWork envelope ("single-owner movable T"): closures capturing true refs are equally unsafe there.
Lifecycle: newTaskManager starts the pool and the scheduler thread. stop/shutdown reject new work and tear down the pool (graceful drain vs discard-queued); close additionally joins the scheduler thread and frees the loop. Teardown is idempotent, but close must finish on the creating thread: called from inside a pool/scheduler callback it stops the work and skips the join, and a later close from the creating thread completes the teardown. stop, shutdown and close join pool threads, so they must run outside pool jobs/callbacks — from inside one, call halt.
Types
JobId = distinct int
- Handle for one immediate (submit) job. JobId(0) is invalid (submission rejected while stopping/closed).
SchedProc = proc (ctl: ptr CtlBlock; arg: pointer) {.nimcall, ...gcsafe.}
- One control operation, run on the scheduler thread by the drain (run) — or on the creating thread for leftover ops at close (abort). Args are unmanaged (allocShared0): crossing them touches no cycle table on either thread.
TaskManager = ref object
TaskStatus = enum taskArmed, ## Timer live, will fire (or re-arm, for chained tasks). taskCancelled, ## `cancelTask` stopped it; the name stays reserved. taskInactive, ## Past `scheduleAt`: tracked but never armed, never fires. taskUnknown ## No such task: fired, removed, or never existed.
- Lifecycle state of a delayed/repeating task, per id or name.
Procs
proc cancel(m: TaskManager; id: TimerId) {....raises: [], tags: [], forbids: [].}
- Alias of cancelTask by id.
proc cancelJob(m: TaskManager; id: JobId): bool {....raises: [KeyError], tags: [], forbids: [].}
- Cancel one immediate job by id. Returns true iff the job was still queued: it will never run and neither callback fires. A running, finished or unknown id returns false — a running job runs to completion and delivers normally (jobs cannot be preempted). Takes effect synchronously (lock-guarded, no heartbeat delay), so a true return is authoritative at call time. Thread-safe, including from inside callbacks.
proc cancelTask(m: TaskManager; id: TimerId) {....raises: [], tags: [], forbids: [].}
- Drop a delayed/repeating/scheduled timer by id. Unknown or already-fired ids are ignored; inactive ids are a no-op; a stale chained id still reaches the live occurrence. Takes effect at the next heartbeat (~10ms); a repeating timer may fire once more first (powpow drops timer nodes lazily). Thread-safe, including from inside callbacks.
proc cancelTask(m: TaskManager; name: string) {....raises: [], tags: [], forbids: [].}
- Drop a named delayed/repeating/scheduled task. Unknown names are ignored; otherwise identical to cancelTask by id. The name resolves on the scheduler thread, so a chain re-arming between the call and the op still stops.
proc close(m: TaskManager) {....raises: [Exception, Exception], tags: [RootEffect], forbids: [].}
- stop, then join the scheduler thread, abort leftover ops (registration waiters wake with a closed error), reclaim armed timer payloads, and free the loop and control block. Idempotent. When called from inside a pool/scheduler callback the join and loop teardown are skipped (they would deadlock); call close again from the creating thread to finish.
proc halt(m: TaskManager; delayMs: int): bool {.discardable, ...raises: [], tags: [], forbids: [].}
- Gracefully stop the manager after delayMs milliseconds (plus one heartbeat for the op to land). Returns false when stopping/closed. Only enqueues and returns, so it is safe from inside a job/callback — the recommended way to stop from there.
proc hasTask(m: TaskManager; id: TimerId): bool {....raises: [], tags: [], forbids: [].}
- True while the timer id is armed (named or not).
proc hasTask(m: TaskManager; name: string): bool {....raises: [], tags: [], forbids: [].}
- True while a named delayed/repeating task is armed. Removed, fired or unknown names return false.
proc isRunning(m: TaskManager): bool {....raises: [], tags: [], forbids: [].}
- False once stop/shutdown began or close completed.
proc newTaskManager(poolSize = 4): TaskManager {. ...raises: [ThreadPoolError, OSError, Exception, ResourceExhaustedError], tags: [TimeEffect, RootEffect], forbids: [].}
- Create a manager with poolSize pool workers and start the private scheduler thread. Blocks briefly until the scheduler loop is running, so the first delayed submit never races startup.
proc nextDailyDelayMsFrom(nowT: Time; hour, minute, second: int; minLeadMs = 1000): int {....raises: [], tags: [], forbids: [].}
- Milliseconds from nowT to the next local hour:minute:second at least minLeadMs out (today when still future enough, else tomorrow, stepping whole days so DST stays calendar-correct). Public so callers can preview when a daily task will fire. The lead exists for chains: a fire landing inside its own target second would otherwise recompute a ~0ms delay and echo the same occurrence twice.
proc nextWeeklyDelayMsFrom(nowT: Time; weekday: WeekDay; hour, minute, second: int; minLeadMs = 1000): int {. ...raises: [], tags: [], forbids: [].}
- Milliseconds from nowT to the next local weekday + hour:minute:second at least minLeadMs out (this week when still future enough, else next week, stepping whole weeks). Same echo protection as nextDailyDelayMsFrom.
proc poolSize(m: TaskManager): int {....raises: [], tags: [], forbids: [].}
- Worker count the pool was created with.
proc rawPool(m: TaskManager): ThreadPool {....raises: [], tags: [], forbids: [].}
- The underlying powpow pool, for advanced use (e.g. submitting with submitWork directly). Prefer the submit helpers.
proc removeTask(m: TaskManager; id: TimerId) {....raises: [], tags: [], forbids: [].}
- Strict removal by id: drop the timer, flag the payload dead (a lazy spurious fire drops it silently instead of running the job) and free a taken name for reuse. Payload memory is reclaimed by the fire path or the close walk. Unknown ids are ignored. Takes effect at the next heartbeat (~10ms). Thread-safe, including from inside callbacks.
proc removeTask(m: TaskManager; name: string) {....raises: [], tags: [], forbids: [].}
- Strict removal by name. Unknown names are ignored; otherwise identical to removeTask by id. Resolves on the scheduler thread, like cancelTask by name.
proc scheduleAt[T](m: TaskManager; at: DateTime; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc (err: ref CatchableError) {.closure.} = nil; name = ""): TimerId
- Run job once at wall-clock at (local time, then like submit). Returns the timer id for cancelTask/removeTask and taskStatus. A past at never fires: the task stays tracked as taskInactive (no warning, name still reserved). An optional name arms it as a named task (duplicate names raise CatchableError). Raises CatchableError when stopping/closed.
proc scheduleDaily[T](m: TaskManager; hour, minute: int; second = 0; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc (err: ref CatchableError) {.closure.} = nil; name = ""): TimerId
- Run job every day at local hour:minute:second (then like submit per fire). Returns the first timer id; each occurrence re-arms for the next day, recomputed from local now() so DST shifts land on one 23h/25h day. cancelTask/removeTask by name stop the chain; a stale id still reaches the live occurrence. Raises CatchableError on out-of-range times or stopping/closed.
proc scheduleWeekly[T](m: TaskManager; weekday: WeekDay; hour, minute: int; second = 0; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc ( err: ref CatchableError) {.closure.} = nil; name = ""): TimerId
- Run job every week on weekday at local hour:minute:second (then like submit per fire). Same chaining, cancellation and error rules as scheduleDaily.
proc shutdown(m: TaskManager) {....raises: [Exception], tags: [RootEffect], forbids: [].}
- Immediate stop: like stop, but jobs still queued (never started) are discarded — their callbacks never fire, and their job cells are reaped here. In-flight jobs finish and deliver. Blocks. Idempotent. Must run outside pool jobs/callbacks (it joins pool threads).
proc stop(m: TaskManager) {....raises: [Exception], tags: [RootEffect], forbids: [].}
- Graceful stop: reject new submissions, drop pending timers, let the pool drain queued jobs (every queued job still runs and delivers, cancelled immediates stay silent). Blocks until the pool is torn down. Idempotent. Must run outside pool jobs/callbacks (it joins pool threads).
proc submit[T](m: TaskManager; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc (err: ref CatchableError) {.closure.} = nil): JobId
-
Queue job for immediate execution on a pool worker. cb(res) fires on the pool dispatch thread; onError(err) instead when the job raises. Returns a JobId for cancelJob, or JobId(0) when stopping/closed or the pool is already torn down — then neither callback fires.
Thread-safe: may be called from any thread, including from inside callbacks. See the module contract about captured references.
proc submitDelayed[T](m: TaskManager; delayMs: int; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc (err: ref CatchableError) {.closure.} = nil; name = ""): TimerId
- Run job once after delayMs milliseconds (then like submit). Returns the timer id for cancelTask/removeTask. An optional name arms it as a named task (duplicate names raise CatchableError). Raises CatchableError when stopping/closed.
proc submitRepeating[T](m: TaskManager; intervalMs: int; job: proc (): T {.closure.}; cb: proc (res: T) {.closure.}; onError: proc ( err: ref CatchableError) {.closure.} = nil; name = ""): TimerId
- Run job every intervalMs milliseconds until cancelled (then like submit per fire). Returns the timer id. An optional name arms it as a named task (duplicate names raise CatchableError). Raises CatchableError when stopping/closed. Stopping the manager drops all repeating timers. Cancelled payloads are reclaimed at close (powpow drops timer nodes lazily, with no hook).
proc taskStatus(m: TaskManager; id: TimerId): TaskStatus {....raises: [KeyError], tags: [], forbids: [].}
- Lifecycle state of one delayed/repeating/scheduled task by id. Thread-safe, including from inside callbacks.
proc taskStatus(m: TaskManager; name: string): TaskStatus {....raises: [KeyError], tags: [], forbids: [].}
- Lifecycle state of one named task. Unknown names are taskUnknown. Thread-safe, including from inside callbacks.