diff --git a/README.md b/README.md index a1526d0..a829e31 100644 --- a/README.md +++ b/README.md @@ -2101,6 +2101,7 @@ Create `config/server.json` next to your `web/` directory, or pass `--config=pat | `rateLimit.enabled` | false | Enable per-IP rate limiting | | `rateLimit.requests` | 100 | Requests per window | | `rateLimit.window` | 60 | Window in seconds | +| `webserver.requestTimeout` | 30 | Seconds before killing a hung HTTP worker (0 = no limit) | | `socket.io` | `"/socket.io"` | Socket.IO endpoint. Protocol detection + client JS at `{path}/socket.io.js`. `false` to disable. | | `socket.js` | `"/Q/socket.js"` | Path to serve the minimal bare-WebSocket client (3KB). `false` to disable. | | `app` | `""` | App name — prefixes handler function names (e.g. `"Chess"` → `Chess_chat_message()`) | @@ -2108,6 +2109,60 @@ Create `config/server.json` next to your `web/` directory, or pass `--config=pat | `webserver.cgi.patterns` | [] | Regex patterns for scripts that use php-cgi (legacy compatibility) | | `webserver.cgi.binary` | auto | Path to php-cgi binary (auto-detected if not set) | +### Scheduler + +Run tasks on intervals or at specific times. Handlers are forked like HTTP +requests — they don't block the event loop and respect `requestTimeout`. + +```json +{ + "Q": { + "scheduler": { + "cleanup": { + "handler": "tasks/cleanup", + "every": 3600 + }, + "daily-report": { + "handler": "tasks/report", + "times": ["09:00"] + }, + "business-check": { + "handler": "tasks/check", + "times": ["09:00", "12:00", "17:00"], + "weekdays": ["mon", "wed", "fri"] + }, + "monthly-invoice": { + "handler": "tasks/invoice", + "times": ["00:00"], + "monthdays": [1] + } + } + } +} +``` + +| Field | What it does | +|---|---| +| `handler` | Handler path — dispatched via `Q::event()`, same as HTTP handlers | +| `every` | Run every N seconds from startup | +| `times` | Run at specific `HH:MM` times (24h format) | +| `weekdays` | Only fire on these days: `mon`, `tue`, `wed`, `thu`, `fri`, `sat`, `sun` | +| `monthdays` | Only fire on these days of the month: `[1]`, `[1, 15]`, etc. | + +The handler receives `$params['task']` (the task name) and `$params['scheduled'] = true`: + +```php + timestamp of last run */ + static $lastRun = array(); + /** @var float Server start time */ + static $startTime = 0; + /** @var string The HH:MM at startup (to skip on restart) */ + static $startMinute = ''; + + /** + * Initialize the scheduler with task definitions. + * Marks the current minute as "already checked" to avoid + * re-firing tasks if the server restarts mid-minute. + */ + static function init($schedule) + { + self::$startTime = microtime(true); + self::$startMinute = date('H:i'); + + foreach ($schedule as $name => $def) { + if (empty($def['handler'])) continue; + self::$tasks[$name] = $def; + + // For interval tasks, set lastRun to now so first fire + // is one full interval after startup + if (isset($def['every'])) { + self::$lastRun[$name] = self::$startTime; + } + + // For time-based tasks, mark current minute as run + // to prevent re-fire on restart + if (isset($def['times'])) { + if (in_array(self::$startMinute, $def['times'])) { + self::$lastRun[$name] = self::$startTime; + } + } + } + } + + /** + * Called every second by the event loop. + * Checks each task and fires if due. + */ + static function tick() + { + $now = microtime(true); + $minute = date('H:i'); + $wday = strtolower(date('D')); // mon, tue, wed... + $mday = (int) date('j'); // 1-31 + + foreach (self::$tasks as $name => $def) { + // Interval-based: "every" seconds + if (isset($def['every'])) { + $last = self::$lastRun[$name] ?? 0; + if ($now - $last >= $def['every']) { + self::run($name, $def); + } + continue; + } + + // Time-based: check if current HH:MM matches + if (isset($def['times'])) { + if (!in_array($minute, $def['times'])) continue; + + // Already ran this minute? + $last = self::$lastRun[$name] ?? 0; + if ($now - $last < 60) continue; + + // Weekday filter + if (isset($def['weekdays'])) { + $allowed = array_map('strtolower', $def['weekdays']); + // Accept both "mon" and "monday" style + $wdayFull = strtolower(date('l')); + if (!in_array($wday, $allowed) && !in_array($wdayFull, $allowed)) { + continue; + } + } + + // Monthday filter + if (isset($def['monthdays'])) { + if (!in_array($mday, $def['monthdays'])) continue; + } + + self::run($name, $def); + } + } + } + + /** + * Run a task by forking a child process. + * Sets lastRun BEFORE launching to err on the side of skipping. + */ + static function run($name, $def) + { + $handler = $def['handler']; + + // Mark as run BEFORE fork — if we crash, we skip rather than double-run + self::$lastRun[$name] = microtime(true); + + if (!function_exists('pcntl_fork')) { + // No fork — run in-process (blocks event loop briefly) + $result = null; + Q::event($handler, array('task' => $name, 'scheduled' => true), false, false, $result); + return; + } + + $pid = pcntl_fork(); + if ($pid === 0) { + // CHILD + $result = null; + try { + Q::event($handler, array('task' => $name, 'scheduled' => true), false, false, $result); + } catch (\Throwable $e) { + // Task failed — log but don't crash + fwrite(STDERR, date('H:i:s') . " scheduler: $name failed: " . $e->getMessage() . "\n"); + } + exit(0); + } elseif ($pid > 0) { + // PARENT — track for timeout enforcement + Q_WebServer::$workerPids[$pid] = microtime(true); + pcntl_waitpid($pid, $st, WNOHANG); + } + } +} diff --git a/src/Q/WebServer.php b/src/Q/WebServer.php index 4844fb3..45f7280 100644 --- a/src/Q/WebServer.php +++ b/src/Q/WebServer.php @@ -349,9 +349,39 @@ class Q_WebServer }); // Reap zombie children from fork-per-request PHP execution Q_Evented::onSignal(SIGCHLD, function () { - while (pcntl_waitpid(-1, $st, WNOHANG) > 0) {} + while (($pid = pcntl_waitpid(-1, $st, WNOHANG)) > 0) { + unset(Q_WebServer::$workerPids[$pid]); + } }); } + // Engine.IO ping timer — keeps Socket.IO connections alive + Q_Evented::repeat(25, function () { + Q_WebSocket::pingSocketIO(); + }); + + // Request timeout — kill workers that exceed the configured limit + $timeout = Q_Config::get('Q', 'webserver', 'requestTimeout', 30); + if ($timeout > 0) { + Q_Evented::repeat(1, function () use ($timeout) { + $now = microtime(true); + foreach (Q_WebServer::$workerPids as $pid => $start) { + if ($now - $start > $timeout) { + @posix_kill($pid, SIGKILL); + unset(Q_WebServer::$workerPids[$pid]); + } + } + }); + } + + // Scheduler — run tasks on intervals or at specific times + $schedule = Q_Config::get('Q', 'scheduler', array()); + if (!empty($schedule)) { + Q_Scheduler::init($schedule); + Q_Evented::repeat(1, function () { + Q_Scheduler::tick(); + }); + } + Q_Evented::run(); } @@ -1040,6 +1070,7 @@ class Q_WebServer Q_Evented::cancel(self::$clientWatchers[$key]); } unset(self::$clientWatchers[$key], self::$clients[$key], self::$buffers[$key]); + self::$workerPids[$pid] = microtime(true); pcntl_waitpid($pid, $st, WNOHANG); self::$lastStatus = 200; list($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE) = $saved; @@ -1155,6 +1186,7 @@ class Q_WebServer Q_Evented::cancel(self::$clientWatchers[$key]); } unset(self::$clientWatchers[$key], self::$clients[$key], self::$buffers[$key]); + self::$workerPids[$pid] = microtime(true); // Non-blocking reap — don't wait for child pcntl_waitpid($pid, $st, WNOHANG); self::$lastStatus = 200; @@ -2460,6 +2492,8 @@ HTML private static $running = false; private static $lastStatus = 200; private static $lastBody = ''; + /** @internal pid => start_time for request timeout enforcement */ + static $workerPids = array(); static $allowedExtensions = array( 'html','htm','txt','md','json','xml','yaml','yml','csv','tsv','log', diff --git a/src/Q/WebSocket.php b/src/Q/WebSocket.php index 4930081..9100e93 100644 --- a/src/Q/WebSocket.php +++ b/src/Q/WebSocket.php @@ -382,6 +382,19 @@ class Q_WebSocket self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $text); } + /** + * Send Engine.IO ping to all Socket.IO clients. + * Called on a 25s timer by the parent process. + */ + static function pingSocketIO() + { + foreach (self::$clients as $sk => $c) { + if (($c['protocol'] ?? 'json') === 'socketio') { + self::sendRaw($sk, '2'); + } + } + } + /** * Send a Socket.IO ACK response: 43[data] */