2 if (!file_exists("boot.php") AND (sizeof($_SERVER["argv"]) != 0)) {
3 $directory = dirname($_SERVER["argv"][0]);
5 if (substr($directory, 0, 1) != "/")
6 $directory = $_SERVER["PWD"]."/".$directory;
8 $directory = realpath($directory."/..");
13 use \Friendica\Core\Config;
14 use \Friendica\Core\PConfig;
16 require_once("boot.php");
18 function poller_run($argv, $argc){
26 @include(".htconfig.php");
27 require_once("include/dba.php");
28 $db = new dba($db_host, $db_user, $db_pass, $db_data);
29 unset($db_host, $db_user, $db_pass, $db_data);
32 // Quit when in maintenance
33 if (Config::get('system', 'maintenance', true)) {
39 if (poller_max_connections_reached()) {
43 if ($a->maxload_reached()) {
47 if(($argc <= 1) OR ($argv[1] != "no_cron")) {
51 if ($a->max_processes_reached()) {
55 // Checking the number of workers
56 if (poller_too_much_workers()) {
57 poller_kill_stale_workers();
63 while ($r = poller_worker_process()) {
65 // Count active workers and compare them with a maximum value that depends on the load
66 if (poller_too_much_workers()) {
70 if (!poller_execute($r[0])) {
74 // Quit the poller once every hour
75 if (time() > ($starttime + 3600))
82 * @brief Execute a worker entry
84 * @param array $queue Workerqueue entry
86 * @return boolean "true" if further processing should be stopped
88 function poller_execute($queue) {
94 $cooldown = Config::get("system", "worker_cooldown", 0);
96 // Quit when in maintenance
97 if (Config::get('system', 'maintenance', true)) {
101 // Constantly check the number of parallel database processes
102 if ($a->max_processes_reached()) {
106 // Constantly check the number of available database connections to let the frontend be accessible at any time
107 if (poller_max_connections_reached()) {
111 $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0",
112 dbesc(datetime_convert()),
114 intval($queue["id"]));
117 logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG);
122 // Assure that there are no tasks executed twice
123 $id = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
125 logger("Queue item ".$queue["id"]." vanished - skip this execution", LOGGER_DEBUG);
128 } elseif ((strtotime($id[0]["executed"]) <= 0) OR ($id[0]["pid"] == 0)) {
129 logger("Entry for queue item ".$queue["id"]." wasn't stored - skip this execution", LOGGER_DEBUG);
132 } elseif ($id[0]["pid"] != $mypid) {
133 logger("Queue item ".$queue["id"]." is to be executed by process ".$id[0]["pid"]." and not by me (".$mypid.") - skip this execution", LOGGER_DEBUG);
139 $argv = json_decode($queue["parameter"]);
141 $argc = count($argv);
143 // Check for existance and validity of the include file
146 if (!validate_include($include)) {
147 logger("Include file ".$argv[0]." is not valid!");
148 q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
152 require_once($include);
154 $funcname = str_replace(".php", "", basename($argv[0]))."_run";
156 if (function_exists($funcname)) {
158 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]);
160 $stamp = (float)microtime(true);
162 if (Config::get("system", "profiler")) {
163 $a->performance["start"] = microtime(true);
164 $a->performance["database"] = 0;
165 $a->performance["database_write"] = 0;
166 $a->performance["network"] = 0;
167 $a->performance["file"] = 0;
168 $a->performance["rendering"] = 0;
169 $a->performance["parser"] = 0;
170 $a->performance["marktime"] = 0;
171 $a->performance["markstart"] = microtime(true);
172 $a->callstack = array();
175 // For better logging create a new process id for every worker call
176 // But preserve the old one for the worker
177 $old_process_id = $a->process_id;
178 $a->process_id = uniqid("wrk", true);
180 $funcname($argv, $argc);
182 $a->process_id = $old_process_id;
184 $duration = number_format(microtime(true) - $stamp, 3);
186 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done in ".$duration." seconds.");
188 if (Config::get("system", "profiler")) {
189 $duration = microtime(true)-$a->performance["start"];
191 if (Config::get("rendertime", "callstack")) {
192 $o = "\nDatabase Read:\n";
193 foreach ($a->callstack["database"] AS $func => $time) {
194 $time = round($time, 3);
196 $o .= $func.": ".$time."\n";
198 $o .= "\nDatabase Write:\n";
199 foreach ($a->callstack["database_write"] AS $func => $time) {
200 $time = round($time, 3);
202 $o .= $func.": ".$time."\n";
205 $o .= "\nNetwork:\n";
206 foreach ($a->callstack["network"] AS $func => $time) {
207 $time = round($time, 3);
209 $o .= $func.": ".$time."\n";
215 logger("ID ".$queue["id"].": ".$funcname.": ".sprintf("DB: %s/%s, Net: %s, I/O: %s, Other: %s, Total: %s".$o,
216 number_format($a->performance["database"] - $a->performance["database_write"], 2),
217 number_format($a->performance["database_write"], 2),
218 number_format($a->performance["network"], 2),
219 number_format($a->performance["file"], 2),
220 number_format($duration - ($a->performance["database"] + $a->performance["network"] + $a->performance["file"]), 2),
221 number_format($duration, 2)),
226 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds");
230 q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
232 logger("Function ".$funcname." does not exist");
239 * @brief Checks if the number of database connections has reached a critical limit.
241 * @return bool Are more than 3/4 of the maximum connections used?
243 function poller_max_connections_reached() {
245 // Fetch the max value from the config. This is needed when the system cannot detect the correct value by itself.
246 $max = Config::get("system", "max_connections");
248 // Fetch the percentage level where the poller will get active
249 $maxlevel = Config::get("system", "max_connections_level", 75);
252 // the maximum number of possible user connections can be a system variable
253 $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
255 $max = $r[0]["Value"];
257 // Or it can be granted. This overrides the system variable
258 $r = q("SHOW GRANTS");
260 foreach ($r AS $grants) {
261 $grant = array_pop($grants);
262 if (stristr($grant, "GRANT USAGE ON"))
263 if (preg_match("/WITH MAX_USER_CONNECTIONS (\d*)/", $grant, $match))
268 // If $max is set we will use the processlist to determine the current number of connections
269 // The processlist only shows entries of the current user
271 $r = q("SHOW PROCESSLIST");
272 if (!dbm::is_result($r))
277 logger("Connection usage (user values): ".$used."/".$max, LOGGER_DEBUG);
279 $level = ($used / $max) * 100;
281 if ($level >= $maxlevel) {
282 logger("Maximum level (".$maxlevel."%) of user connections reached: ".$used."/".$max);
287 // We will now check for the system values.
288 // This limit could be reached although the user limits are fine.
289 $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_connections'");
293 $max = intval($r[0]["Value"]);
297 $r = q("SHOW STATUS WHERE `variable_name` = 'Threads_connected'");
301 $used = intval($r[0]["Value"]);
305 logger("Connection usage (system values): ".$used."/".$max, LOGGER_DEBUG);
307 $level = $used / $max * 100;
309 if ($level < $maxlevel)
312 logger("Maximum level (".$level."%) of system connections reached: ".$used."/".$max);
317 * @brief fix the queue entry if the worker process died
320 function poller_kill_stale_workers() {
321 $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'");
323 if (!dbm::is_result($r)) {
324 // No processing here needed
329 if (!posix_kill($pid["pid"], 0))
330 q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `pid` = 0 WHERE `pid` = %d",
331 intval($pid["pid"]));
333 // Kill long running processes
335 // Check if the priority is in a valid range
336 if (!in_array($pid["priority"], array(PRIORITY_CRITICAL, PRIORITY_HIGH, PRIORITY_MEDIUM, PRIORITY_LOW, PRIORITY_NEGLIGIBLE)))
337 $pid["priority"] = PRIORITY_MEDIUM;
339 // Define the maximum durations
340 $max_duration_defaults = array(PRIORITY_CRITICAL => 360, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 360);
341 $max_duration = $max_duration_defaults[$pid["priority"]];
343 $argv = json_decode($pid["parameter"]);
344 $argv[0] = basename($argv[0]);
346 // How long is the process already running?
347 $duration = (time() - strtotime($pid["executed"])) / 60;
348 if ($duration > $max_duration) {
349 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") took more than ".$max_duration." minutes. It will be killed now.");
350 posix_kill($pid["pid"], SIGTERM);
352 // We killed the stale process.
353 // To avoid a blocking situation we reschedule the process at the beginning of the queue.
354 // Additionally we are lowering the priority.
355 q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `created` = '%s',
356 `priority` = %d, `pid` = 0 WHERE `pid` = %d",
357 dbesc(datetime_convert()),
358 intval(PRIORITY_NEGLIGIBLE),
359 intval($pid["pid"]));
361 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") now runs for ".round($duration)." of ".$max_duration." allowed minutes. That's okay.", LOGGER_DEBUG);
366 * @brief Checks if the number of active workers exceeds the given limits
368 * @return bool Are there too much workers running?
370 function poller_too_much_workers() {
371 $queues = Config::get("system", "worker_queues", 4);
373 $maxqueues = $queues;
375 $active = poller_active_workers();
377 // Decrease the number of workers at higher load
378 $load = current_load();
380 $maxsysload = intval(Config::get("system", "maxloadavg", 50));
382 $maxworkers = $queues;
384 // Some magical mathemathics to reduce the workers
386 $slope = $maxworkers / pow($maxsysload, $exponent);
387 $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent));
389 $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00'");
390 $entries = $s[0]["total"];
392 if (Config::get("system", "worker_fastlane", false) AND ($queues > 0) AND ($entries > 0) AND ($active >= $queues)) {
393 $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority` LIMIT 1");
394 $top_priority = $s[0]["priority"];
396 $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` != '0000-00-00 00:00:00' LIMIT 1",
397 intval($top_priority));
398 $high_running = dbm::is_result($s);
400 if (!$high_running AND ($top_priority > PRIORITY_UNDEFINED) AND ($top_priority < PRIORITY_NEGLIGIBLE)) {
401 logger("There are jobs with priority ".$top_priority." waiting but none is executed. Open a fastlane.", LOGGER_DEBUG);
402 $queues = $active + 1;
406 // Create a list of queue entries grouped by their priority
407 $running = array(PRIORITY_CRITICAL => 0,
409 PRIORITY_MEDIUM => 0,
411 PRIORITY_NEGLIGIBLE => 0);
413 $r = q("SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` GROUP BY `priority`");
414 if (dbm::is_result($r))
415 foreach ($r AS $process)
416 $running[$process["priority"]] = $process["running"];
419 $r = q("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`");
420 if (dbm::is_result($r))
421 foreach ($r as $entry) {
422 if ($processlist != "")
423 $processlist .= ", ";
424 $processlist .= $entry["priority"].":".$running[$entry["priority"]]."/".$entry["entries"];
427 logger("Load: ".$load."/".$maxsysload." - processes: ".$active."/".$entries." (".$processlist.") - maximum: ".$queues."/".$maxqueues, LOGGER_DEBUG);
429 // Are there fewer workers running as possible? Then fork a new one.
430 if (!Config::get("system", "worker_dont_fork") AND ($queues > ($active + 1)) AND ($entries > 1)) {
431 logger("Active workers: ".$active."/".$queues." Fork a new worker.", LOGGER_DEBUG);
432 $args = array("php", "include/poller.php", "no_cron");
438 return($active >= $queues);
442 * @brief Returns the number of active poller processes
444 * @return integer Number of active poller processes
446 function poller_active_workers() {
447 $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php'");
449 return($workers[0]["processes"]);
453 * @brief Check if we should pass some slow processes
455 * When the active processes of the highest priority are using more than 2/3
456 * of all processes, we let pass slower processes.
458 * @param string $highest_priority Returns the currently highest priority
459 * @return bool We let pass a slower process than $highest_priority
461 function poller_passing_slow(&$highest_priority) {
463 $highest_priority = 0;
465 $r = q("SELECT `priority`
467 INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`");
469 // No active processes at all? Fine
470 if (!dbm::is_result($r))
473 $priorities = array();
474 foreach ($r AS $line)
475 $priorities[] = $line["priority"];
478 if (count($priorities) == 0)
481 $highest_priority = min($priorities);
483 // The highest process is already the slowest one?
485 if ($highest_priority == PRIORITY_NEGLIGIBLE)
489 foreach ($priorities AS $priority)
490 if ($priority == $highest_priority)
493 logger("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, LOGGER_DEBUG);
494 $passing_slow = (($high/count($priorities)) > (2/3));
497 logger("Passing slower processes than priority ".$highest_priority, LOGGER_DEBUG);
499 return($passing_slow);
503 * @brief Returns the next worker process
505 * @return string SQL statement
507 function poller_worker_process() {
509 q("START TRANSACTION;");
511 // Check if we should pass some low priority process
512 $highest_priority = 0;
514 if (poller_passing_slow($highest_priority)) {
515 // Are there waiting processes with a higher priority than the currently highest?
516 $r = q("SELECT * FROM `workerqueue`
517 WHERE `executed` = '0000-00-00 00:00:00' AND `priority` < %d
518 ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority));
519 if (dbm::is_result($r))
522 // Give slower processes some processing time
523 $r = q("SELECT * FROM `workerqueue`
524 WHERE `executed` = '0000-00-00 00:00:00' AND `priority` > %d
525 ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority));
528 // If there is no result (or we shouldn't pass lower processes) we check without priority limit
529 if (($highest_priority == 0) OR !dbm::is_result($r))
530 $r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority`, `created` LIMIT 1");
536 * @brief Call the front end worker
538 function call_worker() {
539 if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) {
543 $url = App::get_baseurl()."/worker";
544 fetch_url($url, false, $redirects, 1);
548 * @brief Call the front end worker if there aren't any active
550 function call_worker_if_idle() {
551 if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) {
555 // Do we have "proc_open"? Then we can fork the poller
556 if (function_exists("proc_open")) {
557 // When was the last time that we called the worker?
558 // Less than one minute? Then we quit
559 if ((time() - Config::get("system", "worker_started")) < 60) {
563 set_config("system", "worker_started", time());
565 // Do we have enough running workers? Then we quit here.
566 if (poller_too_much_workers()) {
567 // Cleaning dead processes
568 poller_kill_stale_workers();
569 get_app()->remove_inactive_processes();
576 logger('Call poller', LOGGER_DEBUG);
578 $args = array("php", "include/poller.php", "no_cron");
584 // We cannot execute background processes.
585 // We now run the processes from the frontend.
586 // This won't work with long running processes.
589 clear_worker_processes();
591 $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php'");
593 if ($workers[0]["processes"] == 0) {
599 * @brief Removes long running worker processes
601 function clear_worker_processes() {
602 $timeout = Config::get("system", "frontend_worker_timeout", 10);
604 /// @todo We should clean up the corresponding workerqueue entries as well
605 q("DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php'",
606 dbesc(datetime_convert('UTC','UTC',"now - ".$timeout." minutes")));
610 * @brief Runs the cron processes
612 function poller_run_cron() {
613 logger('Add cron entries', LOGGER_DEBUG);
615 // Check for spooled items
616 proc_run(PRIORITY_HIGH, "include/spool_post.php");
618 // Run the cron job that calls all other jobs
619 proc_run(PRIORITY_MEDIUM, "include/cron.php");
621 // Run the cronhooks job separately from cron for being able to use a different timing
622 proc_run(PRIORITY_MEDIUM, "include/cronhooks.php");
624 // Cleaning dead processes
625 poller_kill_stale_workers();
628 if (array_search(__file__,get_included_files())===0){
629 poller_run($_SERVER["argv"],$_SERVER["argc"]);
631 get_app()->end_process();