include/poller.php

   1 <?php
   2 if (!file_exists("boot.php") AND (sizeof($_SERVER["argv"]) != 0)) {
   3         $directory = dirname($_SERVER["argv"][0]);
   4
   5         if (substr($directory, 0, 1) != "/")
   6                 $directory = $_SERVER["PWD"]."/".$directory;
   7
   8         $directory = realpath($directory."/..");
   9
  10         chdir($directory);
  11 }
  12
  13 use \Friendica\Core\Config;
  14 use \Friendica\Core\PConfig;
  15
  16 require_once("boot.php");
  17
  18 function poller_run($argv, $argc){
  19         global $a, $db;
  20
  21         if(is_null($a)) {
  22                 $a = new App;
  23         }
  24
  25         if(is_null($db)) {
  26                 @include(".htconfig.php");
  27                 require_once("include/dba.php");
  28                 $db = new dba($db_host, $db_user, $db_pass, $db_data);
  29                 unset($db_host, $db_user, $db_pass, $db_data);
  30         };
  31
  32         // Quit when in maintenance
  33         if (get_config('system', 'maintenance', true))
  34                 return;
  35
  36         $a->start_process();
  37
  38         if (poller_max_connections_reached()) {
  39                 return;
  40         }
  41
  42         if (App::maxload_reached()) {
  43                 return;
  44         }
  45
  46         if(($argc <= 1) OR ($argv[1] != "no_cron")) {
  47                 poller_run_cron();
  48         }
  49
  50         if ($a->max_processes_reached()) {
  51                 return;
  52         }
  53
  54         // Checking the number of workers
  55         if (poller_too_much_workers()) {
  56                 poller_kill_stale_workers();
  57                 return;
  58         }
  59
  60         $starttime = time();
  61
  62         while ($r = poller_worker_process()) {
  63
  64                 // Count active workers and compare them with a maximum value that depends on the load
  65                 if (poller_too_much_workers()) {
  66                         return;
  67                 }
  68
  69                 if (!poller_execute($r[0])) {
  70                         return;
  71                 }
  72
  73                 // Quit the poller once every hour
  74                 if (time() > ($starttime + 3600))
  75                         return;
  76         }
  77
  78 }
  79
  80 /**
  81  * @brief Execute a worker entry
  82  *
  83  * @param array $queue Workerqueue entry
  84  *
  85  * @return boolean "true" if further processing should be stopped
  86  */
  87 function poller_execute($queue) {
  88
  89         $a = get_app();
  90
  91         $mypid = getmypid();
  92
  93         $cooldown = Config::get("system", "worker_cooldown", 0);
  94
  95         // Quit when in maintenance
  96         if (get_config('system', 'maintenance', true)) {
  97                 return false;
  98         }
  99
 100         // Constantly check the number of parallel database processes
 101         if ($a->max_processes_reached()) {
 102                 return false;
 103         }
 104
 105         // Constantly check the number of available database connections to let the frontend be accessible at any time
 106         if (poller_max_connections_reached()) {
 107                 return false;
 108         }
 109
 110         $upd = q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `pid` = 0",
 111                 dbesc(datetime_convert()),
 112                 intval($mypid),
 113                 intval($queue["id"]));
 114
 115         if (!$upd) {
 116                 logger("Couldn't update queue entry ".$queue["id"]." - skip this execution", LOGGER_DEBUG);
 117                 q("COMMIT");
 118                 return true;
 119         }
 120
 121         // Assure that there are no tasks executed twice
 122         $id = q("SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
 123         if (!$id) {
 124                 logger("Queue item ".$queue["id"]." vanished - skip this execution", LOGGER_DEBUG);
 125                 q("COMMIT");
 126                 return true;
 127         } elseif ((strtotime($id[0]["executed"]) <= 0) OR ($id[0]["pid"] == 0)) {
 128                 logger("Entry for queue item ".$queue["id"]." wasn't stored - skip this execution", LOGGER_DEBUG);
 129                 q("COMMIT");
 130                 return true;
 131         } elseif ($id[0]["pid"] != $mypid) {
 132                 logger("Queue item ".$queue["id"]." is to be executed by process ".$id[0]["pid"]." and not by me (".$mypid.") - skip this execution", LOGGER_DEBUG);
 133                 q("COMMIT");
 134                 return true;
 135         }
 136         q("COMMIT");
 137
 138         $argv = json_decode($queue["parameter"]);
 139
 140         $argc = count($argv);
 141
 142         // Check for existance and validity of the include file
 143         $include = $argv[0];
 144
 145         if (!validate_include($include)) {
 146                 logger("Include file ".$argv[0]." is not valid!");
 147                 q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
 148                 return true;
 149         }
 150
 151         require_once($include);
 152
 153         $funcname = str_replace(".php", "", basename($argv[0]))."_run";
 154
 155         if (function_exists($funcname)) {
 156                 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]);
 157
 158                 // For better logging create a new process id for every worker call
 159                 // But preserve the old one for the worker
 160                 $old_process_id = $a->process_id;
 161                 $a->process_id = uniqid("wrk", true);
 162
 163                 $funcname($argv, $argc);
 164
 165                 $a->process_id = $old_process_id;
 166
 167                 if ($cooldown > 0) {
 168                         logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds");
 169                         sleep($cooldown);
 170                 }
 171
 172                 logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done");
 173
 174                 q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($queue["id"]));
 175         } else {
 176                 logger("Function ".$funcname." does not exist");
 177         }
 178
 179         return true;
 180 }
 181
 182 /**
 183  * @brief Checks if the number of database connections has reached a critical limit.
 184  *
 185  * @return bool Are more than 3/4 of the maximum connections used?
 186  */
 187 function poller_max_connections_reached() {
 188
 189         // Fetch the max value from the config. This is needed when the system cannot detect the correct value by itself.
 190         $max = get_config("system", "max_connections");
 191
 192         // Fetch the percentage level where the poller will get active
 193         $maxlevel = Config::get("system", "max_connections_level", 75);
 194
 195         if ($max == 0) {
 196                 // the maximum number of possible user connections can be a system variable
 197                 $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
 198                 if ($r)
 199                         $max = $r[0]["Value"];
 200
 201                 // Or it can be granted. This overrides the system variable
 202                 $r = q("SHOW GRANTS");
 203                 if ($r)
 204                         foreach ($r AS $grants) {
 205                                 $grant = array_pop($grants);
 206                                 if (stristr($grant, "GRANT USAGE ON"))
 207                                         if (preg_match("/WITH MAX_USER_CONNECTIONS (\d*)/", $grant, $match))
 208                                                 $max = $match[1];
 209                         }
 210         }
 211
 212         // If $max is set we will use the processlist to determine the current number of connections
 213         // The processlist only shows entries of the current user
 214         if ($max != 0) {
 215                 $r = q("SHOW PROCESSLIST");
 216                 if (!$r)
 217                         return false;
 218
 219                 $used = count($r);
 220
 221                 logger("Connection usage (user values): ".$used."/".$max, LOGGER_DEBUG);
 222
 223                 $level = ($used / $max) * 100;
 224
 225                 if ($level >= $maxlevel) {
 226                         logger("Maximum level (".$maxlevel."%) of user connections reached: ".$used."/".$max);
 227                         return true;
 228                 }
 229         }
 230
 231         // We will now check for the system values.
 232         // This limit could be reached although the user limits are fine.
 233         $r = q("SHOW VARIABLES WHERE `variable_name` = 'max_connections'");
 234         if (!$r)
 235                 return false;
 236
 237         $max = intval($r[0]["Value"]);
 238         if ($max == 0)
 239                 return false;
 240
 241         $r = q("SHOW STATUS WHERE `variable_name` = 'Threads_connected'");
 242         if (!$r)
 243                 return false;
 244
 245         $used = intval($r[0]["Value"]);
 246         if ($used == 0)
 247                 return false;
 248
 249         logger("Connection usage (system values): ".$used."/".$max, LOGGER_DEBUG);
 250
 251         $level = $used / $max * 100;
 252
 253         if ($level < $maxlevel)
 254                 return false;
 255
 256         logger("Maximum level (".$level."%) of system connections reached: ".$used."/".$max);
 257         return true;
 258 }
 259
 260 /**
 261  * @brief fix the queue entry if the worker process died
 262  *
 263  */
 264 function poller_kill_stale_workers() {
 265         $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'");
 266
 267         if (!dbm::is_result($r)) {
 268                 // No processing here needed
 269                 return;
 270         }
 271
 272         foreach($r AS $pid)
 273                 if (!posix_kill($pid["pid"], 0))
 274                         q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `pid` = 0 WHERE `pid` = %d",
 275                                 intval($pid["pid"]));
 276                 else {
 277                         // Kill long running processes
 278
 279                         // Check if the priority is in a valid range
 280                         if (!in_array($pid["priority"], array(PRIORITY_CRITICAL, PRIORITY_HIGH, PRIORITY_MEDIUM, PRIORITY_LOW, PRIORITY_NEGLIGIBLE)))
 281                                 $pid["priority"] = PRIORITY_MEDIUM;
 282
 283                         // Define the maximum durations
 284                         $max_duration_defaults = array(PRIORITY_CRITICAL => 360, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 360);
 285                         $max_duration = $max_duration_defaults[$pid["priority"]];
 286
 287                         $argv = json_decode($pid["parameter"]);
 288                         $argv[0] = basename($argv[0]);
 289
 290                         // How long is the process already running?
 291                         $duration = (time() - strtotime($pid["executed"])) / 60;
 292                         if ($duration > $max_duration) {
 293                                 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") took more than ".$max_duration." minutes. It will be killed now.");
 294                                 posix_kill($pid["pid"], SIGTERM);
 295
 296                                 // We killed the stale process.
 297                                 // To avoid a blocking situation we reschedule the process at the beginning of the queue.
 298                                 // Additionally we are lowering the priority.
 299                                 q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `created` = '%s',
 300                                                         `priority` = %d, `pid` = 0 WHERE `pid` = %d",
 301                                         dbesc(datetime_convert()),
 302                                         intval(PRIORITY_NEGLIGIBLE),
 303                                         intval($pid["pid"]));
 304                         } else
 305                                 logger("Worker process ".$pid["pid"]." (".implode(" ", $argv).") now runs for ".round($duration)." of ".$max_duration." allowed minutes. That's okay.", LOGGER_DEBUG);
 306                 }
 307 }
 308
 309 /**
 310  * @brief Checks if the number of active workers exceeds the given limits
 311  *
 312  * @return bool Are there too much workers running?
 313  */
 314 function poller_too_much_workers() {
 315         $queues = Config::get("system", "worker_queues", 4);
 316
 317         $maxqueues = $queues;
 318
 319         $active = poller_active_workers();
 320
 321         // Decrease the number of workers at higher load
 322         $load = current_load();
 323         if($load) {
 324                 $maxsysload = intval(Config::get("system", "maxloadavg", 50));
 325
 326                 $maxworkers = $queues;
 327
 328                 // Some magical mathemathics to reduce the workers
 329                 $exponent = 3;
 330                 $slope = $maxworkers / pow($maxsysload, $exponent);
 331                 $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent));
 332
 333                 $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00'");
 334                 $entries = $s[0]["total"];
 335
 336                 if (Config::get("system", "worker_fastlane", false) AND ($queues > 0) AND ($entries > 0) AND ($active >= $queues)) {
 337                         $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority` LIMIT 1");
 338                         $top_priority = $s[0]["priority"];
 339
 340                         $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` != '0000-00-00 00:00:00' LIMIT 1",
 341                                 intval($top_priority));
 342                         $high_running = dbm::is_result($s);
 343
 344                         if (!$high_running AND ($top_priority > PRIORITY_UNDEFINED) AND ($top_priority < PRIORITY_NEGLIGIBLE)) {
 345                                 logger("There are jobs with priority ".$top_priority." waiting but none is executed. Open a fastlane.", LOGGER_DEBUG);
 346                                 $queues = $active + 1;
 347                         }
 348                 }
 349
 350                 // Create a list of queue entries grouped by their priority
 351                 $running = array(PRIORITY_CRITICAL => 0,
 352                                 PRIORITY_HIGH => 0,
 353                                 PRIORITY_MEDIUM => 0,
 354                                 PRIORITY_LOW => 0,
 355                                 PRIORITY_NEGLIGIBLE => 0);
 356
 357                 $r = q("SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` GROUP BY `priority`");
 358                 if (dbm::is_result($r))
 359                         foreach ($r AS $process)
 360                                 $running[$process["priority"]] = $process["running"];
 361
 362                 $processlist = "";
 363                 $r = q("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`");
 364                 if (dbm::is_result($r))
 365                         foreach ($r as $entry) {
 366                                 if ($processlist != "")
 367                                         $processlist .= ", ";
 368                                 $processlist .= $entry["priority"].":".$running[$entry["priority"]]."/".$entry["entries"];
 369                         }
 370
 371                 logger("Load: ".$load."/".$maxsysload." - processes: ".$active."/".$entries." (".$processlist.") - maximum: ".$queues."/".$maxqueues, LOGGER_DEBUG);
 372
 373                 // Are there fewer workers running as possible? Then fork a new one.
 374                 if (!get_config("system", "worker_dont_fork") AND ($queues > ($active + 1)) AND ($entries > 1)) {
 375                         logger("Active workers: ".$active."/".$queues." Fork a new worker.", LOGGER_DEBUG);
 376                         $args = array("php", "include/poller.php", "no_cron");
 377                         $a = get_app();
 378                         $a->proc_run($args);
 379                 }
 380         }
 381
 382         return($active >= $queues);
 383 }
 384
 385 /**
 386  * @brief Returns the number of active poller processes
 387  *
 388  * @return integer Number of active poller processes
 389  */
 390 function poller_active_workers() {
 391         $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php'");
 392
 393         return($workers[0]["processes"]);
 394 }
 395
 396 /**
 397  * @brief Check if we should pass some slow processes
 398  *
 399  * When the active processes of the highest priority are using more than 2/3
 400  * of all processes, we let pass slower processes.
 401  *
 402  * @param string $highest_priority Returns the currently highest priority
 403  * @return bool We let pass a slower process than $highest_priority
 404  */
 405 function poller_passing_slow(&$highest_priority) {
 406
 407         $highest_priority = 0;
 408
 409         $r = q("SELECT `priority`
 410                 FROM `process`
 411                 INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`");
 412
 413         // No active processes at all? Fine
 414         if (!dbm::is_result($r))
 415                 return(false);
 416
 417         $priorities = array();
 418         foreach ($r AS $line)
 419                 $priorities[] = $line["priority"];
 420
 421         // Should not happen
 422         if (count($priorities) == 0)
 423                 return(false);
 424
 425         $highest_priority = min($priorities);
 426
 427         // The highest process is already the slowest one?
 428         // Then we quit
 429         if ($highest_priority == PRIORITY_NEGLIGIBLE)
 430                 return(false);
 431
 432         $high = 0;
 433         foreach ($priorities AS $priority)
 434                 if ($priority == $highest_priority)
 435                         ++$high;
 436
 437         logger("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, LOGGER_DEBUG);
 438         $passing_slow = (($high/count($priorities)) > (2/3));
 439
 440         if ($passing_slow)
 441                 logger("Passing slower processes than priority ".$highest_priority, LOGGER_DEBUG);
 442
 443         return($passing_slow);
 444 }
 445
 446 /**
 447  * @brief Returns the next worker process
 448  *
 449  * @return string SQL statement
 450  */
 451 function poller_worker_process() {
 452
 453         q("START TRANSACTION;");
 454
 455         // Check if we should pass some low priority process
 456         $highest_priority = 0;
 457
 458         if (poller_passing_slow($highest_priority)) {
 459                 // Are there waiting processes with a higher priority than the currently highest?
 460                 $r = q("SELECT * FROM `workerqueue`
 461                                 WHERE `executed` = '0000-00-00 00:00:00' AND `priority` < %d
 462                                 ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority));
 463                 if (dbm::is_result($r))
 464                         return $r;
 465
 466                 // Give slower processes some processing time
 467                 $r = q("SELECT * FROM `workerqueue`
 468                                 WHERE `executed` = '0000-00-00 00:00:00' AND `priority` > %d
 469                                 ORDER BY `priority`, `created` LIMIT 1", dbesc($highest_priority));
 470         }
 471
 472         // If there is no result (or we shouldn't pass lower processes) we check without priority limit
 473         if (($highest_priority == 0) OR !dbm::is_result($r))
 474                 $r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `priority`, `created` LIMIT 1");
 475
 476         return $r;
 477 }
 478
 479 /**
 480  * @brief Call the front end worker
 481  */
 482 function call_worker() {
 483         if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) {
 484                 return;
 485         }
 486
 487         $url = get_app()->get_baseurl()."/worker";
 488         fetch_url($url, false, $redirects, 1);
 489 }
 490
 491 /**
 492  * @brief Call the front end worker if there aren't any active
 493  */
 494 function call_worker_if_idle() {
 495         if (!Config::get("system", "frontend_worker") OR !Config::get("system", "worker")) {
 496                 return;
 497         }
 498
 499         // Do we have "proc_open"? Then we can fork the poller
 500         if (function_exists("proc_open")) {
 501                 // When was the last time that we called the worker?
 502                 // Less than one minute? Then we quit
 503                 if ((time() - get_config("system", "worker_started")) < 60) {
 504                         return;
 505                 }
 506
 507                 set_config("system", "worker_started", time());
 508
 509                 // Do we have enough running workers? Then we quit here.
 510                 if (poller_too_much_workers()) {
 511                         // Cleaning dead processes
 512                         poller_kill_stale_workers();
 513                         get_app()->remove_inactive_processes();
 514
 515                         return;
 516                 }
 517
 518                 poller_run_cron();
 519
 520                 logger('Call poller', LOGGER_DEBUG);
 521
 522                 $args = array("php", "include/poller.php", "no_cron");
 523                 $a = get_app();
 524                 $a->proc_run($args);
 525                 return;
 526         }
 527
 528         // We cannot execute background processes.
 529         // We now run the processes from the frontend.
 530         // This won't work with long running processes.
 531         poller_run_cron();
 532
 533         clear_worker_processes();
 534
 535         $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php'");
 536
 537         if ($workers[0]["processes"] == 0) {
 538                 call_worker();
 539         }
 540 }
 541
 542 /**
 543  * @brief Removes long running worker processes
 544  */
 545 function clear_worker_processes() {
 546         $timeout = Config::get("system", "frontend_worker_timeout", 10);
 547
 548         /// @todo We should clean up the corresponding workerqueue entries as well
 549         q("DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php'",
 550                 dbesc(datetime_convert('UTC','UTC',"now - ".$timeout." minutes")));
 551 }
 552
 553 /**
 554  * @brief Runs the cron processes
 555  */
 556 function poller_run_cron() {
 557         logger('Add cron entries', LOGGER_DEBUG);
 558
 559         // Check for spooled items
 560         proc_run(PRIORITY_HIGH, "include/spool_post.php");
 561
 562         // Run the cron job that calls all other jobs
 563         proc_run(PRIORITY_MEDIUM, "include/cron.php");
 564
 565         // Run the cronhooks job separately from cron for being able to use a different timing
 566         proc_run(PRIORITY_MEDIUM, "include/cronhooks.php");
 567
 568         // Cleaning dead processes
 569         poller_kill_stale_workers();
 570 }
 571
 572 if (array_search(__file__,get_included_files())===0){
 573         poller_run($_SERVER["argv"],$_SERVER["argc"]);
 574
 575         get_app()->end_process();
 576
 577         killme();
 578 }
 579 ?>