diff --git a/boot.php b/boot.php index dcf6c65b1..5f633fd87 100644 --- a/boot.php +++ b/boot.php @@ -19,7 +19,7 @@ define ( 'FRIENDICA_PLATFORM', 'Friendica'); define ( 'FRIENDICA_CODENAME', 'Lily of the valley'); define ( 'FRIENDICA_VERSION', '3.4.2' ); define ( 'DFRN_PROTOCOL_VERSION', '2.23' ); -define ( 'DB_UPDATE_VERSION', 1188 ); +define ( 'DB_UPDATE_VERSION', 1189 ); define ( 'EOL', "
\r\n" ); define ( 'ATOM_TIME', 'Y-m-d\TH:i:s\Z' ); @@ -1432,8 +1432,46 @@ if(! function_exists('proc_run')) { if(! $arr['run_cmd']) return; - if(count($args) && $args[0] === 'php') + if(count($args) && $args[0] === 'php') { + + if (get_config("system", "worker")) { + $argv = $args; + array_shift($argv); + + $parameters = json_encode($argv); + $found = q("SELECT `id` FROM `workerqueue` WHERE `parameter` = '%s'", + dbesc($parameters)); + + if (!$found) + q("INSERT INTO `workerqueue` (`parameter`, `created`, `priority`) + VALUES ('%s', '%s', %d)", + dbesc($parameters), + dbesc(datetime_convert()), + intval(0)); + + // Should we quit and wait for the poller to be called as a cronjob? + if (get_config("system", "worker_dont_fork")) + return; + + // Checking number of workers + $workers = q("SELECT COUNT(*) AS `workers` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + + // Get number of allowed number of worker threads + $queues = intval(get_config("system", "worker_queues")); + + if ($queues == 0) + $queues = 4; + + // If there are already enough workers running, don't fork another one + if ($workers[0]["workers"] >= $queues) + return; + + // Now call the poller to execute the jobs that we just added to the queue + $args = array("php", "include/poller.php", "no_cron"); + } + $args[0] = ((x($a->config,'php_path')) && (strlen($a->config['php_path'])) ? $a->config['php_path'] : 'php'); + } // add baseurl to args. cli scripts can't construct it $args[] = $a->get_baseurl(); @@ -1441,9 +1479,8 @@ if(! function_exists('proc_run')) { for($x = 0; $x < count($args); $x ++) $args[$x] = escapeshellarg($args[$x]); - - $cmdline = implode($args," "); + if(get_config('system','proc_windows')) proc_close(proc_open('cmd /c start /b ' . $cmdline,array(),$foo,dirname(__FILE__))); else @@ -1860,3 +1897,31 @@ if(!function_exists('exif_imagetype')) { return($size[2]); } } + +function validate_include(&$file) { + $orig_file = $file; + + $file = realpath($file); + + if (strpos($file, getcwd()) !== 0) + return false; + + $file = str_replace(getcwd()."/", "", $file, $count); + if ($count != 1) + return false; + + if ($orig_file !== $file) + return false; + + $valid = false; + if (strpos($file, "include/") === 0) + $valid = true; + + if (strpos($file, "addon/") === 0) + $valid = true; + + if (!$valid) + return false; + + return true; +} diff --git a/database.sql b/database.sql index 76df6aec1..a6eb71ef3 100644 --- a/database.sql +++ b/database.sql @@ -1,6 +1,6 @@ -- ------------------------------------------ -- Friendica 3.4.1 (Lily of the valley) --- DB_UPDATE_VERSION 1188 +-- DB_UPDATE_VERSION 1189 -- ------------------------------------------ @@ -1020,3 +1020,16 @@ CREATE TABLE IF NOT EXISTS `userd` ( INDEX `username` (`username`) ) DEFAULT CHARSET=utf8; +-- +-- TABLE workerqueue +-- +CREATE TABLE IF NOT EXISTS `workerqueue` ( + `id` int(11) NOT NULL auto_increment PRIMARY KEY, + `parameter` text NOT NULL, + `priority` tinyint(3) unsigned NOT NULL DEFAULT 0, + `created` datetime NOT NULL DEFAULT '0000-00-00 00:00:00', + `pid` int(11) NOT NULL DEFAULT 0, + `executed` datetime NOT NULL DEFAULT '0000-00-00 00:00:00', + INDEX `created` (`created`) +) DEFAULT CHARSET=utf8; + diff --git a/include/cron.php b/include/cron.php new file mode 100644 index 000000000..46067d76d --- /dev/null +++ b/include/cron.php @@ -0,0 +1,341 @@ + $maxsysload) { + logger('system: load ' . $load[0] . ' too high. cron deferred to next scheduled run.'); + return; + } + } + + $last = get_config('system','last_cron'); + + $poll_interval = intval(get_config('system','cron_interval')); + if(! $poll_interval) + $poll_interval = 10; + + if($last) { + $next = $last + ($poll_interval * 60); + if($next > time()) { + logger('cron intervall not reached'); + return; + } + } + + $lockpath = get_lockpath(); + if ($lockpath != '') { + $pidfile = new pidfile($lockpath, 'cron'); + if($pidfile->is_already_running()) { + logger("cron: Already running"); + if ($pidfile->running_time() > 9*60) { + $pidfile->kill(); + logger("cron: killed stale process"); + // Calling a new instance + proc_run('php','include/cron.php'); + } + exit; + } + } + + + + $a->set_baseurl(get_config('system','url')); + + load_hooks(); + + logger('cron: start'); + + // run queue delivery process in the background + + proc_run('php',"include/queue.php"); + + // run diaspora photo queue process in the background + + proc_run('php',"include/dsprphotoq.php"); + + // run the process to discover global contacts in the background + + proc_run('php',"include/discover_poco.php"); + + // run the process to update locally stored global contacts in the background + + proc_run('php',"include/discover_poco.php", "checkcontact"); + + // expire any expired accounts + + q("UPDATE user SET `account_expired` = 1 where `account_expired` = 0 + AND `account_expires_on` != '0000-00-00 00:00:00' + AND `account_expires_on` < UTC_TIMESTAMP() "); + + // delete user and contact records for recently removed accounts + + $r = q("SELECT * FROM `user` WHERE `account_removed` = 1 AND `account_expires_on` < UTC_TIMESTAMP() - INTERVAL 3 DAY"); + if ($r) { + foreach($r as $user) { + q("DELETE FROM `contact` WHERE `uid` = %d", intval($user['uid'])); + q("DELETE FROM `user` WHERE `uid` = %d", intval($user['uid'])); + } + } + + $abandon_days = intval(get_config('system','account_abandon_days')); + if($abandon_days < 1) + $abandon_days = 0; + + // Check OStatus conversations + // Check only conversations with mentions (for a longer time) + check_conversations(true); + + // Check every conversation + check_conversations(false); + + // Follow your friends from your legacy OStatus account + // Doesn't work + // ostatus_check_follow_friends(); + + // update nodeinfo data + nodeinfo_cron(); + + // To-Do: Regenerate usage statistics + // q("ANALYZE TABLE `item`"); + + // once daily run birthday_updates and then expire in background + + $d1 = get_config('system','last_expire_day'); + $d2 = intval(datetime_convert('UTC','UTC','now','d')); + + if($d2 != intval($d1)) { + + update_contact_birthdays(); + + update_suggestions(); + + set_config('system','last_expire_day',$d2); + proc_run('php','include/expire.php'); + } + + $last = get_config('system','cache_last_cleared'); + + if($last) { + $next = $last + (3600); // Once per hour + $clear_cache = ($next <= time()); + } else + $clear_cache = true; + + if ($clear_cache) { + // clear old cache + Cache::clear(); + + // clear old item cache files + clear_cache(); + + // clear cache for photos + clear_cache($a->get_basepath(), $a->get_basepath()."/photo"); + + // clear smarty cache + clear_cache($a->get_basepath()."/view/smarty3/compiled", $a->get_basepath()."/view/smarty3/compiled"); + + // clear cache for image proxy + if (!get_config("system", "proxy_disabled")) { + clear_cache($a->get_basepath(), $a->get_basepath()."/proxy"); + + $cachetime = get_config('system','proxy_cache_time'); + if (!$cachetime) $cachetime = PROXY_DEFAULT_TIME; + + q('DELETE FROM `photo` WHERE `uid` = 0 AND `resource-id` LIKE "pic:%%" AND `created` < NOW() - INTERVAL %d SECOND', $cachetime); + } + + set_config('system','cache_last_cleared', time()); + } + + $manual_id = 0; + $generation = 0; + $force = false; + $restart = false; + + if(($argc > 1) && ($argv[1] == 'force')) + $force = true; + + if(($argc > 1) && ($argv[1] == 'restart')) { + $restart = true; + $generation = intval($argv[2]); + if(! $generation) + killme(); + } + + if(($argc > 1) && intval($argv[1])) { + $manual_id = intval($argv[1]); + $force = true; + } + + $interval = intval(get_config('system','poll_interval')); + if(! $interval) + $interval = ((get_config('system','delivery_interval') === false) ? 3 : intval(get_config('system','delivery_interval'))); + + // If we are using the worker we don't need a delivery interval + if (get_config("system", "worker")) + $interval = false; + + $sql_extra = (($manual_id) ? " AND `id` = $manual_id " : ""); + + reload_plugins(); + + $d = datetime_convert(); + + // Only poll from those with suitable relationships, + // and which have a polling address and ignore Diaspora since + // we are unable to match those posts with a Diaspora GUID and prevent duplicates. + + $abandon_sql = (($abandon_days) + ? sprintf(" AND `user`.`login_date` > UTC_TIMESTAMP() - INTERVAL %d DAY ", intval($abandon_days)) + : '' + ); + + $contacts = q("SELECT `contact`.`id` FROM `contact` INNER JOIN `user` ON `user`.`uid` = `contact`.`uid` + WHERE `rel` IN (%d, %d) AND `poll` != '' AND `network` IN ('%s', '%s', '%s', '%s', '%s', '%s') + $sql_extra + AND NOT `self` AND NOT `contact`.`blocked` AND NOT `contact`.`readonly` AND NOT `contact`.`archive` + AND NOT `user`.`account_expired` AND NOT `user`.`account_removed` $abandon_sql ORDER BY RAND()", + intval(CONTACT_IS_SHARING), + intval(CONTACT_IS_FRIEND), + dbesc(NETWORK_DFRN), + dbesc(NETWORK_ZOT), + dbesc(NETWORK_OSTATUS), + dbesc(NETWORK_FEED), + dbesc(NETWORK_MAIL), + dbesc(NETWORK_MAIL2) + ); + + if(! count($contacts)) { + return; + } + + foreach($contacts as $c) { + + $res = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1", + intval($c['id']) + ); + + if((! $res) || (! count($res))) + continue; + + foreach($res as $contact) { + + $xml = false; + + if($manual_id) + $contact['last-update'] = '0000-00-00 00:00:00'; + + if(in_array($contact['network'], array(NETWORK_DFRN, NETWORK_ZOT, NETWORK_OSTATUS))) + $contact['priority'] = 2; + + if($contact['subhub'] AND in_array($contact['network'], array(NETWORK_DFRN, NETWORK_ZOT, NETWORK_OSTATUS))) { + // We should be getting everything via a hub. But just to be sure, let's check once a day. + // (You can make this more or less frequent if desired by setting 'pushpoll_frequency' appropriately) + // This also lets us update our subscription to the hub, and add or replace hubs in case it + // changed. We will only update hubs once a day, regardless of 'pushpoll_frequency'. + + $poll_interval = get_config('system','pushpoll_frequency'); + $contact['priority'] = (($poll_interval !== false) ? intval($poll_interval) : 3); + } + + if($contact['priority'] AND !$force) { + + $update = false; + + $t = $contact['last-update']; + + /** + * Based on $contact['priority'], should we poll this site now? Or later? + */ + + switch ($contact['priority']) { + case 5: + if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 month")) + $update = true; + break; + case 4: + if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 week")) + $update = true; + break; + case 3: + if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 day")) + $update = true; + break; + case 2: + if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 12 hour")) + $update = true; + break; + case 1: + default: + if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 hour")) + $update = true; + break; + } + if(!$update) + continue; + } + + logger("Polling ".$contact["network"]." ".$contact["id"]." ".$contact["nick"]." ".$contact["name"]); + + proc_run('php','include/onepoll.php',$contact['id']); + + if($interval) + @time_sleep_until(microtime(true) + (float) $interval); + } + } + + logger('cron: end'); + + set_config('system','last_cron', time()); + + return; +} + +if (array_search(__file__,get_included_files())===0){ + cron_run($_SERVER["argv"],$_SERVER["argc"]); + killme(); +} diff --git a/include/cronhooks.php b/include/cronhooks.php index 26cab3cf9..d5b4f3bf6 100644 --- a/include/cronhooks.php +++ b/include/cronhooks.php @@ -11,11 +11,11 @@ function cronhooks_run(&$argv, &$argc){ } if(is_null($db)) { - @include(".htconfig.php"); - require_once("include/dba.php"); - $db = new dba($db_host, $db_user, $db_pass, $db_data); - unset($db_host, $db_user, $db_pass, $db_data); - }; + @include(".htconfig.php"); + require_once("include/dba.php"); + $db = new dba($db_host, $db_user, $db_pass, $db_data); + unset($db_host, $db_user, $db_pass, $db_data); + }; require_once('include/session.php'); require_once('include/datetime.php'); @@ -35,17 +35,31 @@ function cronhooks_run(&$argv, &$argc){ } } + $last = get_config('system','last_cronhook'); + + $poll_interval = intval(get_config('system','cronhook_interval')); + if(! $poll_interval) + $poll_interval = 9; + + if($last) { + $next = $last + ($poll_interval * 60); + if($next > time()) { + logger('cronhook intervall not reached'); + return; + } + } + $lockpath = get_lockpath(); if ($lockpath != '') { $pidfile = new pidfile($lockpath, 'cronhooks'); if($pidfile->is_already_running()) { logger("cronhooks: Already running"); if ($pidfile->running_time() > 19*60) { - $pidfile->kill(); - logger("cronhooks: killed stale process"); + $pidfile->kill(); + logger("cronhooks: killed stale process"); // Calling a new instance proc_run('php','include/cronhooks.php'); - } + } exit; } } @@ -62,10 +76,12 @@ function cronhooks_run(&$argv, &$argc){ logger('cronhooks: end'); + set_config('system','last_cronhook', time()); + return; } if (array_search(__file__,get_included_files())===0){ - cronhooks_run($_SERVER["argv"],$_SERVER["argc"]); - killme(); + cronhooks_run($_SERVER["argv"],$_SERVER["argc"]); + killme(); } diff --git a/include/dbstructure.php b/include/dbstructure.php index 2b1ee84fd..deb6ddf53 100644 --- a/include/dbstructure.php +++ b/include/dbstructure.php @@ -1382,6 +1382,20 @@ function db_definition() { "username" => array("username"), ) ); + $database["workerqueue"] = array( + "fields" => array( + "id" => array("type" => "int(11)", "not null" => "1", "extra" => "auto_increment", "primary" => "1"), + "parameter" => array("type" => "text", "not null" => "1"), + "priority" => array("type" => "tinyint(3) unsigned", "not null" => "1", "default" => "0"), + "created" => array("type" => "datetime", "not null" => "1", "default" => "0000-00-00 00:00:00"), + "pid" => array("type" => "int(11)", "not null" => "1", "default" => "0"), + "executed" => array("type" => "datetime", "not null" => "1", "default" => "0000-00-00 00:00:00"), + ), + "indexes" => array( + "PRIMARY" => array("id"), + "created" => array("created"), + ) + ); return($database); } diff --git a/include/follow.php b/include/follow.php index 217b9d07b..ca0228cc0 100644 --- a/include/follow.php +++ b/include/follow.php @@ -9,13 +9,13 @@ function update_contact($id) { $r = q("SELECT `url`, `nurl`, `addr`, `alias`, `batch`, `notify`, `poll`, `poco`, `network` FROM `contact` WHERE `id` = %d", intval($id)); if (!$r) - return; + return false; $ret = probe_url($r[0]["url"]); // If probe_url fails the network code will be different if ($ret["network"] != $r[0]["network"]) - return; + return false; $update = false; @@ -29,7 +29,7 @@ function update_contact($id) { } if (!$update) - return; + return true; q("UPDATE `contact` SET `url` = '%s', `nurl` = '%s', `addr` = '%s', `alias` = '%s', `batch` = '%s', `notify` = '%s', `poll` = '%s', `poco` = '%s' WHERE `id` = %d", dbesc($ret['url']), @@ -42,6 +42,8 @@ function update_contact($id) { dbesc($ret['poco']), intval($id) ); + + return true; } // diff --git a/include/notifier.php b/include/notifier.php index 002b3c8d7..d4d254f1c 100644 --- a/include/notifier.php +++ b/include/notifier.php @@ -615,6 +615,10 @@ function notifier_run(&$argv, &$argc){ $interval = ((get_config('system','delivery_interval') === false) ? 2 : intval(get_config('system','delivery_interval'))); + // If we are using the worker we don't need a delivery interval + if (get_config("system", "worker")) + $interval = false; + // delivery loop if(count($r)) { @@ -634,7 +638,7 @@ function notifier_run(&$argv, &$argc){ // This controls the number of deliveries to execute with each separate delivery process. // By default we'll perform one delivery per process. Assuming a hostile shared hosting - // provider, this provides the greatest chance of deliveries if processes start getting + // provider, this provides the greatest chance of deliveries if processes start getting // killed. We can also space them out with the delivery_interval to also help avoid them // getting whacked. @@ -642,8 +646,10 @@ function notifier_run(&$argv, &$argc){ // together into a single process. This will reduce the overall number of processes // spawned for each delivery, but they will run longer. + // When using the workerqueue, we don't need this functionality. + $deliveries_per_process = intval(get_config('system','delivery_batch_count')); - if($deliveries_per_process <= 0) + if (($deliveries_per_process <= 0) OR get_config("system", "worker")) $deliveries_per_process = 1; $this_batch = array(); diff --git a/include/onepoll.php b/include/onepoll.php index 1fc861afa..0e58a776c 100644 --- a/include/onepoll.php +++ b/include/onepoll.php @@ -168,8 +168,18 @@ function onepoll_run(&$argv, &$argc){ ); // Update the contact entry - if(($contact['network'] === NETWORK_OSTATUS) || ($contact['network'] === NETWORK_DIASPORA) || ($contact['network'] === NETWORK_DFRN)) - update_contact($contact["id"]); + if(($contact['network'] === NETWORK_OSTATUS) || ($contact['network'] === NETWORK_DIASPORA) || ($contact['network'] === NETWORK_DFRN)) { + if (!poco_reachable($contact['url'])) { + logger("Skipping probably dead contact ".$contact['url']); + return; + } + + if (!update_contact($contact["id"])) { + mark_for_death($contact); + return; + } else + unmark_for_death($contact); + } if($contact['network'] === NETWORK_DFRN) { @@ -360,7 +370,7 @@ function onepoll_run(&$argv, &$argc){ ); logger("Mail: Connected to " . $mailconf[0]['user']); } else - logger("Mail: Connection error ".$mailconf[0]['user']." ".print_r(imap_errors())); + logger("Mail: Connection error ".$mailconf[0]['user']." ".print_r(imap_errors(), true)); } if($mbox) { diff --git a/include/pidfile.php b/include/pidfile.php index 4f5b25ad7..3093e149a 100644 --- a/include/pidfile.php +++ b/include/pidfile.php @@ -7,8 +7,8 @@ class pidfile { $this->_file = "$dir/$name.pid"; if (file_exists($this->_file)) { - $pid = trim(file_get_contents($this->_file)); - if (posix_kill($pid, 0)) { + $pid = trim(@file_get_contents($this->_file)); + if (($pid != "") AND posix_kill($pid, 0)) { $this->_running = true; } } @@ -21,7 +21,7 @@ class pidfile { public function __destruct() { if ((! $this->_running) && file_exists($this->_file)) { - unlink($this->_file); + @unlink($this->_file); } } @@ -30,7 +30,7 @@ class pidfile { } public function running_time() { - return(time() - filectime($this->_file)); + return(time() - @filectime($this->_file)); } public function kill() { diff --git a/include/poller.php b/include/poller.php index 28dc0c0cd..45740dab6 100644 --- a/include/poller.php +++ b/include/poller.php @@ -12,7 +12,6 @@ if (!file_exists("boot.php") AND (sizeof($_SERVER["argv"]) != 0)) { require_once("boot.php"); - function poller_run(&$argv, &$argc){ global $a, $db; @@ -21,303 +20,145 @@ function poller_run(&$argv, &$argc){ } if(is_null($db)) { - @include(".htconfig.php"); - require_once("include/dba.php"); - $db = new dba($db_host, $db_user, $db_pass, $db_data); - unset($db_host, $db_user, $db_pass, $db_data); - }; + @include(".htconfig.php"); + require_once("include/dba.php"); + $db = new dba($db_host, $db_user, $db_pass, $db_data); + unset($db_host, $db_user, $db_pass, $db_data); + }; - - require_once('include/session.php'); - require_once('include/datetime.php'); - require_once('library/simplepie/simplepie.inc'); - require_once('include/items.php'); - require_once('include/Contact.php'); - require_once('include/email.php'); - require_once('include/socgraph.php'); - require_once('include/pidfile.php'); - require_once('mod/nodeinfo.php'); - - load_config('config'); - load_config('system'); - - $maxsysload = intval(get_config('system','maxloadavg')); - if($maxsysload < 1) - $maxsysload = 50; if(function_exists('sys_getloadavg')) { + $maxsysload = intval(get_config('system','maxloadavg')); + if($maxsysload < 1) + $maxsysload = 50; + $load = sys_getloadavg(); if(intval($load[0]) > $maxsysload) { - logger('system: load ' . $load[0] . ' too high. Poller deferred to next scheduled run.'); + logger('system: load ' . $load[0] . ' too high. poller deferred to next scheduled run.'); return; } } - $lockpath = get_lockpath(); - if ($lockpath != '') { - $pidfile = new pidfile($lockpath, 'poller'); - if($pidfile->is_already_running()) { - logger("poller: Already running"); - if ($pidfile->running_time() > 9*60) { - $pidfile->kill(); - logger("poller: killed stale process"); - // Calling a new instance - proc_run('php','include/poller.php'); - } - exit; - } - } + if(($argc <= 1) OR ($argv[1] != "no_cron")) { + // Run the cron job that calls all other jobs + proc_run("php","include/cron.php"); + // Run the cronhooks job separately from cron for being able to use a different timing + proc_run("php","include/cronhooks.php"); + // Cleaning dead processes + $r = q("SELECT DISTINCT(`pid`) FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + foreach($r AS $pid) + if (!posix_kill($pid["pid"], 0)) + q("UPDATE `workerqueue` SET `executed` = '0000-00-00 00:00:00', `pid` = 0 WHERE `pid` = %d", + intval($pid["pid"])); + else { + // To-Do: Kill long running processes + // But: Update processes (like the database update) mustn't be killed + } - $a->set_baseurl(get_config('system','url')); + } else + // Sleep two seconds before checking for running processes to avoid having too many workers + sleep(4); - load_hooks(); - - logger('poller: start'); - - // run queue delivery process in the background - - proc_run('php',"include/queue.php"); - - // run diaspora photo queue process in the background - - proc_run('php',"include/dsprphotoq.php"); - - // run the process to discover global contacts in the background - - proc_run('php',"include/discover_poco.php"); - - // run the process to update locally stored global contacts in the background - - proc_run('php',"include/discover_poco.php", "checkcontact"); - - // expire any expired accounts - - q("UPDATE user SET `account_expired` = 1 where `account_expired` = 0 - AND `account_expires_on` != '0000-00-00 00:00:00' - AND `account_expires_on` < UTC_TIMESTAMP() "); - - // delete user and contact records for recently removed accounts - - $r = q("SELECT * FROM `user` WHERE `account_removed` = 1 AND `account_expires_on` < UTC_TIMESTAMP() - INTERVAL 3 DAY"); - if ($r) { - foreach($r as $user) { - q("DELETE FROM `contact` WHERE `uid` = %d", intval($user['uid'])); - q("DELETE FROM `user` WHERE `uid` = %d", intval($user['uid'])); - } - } - - $abandon_days = intval(get_config('system','account_abandon_days')); - if($abandon_days < 1) - $abandon_days = 0; - - // Check OStatus conversations - // Check only conversations with mentions (for a longer time) - check_conversations(true); - - // Check every conversation - check_conversations(false); - - // Follow your friends from your legacy OStatus account - ostatus_check_follow_friends(); - - // update nodeinfo data - nodeinfo_cron(); - - // To-Do: Regenerate usage statistics - // q("ANALYZE TABLE `item`"); - - // once daily run birthday_updates and then expire in background - - $d1 = get_config('system','last_expire_day'); - $d2 = intval(datetime_convert('UTC','UTC','now','d')); - - if($d2 != intval($d1)) { - - update_contact_birthdays(); - - update_suggestions(); - - set_config('system','last_expire_day',$d2); - proc_run('php','include/expire.php'); - } - - $last = get_config('system','cache_last_cleared'); - - if($last) { - $next = $last + (3600); // Once per hour - $clear_cache = ($next <= time()); - } else - $clear_cache = true; - - if ($clear_cache) { - // clear old cache - Cache::clear(); - - // clear old item cache files - clear_cache(); - - // clear cache for photos - clear_cache($a->get_basepath(), $a->get_basepath()."/photo"); - - // clear smarty cache - clear_cache($a->get_basepath()."/view/smarty3/compiled", $a->get_basepath()."/view/smarty3/compiled"); - - // clear cache for image proxy - if (!get_config("system", "proxy_disabled")) { - clear_cache($a->get_basepath(), $a->get_basepath()."/proxy"); - - $cachetime = get_config('system','proxy_cache_time'); - if (!$cachetime) $cachetime = PROXY_DEFAULT_TIME; - - q('DELETE FROM `photo` WHERE `uid` = 0 AND `resource-id` LIKE "pic:%%" AND `created` < NOW() - INTERVAL %d SECOND', $cachetime); - } - - set_config('system','cache_last_cleared', time()); - } - - $manual_id = 0; - $generation = 0; - $force = false; - $restart = false; - - if(($argc > 1) && ($argv[1] == 'force')) - $force = true; - - if(($argc > 1) && ($argv[1] == 'restart')) { - $restart = true; - $generation = intval($argv[2]); - if(! $generation) - killme(); - } - - if(($argc > 1) && intval($argv[1])) { - $manual_id = intval($argv[1]); - $force = true; - } - - $interval = intval(get_config('system','poll_interval')); - if(! $interval) - $interval = ((get_config('system','delivery_interval') === false) ? 3 : intval(get_config('system','delivery_interval'))); - - $sql_extra = (($manual_id) ? " AND `id` = $manual_id " : ""); - - reload_plugins(); - - $d = datetime_convert(); - - if(! $restart) - proc_run('php','include/cronhooks.php'); - - // Only poll from those with suitable relationships, - // and which have a polling address and ignore Diaspora since - // we are unable to match those posts with a Diaspora GUID and prevent duplicates. - - $abandon_sql = (($abandon_days) - ? sprintf(" AND `user`.`login_date` > UTC_TIMESTAMP() - INTERVAL %d DAY ", intval($abandon_days)) - : '' - ); - - $contacts = q("SELECT `contact`.`id` FROM `contact` INNER JOIN `user` ON `user`.`uid` = `contact`.`uid` - WHERE `rel` IN (%d, %d) AND `poll` != '' AND `network` IN ('%s', '%s', '%s', '%s', '%s', '%s') - $sql_extra - AND NOT `self` AND NOT `contact`.`blocked` AND NOT `contact`.`readonly` AND NOT `contact`.`archive` - AND NOT `user`.`account_expired` AND NOT `user`.`account_removed` $abandon_sql ORDER BY RAND()", - intval(CONTACT_IS_SHARING), - intval(CONTACT_IS_FRIEND), - dbesc(NETWORK_DFRN), - dbesc(NETWORK_ZOT), - dbesc(NETWORK_OSTATUS), - dbesc(NETWORK_FEED), - dbesc(NETWORK_MAIL), - dbesc(NETWORK_MAIL2) - ); - - if(! count($contacts)) { + // Checking number of workers + if (poller_too_much_workers()) return; - } - foreach($contacts as $c) { + $starttime = time(); - $res = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1", - intval($c['id']) - ); + while ($r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `created` LIMIT 1")) { - if((! $res) || (! count($res))) + q("UPDATE `workerqueue` SET `executed` = '%s', `pid` = %d WHERE `id` = %d AND `executed` = '0000-00-00 00:00:00'", + dbesc(datetime_convert()), + intval(getmypid()), + intval($r[0]["id"])); + + // Assure that there are no tasks executed twice + $id = q("SELECT `id` FROM `workerqueue` WHERE `id` = %d AND `pid` = %d", + intval($r[0]["id"]), + intval(getmypid())); + if (!$id) { + logger("Queue item ".$r[0]["id"]." was executed multiple times - skip this execution", LOGGER_DEBUG); continue; - - foreach($res as $contact) { - - $xml = false; - - if($manual_id) - $contact['last-update'] = '0000-00-00 00:00:00'; - - if(in_array($contact['network'], array(NETWORK_DFRN, NETWORK_ZOT, NETWORK_OSTATUS))) - $contact['priority'] = 2; - - if($contact['subhub'] AND in_array($contact['network'], array(NETWORK_DFRN, NETWORK_ZOT, NETWORK_OSTATUS))) { - // We should be getting everything via a hub. But just to be sure, let's check once a day. - // (You can make this more or less frequent if desired by setting 'pushpoll_frequency' appropriately) - // This also lets us update our subscription to the hub, and add or replace hubs in case it - // changed. We will only update hubs once a day, regardless of 'pushpoll_frequency'. - - $poll_interval = get_config('system','pushpoll_frequency'); - $contact['priority'] = (($poll_interval !== false) ? intval($poll_interval) : 3); - } - - if($contact['priority'] AND !$force) { - - $update = false; - - $t = $contact['last-update']; - - /** - * Based on $contact['priority'], should we poll this site now? Or later? - */ - - switch ($contact['priority']) { - case 5: - if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 month")) - $update = true; - break; - case 4: - if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 week")) - $update = true; - break; - case 3: - if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 day")) - $update = true; - break; - case 2: - if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 12 hour")) - $update = true; - break; - case 1: - default: - if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 hour")) - $update = true; - break; - } - if(!$update) - continue; - } - - logger("Polling ".$contact["network"]." ".$contact["id"]." ".$contact["nick"]." ".$contact["name"]); - - proc_run('php','include/onepoll.php',$contact['id']); - - if($interval) - @time_sleep_until(microtime(true) + (float) $interval); } + + $argv = json_decode($r[0]["parameter"]); + + $argc = count($argv); + + // Check for existance and validity of the include file + $include = $argv[0]; + + if (!validate_include($include)) { + logger("Include file ".$argv[0]." is not valid!"); + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); + continue; + } + + require_once($include); + + $funcname=str_replace(".php", "", basename($argv[0]))."_run"; + + if (function_exists($funcname)) { + logger("Process ".getmypid().": ".$funcname." ".$r[0]["parameter"]); + $funcname($argv, $argc); + + logger("Process ".getmypid().": ".$funcname." - done"); + + q("DELETE FROM `workerqueue` WHERE `id` = %d", intval($r[0]["id"])); + } else + logger("Function ".$funcname." does not exist"); + + // Quit the poller once every hour + if (time() > ($starttime + 3600)) + return; + + // Count active workers and compare them with a maximum value that depends on the load + if (poller_too_much_workers()) + return; } - logger('poller: end'); +} - return; +function poller_too_much_workers() { + + $queues = get_config("system", "worker_queues"); + + if ($queues == 0) + $queues = 4; + + $active = poller_active_workers(); + + // Decrease the number of workers at higher load + if(function_exists('sys_getloadavg')) { + $load = max(sys_getloadavg()); + + $maxsysload = intval(get_config('system','maxloadavg')); + if($maxsysload < 1) + $maxsysload = 50; + + $maxworkers = $queues; + + // Some magical mathemathics to reduce the workers + $exponent = 3; + $slope = $maxworkers / pow($maxsysload, $exponent); + $queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent)); + + logger("Current load: ".$load." - maximum: ".$maxsysload." - current queues: ".$active." - maximum: ".$queues, LOGGER_DEBUG); + + } + + return($active >= $queues); +} + +function poller_active_workers() { + $workers = q("SELECT COUNT(*) AS `workers` FROM `workerqueue` WHERE `executed` != '0000-00-00 00:00:00'"); + + return($workers[0]["workers"]); } if (array_search(__file__,get_included_files())===0){ poller_run($_SERVER["argv"],$_SERVER["argc"]); killme(); } +?> diff --git a/include/queue.php b/include/queue.php index 0edd64fdb..cb5fe28ad 100644 --- a/include/queue.php +++ b/include/queue.php @@ -22,6 +22,7 @@ function queue_run(&$argv, &$argc){ require_once('include/items.php'); require_once('include/bbcode.php'); require_once('include/pidfile.php'); + require_once('include/socgraph.php'); load_config('config'); load_config('system'); @@ -59,6 +60,10 @@ function queue_run(&$argv, &$argc){ $interval = ((get_config('system','delivery_interval') === false) ? 2 : intval(get_config('system','delivery_interval'))); + // If we are using the worker we don't need a delivery interval + if (get_config("system", "worker")) + $interval = false; + $r = q("select * from deliverq where 1"); if($r) { foreach($r as $rr) { @@ -88,7 +93,7 @@ function queue_run(&$argv, &$argc){ else { // For the first 12 hours we'll try to deliver every 15 minutes - // After that, we'll only attempt delivery once per hour. + // After that, we'll only attempt delivery once per hour. $r = q("SELECT `id` FROM `queue` WHERE (( `created` > UTC_TIMESTAMP() - INTERVAL 12 HOUR && `last` < UTC_TIMESTAMP() - INTERVAL 15 MINUTE ) OR ( `last` < UTC_TIMESTAMP() - INTERVAL 1 HOUR ))"); } @@ -107,7 +112,7 @@ function queue_run(&$argv, &$argc){ foreach($r as $q_item) { - // queue_predeliver hooks may have changed the queue db details, + // queue_predeliver hooks may have changed the queue db details, // so check again if this entry still needs processing if($queue_id) { @@ -132,12 +137,18 @@ function queue_run(&$argv, &$argc){ continue; } if(in_array($c[0]['notify'],$deadguys)) { - logger('queue: skipping known dead url: ' . $c[0]['notify']); - update_queue_time($q_item['id']); - continue; + logger('queue: skipping known dead url: ' . $c[0]['notify']); + update_queue_time($q_item['id']); + continue; } - $u = q("SELECT `user`.*, `user`.`pubkey` AS `upubkey`, `user`.`prvkey` AS `uprvkey` + if (!poco_reachable($c[0]['url'])) { + logger('queue: skipping probably dead url: ' . $c[0]['url']); + update_queue_time($q_item['id']); + continue; + } + + $u = q("SELECT `user`.*, `user`.`pubkey` AS `upubkey`, `user`.`prvkey` AS `uprvkey` FROM `user` WHERE `uid` = %d LIMIT 1", intval($c[0]['uid']) ); @@ -194,9 +205,9 @@ function queue_run(&$argv, &$argc){ call_hooks('queue_deliver', $a, $params); if($params['result']) - remove_queue_item($q_item['id']); + remove_queue_item($q_item['id']); else - update_queue_time($q_item['id']); + update_queue_time($q_item['id']); break; diff --git a/update.php b/update.php index 761da7273..06aab577a 100644 --- a/update.php +++ b/update.php @@ -1,6 +1,6 @@