[tt-rss.git] / update_daemon2.php

#!/usr/bin/php
<?php
	// This is an experimental multiprocess update daemon.
	// Some configurable variable may be found below.

	// define('DEFAULT_ERROR_LEVEL', E_ALL);
	define('DEFAULT_ERROR_LEVEL', E_ERROR | E_WARNING | E_PARSE);

	declare(ticks = 1);

	define('MAGPIE_CACHE_DIR', '/var/tmp/magpie-ttrss-cache-daemon');
	define('SIMPLEPIE_CACHE_DIR',	'/var/tmp/simplepie-ttrss-cache-daemon');
	define('DISABLE_SESSIONS', true);

	define('MAX_JOBS', 2);

	require_once "version.php";

	if (strpos(VERSION, ".99") !== false || getenv('DAEMON_XDEBUG')) {
		define('DAEMON_EXTENDED_DEBUG', true);
	}

	define('PURGE_INTERVAL', 3600); // seconds

	require_once "sanity_check.php";
	require_once "config.php";

	define('SPAWN_INTERVAL', DAEMON_SLEEP_INTERVAL);

	if (!ENABLE_UPDATE_DAEMON) {
		die("Please enable option ENABLE_UPDATE_DAEMON in config.php\n");
	}
	
	require_once "db.php";
	require_once "db-prefs.php";
	require_once "functions.php";
	require_once "magpierss/rss_fetch.inc";

	error_reporting(DEFAULT_ERROR_LEVEL);

	$children = array();

	$last_checkpoint = -1;

	function reap_children() {
		global $children;

		$tmp = array();

		foreach ($children as $pid) {
			if (pcntl_waitpid($pid, $status, WNOHANG) != $pid) {
				array_push($tmp, $pid);
			} else {
				_debug("[SIGCHLD] child $pid reaped.");
			}
		}

		$children = $tmp;

		return count($tmp);
	}

	function sigalrm_handler() {
		die("received SIGALRM, hang in feed update?\n");
	}

	function sigchld_handler($signal) {
		$running_jobs = reap_children();

		_debug("[SIGCHLD] jobs left: $running_jobs");

		pcntl_waitpid(-1, $status, WNOHANG);
	}

	function sigint_handler() {
		unlink(LOCK_DIRECTORY . "/update_daemon.lock");
		die("Received SIGINT. Exiting.\n");
	}

	pcntl_signal(SIGALRM, 'sigalrm_handler');
	pcntl_signal(SIGCHLD, 'sigchld_handler');

	if (file_is_locked("update_daemon.lock")) {
		die("error: Can't create lockfile. ".
			"Maybe another daemon is already running.\n");
	}

	if (file_is_locked("update_daemon.lock")) {
		die("error: Can't create lockfile. ".
			"Maybe another daemon is already running.\n");
	}

	if (!pcntl_fork()) {
		pcntl_signal(SIGINT, 'sigint_handler');

		$lock_handle = make_lockfile("update_daemon.lock");

		if (!$lock_handle) {
			die("error: Can't create lockfile. ".
				"Maybe another daemon is already running.\n");
		}

		while (true) { sleep(100); }
	}

	// Testing database connection.
	// It is unnecessary to start the fork loop if database is not ok.
	$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);	

	if (!$link) {
		if (DB_TYPE == "mysql") {
			print mysql_error();
		}
		// PG seems to display its own errors just fine by default.		
		return;
	}

	db_close($link);


	while (true) {

		$next_spawn = $last_checkpoint + SPAWN_INTERVAL - time();

		if ($next_spawn % 10 == 0) {
			$running_jobs = count($children);
			_debug("[MASTER] active jobs: $running_jobs, next spawn at $next_spawn sec.");
		}

		if ($last_checkpoint + SPAWN_INTERVAL < time()) {

			reap_children();

			for ($j = count($children); $j < MAX_JOBS; $j++) {
				$pid = pcntl_fork();
				if ($pid == -1) {
					die("fork failed!\n");
				} else if ($pid) {
					_debug("[MASTER] spawned client $j [PID:$pid]...");
					array_push($children, $pid);
				} else {
					pcntl_signal(SIGCHLD, SIG_IGN);
					pcntl_signal(SIGINT, SIG_DFL);

					// ****** Updating RSS code *******
					// Only run in fork process.

					$start_timestamp = time();

					$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);	

					if (!$link) {
						if (DB_TYPE == "mysql") {
							print mysql_error();
						}
						// PG seems to display its own errors just fine by default.		
						return;
					}

					if (DB_TYPE == "pgsql") {
						pg_query("set client_encoding = 'utf-8'");
						pg_set_client_encoding("UNICODE");
					} else {
						if (defined('MYSQL_CHARSET') && MYSQL_CHARSET) {
							db_query($link, "SET NAMES " . MYSQL_CHARSET);
							// db_query($link, "SET CHARACTER SET " . MYSQL_CHARSET);
						}
					}

					// We disable stamp file, since it is of no use in a multiprocess update.
					// not really, tho for the time being -fox
					if (!make_stampfile('update_daemon.stamp')) {
						print "warning: unable to create stampfile";
					}	

					// $last_purge = 0;

					// if (time() - $last_purge > PURGE_INTERVAL) {

					// FIXME : $last_purge is of no use in a multiprocess update.
					// FIXME : We ALWAYS purge old posts.
					_debug("Purging old posts (random 30 feeds)...");
					global_purge_old_posts($link, true, 30);

					// 	$last_purge = time();
					// }

					// Process all other feeds using last_updated and interval parameters

					$random_qpart = sql_random_function();
						
					if (DAEMON_UPDATE_LOGIN_LIMIT > 0) {
						if (DB_TYPE == "pgsql") {
							$login_thresh_qpart = "AND ttrss_users.last_login >= NOW() - INTERVAL '".DAEMON_UPDATE_LOGIN_LIMIT." days'";
						} else {
							$login_thresh_qpart = "AND ttrss_users.last_login >= DATE_SUB(NOW(), INTERVAL ".DAEMON_UPDATE_LOGIN_LIMIT." DAY)";
						}			
					} else {
						$login_thresh_qpart = "";
					}

					//if (DB_TYPE == "pgsql") {
					//	$update_limit_qpart = "AND ttrss_feeds.last_updated < NOW() - INTERVAL '".(DAEMON_SLEEP_INTERVAL*2)." seconds'";
					//} else {
					//	$update_limit_qpart = "AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL ".(DAEMON_SLEEP_INTERVAL*2)." SECOND)";
					//}

					if (DB_TYPE == "pgsql") {
						$update_limit_qpart = "AND ttrss_feeds.last_updated < NOW() - INTERVAL '".(DAEMON_SLEEP_INTERVAL*2)." seconds'";
						$update_limit_qpart = "AND ((
								ttrss_feeds.update_interval = 0
								AND ttrss_feeds.last_updated < NOW() - INTERVAL ttrss_user_prefs.value || ' minutes'
							) OR (
								ttrss_feeds.update_interval > 0
								AND ttrss_feeds.last_updated < NOW() - INTERVAL ttrss_feeds.update_interval || ' minutes'
							))";
					} else {
						$update_limit_qpart = "AND ((
								ttrss_feeds.update_interval = 0
								AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL CONVERT(ttrss_user_prefs.value, SIGNED INTEGER) MINUTE)
							) OR (
								ttrss_feeds.update_interval > 0
								AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL ttrss_feeds.update_interval MINUTE)
							))";
					}


					if (DB_TYPE == "pgsql") {
						$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < NOW() - INTERVAL '120 seconds')";
					} else {
						$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < DATE_SUB(NOW(), INTERVAL 120 SECOND))";
					}

					$result = db_query($link, "SELECT ttrss_feeds.feed_url,ttrss_feeds.id, ttrss_feeds.owner_uid,
							SUBSTRING(ttrss_feeds.last_updated,1,19) AS last_updated,
							ttrss_feeds.update_interval 
						FROM 
							ttrss_feeds, ttrss_users, ttrss_user_prefs
						WHERE
							ttrss_feeds.owner_uid = ttrss_users.id
							AND ttrss_users.id = ttrss_user_prefs.owner_uid
							AND ttrss_user_prefs.pref_name='DEFAULT_UPDATE_INTERVAL'
							$login_thresh_qpart $update_limit_qpart
							 $updstart_thresh_qpart
						ORDER BY $random_qpart DESC LIMIT " . DAEMON_FEED_LIMIT);

					$user_prefs_cache = array();

					_debug(sprintf("Scheduled %d feeds to update...\n", db_num_rows($result)));

					// Here is a little cache magic in order to minimize risk of double feed updates.
					$feeds_to_update = array();
					while ($line = db_fetch_assoc($result)) {
						$feeds_to_update[$line['id']] = $line;
					}

					// We update the feed last update started date before anything else.
					// There is no lag due to feed contents downloads
					// It prevent an other process to update the same feed.
					$feed_ids = array_keys($feeds_to_update);
					if($feed_ids) {
						db_query($link, sprintf("UPDATE ttrss_feeds SET last_update_started = NOW()
							WHERE id IN (%s)", implode(',', $feed_ids)));
					}

					while ($line = array_pop($feeds_to_update)) {

						_debug("Feed: " . $line["feed_url"] . ", " . $line["last_updated"]);

						pcntl_alarm(300);
						update_rss_feed($link, $line["feed_url"], $line["id"], true);	
						pcntl_alarm(0);

						sleep(1); // prevent flood (FIXME make this an option?)
					}

					if (DAEMON_SENDS_DIGESTS) send_headlines_digests($link);

					_debug("Elapsed time: " . (time() - $start_timestamp) . " second(s)");

					db_close($link);

					// We are in a fork.
					// We wait a little before exiting to avoid to be faster than our parent process.
					sleep(1);
					// We exit in order to avoid fork bombing.
					exit(0);
				}

				// We wait a little time before the next fork, in order to let the first fork
				// mark the feeds it update :
				sleep(1);
			}
			$last_checkpoint = time();
		}
		sleep(1);
	}

?>
Commit	Line	Data
02008cb1 AD	1	#!/usr/bin/php
02008cb1 AD	2	<?php
ffa7cbae AD	3	// This is an experimental multiprocess update daemon.
	4	// Some configurable variable may be found below.
	5
	6	// define('DEFAULT_ERROR_LEVEL', E_ALL);
	7	define('DEFAULT_ERROR_LEVEL', E_ERROR \| E_WARNING \| E_PARSE);
02008cb1 AD	8
	9	declare(ticks = 1);
	10
ffa7cbae AD	11	define('MAGPIE_CACHE_DIR', '/var/tmp/magpie-ttrss-cache-daemon');
	12	define('SIMPLEPIE_CACHE_DIR', '/var/tmp/simplepie-ttrss-cache-daemon');
	13	define('DISABLE_SESSIONS', true);
02008cb1 AD	14
02008cb1 AD	15	define('MAX_JOBS', 2);
ffa7cbae AD	16
	17	require_once "version.php";
	18
010c16f1	19	if (strpos(VERSION, ".99") !== false \|\| getenv('DAEMON_XDEBUG')) {
ffa7cbae AD	20	define('DAEMON_EXTENDED_DEBUG', true);
	21	}
	22
	23	define('PURGE_INTERVAL', 3600); // seconds
	24
	25	require_once "sanity_check.php";
	26	require_once "config.php";
	27
02008cb1	28	define('SPAWN_INTERVAL', DAEMON_SLEEP_INTERVAL);
ffa7cbae AD	29
	30	if (!ENABLE_UPDATE_DAEMON) {
	31	die("Please enable option ENABLE_UPDATE_DAEMON in config.php\n");
	32	}
	33
	34	require_once "db.php";
	35	require_once "db-prefs.php";
	36	require_once "functions.php";
	37	require_once "magpierss/rss_fetch.inc";
	38
	39	error_reporting(DEFAULT_ERROR_LEVEL);
02008cb1	40
e9338405 AD	41	$children = array();
e9338405 AD	42
02008cb1 AD	43	$last_checkpoint = -1;
02008cb1 AD	44
5a613536	45	function reap_children() {
e9338405 AD	46	global $children;
	47
	48	$tmp = array();
	49
	50	foreach ($children as $pid) {
	51	if (pcntl_waitpid($pid, $status, WNOHANG) != $pid) {
	52	array_push($tmp, $pid);
	53	} else {
	54	_debug("[SIGCHLD] child $pid reaped.");
	55	}
	56	}
	57
	58	$children = $tmp;
	59
5a613536 AD	60	return count($tmp);
	61	}
	62
	63	function sigalrm_handler() {
	64	die("received SIGALRM, hang in feed update?\n");
	65	}
	66
	67	function sigchld_handler($signal) {
	68	$running_jobs = reap_children();
e9338405 AD	69
e9338405 AD	70	_debug("[SIGCHLD] jobs left: $running_jobs");
5a613536	71
02008cb1 AD	72	pcntl_waitpid(-1, $status, WNOHANG);
	73	}
	74
6a69e61f AD	75	function sigint_handler() {
	76	unlink(LOCK_DIRECTORY . "/update_daemon.lock");
	77	die("Received SIGINT. Exiting.\n");
	78	}
	79
ffa7cbae	80	pcntl_signal(SIGALRM, 'sigalrm_handler');
02008cb1	81	pcntl_signal(SIGCHLD, 'sigchld_handler');
6a69e61f	82
884c0a36 AD	83	if (file_is_locked("update_daemon.lock")) {
884c0a36 AD	84	die("error: Can't create lockfile. ".
6a69e61f AD	85	"Maybe another daemon is already running.\n");
6a69e61f AD	86	}
02008cb1	87
ffa7cbae AD	88	if (file_is_locked("update_daemon.lock")) {
	89	die("error: Can't create lockfile. ".
	90	"Maybe another daemon is already running.\n");
	91	}
	92
884c0a36	93	if (!pcntl_fork()) {
0d6a7147 AD	94	pcntl_signal(SIGINT, 'sigint_handler');
0d6a7147 AD	95
884c0a36 AD	96	$lock_handle = make_lockfile("update_daemon.lock");
	97
	98	if (!$lock_handle) {
	99	die("error: Can't create lockfile. ".
	100	"Maybe another daemon is already running.\n");
	101	}
	102
	103	while (true) { sleep(100); }
	104	}
	105
ffa7cbae AD	106	// Testing database connection.
	107	// It is unnecessary to start the fork loop if database is not ok.
	108	$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);
	109
	110	if (!$link) {
	111	if (DB_TYPE == "mysql") {
	112	print mysql_error();
	113	}
	114	// PG seems to display its own errors just fine by default.
	115	return;
	116	}
	117
	118	db_close($link);
	119
	120
02008cb1 AD	121	while (true) {
	122
	123	$next_spawn = $last_checkpoint + SPAWN_INTERVAL - time();
	124
ce1aa9b7	125	if ($next_spawn % 10 == 0) {
e9338405 AD	126	$running_jobs = count($children);
e9338405 AD	127	_debug("[MASTER] active jobs: $running_jobs, next spawn at $next_spawn sec.");
ce1aa9b7	128	}
02008cb1 AD	129
	130	if ($last_checkpoint + SPAWN_INTERVAL < time()) {
	131
5a613536 AD	132	reap_children();
5a613536 AD	133
e9338405	134	for ($j = count($children); $j < MAX_JOBS; $j++) {
02008cb1 AD	135	$pid = pcntl_fork();
	136	if ($pid == -1) {
	137	die("fork failed!\n");
	138	} else if ($pid) {
e9338405 AD	139	_debug("[MASTER] spawned client $j [PID:$pid]...");
e9338405 AD	140	array_push($children, $pid);
02008cb1 AD	141	} else {
02008cb1 AD	142	pcntl_signal(SIGCHLD, SIG_IGN);
6a69e61f	143	pcntl_signal(SIGINT, SIG_DFL);
ffa7cbae AD	144
	145	// **** Updating RSS code *****
	146	// Only run in fork process.
	147
	148	$start_timestamp = time();
	149
	150	$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);
	151
	152	if (!$link) {
	153	if (DB_TYPE == "mysql") {
	154	print mysql_error();
	155	}
	156	// PG seems to display its own errors just fine by default.
	157	return;
	158	}
	159
	160	if (DB_TYPE == "pgsql") {
	161	pg_query("set client_encoding = 'utf-8'");
	162	pg_set_client_encoding("UNICODE");
	163	} else {
	164	if (defined('MYSQL_CHARSET') && MYSQL_CHARSET) {
	165	db_query($link, "SET NAMES " . MYSQL_CHARSET);
	166	// db_query($link, "SET CHARACTER SET " . MYSQL_CHARSET);
	167	}
	168	}
	169
	170	// We disable stamp file, since it is of no use in a multiprocess update.
	171	// not really, tho for the time being -fox
	172	if (!make_stampfile('update_daemon.stamp')) {
	173	print "warning: unable to create stampfile";
	174	}
	175
	176	// $last_purge = 0;
	177
	178	// if (time() - $last_purge > PURGE_INTERVAL) {
	179
	180	// FIXME : $last_purge is of no use in a multiprocess update.
	181	// FIXME : We ALWAYS purge old posts.
	182	_debug("Purging old posts (random 30 feeds)...");
	183	global_purge_old_posts($link, true, 30);
	184
	185	// $last_purge = time();
	186	// }
	187
	188	// Process all other feeds using last_updated and interval parameters
	189
	190	$random_qpart = sql_random_function();
	191
	192	if (DAEMON_UPDATE_LOGIN_LIMIT > 0) {
	193	if (DB_TYPE == "pgsql") {
	194	$login_thresh_qpart = "AND ttrss_users.last_login >= NOW() - INTERVAL '".DAEMON_UPDATE_LOGIN_LIMIT." days'";
	195	} else {
	196	$login_thresh_qpart = "AND ttrss_users.last_login >= DATE_SUB(NOW(), INTERVAL ".DAEMON_UPDATE_LOGIN_LIMIT." DAY)";
	197	}
	198	} else {
	199	$login_thresh_qpart = "";
	200	}
	201
0d6a7147 AD	202	//if (DB_TYPE == "pgsql") {
	203	// $update_limit_qpart = "AND ttrss_feeds.last_updated < NOW() - INTERVAL '".(DAEMON_SLEEP_INTERVAL*2)." seconds'";
	204	//} else {
	205	// $update_limit_qpart = "AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL ".(DAEMON_SLEEP_INTERVAL*2)." SECOND)";
	206	//}
	207
ffa7cbae AD	208	if (DB_TYPE == "pgsql") {
ffa7cbae AD	209	$update_limit_qpart = "AND ttrss_feeds.last_updated < NOW() - INTERVAL '".(DAEMON_SLEEP_INTERVAL*2)." seconds'";
0d6a7147 AD	210	$update_limit_qpart = "AND ((
	211	ttrss_feeds.update_interval = 0
	212	AND ttrss_feeds.last_updated < NOW() - INTERVAL ttrss_user_prefs.value \|\| ' minutes'
	213	) OR (
	214	ttrss_feeds.update_interval > 0
	215	AND ttrss_feeds.last_updated < NOW() - INTERVAL ttrss_feeds.update_interval \|\| ' minutes'
	216	))";
ffa7cbae	217	} else {
0d6a7147 AD	218	$update_limit_qpart = "AND ((
	219	ttrss_feeds.update_interval = 0
	220	AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL CONVERT(ttrss_user_prefs.value, SIGNED INTEGER) MINUTE)
	221	) OR (
	222	ttrss_feeds.update_interval > 0
	223	AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL ttrss_feeds.update_interval MINUTE)
	224	))";
ffa7cbae AD	225	}
ffa7cbae AD	226
0d6a7147	227
ffa7cbae	228	if (DB_TYPE == "pgsql") {
0d6a7147 AD	229	$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < NOW() - INTERVAL '120 seconds')";
	230	} else {
	231	$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < DATE_SUB(NOW(), INTERVAL 120 SECOND))";
	232	}
ffa7cbae	233
0d6a7147 AD	234	$result = db_query($link, "SELECT ttrss_feeds.feed_url,ttrss_feeds.id, ttrss_feeds.owner_uid,
	235	SUBSTRING(ttrss_feeds.last_updated,1,19) AS last_updated,
	236	ttrss_feeds.update_interval
ffa7cbae	237	FROM
0d6a7147 AD	238	ttrss_feeds, ttrss_users, ttrss_user_prefs
	239	WHERE
	240	ttrss_feeds.owner_uid = ttrss_users.id
	241	AND ttrss_users.id = ttrss_user_prefs.owner_uid
	242	AND ttrss_user_prefs.pref_name='DEFAULT_UPDATE_INTERVAL'
	243	$login_thresh_qpart $update_limit_qpart
	244	$updstart_thresh_qpart
ffa7cbae AD	245	ORDER BY $random_qpart DESC LIMIT " . DAEMON_FEED_LIMIT);
	246
	247	$user_prefs_cache = array();
	248
	249	_debug(sprintf("Scheduled %d feeds to update...\n", db_num_rows($result)));
	250
	251	// Here is a little cache magic in order to minimize risk of double feed updates.
	252	$feeds_to_update = array();
	253	while ($line = db_fetch_assoc($result)) {
	254	$feeds_to_update[$line['id']] = $line;
	255	}
	256
	257	// We update the feed last update started date before anything else.
	258	// There is no lag due to feed contents downloads
	259	// It prevent an other process to update the same feed.
	260	$feed_ids = array_keys($feeds_to_update);
	261	if($feed_ids) {
	262	db_query($link, sprintf("UPDATE ttrss_feeds SET last_update_started = NOW()
	263	WHERE id IN (%s)", implode(',', $feed_ids)));
	264	}
	265
	266	while ($line = array_pop($feeds_to_update)) {
	267
ffa7cbae AD	268	_debug("Feed: " . $line["feed_url"] . ", " . $line["last_updated"]);
ffa7cbae AD	269
0d6a7147 AD	270	pcntl_alarm(300);
	271	update_rss_feed($link, $line["feed_url"], $line["id"], true);
	272	pcntl_alarm(0);
ffa7cbae	273
0d6a7147	274	sleep(1); // prevent flood (FIXME make this an option?)
ffa7cbae AD	275	}
	276
	277	if (DAEMON_SENDS_DIGESTS) send_headlines_digests($link);
	278
0d6a7147	279	_debug("Elapsed time: " . (time() - $start_timestamp) . " second(s)");
ffa7cbae AD	280
	281	db_close($link);
	282
	283	// We are in a fork.
	284	// We wait a little before exiting to avoid to be faster than our parent process.
	285	sleep(1);
	286	// We exit in order to avoid fork bombing.
02008cb1 AD	287	exit(0);
02008cb1 AD	288	}
0d6a7147 AD	289
	290	// We wait a little time before the next fork, in order to let the first fork
	291	// mark the feeds it update :
	292	sleep(1);
02008cb1 AD	293	}
	294	$last_checkpoint = time();
	295	}
	296	sleep(1);
	297	}
	298
	299	?>