[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("[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("[SIGINT] removing lockfile and 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 (!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.update_interval = 0
								AND ttrss_feeds.last_updated < NOW() - CAST((ttrss_user_prefs.value || ' minutes') AS INTERVAL)
							) OR (
								ttrss_feeds.update_interval > 0
								AND ttrss_feeds.last_updated < NOW() - CAST((ttrss_feeds.update_interval || ' minutes') AS INTERVAL)
							))";
					} 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() {
a65a05a7	64	die("[SIGALRM] hang in feed update?\n");
5a613536 AD	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() {
6a69e61f AD	76	unlink(LOCK_DIRECTORY . "/update_daemon.lock");
a65a05a7	77	die("[SIGINT] removing lockfile and exiting.\n");
6a69e61f AD	78	}
6a69e61f AD	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
884c0a36	88	if (!pcntl_fork()) {
0d6a7147 AD	89	pcntl_signal(SIGINT, 'sigint_handler');
0d6a7147 AD	90
884c0a36 AD	91	$lock_handle = make_lockfile("update_daemon.lock");
	92
	93	if (!$lock_handle) {
	94	die("error: Can't create lockfile. ".
	95	"Maybe another daemon is already running.\n");
	96	}
	97
	98	while (true) { sleep(100); }
	99	}
	100
ffa7cbae AD	101	// Testing database connection.
	102	// It is unnecessary to start the fork loop if database is not ok.
	103	$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);
	104
	105	if (!$link) {
	106	if (DB_TYPE == "mysql") {
	107	print mysql_error();
	108	}
	109	// PG seems to display its own errors just fine by default.
	110	return;
	111	}
	112
	113	db_close($link);
	114
	115
02008cb1 AD	116	while (true) {
	117
	118	$next_spawn = $last_checkpoint + SPAWN_INTERVAL - time();
	119
ce1aa9b7	120	if ($next_spawn % 10 == 0) {
e9338405 AD	121	$running_jobs = count($children);
e9338405 AD	122	_debug("[MASTER] active jobs: $running_jobs, next spawn at $next_spawn sec.");
ce1aa9b7	123	}
02008cb1 AD	124
	125	if ($last_checkpoint + SPAWN_INTERVAL < time()) {
	126
5a613536 AD	127	reap_children();
5a613536 AD	128
e9338405	129	for ($j = count($children); $j < MAX_JOBS; $j++) {
02008cb1 AD	130	$pid = pcntl_fork();
	131	if ($pid == -1) {
	132	die("fork failed!\n");
	133	} else if ($pid) {
e9338405 AD	134	_debug("[MASTER] spawned client $j [PID:$pid]...");
e9338405 AD	135	array_push($children, $pid);
02008cb1 AD	136	} else {
02008cb1 AD	137	pcntl_signal(SIGCHLD, SIG_IGN);
6a69e61f	138	pcntl_signal(SIGINT, SIG_DFL);
ffa7cbae AD	139
	140	// **** Updating RSS code *****
	141	// Only run in fork process.
	142
	143	$start_timestamp = time();
	144
	145	$link = db_connect(DB_HOST, DB_USER, DB_PASS, DB_NAME);
	146
	147	if (!$link) {
	148	if (DB_TYPE == "mysql") {
	149	print mysql_error();
	150	}
	151	// PG seems to display its own errors just fine by default.
	152	return;
	153	}
	154
	155	if (DB_TYPE == "pgsql") {
	156	pg_query("set client_encoding = 'utf-8'");
	157	pg_set_client_encoding("UNICODE");
	158	} else {
	159	if (defined('MYSQL_CHARSET') && MYSQL_CHARSET) {
	160	db_query($link, "SET NAMES " . MYSQL_CHARSET);
	161	// db_query($link, "SET CHARACTER SET " . MYSQL_CHARSET);
	162	}
	163	}
	164
	165	// We disable stamp file, since it is of no use in a multiprocess update.
	166	// not really, tho for the time being -fox
	167	if (!make_stampfile('update_daemon.stamp')) {
	168	print "warning: unable to create stampfile";
	169	}
	170
	171	// $last_purge = 0;
	172
	173	// if (time() - $last_purge > PURGE_INTERVAL) {
	174
	175	// FIXME : $last_purge is of no use in a multiprocess update.
	176	// FIXME : We ALWAYS purge old posts.
	177	_debug("Purging old posts (random 30 feeds)...");
	178	global_purge_old_posts($link, true, 30);
	179
	180	// $last_purge = time();
	181	// }
	182
	183	// Process all other feeds using last_updated and interval parameters
	184
	185	$random_qpart = sql_random_function();
	186
	187	if (DAEMON_UPDATE_LOGIN_LIMIT > 0) {
	188	if (DB_TYPE == "pgsql") {
	189	$login_thresh_qpart = "AND ttrss_users.last_login >= NOW() - INTERVAL '".DAEMON_UPDATE_LOGIN_LIMIT." days'";
	190	} else {
	191	$login_thresh_qpart = "AND ttrss_users.last_login >= DATE_SUB(NOW(), INTERVAL ".DAEMON_UPDATE_LOGIN_LIMIT." DAY)";
	192	}
	193	} else {
	194	$login_thresh_qpart = "";
	195	}
	196
	197	if (DB_TYPE == "pgsql") {
0d6a7147 AD	198	$update_limit_qpart = "AND ((
0d6a7147 AD	199	ttrss_feeds.update_interval = 0
2240601c	200	AND ttrss_feeds.last_updated < NOW() - CAST((ttrss_user_prefs.value \|\| ' minutes') AS INTERVAL)
0d6a7147 AD	201	) OR (
0d6a7147 AD	202	ttrss_feeds.update_interval > 0
2240601c	203	AND ttrss_feeds.last_updated < NOW() - CAST((ttrss_feeds.update_interval \|\| ' minutes') AS INTERVAL)
0d6a7147	204	))";
ffa7cbae	205	} else {
0d6a7147 AD	206	$update_limit_qpart = "AND ((
	207	ttrss_feeds.update_interval = 0
	208	AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL CONVERT(ttrss_user_prefs.value, SIGNED INTEGER) MINUTE)
	209	) OR (
	210	ttrss_feeds.update_interval > 0
	211	AND ttrss_feeds.last_updated < DATE_SUB(NOW(), INTERVAL ttrss_feeds.update_interval MINUTE)
	212	))";
ffa7cbae AD	213	}
ffa7cbae AD	214
0d6a7147	215
ffa7cbae	216	if (DB_TYPE == "pgsql") {
0d6a7147 AD	217	$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < NOW() - INTERVAL '120 seconds')";
	218	} else {
	219	$updstart_thresh_qpart = "AND (ttrss_feeds.last_update_started IS NULL OR ttrss_feeds.last_update_started < DATE_SUB(NOW(), INTERVAL 120 SECOND))";
	220	}
ffa7cbae	221
0d6a7147 AD	222	$result = db_query($link, "SELECT ttrss_feeds.feed_url,ttrss_feeds.id, ttrss_feeds.owner_uid,
	223	SUBSTRING(ttrss_feeds.last_updated,1,19) AS last_updated,
	224	ttrss_feeds.update_interval
ffa7cbae	225	FROM
0d6a7147 AD	226	ttrss_feeds, ttrss_users, ttrss_user_prefs
	227	WHERE
	228	ttrss_feeds.owner_uid = ttrss_users.id
	229	AND ttrss_users.id = ttrss_user_prefs.owner_uid
2240601c	230	AND ttrss_user_prefs.pref_name = 'DEFAULT_UPDATE_INTERVAL'
0d6a7147 AD	231	$login_thresh_qpart $update_limit_qpart
0d6a7147 AD	232	$updstart_thresh_qpart
ffa7cbae AD	233	ORDER BY $random_qpart DESC LIMIT " . DAEMON_FEED_LIMIT);
	234
	235	$user_prefs_cache = array();
	236
	237	_debug(sprintf("Scheduled %d feeds to update...\n", db_num_rows($result)));
	238
	239	// Here is a little cache magic in order to minimize risk of double feed updates.
	240	$feeds_to_update = array();
	241	while ($line = db_fetch_assoc($result)) {
	242	$feeds_to_update[$line['id']] = $line;
	243	}
	244
	245	// We update the feed last update started date before anything else.
	246	// There is no lag due to feed contents downloads
	247	// It prevent an other process to update the same feed.
	248	$feed_ids = array_keys($feeds_to_update);
	249	if($feed_ids) {
	250	db_query($link, sprintf("UPDATE ttrss_feeds SET last_update_started = NOW()
	251	WHERE id IN (%s)", implode(',', $feed_ids)));
	252	}
	253
	254	while ($line = array_pop($feeds_to_update)) {
	255
ffa7cbae AD	256	_debug("Feed: " . $line["feed_url"] . ", " . $line["last_updated"]);
ffa7cbae AD	257
0d6a7147 AD	258	pcntl_alarm(300);
	259	update_rss_feed($link, $line["feed_url"], $line["id"], true);
	260	pcntl_alarm(0);
ffa7cbae	261
0d6a7147	262	sleep(1); // prevent flood (FIXME make this an option?)
ffa7cbae AD	263	}
	264
	265	if (DAEMON_SENDS_DIGESTS) send_headlines_digests($link);
	266
0d6a7147	267	_debug("Elapsed time: " . (time() - $start_timestamp) . " second(s)");
abfa57fd	268
ffa7cbae AD	269	db_close($link);
	270
	271	// We are in a fork.
	272	// We wait a little before exiting to avoid to be faster than our parent process.
	273	sleep(1);
	274	// We exit in order to avoid fork bombing.
02008cb1 AD	275	exit(0);
02008cb1 AD	276	}
0d6a7147 AD	277
	278	// We wait a little time before the next fork, in order to let the first fork
	279	// mark the feeds it update :
	280	sleep(1);
02008cb1 AD	281	}
	282	$last_checkpoint = time();
	283	}
	284	sleep(1);
	285	}
	286
	287	?>