diff --git a/boot.php b/boot.php index b93810d774..cfb82f5381 100644 --- 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 ed2c816dea..a452ecfc14 100644 --- 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 76f3e41019..9c2e6aeaa3 100644 --- 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 8ab9bf9d58..9b4921eb83 100644 --- 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 ef6fbd69de..816bcb53ec 100644 --- 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 a9816b06ce..0d347882e2 100644 --- 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 aaa7aa0b9e..09f11918c5 100644 --- a/update.php +++ b/update.php @@ -1,6 +1,6 @@