X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=include%2Fpoller.php;h=bbec43ae7bcc2c1e1f6895ab4a959e6b090cd50e;hb=99fb15037bde3017b8fe270ddcaa802548f5ed8c;hp=714d4c537c896f9ae79f2a01f6ba04d9a2c9e1ef;hpb=4914e8826beb8dc8853580e383f60538158707b0;p=friendica.git diff --git a/include/poller.php b/include/poller.php index 714d4c537c..89ab85a1f5 100644 --- a/include/poller.php +++ b/include/poller.php @@ -10,10 +10,11 @@ if (!file_exists("boot.php") AND (sizeof($_SERVER["argv"]) != 0)) { chdir($directory); } +use \Friendica\Core\Config; + require_once("boot.php"); -require_once("dbm.php"); -function poller_run(&$argv, &$argc){ +function poller_run($argv, $argc){ global $a, $db; if(is_null($a)) { @@ -27,102 +28,240 @@ function poller_run(&$argv, &$argc){ unset($db_host, $db_user, $db_pass, $db_data); }; - $processlist = dbm::processlist(); - if ($processlist != "") - logger("Processlist: ".$processlist, LOGGER_DEBUG); + Config::load(); - if (poller_max_connections_reached()) + // Quit when in maintenance + if (Config::get('system', 'maintenance', true)) { return; + } + + $a->set_baseurl(Config::get('system', 'url')); - if (App::maxload_reached()) + load_hooks(); + + $a->start_process(); + + if (poller_max_connections_reached()) { return; + } - // Checking the number of workers - if (poller_too_much_workers(1)) { - poller_kill_stale_workers(); + if ($a->maxload_reached()) { return; } if(($argc <= 1) OR ($argv[1] != "no_cron")) { - // Run the cron job that calls all other jobs - proc_run("php","include/cron.php"); + poller_run_cron(); + } - // Run the cronhooks job separately from cron for being able to use a different timing - proc_run("php","include/cronhooks.php"); + if ($a->max_processes_reached()) { + return; + } - // Cleaning dead processes + // Checking the number of workers + if (poller_too_much_workers()) { poller_kill_stale_workers(); - } else - // Sleep four seconds before checking for running processes again to avoid having too many workers - sleep(4); - - // Checking number of workers - if (poller_too_much_workers(2)) return; + } $starttime = time(); - while ($r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `created` LIMIT 1")) { + while ($r = poller_worker_process()) { - // Log the type of database processes - $processlist = dbm::processlist(); - if ($processlist != "") - logger("Processlist: ".$processlist, LOGGER_DEBUG); + // Count active workers and compare them with a maximum value that depends on the load + if (poller_too_much_workers()) { + return; + } - // Constantly check the number of available database connections to let the frontend be accessible at any time - if (poller_max_connections_reached()) + if (!poller_execute($r[0])) { return; + } - // Count active workers and compare them with a maximum value that depends on the load - if (poller_too_much_workers(3)) + // Quit the poller once every hour + if (time() > ($starttime + 3600)) return; + } - q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `executed` = '0000-00-00 00:00:00'", - dbesc(datetime_convert()), - intval(getmypid()), - intval($r[0]["id"])); - - // Assure that there are no tasks executed twice - $id = q("SELECT `id` FROM `workerqueue` WHERE `id` = %d AND `pid` = %d", - intval($r[0]["id"]), - intval(getmypid())); - if (!$id) { - logger("Queue item ".$r[0]["id"]." was executed multiple times - skip this execution", LOGGER_DEBUG); - continue; - } +} - $argv = json_decode($r[0]["parameter"]); +/** + * @brief Execute a worker entry + * + * @param array $queue Workerqueue entry + * + * @return boolean "true" if further processing should be stopped + */ +function poller_execute($queue) { - $argc = count($argv); + $a = get_app(); - // Check for existance and validity of the include file - $include = $argv[0]; + $mypid = getmypid(); - if (!validate_include($include)) { - logger("Include file ".$argv[0]." is not valid!"); - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); - continue; - } + // Quit when in maintenance + if (Config::get('system', 'maintenance', true)) { + return false; + } - require_once($include); + // Constantly check the number of parallel database processes + if ($a->max_processes_reached()) { + return false; + } - $funcname=str_replace(".php", "", basename($argv[0]))."_run"; + // Constantly check the number of available database connections to let the frontend be accessible at any time + if (poller_max_connections_reached()) { + return false; + } - if (function_exists($funcname)) { - logger("Process ".getmypid()." - ID ".$r[0]["id"].": ".$funcname." ".$r[0]["parameter"]); - $funcname($argv, $argc); + $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0", + dbesc(datetime_convert()), + intval($mypid), + intval($queue["id"])); - logger("Process ".getmypid()." - ID ".$r[0]["id"].": ".$funcname." - done"); + if (!$upd) { + logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + return true; + } - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); - } else - logger("Function ".$funcname." does not exist"); + // Assure that there are no tasks executed twice + $id = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + if (!$id) { + logger("Queue item ".$queue["id"]." vanished - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + return true; + } elseif ((strtotime($id[0]["executed"]) <= 0) OR ($id[0]["pid"] == 0)) { + logger("Entry for queue item ".$queue["id"]." wasn't stored - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + return true; + } elseif ($id[0]["pid"] != $mypid) { + logger("Queue item ".$queue["id"]." is to be executed by process ".$id[0]["pid"]." and not by me (".$mypid.") - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + return true; + } + q("COMMIT"); - // Quit the poller once every hour - if (time() > ($starttime + 3600)) - return; + $argv = json_decode($queue["parameter"]); + + // Check for existance and validity of the include file + $include = $argv[0]; + + if (!validate_include($include)) { + logger("Include file ".$argv[0]." is not valid!"); + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + return true; + } + + require_once($include); + + $funcname = str_replace(".php", "", basename($argv[0]))."_run"; + + if (function_exists($funcname)) { + + poller_exec_function($queue, $funcname, $argv); + + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + } else { + logger("Function ".$funcname." does not exist"); } + return true; +} + +/** + * @brief Execute a function from the queue + * + * @param array $queue Workerqueue entry + * @param string $funcname name of the function + * @param array $argv Array of values to be passed to the function + */ +function poller_exec_function($queue, $funcname, $argv) { + + $a = get_app(); + + $mypid = getmypid(); + + $argc = count($argv); + + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]); + + $stamp = (float)microtime(true); + + // We use the callstack here to analyze the performance of executed worker entries. + // For this reason the variables have to be initialized. + if (Config::get("system", "profiler")) { + $a->performance["start"] = microtime(true); + $a->performance["database"] = 0; + $a->performance["database_write"] = 0; + $a->performance["network"] = 0; + $a->performance["file"] = 0; + $a->performance["rendering"] = 0; + $a->performance["parser"] = 0; + $a->performance["marktime"] = 0; + $a->performance["markstart"] = microtime(true); + $a->callstack = array(); + } + + // For better logging create a new process id for every worker call + // But preserve the old one for the worker + $old_process_id = $a->process_id; + $a->process_id = uniqid("wrk", true); + + $funcname($argv, $argc); + + $a->process_id = $old_process_id; + + $duration = number_format(microtime(true) - $stamp, 3); + + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done in ".$duration." seconds."); + + // Write down the performance values into the log + if (Config::get("system", "profiler")) { + $duration = microtime(true)-$a->performance["start"]; + + if (Config::get("rendertime", "callstack")) { + if (isset($a->callstack["database"])) { + $o = "\nDatabase Read:\n"; + foreach ($a->callstack["database"] AS $func => $time) { + $time = round($time, 3); + if ($time > 0) + $o .= $func.": ".$time."\n"; + } + } + if (isset($a->callstack["database_write"])) { + $o .= "\nDatabase Write:\n"; + foreach ($a->callstack["database_write"] AS $func => $time) { + $time = round($time, 3); + if ($time > 0) + $o .= $func.": ".$time."\n"; + } + } + if (isset($a->callstack["network"])) { + $o .= "\nNetwork:\n"; + foreach ($a->callstack["network"] AS $func => $time) { + $time = round($time, 3); + if ($time > 0) + $o .= $func.": ".$time."\n"; + } + } + } else { + $o = ''; + } + + logger("ID ".$queue["id"].": ".$funcname.": ".sprintf("DB: %s/%s, Net: %s, I/O: %s, Other: %s, Total: %s".$o, + number_format($a->performance["database"] - $a->performance["database_write"], 2), + number_format($a->performance["database_write"], 2), + number_format($a->performance["network"], 2), + number_format($a->performance["file"], 2), + number_format($duration - ($a->performance["database"] + $a->performance["network"] + $a->performance["file"]), 2), + number_format($duration, 2)), + LOGGER_DEBUG); + } + + $cooldown = Config::get("system", "worker_cooldown", 0); + + if ($cooldown > 0) { + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); + sleep($cooldown); + } } /** @@ -133,12 +272,10 @@ function poller_run(&$argv, &$argc){ function poller_max_connections_reached() { // Fetch the max value from the config. This is needed when the system cannot detect the correct value by itself. - $max = get_config("system", "max_connections"); + $max = Config::get("system", "max_connections"); // Fetch the percentage level where the poller will get active - $maxlevel = get_config("system", "max_connections_level"); - if ($maxlevel == 0) - $maxlevel = 75; + $maxlevel = Config::get("system", "max_connections_level", 75); if ($max == 0) { // the maximum number of possible user connections can be a system variable @@ -161,7 +298,7 @@ function poller_max_connections_reached() { // The processlist only shows entries of the current user if ($max != 0) { $r = q("SHOW PROCESSLIST"); - if (!$r) + if (!dbm::is_result($r)) return false; $used = count($r); @@ -210,9 +347,9 @@ function poller_max_connections_reached() { * */ function poller_kill_stale_workers() { - $r = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); - if (!is_array($r) || count($r) == 0) { + if (!dbm::is_result($r)) { // No processing here needed return; } @@ -223,35 +360,53 @@ function poller_kill_stale_workers() { intval($pid["pid"])); else { // Kill long running processes + + // Check if the priority is in a valid range + if (!in_array($pid["priority"], array(PRIORITY_CRITICAL, PRIORITY_HIGH, PRIORITY_MEDIUM, PRIORITY_LOW, PRIORITY_NEGLIGIBLE))) + $pid["priority"] = PRIORITY_MEDIUM; + + // Define the maximum durations + $max_duration_defaults = array(PRIORITY_CRITICAL => 360, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 360); + $max_duration = $max_duration_defaults[$pid["priority"]]; + + $argv = json_decode($pid["parameter"]); + $argv[0] = basename($argv[0]); + + // How long is the process already running? $duration = (time() - strtotime($pid["executed"])) / 60; - if ($duration > 180) { - logger("Worker process ".$pid["pid"]." took more than 3 hours. It will be killed now."); + if ($duration > $max_duration) { + logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") took more than ".$max_duration." minutes. It will be killed now."); posix_kill($pid["pid"], SIGTERM); - // Question: If a process is stale: Should we remove it or should we reschedule it? - // By now we rescheduling it. It's maybe not the wisest decision? - q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `pid` = 0 WHERE `pid` = %d", + // We killed the stale process. + // To avoid a blocking situation we reschedule the process at the beginning of the queue. + // Additionally we are lowering the priority. + q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `created` = '%s', + `priority` = %d, `pid` = 0 WHERE `pid` = %d", + dbesc(datetime_convert()), + intval(PRIORITY_NEGLIGIBLE), intval($pid["pid"])); } else - logger("Worker process ".$pid["pid"]." now runs for ".round($duration)." minutes. That's okay.", LOGGER_DEBUG); + logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") now runs for ".round($duration)." of ".$max_duration." allowed minutes. That's okay.", LOGGER_DEBUG); } } -function poller_too_much_workers($stage) { - - $queues = get_config("system", "worker_queues"); +/** + * @brief Checks if the number of active workers exceeds the given limits + * + * @return bool Are there too much workers running? + */ +function poller_too_much_workers() { + $queues = Config::get("system", "worker_queues", 4); - if ($queues == 0) - $queues = 4; + $maxqueues = $queues; $active = poller_active_workers(); // Decrease the number of workers at higher load $load = current_load(); if($load) { - $maxsysload = intval(get_config('system','maxloadavg')); - if($maxsysload < 1) - $maxsysload = 50; + $maxsysload = intval(Config::get("system", "maxloadavg", 50)); $maxworkers = $queues; @@ -260,21 +415,250 @@ function poller_too_much_workers($stage) { $slope = $maxworkers / pow($maxsysload, $exponent); $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent)); - logger("Current load stage ".$stage.": ".$load." - maximum: ".$maxsysload." - current queues: ".$active." - maximum: ".$queues, LOGGER_DEBUG); + $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00'"); + $entries = $s[0]["total"]; + + if (Config::get("system", "worker_fastlane", false) AND ($queues > 0) AND ($entries > 0) AND ($active >= $queues)) { + $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority` LIMIT 1"); + $top_priority = $s[0]["priority"]; + $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` != '0000-00-00 00:00:00' LIMIT 1", + intval($top_priority)); + $high_running = dbm::is_result($s); + + if (!$high_running AND ($top_priority > PRIORITY_UNDEFINED) AND ($top_priority < PRIORITY_NEGLIGIBLE)) { + logger("There are jobs with priority ".$top_priority." waiting but none is executed. Open a fastlane.", LOGGER_DEBUG); + $queues = $active + 1; + } + } + + // Create a list of queue entries grouped by their priority + $running = array(PRIORITY_CRITICAL => 0, + PRIORITY_HIGH => 0, + PRIORITY_MEDIUM => 0, + PRIORITY_LOW => 0, + PRIORITY_NEGLIGIBLE => 0); + + $r = q("SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` GROUP BY `priority`"); + if (dbm::is_result($r)) + foreach ($r AS $process) + $running[$process["priority"]] = $process["running"]; + + $processlist = ""; + $r = q("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`"); + if (dbm::is_result($r)) + foreach ($r as $entry) { + if ($processlist != "") + $processlist .= ", "; + $processlist .= $entry["priority"].":".$running[$entry["priority"]]."/".$entry["entries"]; + } + + logger("Load: ".$load."/".$maxsysload." - processes: ".$active."/".$entries." (".$processlist.") - maximum: ".$queues."/".$maxqueues, LOGGER_DEBUG); + + // Are there fewer workers running as possible? Then fork a new one. + if (!Config::get("system", "worker_dont_fork") AND ($queues > ($active + 1)) AND ($entries > 1)) { + logger("Active workers: ".$active."/".$queues." Fork a new worker.", LOGGER_DEBUG); + $args = array("include/poller.php", "no_cron"); + $a = get_app(); + $a->proc_run($args); + } } return($active >= $queues); } +/** + * @brief Returns the number of active poller processes + * + * @return integer Number of active poller processes + */ function poller_active_workers() { - $workers = q("SELECT COUNT(*) AS `workers` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php'"); + + return($workers[0]["processes"]); +} + +/** + * @brief Check if we should pass some slow processes + * + * When the active processes of the highest priority are using more than 2/3 + * of all processes, we let pass slower processes. + * + * @param string $highest_priority Returns the currently highest priority + * @return bool We let pass a slower process than $highest_priority + */ +function poller_passing_slow(&$highest_priority) { + + $highest_priority = 0; + + $r = q("SELECT `priority` + FROM `process` + INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`"); + + // No active processes at all? Fine + if (!dbm::is_result($r)) + return(false); + + $priorities = array(); + foreach ($r AS $line) + $priorities[] = $line["priority"]; + + // Should not happen + if (count($priorities) == 0) + return(false); + + $highest_priority = min($priorities); + + // The highest process is already the slowest one? + // Then we quit + if ($highest_priority == PRIORITY_NEGLIGIBLE) + return(false); + + $high = 0; + foreach ($priorities AS $priority) + if ($priority == $highest_priority) + ++$high; + + logger("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, LOGGER_DEBUG); + $passing_slow = (($high/count($priorities)) > (2/3)); + + if ($passing_slow) + logger("Passing slower processes than priority ".$highest_priority, LOGGER_DEBUG); + + return($passing_slow); +} + +/** + * @brief Returns the next worker process + * + * @return string SQL statement + */ +function poller_worker_process() { + + q("START TRANSACTION;"); + + // Check if we should pass some low priority process + $highest_priority = 0; + + if (poller_passing_slow($highest_priority)) { + // Are there waiting processes with a higher priority than the currently highest? + $r = q("SELECT * FROM `workerqueue` + WHERE `executed` = '0000-00-00 00:00:00' AND `priority` < %d + ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority)); + if (dbm::is_result($r)) + return $r; + + // Give slower processes some processing time + $r = q("SELECT * FROM `workerqueue` + WHERE `executed` = '0000-00-00 00:00:00' AND `priority` > %d + ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority)); + } + + // If there is no result (or we shouldn't pass lower processes) we check without priority limit + if (($highest_priority == 0) OR !dbm::is_result($r)) + $r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority`, `created` LIMIT 1"); + + return $r; +} + +/** + * @brief Call the front end worker + */ +function call_worker() { + if (!Config::get("system", "frontend_worker")) { + return; + } - return($workers[0]["workers"]); + $url = App::get_baseurl()."/worker"; + fetch_url($url, false, $redirects, 1); +} + +/** + * @brief Call the front end worker if there aren't any active + */ +function call_worker_if_idle() { + if (!Config::get("system", "frontend_worker")) { + return; + } + + // Do we have "proc_open"? Then we can fork the poller + if (function_exists("proc_open")) { + // When was the last time that we called the worker? + // Less than one minute? Then we quit + if ((time() - Config::get("system", "worker_started")) < 60) { + return; + } + + set_config("system", "worker_started", time()); + + // Do we have enough running workers? Then we quit here. + if (poller_too_much_workers()) { + // Cleaning dead processes + poller_kill_stale_workers(); + get_app()->remove_inactive_processes(); + + return; + } + + poller_run_cron(); + + logger('Call poller', LOGGER_DEBUG); + + $args = array("include/poller.php", "no_cron"); + $a = get_app(); + $a->proc_run($args); + return; + } + + // We cannot execute background processes. + // We now run the processes from the frontend. + // This won't work with long running processes. + poller_run_cron(); + + clear_worker_processes(); + + $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php'"); + + if ($workers[0]["processes"] == 0) { + call_worker(); + } +} + +/** + * @brief Removes long running worker processes + */ +function clear_worker_processes() { + $timeout = Config::get("system", "frontend_worker_timeout", 10); + + /// @todo We should clean up the corresponding workerqueue entries as well + q("DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php'", + dbesc(datetime_convert('UTC','UTC',"now - ".$timeout." minutes"))); +} + +/** + * @brief Runs the cron processes + */ +function poller_run_cron() { + logger('Add cron entries', LOGGER_DEBUG); + + // Check for spooled items + proc_run(PRIORITY_HIGH, "include/spool_post.php"); + + // Run the cron job that calls all other jobs + proc_run(PRIORITY_MEDIUM, "include/cron.php"); + + // Run the cronhooks job separately from cron for being able to use a different timing + proc_run(PRIORITY_MEDIUM, "include/cronhooks.php"); + + // Cleaning dead processes + poller_kill_stale_workers(); } if (array_search(__file__,get_included_files())===0){ - poller_run($_SERVER["argv"],$_SERVER["argc"]); - killme(); + poller_run($_SERVER["argv"],$_SERVER["argc"]); + + get_app()->end_process(); + + killme(); } ?>