From e856ab9a09c74e621b2df5a7d325d71575f06b5e Mon Sep 17 00:00:00 2001 From: Michael Date: Sun, 27 Nov 2016 00:55:05 +0000 Subject: [PATCH] The worker can now run from the frontend as well --- boot.php | 4 + doc/htconfig.md | 1 + include/poller.php | 213 ++++++++++++++++++++++++++++----------------- index.php | 4 + mod/worker.php | 38 ++++++++ 5 files changed, 181 insertions(+), 79 deletions(-) create mode 100644 mod/worker.php diff --git a/boot.php b/boot.php index b44c063f8a..866e7165c7 100644 --- a/boot.php +++ b/boot.php @@ -1385,6 +1385,10 @@ class App { function proc_run($args) { + if (!function_exists("proc_open")) { + return; + } + // Add the php path if it is a php call if (count($args) && ($args[0] === 'php' OR !is_string($args[0]))) { diff --git a/doc/htconfig.md b/doc/htconfig.md index b6f98b2ef0..8e0b890926 100644 --- a/doc/htconfig.md +++ b/doc/htconfig.md @@ -29,6 +29,7 @@ Example: To set the directory value please add this line to your .htconfig.php: * disable_email_validation (Boolean) - Disables the check if a mail address is in a valid format and can be resolved via DNS. * disable_url_validation (Boolean) - Disables the DNS lookup of an URL. * event_input_format - Default value is "ymd". +* frontend_worker (Boolean) - Activates the frontend worker which acts as a replacement for running the poller via the command line. * ignore_cache (Boolean) - For development only. Disables the item cache. * like_no_comment (Boolean) - Don't update the "commented" value of an item when it is liked. * local_block (Boolean) - Used in conjunction with "block_public". diff --git a/include/poller.php b/include/poller.php index 2e036247e2..b4f5df4bf5 100644 --- a/include/poller.php +++ b/include/poller.php @@ -15,7 +15,7 @@ use \Friendica\Core\PConfig; require_once("boot.php"); -function poller_run(&$argv, &$argc){ +function poller_run($argv, $argc){ global $a, $db; if(is_null($a)) { @@ -35,8 +35,6 @@ function poller_run(&$argv, &$argc){ $a->start_process(); - $mypid = getmypid(); - if ($a->max_processes_reached()) return; @@ -53,14 +51,7 @@ function poller_run(&$argv, &$argc){ } if(($argc <= 1) OR ($argv[1] != "no_cron")) { - // 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(); + poller_run_cron(); } else // Sleep four seconds before checking for running processes again to avoid having too many workers sleep(4); @@ -69,101 +60,126 @@ function poller_run(&$argv, &$argc){ if (poller_too_much_workers()) return; - $cooldown = Config::get("system", "worker_cooldown", 0); - $starttime = time(); while ($r = poller_worker_process()) { - // Quit when in maintenance - if (get_config('system', 'maintenance', true)) + // 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 parallel database processes - if ($a->max_processes_reached()) + if (!poller_execute($r[0])) { return; + } - // Constantly check the number of available database connections to let the frontend be accessible at any time - if (poller_max_connections_reached()) + // Quit the poller once every hour + if (time() > ($starttime + 3600)) return; + } - // Count active workers and compare them with a maximum value that depends on the load - if (poller_too_much_workers()) - return; +} + +/** + * @brief Execute a worker entry + * + * @param array $queue Workerqueue entry + * + * @return boolean "true" if further processing should be stopped + */ +function poller_execute($queue) { - $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0", - dbesc(datetime_convert()), - intval($mypid), - intval($r[0]["id"])); + $a = get_app(); - if (!$upd) { - logger("Couldn't update queue entry ".$r[0]["id"]." - skip this execution", LOGGER_DEBUG); - q("COMMIT"); - continue; - } + $mypid = getmypid(); - // 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 - 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; - } + $cooldown = Config::get("system", "worker_cooldown", 0); + + // Quit when in maintenance + if (get_config('system', 'maintenance', true)) { + return false; + } + + // Constantly check the number of parallel database processes + if ($a->max_processes_reached()) { + return false; + } + + // Constantly check the number of available database connections to let the frontend be accessible at any time + if (poller_max_connections_reached()) { + 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; + } - $argv = json_decode($r[0]["parameter"]); + // 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"); - $argc = count($argv); + $argv = json_decode($queue["parameter"]); - // Check for existance and validity of the include file - $include = $argv[0]; + $argc = count($argv); - if (!validate_include($include)) { - logger("Include file ".$argv[0]." is not valid!"); - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); - continue; - } + // Check for existance and validity of the include file + $include = $argv[0]; - require_once($include); + if (!validate_include($include)) { + logger("Include file ".$argv[0]." is not valid!"); + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + return true; + } - $funcname = str_replace(".php", "", basename($argv[0]))."_run"; + require_once($include); - if (function_exists($funcname)) { - logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." ".$r[0]["parameter"]); + $funcname = str_replace(".php", "", basename($argv[0]))."_run"; - // 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); + if (function_exists($funcname)) { + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]); - $funcname($argv, $argc); + // 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); - $a->process_id = $old_process_id; + $funcname($argv, $argc); - if ($cooldown > 0) { - logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); - sleep($cooldown); - } + $a->process_id = $old_process_id; - logger("Process ".$mypid." - Prio ".$r[0]["priority"]." - ID ".$r[0]["id"].": ".$funcname." - done"); + if ($cooldown > 0) { + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds"); + sleep($cooldown); + } - q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); - } else - logger("Function ".$funcname." does not exist"); + logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done"); - // Quit the poller once every hour - if (time() > ($starttime + 3600)) - return; + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"])); + } else { + logger("Function ".$funcname." does not exist"); } + return true; } /** @@ -394,8 +410,7 @@ function poller_passing_slow(&$highest_priority) { $r = q("SELECT `priority` FROM `process` - INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` - WHERE `process`.`command` = 'poller.php'"); + INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`"); // No active processes at all? Fine if (!dbm::is_result($r)) @@ -435,7 +450,6 @@ function poller_passing_slow(&$highest_priority) { * * @return string SQL statement */ - function poller_worker_process() { q("START TRANSACTION;"); @@ -464,6 +478,47 @@ function poller_worker_process() { return $r; } +function call_worker() { + if (!get_config("system", "frontend_worker")) { + return; + } + + $url = get_app()->get_baseurl()."/worker"; + fetch_url($url, false, $redirects, 1); +} + +function call_worker_if_idle() { + if (!get_config("system", "frontend_worker")) { + return; + } + + 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(); + } +} + +function clear_worker_processes() { + q("DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php'", + dbesc(datetime_convert('UTC','UTC',"now - 10 minutes"))); +} + +function poller_run_cron() { + // 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"]); diff --git a/index.php b/index.php index 13c3f6318f..08f24af60f 100644 --- a/index.php +++ b/index.php @@ -99,6 +99,10 @@ if (!$a->is_backend()) { $stamp1 = microtime(true); session_start(); $a->save_timestamp($stamp1, "parser"); +} else { + require_once "include/poller.php"; + + call_worker_if_idle(); } /** diff --git a/mod/worker.php b/mod/worker.php new file mode 100644 index 0000000000..1d1ab2af33 --- /dev/null +++ b/mod/worker.php @@ -0,0 +1,38 @@ + Config::get("system", "worker_queues", 4)) { + return; + } + + $a->start_process(); + + logger("Front end worker started: ".getmypid()); + + call_worker(); + + if ($r = poller_worker_process()) { + poller_execute($r[0]); + } + + call_worker(); + + $a->end_process(); + + logger("Front end worker ended: ".getmypid()); + + killme(); +} -- 2.39.5