Some more worker fine tuning

diff --git a/boot.php b/boot.php
index b93810d774309a10c83cf3d639197b4645363565..cfb82f5381b36c2390f4878c64e81c6dc335195f 100644 (file)
--- a/boot.php
+++ b/boot.php
@@ -42,7 +42,7 @@ define ( 'FRIENDICA_PLATFORM',     'Friendica');
 define ( 'FRIENDICA_CODENAME',     'Asparagus');
 define ( 'FRIENDICA_VERSION',      '3.5.3-dev' );
 define ( 'DFRN_PROTOCOL_VERSION',  '2.23'    );
-define ( 'DB_UPDATE_VERSION',      1229      );
+define ( 'DB_UPDATE_VERSION',      1230      );
 
 /**
  * @brief Constant with a HTML line break.

diff --git a/doc/htconfig.md b/doc/htconfig.md
index ed2c816dea75c6ca845e863579287e0b80fa15f1..a452ecfc14f7bc5a70dde7163340f41d699388a4 100644 (file)
--- a/doc/htconfig.md
+++ b/doc/htconfig.md
@@ -37,7 +37,6 @@ Example: To set the directory value please add this line to your .htconfig.php:
 * **dbclean** (Boolean) - Enable the automatic database cleanup process
 * **dbclean-expire-days** (Integer) - Days after which remote items will be deleted. Own items, and marked or filed items are kept.
 * **default_service_class** -
-* **delivery_batch_count** - Number of deliveries per process. Default value is 1. (Disabled when using the worker)
 * **diaspora_test** (Boolean) - For development only. Disables the message transfer.
 * **directory** - The path to global directory. If not set then "http://dir.friendica.social" is used.
 * **disable_email_validation** (Boolean) - Disables the check if a mail address is in a valid format and can be resolved via DNS.
@@ -69,6 +68,7 @@ Example: To set the directory value please add this line to your .htconfig.php:
 * **paranoia** (Boolean) - Log out users if their IP address changed.
 * **permit_crawling** (Boolean) - Restricts the search for not logged in users to one search per minute.
 * **worker_debug** (Boolean) - If enabled, it prints out the number of running processes split by priority.
+* **worker_fetch_limit** - Number of worker tasks that are fetched in a single query. Default is 5.
 * **profiler** (Boolean) - Enable internal timings to help optimize code. Needed for "rendertime" addon. Default is false.
 * **free_crawls** - Number of "free" searches when "permit_crawling" is activated (Default value is 10)
 * **crawl_permit_period** - Period in seconds between allowed searches when the number of free searches is reached and "permit_crawling" is activated (Default value is 60)

diff --git a/include/cron.php b/include/cron.php
index 76f3e410196185514b9c24a12ef8ff3c6d8e8d62..9c2e6aeaa30d5f68f5b3f972befdb0e4845896c3 100644 (file)
--- a/include/cron.php
+++ b/include/cron.php
@@ -82,6 +82,9 @@ function cron_run(&$argv, &$argc){
                proc_run(PRIORITY_MEDIUM, 'include/dbclean.php');
 
                proc_run(PRIORITY_LOW, "include/cronjobs.php", "update_photo_albums");
+
+               // Delete all done workerqueue entries
+               dba::delete('workerqueue', array('done' => true));
        }
 
        // Poll contacts

diff --git a/include/dbstructure.php b/include/dbstructure.php
index 8ab9bf9d583155d9da2fc15d55b2a08e08bdb75e..9b4921eb83888b598af4406e832559e4b6074b3e 100644 (file)
--- a/include/dbstructure.php
+++ b/include/dbstructure.php
@@ -1739,6 +1739,7 @@ function db_definition() {
                                        "created" => array("type" => "datetime", "not null" => "1", "default" => NULL_DATE),
                                        "pid" => array("type" => "int(11)", "not null" => "1", "default" => "0"),
                                        "executed" => array("type" => "datetime", "not null" => "1", "default" => NULL_DATE),
+                                       "done" => array("type" => "tinyint(1)", "not null" => "1", "default" => "0"),
                                        ),
                        "indexes" => array(
                                        "PRIMARY" => array("id"),

diff --git a/include/poller.php b/include/poller.php
index ef6fbd69de1d44bc9d8966fd87776a928d14c7c7..816bcb53eceade70f001f7d79e9d5f22933cd19b 100644 (file)
--- a/include/poller.php
+++ b/include/poller.php
@@ -18,7 +18,9 @@ if (!file_exists("boot.php") && (sizeof($_SERVER["argv"]) != 0)) {
 require_once("boot.php");
 
 function poller_run($argv, $argc){
-       global $a, $db;
+       global $a, $db, $poller_up_start;
+
+       $poller_up_start = microtime(true);
 
        $a = new App(dirname(__DIR__));
 
@@ -129,7 +131,7 @@ function poller_run($argv, $argc){
  * @return integer Number of non executed entries in the worker queue
  */
 function poller_total_entries() {
-       $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= '%s'", dbesc(NULL_DATE));
+       $s = q("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= '%s' AND NOT `done`", dbesc(NULL_DATE));
        if (dbm::is_result($s)) {
                return $s[0]["total"];
        } else {
@@ -143,7 +145,7 @@ function poller_total_entries() {
  * @return integer Number of active poller processes
  */
 function poller_highest_priority() {
-       $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority` LIMIT 1", dbesc(NULL_DATE));
+       $s = q("SELECT `priority` FROM `workerqueue` WHERE `executed` <= '%s' AND NOT `done` ORDER BY `priority` LIMIT 1", dbesc(NULL_DATE));
        if (dbm::is_result($s)) {
                return $s[0]["priority"];
        } else {
@@ -159,7 +161,7 @@ function poller_highest_priority() {
  * @return integer Is there a process running with that priority?
  */
 function poller_process_with_priority_active($priority) {
-       $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` > '%s' LIMIT 1",
+       $s = q("SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` > '%s' AND NOT `done` LIMIT 1",
                        intval($priority), dbesc(NULL_DATE));
        return dbm::is_result($s);
 }
@@ -172,6 +174,7 @@ function poller_process_with_priority_active($priority) {
  * @return boolean "true" if further processing should be stopped
  */
 function poller_execute($queue) {
+       global $poller_db_duration;
 
        $a = get_app();
 
@@ -212,9 +215,13 @@ function poller_execute($queue) {
 
        if (function_exists($funcname)) {
                poller_exec_function($queue, $funcname, $argv);
-               dba::delete('workerqueue', array('id' => $queue["id"]));
+
+               $stamp = (float)microtime(true);
+               dba::update('workerqueue', array('done' => true), array('id' => $queue["id"]));
+               $poller_db_duration = (microtime(true) - $stamp);
        } else {
                logger("Function ".$funcname." does not exist");
+               dba::delete('workerqueue', array('id' => $queue["id"]));
        }
 
        return true;
@@ -228,6 +235,7 @@ function poller_execute($queue) {
  * @param array $argv Array of values to be passed to the function
  */
 function poller_exec_function($queue, $funcname, $argv) {
+       global $poller_up_start, $poller_db_duration;
 
        $a = get_app();
 
@@ -260,6 +268,8 @@ function poller_exec_function($queue, $funcname, $argv) {
        $a->process_id = uniqid("wrk", true);
        $a->queue = $queue;
 
+       $up_duration = number_format(microtime(true) - $poller_up_start, 3);
+
        $funcname($argv, $argc);
 
        $a->process_id = $old_process_id;
@@ -267,6 +277,15 @@ function poller_exec_function($queue, $funcname, $argv) {
 
        $duration = number_format(microtime(true) - $stamp, 3);
 
+       $poller_up_start = microtime(true);
+
+       /** With these values we can analyze how effective the worker is.
+        * The database and rest time should be low since this is the unproductive time.
+        * The execution time is the productive time.
+        * By changing parameters like the maximum number of workers we can check the effectivness.
+       */
+       logger('DB: '.number_format($poller_db_duration, 2).' - Rest: '.number_format($up_duration - $poller_db_duration, 2).' - Execution: '.number_format($duration, 2), LOGGER_DEBUG);
+
        if ($duration > 3600) {
                logger("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 1 hour (".round($duration/60, 3).")", LOGGER_DEBUG);
        } elseif ($duration > 600) {
@@ -419,7 +438,7 @@ function poller_max_connections_reached() {
  *
  */
 function poller_kill_stale_workers() {
-       $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` > '%s'", dbesc(NULL_DATE));
+       $r = q("SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` > '%s' AND NOT `done`", dbesc(NULL_DATE));
 
        if (!dbm::is_result($r)) {
                // No processing here needed
@@ -493,16 +512,16 @@ function poller_too_much_workers() {
                        $listitem = array();
 
                        // Adding all processes with no workerqueue entry
-                       $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` WHERE NOT EXISTS (SELECT id FROM `workerqueue` WHERE `workerqueue`.`pid` = `process`.`pid`)");
+                       $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` WHERE NOT EXISTS (SELECT id FROM `workerqueue` WHERE `workerqueue`.`pid` = `process`.`pid` AND NOT `done`)");
                        if ($process = dba::fetch($processes)) {
                                $listitem[0] = "0:".$process["running"];
                        }
                        dba::close($processes);
 
                        // Now adding all processes with workerqueue entries
-                       $entries = dba::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority`");
+                       $entries = dba::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` AND NOT `done` GROUP BY `priority`");
                        while ($entry = dba::fetch($entries)) {
-                               $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` WHERE `priority` = ?", $entry["priority"]);
+                               $processes = dba::p("SELECT COUNT(*) AS `running` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `done` WHERE `priority` = ?", $entry["priority"]);
                                if ($process = dba::fetch($processes)) {
                                        $listitem[$entry["priority"]] = $entry["priority"].":".$process["running"]."/".$entry["entries"];
                                }
@@ -564,7 +583,7 @@ function poller_passing_slow(&$highest_priority) {
 
        $r = q("SELECT `priority`
                FROM `process`
-               INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`");
+               INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `done`");
 
        // No active processes at all? Fine
        if (!dbm::is_result($r)) {
@@ -609,12 +628,13 @@ function find_worker_processes() {
        // Check if we should pass some low priority process
        $highest_priority = 0;
        $found = false;
+       $limit = Config::get('system', 'worker_fetch_limit', 5);
 
        if (poller_passing_slow($highest_priority)) {
                // Are there waiting processes with a higher priority than the currently highest?
                $result = dba::e("UPDATE `workerqueue` SET `executed` = ?, `pid` = ?
-                                       WHERE `executed` <= ? AND `priority` < ?
-                                       ORDER BY `priority`, `created` LIMIT 5",
+                                       WHERE `executed` <= ? AND `priority` < ? AND NOT `done`
+                                       ORDER BY `priority`, `created` LIMIT ".intval($limit),
                                datetime_convert(), getmypid(), NULL_DATE, $highest_priority);
                if ($result) {
                        $found = (dba::affected_rows() > 0);
@@ -623,7 +643,7 @@ function find_worker_processes() {
                if (!$found) {
                        // Give slower processes some processing time
                        $result = dba::e("UPDATE `workerqueue` SET `executed` = ?, `pid` = ?
-                                               WHERE `executed` <= ? AND `priority` > ?
+                                               WHERE `executed` <= ? AND `priority` > ? AND NOT `done`
                                                ORDER BY `priority`, `created` LIMIT 1",
                                        datetime_convert(), getmypid(), NULL_DATE, $highest_priority);
                        if ($result) {
@@ -634,7 +654,7 @@ function find_worker_processes() {
 
        // If there is no result (or we shouldn't pass lower processes) we check without priority limit
        if (!$found) {
-               $result = dba::e("UPDATE `workerqueue` SET `executed` = ?, `pid` = ? WHERE `executed` <= ? ORDER BY `priority`, `created` LIMIT 5",
+               $result = dba::e("UPDATE `workerqueue` SET `executed` = ?, `pid` = ? WHERE `executed` <= ? AND NOT `done` ORDER BY `priority`, `created` LIMIT ".intval($limit),
                                datetime_convert(), getmypid(), NULL_DATE);
                if ($result) {
                        $found = (dba::affected_rows() > 0);
@@ -649,12 +669,13 @@ function find_worker_processes() {
  * @return string SQL statement
  */
 function poller_worker_process() {
+       global $poller_db_duration;
 
        $stamp = (float)microtime(true);
 
        $found = find_worker_processes();
 
-       logger('Duration: '.number_format(microtime(true) - $stamp, 3), LOGGER_DEBUG);
+       $poller_db_duration += (microtime(true) - $stamp);
 
        if ($found) {
                $r = q("SELECT * FROM `workerqueue` WHERE `pid` = %d", intval(getmypid()));

diff --git a/mod/display.php b/mod/display.php
index a9816b06cea4f582da32327406da37bc2352ce21..0d347882e2b16c95120a5fcbd4f16e5c146d96ab 100644 (file)
--- a/mod/display.php
+++ b/mod/display.php
@@ -17,9 +17,12 @@ function display_init(App $a) {
                if (substr($a->argv[2], -5) == '.atom') {
                        $item_id = substr($a->argv[2], 0, -5);
                        $xml = dfrn::itemFeed($item_id);
+                       if ($xml == '') {
+                               http_status_exit(500);
+                       }
                        header("Content-type: application/atom+xml");
                        echo $xml;
-                       http_status_exit(($xml) ? 200 : 500);
+                       killme();
                }
        }
 

diff --git a/update.php b/update.php
index aaa7aa0b9e62161b0cb74f7010d122a60a1d5487..09f11918c5ba026831c8c4928b5d653932a7407e 100644 (file)
--- a/update.php
+++ b/update.php
@@ -1,6 +1,6 @@
 <?php
 
-define('UPDATE_VERSION' , 1229);
+define('UPDATE_VERSION' , 1230);
 
 /**
  *