Merge pull request #3538 from annando/1706-some-more-performance

Some more worker fine tuning

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.

doc/database/db_workerqueue.md

@@ -9,5 +9,6 @@ Table workerqueue
 | created   |                  | datetime            | NO   | MUL | 0001-01-01 00:00:00 |                |
 | pid       |                  | int(11)             | NO   |     | 0                   |                |
 | executed  |                  | datetime            | NO   |     | 0001-01-01 00:00:00 |                |
+| done      | set to 1 if done | tinyint(1)          | NO   |     | 0                   |                |
 
 Return to [database documentation](help/database)

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)

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

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"),

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` WHERE 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` < ?
+					WHERE `executed` <= ? AND `priority` < ? AND NOT `done`
-					ORDER BY `priority`, `created` LIMIT 5",
+					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()));

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();
 		}
 	}
 

update.php

@@ -1,6 +1,6 @@
 <?php
 
-define('UPDATE_VERSION' , 1229);
+define('UPDATE_VERSION' , 1230);
 
 /**
  *