2010-07-18 23:49:54 -04:00
< ? php
2017-04-30 00:01:26 -04:00
2017-04-30 00:07:00 -04:00
use Friendica\App ;
2017-04-30 00:01:26 -04:00
use Friendica\Core\Config ;
2015-02-19 04:45:46 -05:00
if ( ! file_exists ( " boot.php " ) AND ( sizeof ( $_SERVER [ " argv " ]) != 0 )) {
$directory = dirname ( $_SERVER [ " argv " ][ 0 ]);
2015-02-19 04:26:49 -05:00
2015-02-19 04:45:46 -05:00
if ( substr ( $directory , 0 , 1 ) != " / " )
$directory = $_SERVER [ " PWD " ] . " / " . $directory ;
2015-02-19 04:26:49 -05:00
2015-02-19 04:45:46 -05:00
$directory = realpath ( $directory . " /.. " );
2015-02-19 04:26:49 -05:00
2015-02-19 04:45:46 -05:00
chdir ( $directory );
}
2011-04-16 02:40:43 -04:00
2011-01-28 08:04:18 -05:00
require_once ( " boot.php " );
2016-11-26 19:55:05 -05:00
function poller_run ( $argv , $argc ){
2011-03-04 23:55:32 -05:00
global $a , $db ;
2011-01-28 08:04:18 -05:00
2017-05-02 22:42:42 -04:00
if ( is_null ( $a )) {
$a = new App ( dirname ( __DIR__ ));
2011-03-04 23:55:32 -05:00
}
2013-01-20 08:08:28 -05:00
2017-04-04 13:48:13 -04:00
if ( is_null ( $db )) {
2015-09-10 17:32:56 -04:00
@ 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 );
};
2013-01-20 08:08:28 -05:00
2017-02-25 08:59:24 -05:00
Config :: load ();
2016-10-02 09:52:52 -04:00
// Quit when in maintenance
2017-01-21 01:06:29 -05:00
if ( Config :: get ( 'system' , 'maintenance' , true )) {
2016-10-02 09:52:52 -04:00
return ;
2017-01-21 01:06:29 -05:00
}
2016-10-02 09:52:52 -04:00
2017-02-26 18:16:49 -05:00
$a -> set_baseurl ( Config :: get ( 'system' , 'url' ));
load_hooks ();
2016-09-09 16:33:54 -04:00
$a -> start_process ();
2017-04-29 13:36:46 -04:00
if ( $a -> min_memory_reached ()) {
return ;
}
2016-11-30 14:24:58 -05:00
if ( poller_max_connections_reached ()) {
2016-07-02 03:31:28 -04:00
return ;
2016-11-30 14:24:58 -05:00
}
2016-06-01 15:54:02 -04:00
2016-12-13 04:16:36 -05:00
if ( $a -> maxload_reached ()) {
2016-02-09 00:42:00 -05:00
return ;
2016-11-30 14:24:58 -05:00
}
2017-04-04 13:48:13 -04:00
if (( $argc <= 1 ) OR ( $argv [ 1 ] != " no_cron " )) {
2016-11-30 14:24:58 -05:00
poller_run_cron ();
}
2016-02-09 00:42:00 -05:00
2016-11-30 14:24:58 -05:00
if ( $a -> max_processes_reached ()) {
2016-03-08 14:28:09 -05:00
return ;
2016-11-30 14:24:58 -05:00
}
2015-09-23 02:56:48 -04:00
2015-12-06 10:40:31 -05:00
// Checking the number of workers
2016-07-24 14:25:11 -04:00
if ( poller_too_much_workers ()) {
2016-01-23 20:53:21 -05:00
poller_kill_stale_workers ();
2015-12-06 10:40:31 -05:00
return ;
2016-01-23 20:53:21 -05:00
}
2015-12-06 10:40:31 -05:00
2015-09-25 11:38:56 -04:00
$starttime = time ();
2016-10-10 17:01:36 -04:00
while ( $r = poller_worker_process ()) {
2015-09-23 02:56:48 -04:00
2017-04-29 13:45:09 -04:00
// Check free memory
if ( $a -> min_memory_reached ()) {
return ;
}
2016-11-26 19:55:05 -05:00
// Count active workers and compare them with a maximum value that depends on the load
if ( poller_too_much_workers ()) {
2016-10-02 09:52:52 -04:00
return ;
2016-11-26 19:55:05 -05:00
}
2016-10-02 09:52:52 -04:00
2016-11-26 19:55:05 -05:00
if ( ! poller_execute ( $r [ 0 ])) {
2016-07-02 03:31:28 -04:00
return ;
2016-11-26 19:55:05 -05:00
}
2016-06-01 15:54:02 -04:00
2016-11-26 19:55:05 -05:00
// Quit the poller once every hour
if ( time () > ( $starttime + 3600 ))
2016-02-11 15:39:34 -05:00
return ;
2016-11-26 19:55:05 -05:00
}
}
/**
* @ brief Execute a worker entry
*
* @ param array $queue Workerqueue entry
*
* @ return boolean " true " if further processing should be stopped
*/
function poller_execute ( $queue ) {
2015-12-15 18:14:53 -05:00
2016-11-26 19:55:05 -05:00
$a = get_app ();
2011-04-16 02:40:43 -04:00
2016-11-26 19:55:05 -05:00
$mypid = getmypid ();
2016-10-17 14:38:51 -04:00
2016-11-26 19:55:05 -05:00
// Quit when in maintenance
2017-01-21 01:06:29 -05:00
if ( Config :: get ( 'system' , 'maintenance' , true )) {
2016-11-26 19:55:05 -05:00
return false ;
}
// Constantly check the number of parallel database processes
if ( $a -> max_processes_reached ()) {
return false ;
}
// Constantly check the number of available database connections to let the frontend be accessible at any time
if ( poller_max_connections_reached ()) {
return false ;
}
2017-05-12 12:09:25 -04:00
if ( ! dba :: update ( 'workerqueue' , array ( 'executed' => datetime_convert (), 'pid' => $mypid ),
array ( 'id' => $queue [ " id " ], 'pid' => 0 ))) {
2016-11-26 19:55:05 -05:00
logger ( " Couldn't update queue entry " . $queue [ " id " ] . " - skip this execution " , LOGGER_DEBUG );
2017-05-11 16:13:45 -04:00
dba :: commit ();
2016-11-26 19:55:05 -05:00
return true ;
}
2015-09-28 15:58:58 -04:00
2016-11-26 19:55:05 -05:00
// Assure that there are no tasks executed twice
$id = q ( " SELECT `pid`, `executed` FROM `workerqueue` WHERE `id` = %d " , intval ( $queue [ " id " ]));
if ( ! $id ) {
logger ( " Queue item " . $queue [ " id " ] . " vanished - skip this execution " , LOGGER_DEBUG );
2017-05-11 16:13:45 -04:00
dba :: commit ();
2016-11-26 19:55:05 -05:00
return true ;
} elseif (( strtotime ( $id [ 0 ][ " executed " ]) <= 0 ) OR ( $id [ 0 ][ " pid " ] == 0 )) {
logger ( " Entry for queue item " . $queue [ " id " ] . " wasn't stored - skip this execution " , LOGGER_DEBUG );
2017-05-11 16:13:45 -04:00
dba :: commit ();
2016-11-26 19:55:05 -05:00
return true ;
} elseif ( $id [ 0 ][ " pid " ] != $mypid ) {
logger ( " Queue item " . $queue [ " id " ] . " is to be executed by process " . $id [ 0 ][ " pid " ] . " and not by me ( " . $mypid . " ) - skip this execution " , LOGGER_DEBUG );
2017-05-11 16:13:45 -04:00
dba :: commit ();
2016-11-26 19:55:05 -05:00
return true ;
}
2017-05-11 16:13:45 -04:00
dba :: commit ();
2016-07-24 14:25:11 -04:00
2016-11-26 19:55:05 -05:00
$argv = json_decode ( $queue [ " parameter " ]);
2016-07-24 14:25:11 -04:00
2016-11-26 19:55:05 -05:00
// Check for existance and validity of the include file
$include = $argv [ 0 ];
2015-09-13 12:47:10 -04:00
2016-11-26 19:55:05 -05:00
if ( ! validate_include ( $include )) {
logger ( " Include file " . $argv [ 0 ] . " is not valid! " );
2017-05-12 12:09:25 -04:00
dba :: delete ( 'workerqueue' , array ( 'id' => $queue [ " id " ]));
2016-11-26 19:55:05 -05:00
return true ;
}
2010-12-21 17:51:26 -05:00
2016-11-26 19:55:05 -05:00
require_once ( $include );
2016-07-24 14:25:11 -04:00
2016-11-26 19:55:05 -05:00
$funcname = str_replace ( " .php " , " " , basename ( $argv [ 0 ])) . " _run " ;
2016-10-07 02:05:43 -04:00
2016-11-26 19:55:05 -05:00
if ( function_exists ( $funcname )) {
2017-01-19 12:06:23 -05:00
2017-01-21 14:50:56 -05:00
poller_exec_function ( $queue , $funcname , $argv );
2017-05-12 12:09:25 -04:00
dba :: delete ( 'workerqueue' , array ( 'id' => $queue [ " id " ]));
2017-01-21 14:50:56 -05:00
} else {
logger ( " Function " . $funcname . " does not exist " );
}
2017-01-20 16:58:53 -05:00
2017-01-21 14:50:56 -05:00
return true ;
}
2015-09-11 15:35:58 -04:00
2017-01-21 14:50:56 -05:00
/**
* @ brief Execute a function from the queue
*
* @ param array $queue Workerqueue entry
* @ param string $funcname name of the function
2017-01-27 10:13:37 -05:00
* @ param array $argv Array of values to be passed to the function
2017-01-21 14:50:56 -05:00
*/
function poller_exec_function ( $queue , $funcname , $argv ) {
2016-10-07 02:05:43 -04:00
2017-01-21 14:50:56 -05:00
$a = get_app ();
2016-07-13 13:43:16 -04:00
2017-01-21 14:50:56 -05:00
$mypid = getmypid ();
2017-01-19 12:06:23 -05:00
2017-01-21 14:50:56 -05:00
$argc = count ( $argv );
2015-09-29 00:15:26 -04:00
2017-01-21 14:50:56 -05:00
logger ( " Process " . $mypid . " - Prio " . $queue [ " priority " ] . " - ID " . $queue [ " id " ] . " : " . $funcname . " " . $queue [ " parameter " ]);
$stamp = ( float ) microtime ( true );
2017-01-21 15:15:49 -05:00
// We use the callstack here to analyze the performance of executed worker entries.
// For this reason the variables have to be initialized.
2017-01-21 14:50:56 -05:00
if ( Config :: get ( " system " , " profiler " )) {
$a -> performance [ " start " ] = microtime ( true );
$a -> performance [ " database " ] = 0 ;
$a -> performance [ " database_write " ] = 0 ;
$a -> performance [ " network " ] = 0 ;
$a -> performance [ " file " ] = 0 ;
$a -> performance [ " rendering " ] = 0 ;
$a -> performance [ " parser " ] = 0 ;
$a -> performance [ " marktime " ] = 0 ;
$a -> performance [ " markstart " ] = microtime ( true );
$a -> callstack = array ();
}
2017-01-20 16:58:53 -05:00
2017-01-21 14:50:56 -05:00
// For better logging create a new process id for every worker call
// But preserve the old one for the worker
$old_process_id = $a -> process_id ;
$a -> process_id = uniqid ( " wrk " , true );
$funcname ( $argv , $argc );
$a -> process_id = $old_process_id ;
$duration = number_format ( microtime ( true ) - $stamp , 3 );
2017-01-31 14:39:09 -05:00
if ( $duration > 3600 ) {
logger ( " Prio " . $queue [ " priority " ] . " : " . $queue [ " parameter " ] . " - longer than 1 hour ( " . round ( $duration / 60 , 3 ) . " ) " , LOGGER_DEBUG );
} elseif ( $duration > 600 ) {
logger ( " Prio " . $queue [ " priority " ] . " : " . $queue [ " parameter " ] . " - longer than 10 minutes ( " . round ( $duration / 60 , 3 ) . " ) " , LOGGER_DEBUG );
} elseif ( $duration > 300 ) {
logger ( " Prio " . $queue [ " priority " ] . " : " . $queue [ " parameter " ] . " - longer than 5 minutes ( " . round ( $duration / 60 , 3 ) . " ) " , LOGGER_DEBUG );
} elseif ( $duration > 120 ) {
logger ( " Prio " . $queue [ " priority " ] . " : " . $queue [ " parameter " ] . " - longer than 2 minutes ( " . round ( $duration / 60 , 3 ) . " ) " , LOGGER_DEBUG );
}
2017-01-21 14:50:56 -05:00
logger ( " Process " . $mypid . " - Prio " . $queue [ " priority " ] . " - ID " . $queue [ " id " ] . " : " . $funcname . " - done in " . $duration . " seconds. " );
2017-01-21 15:15:49 -05:00
// Write down the performance values into the log
2017-01-21 14:50:56 -05:00
if ( Config :: get ( " system " , " profiler " )) {
$duration = microtime ( true ) - $a -> performance [ " start " ];
if ( Config :: get ( " rendertime " , " callstack " )) {
if ( isset ( $a -> callstack [ " database " ])) {
2017-01-21 01:16:15 -05:00
$o = " \n Database Read: \n " ;
2017-01-21 00:05:41 -05:00
foreach ( $a -> callstack [ " database " ] AS $func => $time ) {
$time = round ( $time , 3 );
if ( $time > 0 )
$o .= $func . " : " . $time . " \n " ;
}
2017-01-21 14:50:56 -05:00
}
if ( isset ( $a -> callstack [ " database_write " ])) {
2017-01-21 00:05:41 -05:00
$o .= " \n Database Write: \n " ;
foreach ( $a -> callstack [ " database_write " ] AS $func => $time ) {
$time = round ( $time , 3 );
if ( $time > 0 )
$o .= $func . " : " . $time . " \n " ;
}
2017-01-21 14:50:56 -05:00
}
if ( isset ( $a -> callstack [ " network " ])) {
2017-01-21 00:05:41 -05:00
$o .= " \n Network: \n " ;
foreach ( $a -> callstack [ " network " ] AS $func => $time ) {
$time = round ( $time , 3 );
if ( $time > 0 )
$o .= $func . " : " . $time . " \n " ;
}
}
2017-01-21 14:50:56 -05:00
} else {
$o = '' ;
2017-01-20 16:58:53 -05:00
}
2017-01-21 14:50:56 -05:00
logger ( " ID " . $queue [ " id " ] . " : " . $funcname . " : " . sprintf ( " DB: %s/%s, Net: %s, I/O: %s, Other: %s, Total: %s " . $o ,
number_format ( $a -> performance [ " database " ] - $a -> performance [ " database_write " ], 2 ),
number_format ( $a -> performance [ " database_write " ], 2 ),
number_format ( $a -> performance [ " network " ], 2 ),
number_format ( $a -> performance [ " file " ], 2 ),
number_format ( $duration - ( $a -> performance [ " database " ] + $a -> performance [ " network " ] + $a -> performance [ " file " ]), 2 ),
number_format ( $duration , 2 )),
LOGGER_DEBUG );
2011-03-04 23:55:32 -05:00
}
2011-04-16 11:45:08 -04:00
2017-01-21 14:50:56 -05:00
$cooldown = Config :: get ( " system " , " worker_cooldown " , 0 );
if ( $cooldown > 0 ) {
logger ( " Process " . $mypid . " - Prio " . $queue [ " priority " ] . " - ID " . $queue [ " id " ] . " : " . $funcname . " - in cooldown for " . $cooldown . " seconds " );
sleep ( $cooldown );
}
2011-01-28 08:04:18 -05:00
}
2010-07-18 23:49:54 -04:00
2016-02-09 00:42:00 -05:00
/**
* @ brief Checks if the number of database connections has reached a critical limit .
*
* @ return bool Are more than 3 / 4 of the maximum connections used ?
*/
function poller_max_connections_reached () {
2016-02-09 17:28:33 -05:00
// Fetch the max value from the config. This is needed when the system cannot detect the correct value by itself.
2017-01-21 01:06:29 -05:00
$max = Config :: get ( " system " , " max_connections " );
2016-02-09 17:28:33 -05:00
2016-04-23 04:11:09 -04:00
// Fetch the percentage level where the poller will get active
2016-11-30 14:24:58 -05:00
$maxlevel = Config :: get ( " system " , " max_connections_level " , 75 );
2016-04-23 04:11:09 -04:00
2016-02-09 17:28:33 -05:00
if ( $max == 0 ) {
2016-02-11 15:39:34 -05:00
// the maximum number of possible user connections can be a system variable
$r = q ( " SHOW VARIABLES WHERE `variable_name` = 'max_user_connections' " );
if ( $r )
$max = $r [ 0 ][ " Value " ];
// Or it can be granted. This overrides the system variable
$r = q ( " SHOW GRANTS " );
if ( $r )
foreach ( $r AS $grants ) {
$grant = array_pop ( $grants );
if ( stristr ( $grant , " GRANT USAGE ON " ))
if ( preg_match ( " /WITH MAX_USER_CONNECTIONS ( \ d*)/ " , $grant , $match ))
$max = $match [ 1 ];
}
}
// If $max is set we will use the processlist to determine the current number of connections
// The processlist only shows entries of the current user
if ( $max != 0 ) {
$r = q ( " SHOW PROCESSLIST " );
2016-12-19 08:42:20 -05:00
if ( ! dbm :: is_result ( $r ))
2016-02-11 15:39:34 -05:00
return false ;
$used = count ( $r );
logger ( " Connection usage (user values): " . $used . " / " . $max , LOGGER_DEBUG );
2016-04-23 04:11:09 -04:00
$level = ( $used / $max ) * 100 ;
2016-02-09 00:42:00 -05:00
2016-04-23 04:11:09 -04:00
if ( $level >= $maxlevel ) {
logger ( " Maximum level ( " . $maxlevel . " %) of user connections reached: " . $used . " / " . $max );
2016-02-12 05:04:25 -05:00
return true ;
}
}
2016-02-09 00:42:00 -05:00
2016-02-12 05:04:25 -05:00
// We will now check for the system values.
// This limit could be reached although the user limits are fine.
$r = q ( " SHOW VARIABLES WHERE `variable_name` = 'max_connections' " );
if ( ! $r )
return false ;
2016-02-11 05:33:45 -05:00
2016-02-12 05:04:25 -05:00
$max = intval ( $r [ 0 ][ " Value " ]);
if ( $max == 0 )
return false ;
$r = q ( " SHOW STATUS WHERE `variable_name` = 'Threads_connected' " );
if ( ! $r )
return false ;
$used = intval ( $r [ 0 ][ " Value " ]);
if ( $used == 0 )
return false ;
logger ( " Connection usage (system values): " . $used . " / " . $max , LOGGER_DEBUG );
2016-02-09 00:42:00 -05:00
2016-04-23 04:11:09 -04:00
$level = $used / $max * 100 ;
2016-02-09 00:42:00 -05:00
2016-04-23 04:11:09 -04:00
if ( $level < $maxlevel )
2016-02-09 00:42:00 -05:00
return false ;
2016-04-23 04:11:09 -04:00
logger ( " Maximum level ( " . $level . " %) of system connections reached: " . $used . " / " . $max );
2016-02-09 00:42:00 -05:00
return true ;
}
2016-01-23 20:53:21 -05:00
/**
* @ brief fix the queue entry if the worker process died
*
*/
function poller_kill_stale_workers () {
2017-02-27 18:37:15 -05:00
$r = q ( " SELECT `pid`, `executed`, `priority`, `parameter` FROM `workerqueue` WHERE `executed` > '%s' " , dbesc ( NULL_DATE ));
2016-03-04 16:38:18 -05:00
2016-07-02 08:00:42 -04:00
if ( ! dbm :: is_result ( $r )) {
2016-03-04 16:38:18 -05:00
// No processing here needed
return ;
}
2017-03-21 12:02:59 -04:00
foreach ( $r AS $pid ) {
2017-01-26 10:07:30 -05:00
if ( ! posix_kill ( $pid [ " pid " ], 0 )) {
2017-05-12 12:09:25 -04:00
dba :: update ( 'workerqueue' , array ( 'executed' => NULL_DATE , 'pid' => 0 ),
array ( 'pid' => $pid [ " pid " ]));
2017-01-26 10:07:30 -05:00
} else {
2016-01-23 20:53:21 -05:00
// Kill long running processes
2016-08-08 13:20:40 -04:00
// Check if the priority is in a valid range
2017-04-04 13:48:13 -04:00
if ( ! in_array ( $pid [ " priority " ], array ( PRIORITY_CRITICAL , PRIORITY_HIGH , PRIORITY_MEDIUM , PRIORITY_LOW , PRIORITY_NEGLIGIBLE ))) {
2016-08-08 13:20:40 -04:00
$pid [ " priority " ] = PRIORITY_MEDIUM ;
2017-04-04 13:48:13 -04:00
}
2016-08-08 13:20:40 -04:00
// Define the maximum durations
2016-08-18 06:33:17 -04:00
$max_duration_defaults = array ( PRIORITY_CRITICAL => 360 , PRIORITY_HIGH => 10 , PRIORITY_MEDIUM => 60 , PRIORITY_LOW => 180 , PRIORITY_NEGLIGIBLE => 360 );
2016-08-08 13:20:40 -04:00
$max_duration = $max_duration_defaults [ $pid [ " priority " ]];
2016-08-14 15:02:29 -04:00
$argv = json_decode ( $pid [ " parameter " ]);
$argv [ 0 ] = basename ( $argv [ 0 ]);
2016-08-08 13:20:40 -04:00
// How long is the process already running?
2016-01-23 20:53:21 -05:00
$duration = ( time () - strtotime ( $pid [ " executed " ])) / 60 ;
2016-08-08 13:20:40 -04:00
if ( $duration > $max_duration ) {
2016-08-14 15:02:29 -04:00
logger ( " Worker process " . $pid [ " pid " ] . " ( " . implode ( " " , $argv ) . " ) took more than " . $max_duration . " minutes. It will be killed now. " );
2016-01-23 20:53:21 -05:00
posix_kill ( $pid [ " pid " ], SIGTERM );
2016-08-04 09:33:15 -04:00
// We killed the stale process.
// To avoid a blocking situation we reschedule the process at the beginning of the queue.
2016-08-04 09:41:32 -04:00
// Additionally we are lowering the priority.
2017-05-12 12:09:25 -04:00
dba :: update ( 'workerqueue' ,
array ( 'executed' => NULL_DATE , 'created' => datetime_convert (), 'priority' => PRIORITY_NEGLIGIBLE , 'pid' => 0 ),
array ( 'pid' => $pid [ " pid " ]));
2017-01-26 10:07:30 -05:00
} else {
2016-08-14 15:02:29 -04:00
logger ( " Worker process " . $pid [ " pid " ] . " ( " . implode ( " " , $argv ) . " ) now runs for " . round ( $duration ) . " of " . $max_duration . " allowed minutes. That's okay. " , LOGGER_DEBUG );
2017-01-26 10:07:30 -05:00
}
2016-01-23 20:53:21 -05:00
}
2017-03-19 04:04:04 -04:00
}
2016-01-23 20:53:21 -05:00
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Checks if the number of active workers exceeds the given limits
*
* @ return bool Are there too much workers running ?
*/
2016-07-24 14:25:11 -04:00
function poller_too_much_workers () {
2016-11-30 14:24:58 -05:00
$queues = Config :: get ( " system " , " worker_queues " , 4 );
2015-09-28 01:54:28 -04:00
2016-07-23 16:57:22 -04:00
$maxqueues = $queues ;
2015-09-28 01:54:28 -04:00
$active = poller_active_workers ();
// Decrease the number of workers at higher load
2015-12-15 17:26:58 -05:00
$load = current_load ();
2017-04-04 13:48:13 -04:00
if ( $load ) {
2016-11-30 14:24:58 -05:00
$maxsysload = intval ( Config :: get ( " system " , " maxloadavg " , 50 ));
2015-09-28 01:54:28 -04:00
2015-09-28 13:14:07 -04:00
$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 ));
2015-09-28 01:54:28 -04:00
2017-02-27 18:37:15 -05:00
$s = q ( " SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= '%s' " , dbesc ( NULL_DATE ));
2016-08-03 09:59:25 -04:00
$entries = $s [ 0 ][ " total " ];
if ( Config :: get ( " system " , " worker_fastlane " , false ) AND ( $queues > 0 ) AND ( $entries > 0 ) AND ( $active >= $queues )) {
2017-02-27 18:37:15 -05:00
$s = q ( " SELECT `priority` FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority` LIMIT 1 " , dbesc ( NULL_DATE ));
2016-08-03 09:59:25 -04:00
$top_priority = $s [ 0 ][ " priority " ];
2016-08-03 04:03:05 -04:00
2017-02-27 18:37:15 -05:00
$s = q ( " SELECT `id` FROM `workerqueue` WHERE `priority` <= %d AND `executed` > '%s' LIMIT 1 " ,
intval ( $top_priority ), dbesc ( NULL_DATE ));
2016-08-18 06:33:17 -04:00
$high_running = dbm :: is_result ( $s );
2016-08-03 04:03:05 -04:00
2016-08-18 06:33:17 -04:00
if ( ! $high_running AND ( $top_priority > PRIORITY_UNDEFINED ) AND ( $top_priority < PRIORITY_NEGLIGIBLE )) {
2016-08-03 09:59:25 -04:00
logger ( " There are jobs with priority " . $top_priority . " waiting but none is executed. Open a fastlane. " , LOGGER_DEBUG );
2016-08-03 04:19:46 -04:00
$queues = $active + 1 ;
2016-08-03 04:03:05 -04:00
}
}
2016-10-11 02:30:53 -04:00
// Create a list of queue entries grouped by their priority
$running = array ( PRIORITY_CRITICAL => 0 ,
PRIORITY_HIGH => 0 ,
PRIORITY_MEDIUM => 0 ,
PRIORITY_LOW => 0 ,
PRIORITY_NEGLIGIBLE => 0 );
$r = q ( " SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` GROUP BY `priority` " );
if ( dbm :: is_result ( $r ))
foreach ( $r AS $process )
$running [ $process [ " priority " ]] = $process [ " running " ];
$processlist = " " ;
$r = q ( " SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` GROUP BY `priority` " );
if ( dbm :: is_result ( $r ))
foreach ( $r as $entry ) {
if ( $processlist != " " )
$processlist .= " , " ;
$processlist .= $entry [ " priority " ] . " : " . $running [ $entry [ " priority " ]] . " / " . $entry [ " entries " ];
}
logger ( " Load: " . $load . " / " . $maxsysload . " - processes: " . $active . " / " . $entries . " ( " . $processlist . " ) - maximum: " . $queues . " / " . $maxqueues , LOGGER_DEBUG );
2015-09-27 07:56:20 -04:00
2016-08-01 01:48:43 -04:00
// Are there fewer workers running as possible? Then fork a new one.
2017-01-21 01:06:29 -05:00
if ( ! Config :: get ( " system " , " worker_dont_fork " ) AND ( $queues > ( $active + 1 )) AND ( $entries > 1 )) {
2016-08-01 01:48:43 -04:00
logger ( " Active workers: " . $active . " / " . $queues . " Fork a new worker. " , LOGGER_DEBUG );
2017-02-27 01:27:22 -05:00
$args = array ( " include/poller.php " , " no_cron " );
2016-08-01 01:48:43 -04:00
$a = get_app ();
$a -> proc_run ( $args );
}
2015-09-27 07:56:20 -04:00
}
2015-09-28 01:54:28 -04:00
return ( $active >= $queues );
2015-09-27 07:56:20 -04:00
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Returns the number of active poller processes
*
* @ return integer Number of active poller processes
*/
2015-09-27 07:56:20 -04:00
function poller_active_workers () {
2016-09-09 16:33:54 -04:00
$workers = q ( " SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php' " );
2015-09-27 07:56:20 -04:00
2016-09-09 16:33:54 -04:00
return ( $workers [ 0 ][ " processes " ]);
2015-09-27 07:56:20 -04:00
}
2016-10-10 17:01:36 -04:00
/**
* @ brief Check if we should pass some slow processes
*
* When the active processes of the highest priority are using more than 2 / 3
* of all processes , we let pass slower processes .
*
* @ param string $highest_priority Returns the currently highest priority
* @ return bool We let pass a slower process than $highest_priority
*/
function poller_passing_slow ( & $highest_priority ) {
$highest_priority = 0 ;
$r = q ( " SELECT `priority`
FROM `process`
2016-11-26 19:55:05 -05:00
INNER JOIN `workerqueue` ON `workerqueue` . `pid` = `process` . `pid` " );
2016-10-10 17:01:36 -04:00
// No active processes at all? Fine
if ( ! dbm :: is_result ( $r ))
return ( false );
$priorities = array ();
foreach ( $r AS $line )
$priorities [] = $line [ " priority " ];
// Should not happen
if ( count ( $priorities ) == 0 )
return ( false );
$highest_priority = min ( $priorities );
// The highest process is already the slowest one?
// Then we quit
if ( $highest_priority == PRIORITY_NEGLIGIBLE )
return ( false );
$high = 0 ;
foreach ( $priorities AS $priority )
if ( $priority == $highest_priority )
++ $high ;
logger ( " Highest priority: " . $highest_priority . " Total processes: " . count ( $priorities ) . " Count high priority processes: " . $high , LOGGER_DEBUG );
$passing_slow = (( $high / count ( $priorities )) > ( 2 / 3 ));
if ( $passing_slow )
logger ( " Passing slower processes than priority " . $highest_priority , LOGGER_DEBUG );
return ( $passing_slow );
}
/**
* @ brief Returns the next worker process
*
* @ return string SQL statement
*/
function poller_worker_process () {
2017-05-11 16:13:45 -04:00
dba :: transaction ();
2016-10-23 19:31:56 -04:00
2016-10-10 17:01:36 -04:00
// Check if we should pass some low priority process
$highest_priority = 0 ;
if ( poller_passing_slow ( $highest_priority )) {
// Are there waiting processes with a higher priority than the currently highest?
$r = q ( " SELECT * FROM `workerqueue`
2017-02-27 18:37:15 -05:00
WHERE `executed` <= '%s' AND `priority` < % d
ORDER BY `priority` , `created` LIMIT 1 " ,
dbesc ( NULL_DATE ),
intval ( $highest_priority ));
2017-03-19 04:04:04 -04:00
if ( dbm :: is_result ( $r )) {
2016-10-10 17:01:36 -04:00
return $r ;
2017-03-19 04:04:04 -04:00
}
2016-10-10 17:01:36 -04:00
// Give slower processes some processing time
$r = q ( " SELECT * FROM `workerqueue`
2017-02-27 18:37:15 -05:00
WHERE `executed` <= '%s' AND `priority` > % d
ORDER BY `priority` , `created` LIMIT 1 " ,
dbesc ( NULL_DATE ),
intval ( $highest_priority ));
2016-10-10 17:01:36 -04:00
}
// If there is no result (or we shouldn't pass lower processes) we check without priority limit
2017-03-19 04:04:04 -04:00
if (( $highest_priority == 0 ) OR ! dbm :: is_result ( $r )) {
2017-02-27 18:37:15 -05:00
$r = q ( " SELECT * FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority`, `created` LIMIT 1 " , dbesc ( NULL_DATE ));
2017-03-19 04:04:04 -04:00
}
2016-10-10 17:01:36 -04:00
return $r ;
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Call the front end worker
*/
2016-11-26 19:55:05 -05:00
function call_worker () {
2017-02-26 18:16:49 -05:00
if ( ! Config :: get ( " system " , " frontend_worker " )) {
2016-11-26 19:55:05 -05:00
return ;
}
2016-12-20 15:51:25 -05:00
$url = App :: get_baseurl () . " /worker " ;
2016-11-26 19:55:05 -05:00
fetch_url ( $url , false , $redirects , 1 );
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Call the front end worker if there aren ' t any active
*/
2016-11-26 19:55:05 -05:00
function call_worker_if_idle () {
2017-02-26 18:16:49 -05:00
if ( ! Config :: get ( " system " , " frontend_worker " )) {
2016-11-26 19:55:05 -05:00
return ;
}
2016-11-29 17:40:19 -05:00
// Do we have "proc_open"? Then we can fork the poller
if ( function_exists ( " proc_open " )) {
2016-11-30 14:24:58 -05:00
// When was the last time that we called the worker?
// Less than one minute? Then we quit
2017-01-21 01:06:29 -05:00
if (( time () - Config :: get ( " system " , " worker_started " )) < 60 ) {
2016-11-29 17:40:19 -05:00
return ;
}
2016-11-30 14:24:58 -05:00
set_config ( " system " , " worker_started " , time ());
// Do we have enough running workers? Then we quit here.
if ( poller_too_much_workers ()) {
// Cleaning dead processes
poller_kill_stale_workers ();
get_app () -> remove_inactive_processes ();
2016-11-29 17:40:19 -05:00
return ;
}
2016-11-30 14:24:58 -05:00
poller_run_cron ();
logger ( 'Call poller' , LOGGER_DEBUG );
2017-02-27 01:27:22 -05:00
$args = array ( " include/poller.php " , " no_cron " );
2016-11-30 14:24:58 -05:00
$a = get_app ();
$a -> proc_run ( $args );
return ;
2016-11-29 17:40:19 -05:00
}
// We cannot execute background processes.
// We now run the processes from the frontend.
// This won't work with long running processes.
2016-11-26 19:55:05 -05:00
poller_run_cron ();
clear_worker_processes ();
$workers = q ( " SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php' " );
if ( $workers [ 0 ][ " processes " ] == 0 ) {
call_worker ();
}
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Removes long running worker processes
*/
2016-11-26 19:55:05 -05:00
function clear_worker_processes () {
2016-11-27 04:02:08 -05:00
$timeout = Config :: get ( " system " , " frontend_worker_timeout " , 10 );
/// @todo We should clean up the corresponding workerqueue entries as well
2016-11-26 19:55:05 -05:00
q ( " DELETE FROM `process` WHERE `created` < '%s' AND `command` = 'worker.php' " ,
2016-11-27 04:02:08 -05:00
dbesc ( datetime_convert ( 'UTC' , 'UTC' , " now - " . $timeout . " minutes " )));
2016-11-26 19:55:05 -05:00
}
2016-11-26 20:11:22 -05:00
/**
* @ brief Runs the cron processes
*/
2016-11-26 19:55:05 -05:00
function poller_run_cron () {
2016-11-30 14:24:58 -05:00
logger ( 'Add cron entries' , LOGGER_DEBUG );
2016-12-01 15:53:18 -05:00
// Check for spooled items
proc_run ( PRIORITY_HIGH , " include/spool_post.php " );
2016-11-26 19:55:05 -05:00
// Run the cron job that calls all other jobs
proc_run ( PRIORITY_MEDIUM , " include/cron.php " );
// Run the cronhooks job separately from cron for being able to use a different timing
proc_run ( PRIORITY_MEDIUM , " include/cronhooks.php " );
// Cleaning dead processes
poller_kill_stale_workers ();
}
2011-01-28 08:04:18 -05:00
if ( array_search ( __file__ , get_included_files ()) === 0 ){
2016-09-09 16:55:49 -04:00
poller_run ( $_SERVER [ " argv " ], $_SERVER [ " argc " ]);
get_app () -> end_process ();
killme ();
2011-01-28 08:04:18 -05:00
}