Merge pull request #1912 from annando/1509-worker

New way of background processing

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',                    "<br />\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;
+}

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;
+

include/cron.php

@@ -0,0 +1,341 @@
+<?php
+if (!file_exists("boot.php") AND (sizeof($_SERVER["argv"]) != 0)) {
+	$directory = dirname($_SERVER["argv"][0]);
+
+	if (substr($directory, 0, 1) != "/")
+		$directory = $_SERVER["PWD"]."/".$directory;
+
+	$directory = realpath($directory."/..");
+
+	chdir($directory);
+}
+
+require_once("boot.php");
+
+
+function cron_run(&$argv, &$argc){
+	global $a, $db;
+
+	if(is_null($a)) {
+		$a = new App;
+	}
+
+	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);
+	};
+
+
+	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')) {
+		$load = sys_getloadavg();
+		if(intval($load[0]) > $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();
+}

include/cronhooks.php

@@ -11,11 +11,11 @@ function cronhooks_run(&$argv, &$argc){
 	}
 
 	if(is_null($db)) {
-	    @include(".htconfig.php");
+		@include(".htconfig.php");
-    	require_once("include/dba.php");
+		require_once("include/dba.php");
-	    $db = new dba($db_host, $db_user, $db_pass, $db_data);
+		$db = new dba($db_host, $db_user, $db_pass, $db_data);
-    	unset($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();
+				$pidfile->kill();
-                                logger("cronhooks: killed stale process");
+				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"]);
+	cronhooks_run($_SERVER["argv"],$_SERVER["argc"]);
-  killme();
+	killme();
 }

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

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;
 }
 
 //

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

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))
+	if(($contact['network'] === NETWORK_OSTATUS) || ($contact['network'] === NETWORK_DIASPORA) || ($contact['network'] === NETWORK_DFRN)) {
-		update_contact($contact["id"]);
+		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) {
 

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));
+			$pid = trim(@file_get_contents($this->_file));
-			if (posix_kill($pid, 0)) {
+			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() {

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");
+		@include(".htconfig.php");
-    	require_once("include/dba.php");
+		require_once("include/dba.php");
-	    $db = new dba($db_host, $db_user, $db_pass, $db_data);
+		$db = new dba($db_host, $db_user, $db_pass, $db_data);
-    	unset($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(($argc <= 1) OR ($argv[1] != "no_cron")) {
-	if ($lockpath != '') {
+		// Run the cron job that calls all other jobs
-		$pidfile = new pidfile($lockpath, 'poller');
+		proc_run("php","include/cron.php");
-		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;
-		}
-	}
 
+		// 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();
+	// Checking number of workers
-
+	if (poller_too_much_workers())
-	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)) {
 		return;
-	}
 
-	foreach($contacts as $c) {
+	$starttime = time();
 
-		$res = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1",
+	while ($r = q("SELECT * FROM `workerqueue` WHERE `executed` = '0000-00-00 00:00:00' ORDER BY `created` LIMIT 1")) {
-			intval($c['id'])
-		);
 
-		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();
 }
+?>

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']);
+			logger('queue: skipping known dead url: ' . $c[0]['notify']);
-				update_queue_time($q_item['id']);
+			update_queue_time($q_item['id']);
-				continue;
+			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;
 

update.php

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