+}
+
+/**
+ * @brief Execute a worker entry
+ *
+ * @param array $queue Workerqueue entry
+ *
+ * @return boolean "true" if further processing should be stopped
+ */
+function poller_execute($queue) {
+
+ $a = get_app();
+
+ $mypid = getmypid();
+
+ // Quit when in maintenance
+ if (Config::get('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;
+ }
+
+ // 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
+ $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);
+
+ if ($duration > 3600) {
+ logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 1 hour (".round($duration/60, 3).")", LOGGER_DEBUG);
+ } elseif ($duration > 600) {
+ logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 10 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
+ } elseif ($duration > 300) {
+ logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 5 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
+ } elseif ($duration > 120) {
+ logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 2 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
+ }
+
+ 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);
+ }
+}
+
+/**
+ * @brief Checks if the number of database connections has reached a critical limit.
+ *
+ * @return bool Are more than 3/4 of the maximum connections used?
+ */
+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 = Config::get("system", "max_connections");
+
+ // Fetch the percentage level where the poller will get active
+ $maxlevel = Config::get("system", "max_connections_level", 75);
+
+ if ($max == 0) {
+ // the maximum number of possible user connections can be a system variable
+ $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
+ if ($r)
+ $max = $r[0]["Value"];
+
+ // Or it can be granted. This overrides the system variable
+ $r = q("SHOW GRANTS");
+ if ($r)
+ foreach ($r AS $grants) {
+ $grant = array_pop($grants);
+ if (stristr($grant, "GRANT USAGE ON"))
+ if (preg_match("/WITH MAX_USER_CONNECTIONS (\d*)/", $grant, $match))
+ $max = $match[1];
+ }
+ }