include/poller.php

   1 <?php
   2
   3 use Friendica\App;
   4 use Friendica\Core\Config;
   5 use Friendica\Util\Lock;
   6
   7 if (!file_exists("boot.php") && (sizeof($_SERVER["argv"]) != 0)) {
   8         $directory = dirname($_SERVER["argv"][0]);
   9
  10         if (substr($directory, 0, 1) != "/") {
  11                 $directory = $_SERVER["PWD"]."/".$directory;
  12         }
  13         $directory = realpath($directory."/..");
  14
  15         chdir($directory);
  16 }
  17
  18 require_once("boot.php");
  19
  20 function poller_run($argv, $argc){
  21         global $a, $db;
  22
  23         $a = new App(dirname(__DIR__));
  24
  25         @include(".htconfig.php");
  26         require_once("include/dba.php");
  27         $db = new dba($db_host, $db_user, $db_pass, $db_data);
  28         unset($db_host, $db_user, $db_pass, $db_data);
  29
  30         Config::load();
  31
  32         // Quit when in maintenance
  33         if (Config::get('system', 'maintenance', true)) {
  34                 return;
  35         }
  36
  37         $a->set_baseurl(Config::get('system', 'url'));
  38
  39         load_hooks();
  40
  41         // At first check the maximum load. We shouldn't continue with a high load
  42         if ($a->maxload_reached()) {
  43                 logger('Pre check: maximum load reached, quitting.', LOGGER_DEBUG);
  44                 return;
  45         }
  46
  47         // We now start the process. This is done after the load check since this could increase the load.
  48         $a->start_process();
  49
  50         // At first we check the number of workers and quit if there are too much of them
  51         // This is done at the top to avoid that too much code is executed without a need to do so,
  52         // since the poller mostly quits here.
  53         if (poller_too_much_workers()) {
  54                 poller_kill_stale_workers();
  55                 logger('Pre check: Active worker limit reached, quitting.', LOGGER_DEBUG);
  56                 return;
  57         }
  58
  59         // Do we have too few memory?
  60         if ($a->min_memory_reached()) {
  61                 logger('Pre check: Memory limit reached, quitting.', LOGGER_DEBUG);
  62                 return;
  63         }
  64
  65         // Possibly there are too much database connections
  66         if (poller_max_connections_reached()) {
  67                 logger('Pre check: maximum connections reached, quitting.', LOGGER_DEBUG);
  68                 return;
  69         }
  70
  71         // Possibly there are too much database processes that block the system
  72         if ($a->max_processes_reached()) {
  73                 logger('Pre check: maximum processes reached, quitting.', LOGGER_DEBUG);
  74                 return;
  75         }
  76
  77         // Now we start additional cron processes if we should do so
  78         if (($argc <= 1) || ($argv[1] != "no_cron")) {
  79                 poller_run_cron();
  80         }
  81
  82         $starttime = time();
  83
  84         // We fetch the next queue entry that is about to be executed
  85         while ($r = poller_worker_process()) {
  86
  87                 // If we got that queue entry we claim it for us
  88                 if (!poller_claim_process($r[0])) {
  89                         dba::unlock();
  90                         continue;
  91                 } else {
  92                         // Fetch all workerqueue data while the table is still locked
  93                         // This is redundant, but this speeds up the processing
  94                         $entries = poller_total_entries();
  95                         $top_priority = poller_highest_priority();
  96                         $high_running = poller_process_with_priority_active($top_priority);
  97                         dba::unlock();
  98                 }
  99
 100                 // To avoid the quitting of multiple pollers only one poller at a time will execute the check
 101                 if (Lock::set('poller_worker', 0)) {
 102                         // Count active workers and compare them with a maximum value that depends on the load
 103                         if (poller_too_much_workers($entries, $top_priority, $high_running)) {
 104                                 logger('Active worker limit reached, quitting.', LOGGER_DEBUG);
 105                                 return;
 106                         }
 107
 108                         // Check free memory
 109                         if ($a->min_memory_reached()) {
 110                                 logger('Memory limit reached, quitting.', LOGGER_DEBUG);
 111                                 return;
 112                         }
 113                         Lock::remove('poller_worker');
 114                 }
 115
 116                 // finally the work will be done
 117                 if (!poller_execute($r[0])) {
 118                         logger('Process execution failed, quitting.', LOGGER_DEBUG);
 119                         return;
 120                 }
 121
 122                 // Quit the poller once every hour
 123                 if (time() > ($starttime + 3600)) {
 124                         logger('Process lifetime reached, quitting.', LOGGER_DEBUG);
 125                         return;
 126                 }
 127         }
 128         logger("Couldn't select a workerqueue entry, quitting.", LOGGER_DEBUG);
 129 }
 130
 131 /**
 132  * @brief Returns the number of non executed entries in the worker queue
 133  *
 134  * @return integer Number of non executed entries in the worker queue
 135  */
 136 function poller_total_entries() {
 137         $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= '%s'", dbesc(NULL_DATE));
 138         return $s[0]["total"];
 139 }
 140
 141 /**
 142  * @brief Returns the highest priority in the worker queue that isn't executed
 143  *
 144  * @return integer Number of active poller processes
 145  */
 146 function poller_highest_priority() {
 147         $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority` LIMIT 1", dbesc(NULL_DATE));
 148         return $s[0]["priority"];
 149 }
 150
 151 /**
 152  * @brief Returns if a process with the given priority is running
 153  *
 154  * @param integer $priority The priority that should be checked
 155  *
 156  * @return integer Is there a process running with that priority?
 157  */
 158 function poller_process_with_priority_active($priority) {
 159         $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` > '%s' LIMIT 1",
 160                         intval($priority), dbesc(NULL_DATE));
 161         return dbm::is_result($s);
 162 }
 163
 164 /**
 165  * @brief Execute a worker entry
 166  *
 167  * @param array $queue Workerqueue entry
 168  *
 169  * @return boolean "true" if further processing should be stopped
 170  */
 171 function poller_execute($queue) {
 172
 173         $a = get_app();
 174
 175         $mypid = getmypid();
 176
 177         // Quit when in maintenance
 178         if (Config::get('system', 'maintenance', true)) {
 179                 logger("Maintenance mode - quit process ".$mypid, LOGGER_DEBUG);
 180                 return false;
 181         }
 182
 183         // Constantly check the number of parallel database processes
 184         if ($a->max_processes_reached()) {
 185                 logger("Max processes reached for process ".$mypid, LOGGER_DEBUG);
 186                 return false;
 187         }
 188
 189         // Constantly check the number of available database connections to let the frontend be accessible at any time
 190         if (poller_max_connections_reached()) {
 191                 logger("Max connection reached for process ".$mypid, LOGGER_DEBUG);
 192                 return false;
 193         }
 194
 195         $argv = json_decode($queue["parameter"]);
 196
 197         // Check for existance and validity of the include file
 198         $include = $argv[0];
 199
 200         if (!validate_include($include)) {
 201                 logger("Include file ".$argv[0]." is not valid!");
 202                 dba::delete('workerqueue', array('id' => $queue["id"]));
 203                 return true;
 204         }
 205
 206         require_once($include);
 207
 208         $funcname = str_replace(".php", "", basename($argv[0]))."_run";
 209
 210         if (function_exists($funcname)) {
 211                 poller_exec_function($queue, $funcname, $argv);
 212                 dba::delete('workerqueue', array('id' => $queue["id"]));
 213         } else {
 214                 logger("Function ".$funcname." does not exist");
 215         }
 216
 217         return true;
 218 }
 219
 220 /**
 221  * @brief Execute a function from the queue
 222  *
 223  * @param array $queue Workerqueue entry
 224  * @param string $funcname name of the function
 225  * @param array $argv Array of values to be passed to the function
 226  */
 227 function poller_exec_function($queue, $funcname, $argv) {
 228
 229         $a = get_app();
 230
 231         $mypid = getmypid();
 232
 233         $argc = count($argv);
 234
 235         logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]);
 236
 237         $stamp = (float)microtime(true);
 238
 239         // We use the callstack here to analyze the performance of executed worker entries.
 240         // For this reason the variables have to be initialized.
 241         if (Config::get("system", "profiler")) {
 242                 $a->performance["start"] = microtime(true);
 243                 $a->performance["database"] = 0;
 244                 $a->performance["database_write"] = 0;
 245                 $a->performance["network"] = 0;
 246                 $a->performance["file"] = 0;
 247                 $a->performance["rendering"] = 0;
 248                 $a->performance["parser"] = 0;
 249                 $a->performance["marktime"] = 0;
 250                 $a->performance["markstart"] = microtime(true);
 251                 $a->callstack = array();
 252         }
 253
 254         // For better logging create a new process id for every worker call
 255         // But preserve the old one for the worker
 256         $old_process_id = $a->process_id;
 257         $a->process_id = uniqid("wrk", true);
 258
 259         $funcname($argv, $argc);
 260
 261         $a->process_id = $old_process_id;
 262
 263         $duration = number_format(microtime(true) - $stamp, 3);
 264
 265         if ($duration > 3600) {
 266                 logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 1 hour (".round($duration/60, 3).")", LOGGER_DEBUG);
 267         } elseif ($duration > 600) {
 268                 logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 10 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
 269         } elseif ($duration > 300) {
 270                 logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 5 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
 271         } elseif ($duration > 120) {
 272                 logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 2 minutes (".round($duration/60, 3).")", LOGGER_DEBUG);
 273         }
 274
 275         logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done in ".$duration." seconds.");
 276
 277         // Write down the performance values into the log
 278         if (Config::get("system", "profiler")) {
 279                 $duration = microtime(true)-$a->performance["start"];
 280
 281                 if (Config::get("rendertime", "callstack")) {
 282                         if (isset($a->callstack["database"])) {
 283                                 $o = "\nDatabase Read:\n";
 284                                 foreach ($a->callstack["database"] AS $func => $time) {
 285                                         $time = round($time, 3);
 286                                         if ($time > 0) {
 287                                                 $o .= $func.": ".$time."\n";
 288                                         }
 289                                 }
 290                         }
 291                         if (isset($a->callstack["database_write"])) {
 292                                 $o .= "\nDatabase Write:\n";
 293                                 foreach ($a->callstack["database_write"] AS $func => $time) {
 294                                         $time = round($time, 3);
 295                                         if ($time > 0) {
 296                                                 $o .= $func.": ".$time."\n";
 297                                         }
 298                                 }
 299                         }
 300                         if (isset($a->callstack["network"])) {
 301                                 $o .= "\nNetwork:\n";
 302                                 foreach ($a->callstack["network"] AS $func => $time) {
 303                                         $time = round($time, 3);
 304                                         if ($time > 0) {
 305                                                 $o .= $func.": ".$time."\n";
 306                                         }
 307                                 }
 308                         }
 309                 } else {
 310                         $o = '';
 311                 }
 312
 313                 logger("ID ".$queue["id"].": ".$funcname.": ".sprintf("DB: %s/%s, Net: %s, I/O: %s, Other: %s, Total: %s".$o,
 314                         number_format($a->performance["database"] - $a->performance["database_write"], 2),
 315                         number_format($a->performance["database_write"], 2),
 316                         number_format($a->performance["network"], 2),
 317                         number_format($a->performance["file"], 2),
 318                         number_format($duration - ($a->performance["database"] + $a->performance["network"] + $a->performance["file"]), 2),
 319                         number_format($duration, 2)),
 320                         LOGGER_DEBUG);
 321         }
 322
 323         $cooldown = Config::get("system", "worker_cooldown", 0);
 324
 325         if ($cooldown > 0) {
 326                 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds");
 327                 sleep($cooldown);
 328         }
 329 }
 330
 331 /**
 332  * @brief Checks if the number of database connections has reached a critical limit.
 333  *
 334  * @return bool Are more than 3/4 of the maximum connections used?
 335  */
 336 function poller_max_connections_reached() {
 337
 338         // Fetch the max value from the config. This is needed when the system cannot detect the correct value by itself.
 339         $max = Config::get("system", "max_connections");
 340
 341         // Fetch the percentage level where the poller will get active
 342         $maxlevel = Config::get("system", "max_connections_level", 75);
 343
 344         if ($max == 0) {
 345                 // the maximum number of possible user connections can be a system variable
 346                 $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
 347                 if (dbm::is_result($r)) {
 348                         $max = $r[0]["Value"];
 349                 }
 350                 // Or it can be granted. This overrides the system variable
 351                 $r = q("SHOW GRANTS");
 352                 if (dbm::is_result($r)) {
 353                         foreach ($r AS $grants) {
 354                                 $grant = array_pop($grants);
 355                                 if (stristr($grant, "GRANT USAGE ON")) {
 356                                         if (preg_match("/WITH MAX_USER_CONNECTIONS (\d*)/", $grant, $match)) {
 357                                                 $max = $match[1];
 358                                         }
 359                                 }
 360                         }
 361                 }
 362         }
 363
 364         // If $max is set we will use the processlist to determine the current number of connections
 365         // The processlist only shows entries of the current user
 366         if ($max != 0) {
 367                 $r = q("SHOW PROCESSLIST");
 368                 if (!dbm::is_result($r)) {
 369                         return false;
 370                 }
 371                 $used = count($r);
 372
 373                 logger("Connection usage (user values): ".$used."/".$max, LOGGER_DEBUG);
 374
 375                 $level = ($used / $max) * 100;
 376
 377                 if ($level >= $maxlevel) {
 378                         logger("Maximum level (".$maxlevel."%) of user connections reached: ".$used."/".$max);
 379                         return true;
 380                 }
 381         }
 382
 383         // We will now check for the system values.
 384         // This limit could be reached although the user limits are fine.
 385         $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_connections'");
 386         if (!dbm::is_result($r)) {
 387                 return false;
 388         }
 389         $max = intval($r[0]["Value"]);
 390         if ($max == 0) {
 391                 return false;
 392         }
 393         $r = q("SHOW STATUS WHERE `variable_name` = 'Threads_connected'");
 394         if (!dbm::is_result($r)) {
 395                 return false;
 396         }
 397         $used = intval($r[0]["Value"]);
 398         if ($used == 0) {
 399                 return false;
 400         }
 401         logger("Connection usage (system values): ".$used."/".$max, LOGGER_DEBUG);
 402
 403         $level = $used / $max * 100;
 404
 405         if ($level < $maxlevel) {
 406                 return false;
 407         }
 408         logger("Maximum level (".$level."%) of system connections reached: ".$used."/".$max);
 409         return true;
 410 }
 411
 412 /**
 413  * @brief fix the queue entry if the worker process died
 414  *
 415  */
 416 function poller_kill_stale_workers() {
 417         $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` > '%s'", dbesc(NULL_DATE));
 418
 419         if (!dbm::is_result($r)) {
 420                 // No processing here needed
 421                 return;
 422         }
 423
 424         foreach ($r AS $pid) {
 425                 if (!posix_kill($pid["pid"], 0)) {
 426                         dba::update('workerqueue', array('executed' => NULL_DATE, 'pid' => 0),
 427                                         array('pid' => $pid["pid"]));
 428                 } else {
 429                         // Kill long running processes
 430
 431                         // Check if the priority is in a valid range
 432                         if (!in_array($pid["priority"], array(PRIORITY_CRITICAL, PRIORITY_HIGH, PRIORITY_MEDIUM, PRIORITY_LOW, PRIORITY_NEGLIGIBLE))) {
 433                                 $pid["priority"] = PRIORITY_MEDIUM;
 434                         }
 435
 436                         // Define the maximum durations
 437                         $max_duration_defaults = array(PRIORITY_CRITICAL => 360, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 360);
 438                         $max_duration = $max_duration_defaults[$pid["priority"]];
 439
 440                         $argv = json_decode($pid["parameter"]);
 441                         $argv[0] = basename($argv[0]);
 442
 443                         // How long is the process already running?
 444                         $duration = (time() - strtotime($pid["executed"])) / 60;
 445                         if ($duration > $max_duration) {
 446                                 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") took more than ".$max_duration." minutes. It will be killed now.");
 447                                 posix_kill($pid["pid"], SIGTERM);
 448
 449                                 // We killed the stale process.
 450                                 // To avoid a blocking situation we reschedule the process at the beginning of the queue.
 451                                 // Additionally we are lowering the priority.
 452                                 dba::update('workerqueue',
 453                                                 array('executed' => NULL_DATE, 'created' => datetime_convert(), 'priority' => PRIORITY_NEGLIGIBLE, 'pid' => 0),
 454                                                 array('pid' => $pid["pid"]));
 455                         } else {
 456                                 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") now runs for ".round($duration)." of ".$max_duration." allowed minutes. That's okay.", LOGGER_DEBUG);
 457                         }
 458                 }
 459         }
 460 }
 461
 462 /**
 463  * @brief Checks if the number of active workers exceeds the given limits
 464  *
 465  * @param integer $entries The number of not executed entries in the worker queue
 466  * @param integer $top_priority The highest not executed priority in the worker queue
 467  * @param boolean $high_running Is a process with priority "$top_priority" running?
 468  *
 469  * @return bool Are there too much workers running?
 470  */
 471 function poller_too_much_workers($entries = NULL, $top_priority = NULL, $high_running = NULL) {
 472         $queues = Config::get("system", "worker_queues", 4);
 473
 474         $maxqueues = $queues;
 475
 476         $active = poller_active_workers();
 477
 478         // Decrease the number of workers at higher load
 479         $load = current_load();
 480         if ($load) {
 481                 $maxsysload = intval(Config::get("system", "maxloadavg", 50));
 482
 483                 $maxworkers = $queues;
 484
 485                 // Some magical mathemathics to reduce the workers
 486                 $exponent = 3;
 487                 $slope = $maxworkers / pow($maxsysload, $exponent);
 488                 $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent));
 489
 490                 if (Config::get('system', 'worker_debug')) {
 491                         // Create a list of queue entries grouped by their priority
 492                         $listitem = array();
 493
 494                         // Adding all processes with no workerqueue entry
 495                         $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` WHERE NOT EXISTS (SELECT id FROM `workerqueue` WHERE `workerqueue`.`pid` = `process`.`pid`)");
 496                         if ($process = dba::fetch($processes)) {
 497                                 $listitem[0] = "0:".$process["running"];
 498                         }
 499                         dba::close($processes);
 500
 501                         // Now adding all processes with workerqueue entries
 502                         $entries = dba::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`");
 503                         while ($entry = dba::fetch($entries)) {
 504                                 $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` WHERE `priority` = ?", $entry["priority"]);
 505                                 if ($process = dba::fetch($processes)) {
 506                                         $listitem[$entry["priority"]] = $entry["priority"].":".$process["running"]."/".$entry["entries"];
 507                                 }
 508                                 dba::close($processes);
 509                         }
 510                         dba::close($entries);
 511                         $processlist = ' ('.implode(', ', $listitem).')';
 512                 }
 513
 514                 if (is_null($entries)) {
 515                         $entries = poller_total_entries();
 516                 }
 517                 if (Config::get("system", "worker_fastlane", false) && ($queues > 0) && ($entries > 0) && ($active >= $queues)) {
 518                         if (is_null($top_priority)) {
 519                                 $top_priority = poller_highest_priority();
 520                         }
 521                         if (is_null($high_running)) {
 522                                 $high_running = poller_process_with_priority_active($top_priority);
 523                         }
 524                         if (!$high_running && ($top_priority > PRIORITY_UNDEFINED) && ($top_priority < PRIORITY_NEGLIGIBLE)) {
 525                                 logger("There are jobs with priority ".$top_priority." waiting but none is executed. Open a fastlane.", LOGGER_DEBUG);
 526                                 $queues = $active + 1;
 527                         }
 528                 }
 529
 530                 logger("Load: ".$load."/".$maxsysload." - processes: ".$active."/".$entries.$processlist." - maximum: ".$queues."/".$maxqueues, LOGGER_DEBUG);
 531
 532                 // Are there fewer workers running as possible? Then fork a new one.
 533                 if (!Config::get("system", "worker_dont_fork") && ($queues > ($active + 1)) && ($entries > 1)) {
 534                         logger("Active workers: ".$active."/".$queues." Fork a new worker.", LOGGER_DEBUG);
 535                         $args = array("include/poller.php", "no_cron");
 536                         $a = get_app();
 537                         $a->proc_run($args);
 538                 }
 539         }
 540
 541         return $active >= $queues;
 542 }
 543
 544 /**
 545  * @brief Returns the number of active poller processes
 546  *
 547  * @return integer Number of active poller processes
 548  */
 549 function poller_active_workers() {
 550         $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php'");
 551
 552         return $workers[0]["processes"];
 553 }
 554
 555 /**
 556  * @brief Check if we should pass some slow processes
 557  *
 558  * When the active processes of the highest priority are using more than 2/3
 559  * of all processes, we let pass slower processes.
 560  *
 561  * @param string $highest_priority Returns the currently highest priority
 562  * @return bool We let pass a slower process than $highest_priority
 563  */
 564 function poller_passing_slow(&$highest_priority) {
 565
 566         $highest_priority = 0;
 567
 568         $r = q("SELECT `priority`
 569                 FROM `process`
 570                 INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`");
 571
 572         // No active processes at all? Fine
 573         if (!dbm::is_result($r)) {
 574                 return false;
 575         }
 576         $priorities = array();
 577         foreach ($r AS $line) {
 578                 $priorities[] = $line["priority"];
 579         }
 580         // Should not happen
 581         if (count($priorities) == 0) {
 582                 return false;
 583         }
 584         $highest_priority = min($priorities);
 585
 586         // The highest process is already the slowest one?
 587         // Then we quit
 588         if ($highest_priority == PRIORITY_NEGLIGIBLE) {
 589                 return false;
 590         }
 591         $high = 0;
 592         foreach ($priorities AS $priority) {
 593                 if ($priority == $highest_priority) {
 594                         ++$high;
 595                 }
 596         }
 597         logger("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, LOGGER_DEBUG);
 598         $passing_slow = (($high/count($priorities)) > (2/3));
 599
 600         if ($passing_slow) {
 601                 logger("Passing slower processes than priority ".$highest_priority, LOGGER_DEBUG);
 602         }
 603         return $passing_slow;
 604 }
 605
 606 /**
 607  * @brief Returns the next worker process
 608  *
 609  * @return string SQL statement
 610  */
 611 function poller_worker_process() {
 612
 613         // Check if we should pass some low priority process
 614         $highest_priority = 0;
 615
 616         if (poller_passing_slow($highest_priority)) {
 617                 dba::lock('workerqueue');
 618
 619                 // Are there waiting processes with a higher priority than the currently highest?
 620                 $r = q("SELECT * FROM `workerqueue`
 621                                 WHERE `executed` <= '%s' AND `priority` < %d
 622                                 ORDER BY `priority`, `created` LIMIT 1",
 623                                 dbesc(NULL_DATE),
 624                                 intval($highest_priority));
 625                 if (dbm::is_result($r)) {
 626                         return $r;
 627                 }
 628                 // Give slower processes some processing time
 629                 $r = q("SELECT * FROM `workerqueue`
 630                                 WHERE `executed` <= '%s' AND `priority` > %d
 631                                 ORDER BY `priority`, `created` LIMIT 1",
 632                                 dbesc(NULL_DATE),
 633                                 intval($highest_priority));
 634
 635                 if (dbm::is_result($r)) {
 636                         return $r;
 637                 }
 638         } else {
 639                 dba::lock('workerqueue');
 640         }
 641
 642         // If there is no result (or we shouldn't pass lower processes) we check without priority limit
 643         if (!dbm::is_result($r)) {
 644                 $r = q("SELECT * FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority`, `created` LIMIT 1", dbesc(NULL_DATE));
 645         }
 646
 647         // We only unlock the tables here, when we got no data
 648         if (!dbm::is_result($r)) {
 649                 dba::unlock();
 650         }
 651
 652         return $r;
 653 }
 654
 655 /**
 656  * @brief Assigns a workerqueue entry to the current process
 657  *
 658  * When we are sure that the table locks are working correctly, we can remove the checks from here
 659  *
 660  * @param array $queue Workerqueue entry
 661  *
 662  * @return boolean "true" if the claiming was successful
 663  */
 664 function poller_claim_process($queue) {
 665         $mypid = getmypid();
 666
 667         $success = dba::update('workerqueue', array('executed' => datetime_convert(), 'pid' => $mypid),
 668                         array('id' => $queue["id"], 'pid' => 0));
 669
 670         if (!$success) {
 671                 logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG);
 672                 return false;
 673         }
 674
 675         // Assure that there are no tasks executed twice
 676         $id = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
 677         if (!$id) {
 678                 logger("Queue item ".$queue["id"]." vanished - skip this execution", LOGGER_DEBUG);
 679                 return false;
 680         } elseif ((strtotime($id[0]["executed"]) <= 0) || ($id[0]["pid"] == 0)) {
 681                 logger("Entry for queue item ".$queue["id"]." wasn't stored - skip this execution", LOGGER_DEBUG);
 682                 return false;
 683         } elseif ($id[0]["pid"] != $mypid) {
 684                 logger("Queue item ".$queue["id"]." is to be executed by process ".$id[0]["pid"]." and not by me (".$mypid.") - skip this execution", LOGGER_DEBUG);
 685                 return false;
 686         }
 687         return true;
 688 }
 689
 690 /**
 691  * @brief Removes a workerqueue entry from the current process
 692  */
 693 function poller_unclaim_process() {
 694         $mypid = getmypid();
 695
 696         dba::update('workerqueue', array('executed' => NULL_DATE, 'pid' => 0), array('pid' => $mypid));
 697 }
 698
 699 /**
 700  * @brief Call the front end worker
 701  */
 702 function call_worker() {
 703         if (!Config::get("system", "frontend_worker")) {
 704                 return;
 705         }
 706
 707         $url = App::get_baseurl()."/worker";
 708         fetch_url($url, false, $redirects, 1);
 709 }
 710
 711 /**
 712  * @brief Call the front end worker if there aren't any active
 713  */
 714 function call_worker_if_idle() {
 715         if (!Config::get("system", "frontend_worker")) {
 716                 return;
 717         }
 718
 719         // Do we have "proc_open"? Then we can fork the poller
 720         if (function_exists("proc_open")) {
 721                 // When was the last time that we called the worker?
 722                 // Less than one minute? Then we quit
 723                 if ((time() - Config::get("system", "worker_started")) < 60) {
 724                         return;
 725                 }
 726
 727                 set_config("system", "worker_started", time());
 728
 729                 // Do we have enough running workers? Then we quit here.
 730                 if (poller_too_much_workers()) {
 731                         // Cleaning dead processes
 732                         poller_kill_stale_workers();
 733                         get_app()->remove_inactive_processes();
 734
 735                         return;
 736                 }
 737
 738                 poller_run_cron();
 739
 740                 logger('Call poller', LOGGER_DEBUG);
 741
 742                 $args = array("include/poller.php", "no_cron");
 743                 $a = get_app();
 744                 $a->proc_run($args);
 745                 return;
 746         }
 747
 748         // We cannot execute background processes.
 749         // We now run the processes from the frontend.
 750         // This won't work with long running processes.
 751         poller_run_cron();
 752
 753         clear_worker_processes();
 754
 755         $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php'");
 756
 757         if ($workers[0]["processes"] == 0) {
 758                 call_worker();
 759         }
 760 }
 761
 762 /**
 763  * @brief Removes long running worker processes
 764  */
 765 function clear_worker_processes() {
 766         $timeout = Config::get("system", "frontend_worker_timeout", 10);
 767
 768         /// @todo We should clean up the corresponding workerqueue entries as well
 769         q("DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php'",
 770                 dbesc(datetime_convert('UTC','UTC',"now - ".$timeout." minutes")));
 771 }
 772
 773 /**
 774  * @brief Runs the cron processes
 775  */
 776 function poller_run_cron() {
 777         logger('Add cron entries', LOGGER_DEBUG);
 778
 779         // Check for spooled items
 780         proc_run(PRIORITY_HIGH, "include/spool_post.php");
 781
 782         // Run the cron job that calls all other jobs
 783         proc_run(PRIORITY_MEDIUM, "include/cron.php");
 784
 785         // Run the cronhooks job separately from cron for being able to use a different timing
 786         proc_run(PRIORITY_MEDIUM, "include/cronhooks.php");
 787
 788         // Cleaning dead processes
 789         poller_kill_stale_workers();
 790 }
 791
 792 if (array_search(__file__,get_included_files())===0){
 793         poller_run($_SERVER["argv"],$_SERVER["argc"]);
 794
 795         poller_unclaim_process();
 796
 797         get_app()->end_process();
 798
 799         Lock::remove('poller_worker');
 800
 801         killme();
 802 }