X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=include%2Fpoller.php;h=0c5bebbda57f911de9e82da7ff326cd28dc372b2;hb=81c87885cf61247608a0fa8e738707510e737ace;hp=82befae51842dc4bd1dd49c2f5f3dabeaa144216;hpb=8e12b159d947fb27ad6bdd5e06e3a90c1713687e;p=friendica.git diff --git a/include/poller.php b/include/poller.php index 82befae518..0c5bebbda5 100644 --- a/include/poller.php +++ b/include/poller.php @@ -1,4 +1,8 @@ set_baseurl(Config::get('system', 'url')); + + load_hooks(); + $a->start_process(); + if ($a->min_memory_reached()) { + return; + } + if (poller_max_connections_reached()) { return; } @@ -63,6 +73,15 @@ function poller_run($argv, $argc){ while ($r = poller_worker_process()) { + if (!poller_claim_process($r[0])) { + continue; + } + + // Check free memory + if ($a->min_memory_reached()) { + return; + } + // Count active workers and compare them with a maximum value that depends on the load if (poller_too_much_workers()) { return; @@ -76,7 +95,6 @@ function poller_run($argv, $argc){ if (time() > ($starttime + 3600)) return; } - } /** @@ -94,47 +112,22 @@ function poller_execute($queue) { // Quit when in maintenance if (Config::get('system', 'maintenance', true)) { + logger("Maintenance mode - quit process ".$mypid, LOGGER_DEBUG); return false; } // Constantly check the number of parallel database processes if ($a->max_processes_reached()) { + logger("Max processes reached for process ".$mypid, LOGGER_DEBUG); return false; } // Constantly check the number of available database connections to let the frontend be accessible at any time if (poller_max_connections_reached()) { + logger("Max connection reached for process ".$mypid, LOGGER_DEBUG); return false; } - $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0", - dbesc(datetime_convert()), - intval($mypid), - intval($queue["id"])); - - if (!$upd) { - logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG); - q("COMMIT"); - return true; - } - - // 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"); - $argv = json_decode($queue["parameter"]); // Check for existance and validity of the include file @@ -142,7 +135,7 @@ function poller_execute($queue) { if (!validate_include($include)) { logger("Include file ".$argv[0]." is not valid!"); - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + dba::delete('workerqueue', array('id' => $queue["id"])); return true; } @@ -154,7 +147,7 @@ function poller_execute($queue) { poller_exec_function($queue, $funcname, $argv); - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + dba::delete('workerqueue', array('id' => $queue["id"])); } else { logger("Function ".$funcname." does not exist"); } @@ -353,18 +346,18 @@ function poller_max_connections_reached() { * */ function poller_kill_stale_workers() { - $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` > '%s'", dbesc(NULL_DATE)); if (!dbm::is_result($r)) { // No processing here needed return; } - foreach($r AS $pid) - if (!posix_kill($pid["pid"], 0)) - q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `pid` = 0 WHERE `pid` = %d", - intval($pid["pid"])); - else { + foreach ($r AS $pid) { + if (!posix_kill($pid["pid"], 0)) { + dba::update('workerqueue', array('executed' => NULL_DATE, 'pid' => 0), + array('pid' => $pid["pid"])); + } else { // Kill long running processes // Check if the priority is in a valid range @@ -387,14 +380,14 @@ function poller_kill_stale_workers() { // 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 + dba::update('workerqueue', + array('executed' => NULL_DATE, 'created' => datetime_convert(), 'priority' => PRIORITY_NEGLIGIBLE, 'pid' => 0), + array('pid' => $pid["pid"])); + } else { logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") now runs for ".round($duration)." of ".$max_duration." allowed minutes. That's okay.", LOGGER_DEBUG); + } } + } } /** @@ -407,11 +400,9 @@ function poller_too_much_workers() { $maxqueues = $queues; - $active = poller_active_workers(); - // Decrease the number of workers at higher load $load = current_load(); - if($load) { + if ($load) { $maxsysload = intval(Config::get("system", "maxloadavg", 50)); $maxworkers = $queues; @@ -421,15 +412,42 @@ function poller_too_much_workers() { $slope = $maxworkers / pow($maxsysload, $exponent); $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent)); - $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00'"); + $active = 0; + + // Create a list of queue entries grouped by their priority + $listitem = array(); + + // Adding all processes with no workerqueue entry + $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` WHERE NOT EXISTS (SELECT id FROM `workerqueue` WHERE `workerqueue`.`pid` = `process`.`pid`)"); + if ($process = dba::fetch($processes)) { + $listitem[0] = "0:".$process["running"]; + $active += $process["running"]; + } + dba::close($processes); + + // Now adding all processes with workerqueue entries + $entries = dba::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`"); + while ($entry = dba::fetch($entries)) { + $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` LEFT JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` WHERE `priority` = ?", $entry["priority"]); + if ($process = dba::fetch($processes)) { + $listitem[$entry["priority"]] = $entry["priority"].":".$process["running"]."/".$entry["entries"]; + $active += $process["running"]; + } + dba::close($processes); + } + dba::close($entries); + + $processlist = implode(', ', $listitem); + + $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= '%s'", dbesc(NULL_DATE)); $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"); + $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority` LIMIT 1", dbesc(NULL_DATE)); $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)); + $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` > '%s' LIMIT 1", + intval($top_priority), dbesc(NULL_DATE)); $high_running = dbm::is_result($s); if (!$high_running AND ($top_priority > PRIORITY_UNDEFINED) AND ($top_priority < PRIORITY_NEGLIGIBLE)) { @@ -438,36 +456,17 @@ function poller_too_much_workers() { } } - // 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("php", "include/poller.php", "no_cron"); + $args = array("include/poller.php", "no_cron"); $a = get_app(); $a->proc_run($args); } + } else { + $active = poller_active_workers(); } return($active >= $queues); @@ -541,7 +540,7 @@ function poller_passing_slow(&$highest_priority) { */ function poller_worker_process() { - q("START TRANSACTION;"); + dba::transaction(); // Check if we should pass some low priority process $highest_priority = 0; @@ -549,29 +548,85 @@ function poller_worker_process() { 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)) + WHERE `executed` <= '%s' AND `priority` < %d + ORDER BY `priority`, `created` LIMIT 1 FOR UPDATE", + dbesc(NULL_DATE), + intval($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)); + WHERE `executed` <= '%s' AND `priority` > %d + ORDER BY `priority`, `created` LIMIT 1 FOR UPDATE", + dbesc(NULL_DATE), + intval($highest_priority)); + + if (dbm::is_result($r)) { + return $r; + } } // 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"); + if (!dbm::is_result($r)) { + $r = q("SELECT * FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority`, `created` LIMIT 1 FOR UPDATE", dbesc(NULL_DATE)); + } return $r; } +/** + * @brief Assigns a workerqueue entry to the current process + * + * All the checks after the update are only needed with MyISAM. + * + * @param array $queue Workerqueue entry + * + * @return boolean "true" if the claiming was successful + */ +function poller_claim_process($queue) { + $mypid = getmypid(); + + if (!dba::update('workerqueue', array('executed' => datetime_convert(), 'pid' => $mypid), + array('id' => $queue["id"], 'pid' => 0))) { + logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG); + dba::commit(); + return false; + } + + // 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); + dba::commit(); + return false; + } 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); + dba::commit(); + return false; + } 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); + dba::commit(); + return false; + } + dba::commit(); + return true; +} + +/** + * @brief Removes a workerqueue entry from the current process + */ +function poller_unclaim_process() { + $mypid = getmypid(); + + dba::update('workerqueue', array('executed' => NULL_DATE, 'pid' => 0), array('pid' => $mypid)); +} + /** * @brief Call the front end worker */ function call_worker() { - if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) { + if (!Config::get("system", "frontend_worker")) { return; } @@ -583,7 +638,7 @@ function call_worker() { * @brief Call the front end worker if there aren't any active */ function call_worker_if_idle() { - if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) { + if (!Config::get("system", "frontend_worker")) { return; } @@ -610,7 +665,7 @@ function call_worker_if_idle() { logger('Call poller', LOGGER_DEBUG); - $args = array("php", "include/poller.php", "no_cron"); + $args = array("include/poller.php", "no_cron"); $a = get_app(); $a->proc_run($args); return; @@ -663,8 +718,9 @@ function poller_run_cron() { if (array_search(__file__,get_included_files())===0){ poller_run($_SERVER["argv"],$_SERVER["argc"]); + poller_unclaim_process(); + get_app()->end_process(); killme(); } -?>