X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=include%2Fpoller.php;h=2e036247e237cf06c40d5f514300b2c57e1a1525;hb=05f71f4a701ece84cb435e7e9a37cc52ae7ce0d6;hp=fe4d4245af8df0a3a3d6fc4c6f94accbe2c5e7b5;hpb=b3ac6189b35c9007e763cd9860841a78a614ff85;p=friendica.git diff --git a/include/poller.php b/include/poller.php index fe4d4245af..2e036247e2 100644 --- a/include/poller.php +++ b/include/poller.php @@ -73,7 +73,7 @@ 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)) @@ -91,23 +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($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 `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); if (!$id) { 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 - we better stop here", LOGGER_DEBUG); - return; + 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"]); @@ -128,8 +138,16 @@ function poller_run(&$argv, &$argc){ if (function_exists($funcname)) { 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 ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); sleep($cooldown); @@ -320,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)) { @@ -340,6 +379,91 @@ function poller_active_workers() { 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` + 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"]);