X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=include%2Fpoller.php;h=2e036247e237cf06c40d5f514300b2c57e1a1525;hb=05f71f4a701ece84cb435e7e9a37cc52ae7ce0d6;hp=0835528717cbe81ab56b26d599e410a6f5a4756b;hpb=7092bfed0cd71d95d41f6d25076d8dbc2a9947ff;p=friendica.git diff --git a/include/poller.php b/include/poller.php index 0835528717..2e036247e2 100644 --- a/include/poller.php +++ b/include/poller.php @@ -29,6 +29,14 @@ function poller_run(&$argv, &$argc){ unset($db_host, $db_user, $db_pass, $db_data); }; + // Quit when in maintenance + if (get_config('system', 'maintenance', true)) + return; + + $a->start_process(); + + $mypid = getmypid(); + if ($a->max_processes_reached()) return; @@ -65,7 +73,11 @@ function poller_run(&$argv, &$argc){ $starttime = time(); - while ($r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority`, `created` LIMIT 1")) { + while ($r = poller_worker_process()) { + + // Quit when in maintenance + if (get_config('system', 'maintenance', true)) + return; // Constantly check the number of parallel database processes if ($a->max_processes_reached()) @@ -79,19 +91,33 @@ function poller_run(&$argv, &$argc){ if (poller_too_much_workers()) return; - q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `executed` = '0000-00-00 00:00:00'", + $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0", dbesc(datetime_convert()), - intval(getmypid()), + intval($mypid), intval($r[0]["id"])); + if (!$upd) { + logger("Couldn't update queue entry ".$r[0]["id"]." - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + continue; + } + // 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())); + $id = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); if (!$id) { - logger("Queue item ".$r[0]["id"]." was executed multiple times - skip this execution", LOGGER_DEBUG); + logger("Queue item ".$r[0]["id"]." vanished - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + continue; + } elseif ((strtotime($id[0]["executed"]) <= 0) OR ($id[0]["pid"] == 0)) { + logger("Entry for queue item ".$r[0]["id"]." wasn't stored - skip this execution", LOGGER_DEBUG); + q("COMMIT"); + continue; + } elseif ($id[0]["pid"] != $mypid) { + logger("Queue item ".$r[0]["id"]." is to be executed by process ".$id[0]["pid"]." and not by me (".$mypid.") - skip this execution", LOGGER_DEBUG); + q("COMMIT"); continue; } + q("COMMIT"); $argv = json_decode($r[0]["parameter"]); @@ -111,15 +137,23 @@ function poller_run(&$argv, &$argc){ $funcname = str_replace(".php", "", basename($argv[0]))."_run"; if (function_exists($funcname)) { - logger("Process ".getmypid()." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." ".$r[0]["parameter"]); + logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." ".$r[0]["parameter"]); + + // 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; + if ($cooldown > 0) { - logger("Process ".getmypid()." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); + logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); sleep($cooldown); } - logger("Process ".getmypid()." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - done"); + logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - done"); q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); } else @@ -304,7 +338,28 @@ function poller_too_much_workers() { } } - logger("Current load: ".$load." - maximum: ".$maxsysload." - current queues: ".$active."/".$entries." - maximum: ".$queues."/".$maxqueues, LOGGER_DEBUG); + // 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 (!get_config("system", "worker_dont_fork") AND ($queues > ($active + 1)) AND ($entries > 1)) { @@ -319,13 +374,101 @@ function poller_too_much_workers() { } 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; - return($workers[0]["workers"]); + $r = q("SELECT `priority` + FROM `process` + INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` + WHERE `process`.`command` = 'poller.php'"); + + // 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; } 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(); } ?>