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