hazza/Phraseanet
1
0
Fork 0
mirror of https://github.com/alchemy-fr/Phraseanet.git synced 2025-10-11 12:03:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

-

Browse Source
This commit is contained in:
jygaulier
2012-01-19 21:01:03 +01:00
parent e5828f06bb
commit 5a5f86a97b
7 changed files with 1079 additions and 653 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
lib/classes/module/console/taskrun.class.php
View File

@@ -25,11 +25,16 @@ use Symfony\Component\Console\Command\Command;
class module_console_taskrun extends Command
{
private $task;
private $shedulerPID;
public function __construct($name = null)
{
parent::__construct($name);
$this->task = NULL;
$this->shedulerPID = NULL;
$this->addArgument('task_id', InputArgument::REQUIRED, 'The task_id to run');
$this->addOption(
'runner'
@@ -54,21 +59,56 @@ class module_console_taskrun extends Command
$task_id = (int) $input->getArgument('task_id');
if ($task_id <= 0 || strlen($task_id) !== strlen($input->getArgument('task_id')))
if($task_id <= 0 || strlen($task_id) !== strlen($input->getArgument('task_id')))
throw new \RuntimeException('Argument must be an Id.');
$appbox = appbox::get_instance();
$task_manager = new task_manager($appbox);
$task = $task_manager->get_task($task_id);
$this->task = $task_manager->get_task($task_id);
$runner = task_abstract::RUNNER_SCHEDULER;
if ($input->getOption('runner') === task_abstract::RUNNER_MANUAL)
if($input->getOption('runner') === task_abstract::RUNNER_MANUAL)
{
$runner = task_abstract::RUNNER_MANUAL;
}
else
{
$runner = task_abstract::RUNNER_SCHEDULER;
$registry = $appbox->get_registry();
$schedFile = $registry->get('GV_RootPath') . 'tmp/locks/scheduler.lock';
if(file_exists($schedFile))
$this->shedulerPID = (int) (trim(file_get_contents($schedFile)));
}
$task->run($runner);
register_tick_function(array($this, 'tick_handler'), true);
declare(ticks=1);
$this->task->run($runner);
printf("TASK QUIT\n");
return $this;
}
public function tick_handler()
{
static $start = FALSE;
if($start === FALSE)
$start = time();
if(time() - $start > 0)
{
if($this->shedulerPID)
{
if(!posix_kill($this->shedulerPID, 0))
{
if(method_exists($this->task, 'signal'))
$this->task->signal('SIGNAL_SCHEDULER_DIED');
else
$this->task->set_status(task_abstract::STATUS_TOSTOP);
}
}
$start = time();
}
}
}

703
lib/classes/task/Scheduler.class.php
View File

@@ -21,25 +21,31 @@ class task_Scheduler
{
const TASKDELAYTOQUIT = 60;
// how to schedule tasks (choose in 'run' method)
const METHOD_FORK = 'METHOD_FORK';
const METHOD_PROC_OPEN = 'METHOD_PROC_OPEN';
private $method;
protected $output;
protected function log($message)
{
if ($this->output instanceof OutputInterface)
{
$this->output->writeln($message);
}
$registry = registry::get_instance();
$logdir = $registry->get('GV_RootPath') . 'logs/';
logs::rotate($logdir . "scheduler.log");
$date_obj = new DateTime();
$message = sprintf("%s %s \n", $date_obj->format(DATE_ATOM), $message);
file_put_contents($logdir . "scheduler.log", $message, FILE_APPEND);
$message = sprintf("%s\t%s", $date_obj->format(DATE_ATOM), $message);
if($this->output instanceof OutputInterface)
{
$this->output->writeln($message);
}
// else
{
file_put_contents($logdir . "scheduler.log", $message."\n", FILE_APPEND);
}
return $this;
}
@@ -52,6 +58,8 @@ class task_Scheduler
public function run(OutputInterface $output = null, $log_tasks = true)
{
$this->method = self::METHOD_FORK;
require_once dirname(__FILE__) . '/../../bootstrap.php';
$this->output = $output;
$appbox = appbox::get_instance();
@@ -61,13 +69,14 @@ class task_Scheduler
$lockdir = $registry->get('GV_RootPath') . 'tmp/locks/';
for ($try = 0; true; $try++)
for($try = 0; true; $try++)
{
$schedlock = fopen(($lockfile = ($lockdir . 'scheduler.lock')), 'a+');
if (flock($schedlock, LOCK_EX | LOCK_NB) != true)
if(($schedlock = fopen(($lockfile = ($lockdir . 'scheduler.lock')), 'a+')))
{
if(flock($schedlock, LOCK_EX | LOCK_NB) === FALSE)
{
$this->log(sprintf("failed to lock '%s' (try=%s/4)", $lockfile, $try));
if ($try == 4)
if($try == 4)
{
$this->log("scheduler already running.");
fclose($schedlock);
@@ -81,46 +90,52 @@ class task_Scheduler
}
else
{
// locked
ftruncate($schedlock, 0);
fwrite($schedlock, '' . getmypid());
fflush($schedlock);
break;
}
}
}
$this->log(sprintf("running scheduler with method %s", $this->method));
if($this->method == self::METHOD_FORK)
pcntl_signal(SIGCHLD, SIG_IGN);
$logdir = $registry->get('GV_RootPath') . 'logs/';
$conn = self::get_connection();
$ttask = array();
$taskPoll = array(); // the poll of tasks
$sleeptime = 3;
$sql = "UPDATE sitepreff SET schedstatus='started', schedpid = :pid";
$stmt = $conn->prepare($sql);
$stmt->execute(array(':pid' => getmypid()));
$stmt->closeCursor();
$sql = "UPDATE sitepreff SET schedstatus='started'";
$conn->exec($sql);
$task_manager = new task_manager($appbox);
$tlist = array();
foreach ($task_manager->get_tasks() as $task)
// set every 'auto-start' task to start
foreach($task_manager->get_tasks() as $task)
{
if (!$task->is_active())
if($task->is_active())
{
continue;
}
$tid = $task->get_task_id();
if (!$task->is_running())
if(!$task->get_pid())
{
/* @var $task task_abstract */
$task->reset_crash_counter();
$task->set_status(task_abstract::STATUS_TOSTART);
}
}
}
$tlist = array();
$schedstatus = 'started';
@@ -129,12 +144,12 @@ class task_Scheduler
$last_log_check = array();
while ($schedstatus == 'started' || $runningtask > 0)
while($schedstatus == 'started' || $runningtask > 0)
{
while (!$conn->ping())
while(!$conn->ping())
{
unset($conn);
if (!$connwaslost)
if(!$connwaslost)
{
$this->log(sprintf("Warning : abox connection lost, restarting in 10 min."));
}
@@ -142,7 +157,7 @@ class task_Scheduler
$conn = self::get_connection();
$connwaslost = true;
}
if ($connwaslost)
if($connwaslost)
{
$this->log("abox connection restored");
@@ -161,12 +176,12 @@ class task_Scheduler
$row = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
if ($row)
if($row)
{
$schedstatus = $row["schedstatus"];
}
if ($schedstatus == 'tostop')
if($schedstatus == 'tostop')
{
$sql = 'UPDATE sitepreff SET schedstatus = "stopping"';
$stmt = $conn->prepare($sql);
@@ -186,87 +201,86 @@ class task_Scheduler
logs::rotate($logdir . "scheduler.error.log");
/**
* potentially, all tasks are supposed to be removed
*/
foreach ($ttask as $tkey => $tv)
{
$ttask[$tkey]["todel"] = true;
}
// initialy, all tasks are supposed to be removed from the poll
foreach($taskPoll as $tkey => $task)
$taskPoll[$tkey]["todel"] = true;
$sql = "SELECT * FROM task2";
$stmt = $conn->prepare($sql);
$stmt->execute();
$rs = $stmt->fetchAll(PDO::FETCH_ASSOC);
$stmt->closeCursor();
foreach ($task_manager->get_tasks(true) as $task)
foreach($task_manager->get_tasks(true) as $task)
{
$tkey = "t_" . $task->get_task_id();
logs::rotate($logdir . "task_$tkey.log");
logs::rotate($logdir . "task_$tkey.error.log");
if (!isset($ttask[$tkey]))
if(!isset($taskPoll[$tkey]))
{
// the task is not in the poll, add it
$phpcli = $registry->get('GV_cli');
switch ($system)
switch($system)
{
default:
case "DARWIN":
case "WINDOWS":
case "LINUX":
$cmd = $phpcli . ' -f '
. $registry->get('GV_RootPath')
. "bin/console task:run "
. $task->get_task_id()
. " --runner=scheduler ";
$cmd = $phpcli;
$args = array('-f', $registry->get('GV_RootPath') . 'bin/console', 'task:run', $task->get_task_id(), '--runner=scheduler');
break;
}
$ttask[$tkey] = array(
$taskPoll[$tkey] = array(
"task" => $task,
"current_status" => $task->get_status(),
"process" => null,
"cmd" => $cmd,
"killat" => null,
"pipes" => null
"args" => $args,
"killat" => null
);
if($this->method == self::METHOD_PROC_OPEN)
{
$taskPoll[$tkey]['process'] = NULL;
$taskPoll[$tkey]['pipes'] = NULL;
}
$this->log(
sprintf(
"new Task %s, status=%s"
, $ttask[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["task"]->get_task_id()
, $task->get_status()
)
);
}
else
{
if ($ttask[$tkey]["current_status"] != $task->get_status())
// the task is already in the poll, update its status
if($taskPoll[$tkey]["current_status"] != $task->get_status())
{
$this->log(
sprintf(
"Task %s, oldstatus=%s, newstatus=%s"
, $ttask[$tkey]["task"]->get_task_id()
, $ttask[$tkey]["current_status"]
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["current_status"]
, $task->get_status()
)
);
$ttask[$tkey]["current_status"] = $task->get_status();
$taskPoll[$tkey]["current_status"] = $task->get_status();
}
// update the whole task object
unset($taskPoll[$tkey]["task"]);
$taskPoll[$tkey]["task"] = $task;
}
$ttask[$tkey]["task"] = $task;
}
$ttask[$tkey]["todel"] = false;
unset($task);
$taskPoll[$tkey]["todel"] = false; // this task exists, do not remove from poll
}
foreach ($ttask as $tkey => $tv)
// remove not-existing task from poll
foreach($taskPoll as $tkey => $task)
{
if ($tv["todel"])
if($task["todel"])
{
$this->log(sprintf("Task %s deleted", $ttask[$tkey]["task"]->get_task_id()));
unset($ttask[$tkey]);
$this->log(sprintf("Task %s deleted", $taskPoll[$tkey]["task"]->get_task_id()));
unset($taskPoll[$tkey]);
}
}
@@ -276,54 +290,54 @@ class task_Scheduler
*/
$runningtask = 0;
$common_status = array(
task_abstract::STATUS_STARTED
, task_abstract::RETURNSTATUS_STOPPED
);
foreach ($ttask as $tkey => $tv)
foreach($taskPoll as $tkey => $tv)
{
if (!in_array($ttask[$tkey]["task"]->get_status(), $common_status))
{
$this->log(
sprintf(
'task %s has status %s'
, $ttask[$tkey]["task"]->get_task_id()
, $ttask[$tkey]["task"]->get_status()
)
);
}
switch ($ttask[$tkey]["task"]->get_status())
switch($tv['task']->get_status())
{
default:
$this->log(
sprintf(
'Unknow status `%s`'
, $ttask[$tkey]["task"]->get_status()
)
);
$this->log(sprintf('Unknow status `%s`', $tv['task']->get_status()));
break;
case task_abstract::RETURNSTATUS_TORESTART:
@fclose($ttask[$tkey]["pipes"][1]);
@fclose($ttask[$tkey]["pipes"][2]);
@proc_close($ttask[$tkey]["process"]);
$ttask[$tkey]["process"] = null;
if ($schedstatus == 'started')
case task_abstract::RETURNSTATUS_TORESTART:
if(!$taskPoll[$tkey]['task']->get_pid())
{
$ttask[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
if($this->method == self::METHOD_PROC_OPEN)
{
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
}
if($schedstatus == 'started')
{
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
// trick to start the task immediatly : DON'T break if ending with 'tostart'
// so it will continue with 'tostart' case !
}
else
{
break;
}
case task_abstract::STATUS_TOSTART:
$ttask[$tkey]["killat"] = NULL;
if ($schedstatus == 'started' && !$ttask[$tkey]["process"])
// if scheduler is 'tostop', don't launch a new task !
if($schedstatus != 'started')
break;
$taskPoll[$tkey]["killat"] = NULL;
if($this->method == self::METHOD_PROC_OPEN)
{
if(!$taskPoll[$tkey]["process"])
{
$descriptors = array(
1 => array("pipe", "w")
, 2 => array("pipe", "w")
1 => array("pipe", "w"),
2 => array("pipe", "w")
);
if ($log_tasks === true)
if($log_tasks === true)
{
$descriptors[1] = array(
"file"
@@ -337,60 +351,377 @@ class task_Scheduler
);
}
$ttask[$tkey]["process"] = proc_open(
$ttask[$tkey]["cmd"]
$taskPoll[$tkey]["process"] = proc_open(
$taskPoll[$tkey]["cmd"] . ' ' . implode(' ', $taskPoll[$tkey]["args"])
, $descriptors
, $ttask[$tkey]["pipes"]
, $taskPoll[$tkey]["pipes"]
, $registry->get('GV_RootPath') . "bin/"
, null
, array('bypass_shell' => true)
);
if (is_resource($ttask[$tkey]["process"]))
if(is_resource($taskPoll[$tkey]["process"]))
{
// proc_status['pid'] if the pid of the sh !!!
// $proc_status = proc_get_status($ttask[$tkey]["process"]);
// if ($proc_status['running'])
// $ttask[$tkey]['task']->set_pid($proc_status['pid']);
sleep(2); // let the process lock and write it's pid
}
if ($ttask[$tkey]['task']->get_pid() !== null)
if(is_resource($taskPoll[$tkey]["process"]) && $taskPoll[$tkey]['task']->get_pid() !== null)
{
$this->log(
sprintf(
"Task %s '%s' started (pid=%s)"
, $ttask[$tkey]['task']->get_task_id()
, $ttask[$tkey]["cmd"]
, $ttask[$tkey]['task']->get_pid()
, $taskPoll[$tkey]['task']->get_task_id()
, $taskPoll[$tkey]["cmd"]
, $taskPoll[$tkey]['task']->get_pid()
)
);
$runningtask++;
}
else
{
$ttask[$tkey]["task"]->increment_crash_counter();
$taskPoll[$tkey]["task"]->increment_crash_counter();
@fclose($ttask[$tkey]["pipes"][1]);
@fclose($ttask[$tkey]["pipes"][2]);
@proc_close($ttask[$tkey]["process"]);
$ttask[$tkey]["process"] = null;
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
$this->log(
sprintf(
"Task %s '%s' failed to start %d times"
, $ttask[$tkey]["task"]->get_task_id()
, $ttask[$tkey]["cmd"]
, $ttask[$tkey]["task"]->get_crash_counter()
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["cmd"]
, $taskPoll[$tkey]["task"]->get_crash_counter()
)
);
if ($ttask[$tkey]["task"]->get_crash_counter() > 5)
if($taskPoll[$tkey]["task"]->get_crash_counter() > 5)
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
else
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
}
}
elseif($this->method == self::METHOD_FORK)
{
$ttask[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
// printf("forking pid %d\n", getmypid());
$pid = pcntl_fork();
if($pid == -1)
{
die("failed to fork");
}
elseif($pid == 0)
{
// child
// printf("hello i am child pid=%d\n", getmypid());
// printf("%s %s \n", $taskPoll[$tkey]["cmd"], implode(' ', $taskPoll[$tkey]["args"]));
// ;
umask(0);
openlog('MyLog', LOG_PID | LOG_PERROR, LOG_LOCAL0);
if(posix_setsid() < 0)
die("Forked process could not detach from terminal\n");
//chdir(dirname(__FILE__));
fclose(STDIN);
fclose(STDOUT);
fclose(STDERR);
$fdIN = fopen('/dev/null', 'r');
$fdOUT = fopen($logdir . "task_$tkey.log", 'a+');
$fdERR = fopen($logdir . "task_$tkey.error.log", 'a+');
pcntl_exec($taskPoll[$tkey]["cmd"], $taskPoll[$tkey]["args"]);
sleep(2);
}
else
{
$ttask[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
// parent
// printf("hello i am parent pid=%d\n", getmypid());
}
}
break;
case task_abstract::STATUS_STARTED:
$crashed = false;
// If no process, the task is probably manually ran
if($this->method == self::METHOD_PROC_OPEN)
{
if($taskPoll[$tkey]["process"])
{
$taskPoll[$tkey]["killat"] = NULL;
if(is_resource($taskPoll[$tkey]["process"]))
{
$proc_status = proc_get_status($taskPoll[$tkey]["process"]);
if($proc_status['running'])
$runningtask++;
else
$crashed = true;
}
else
{
$crashed = true;
}
}
}
if(!$crashed && !$taskPoll[$tkey]['task']->get_pid())
$crashed = true;
if(!$crashed)
{
$taskPoll[$tkey]["killat"] = NULL;
$runningtask++;
}
else
{
// crashed !
$taskPoll[$tkey]["task"]->increment_crash_counter();
if($this->method == self::METHOD_PROC_OPEN)
{
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
}
$this->log(
sprintf(
"Task %s crashed %d times"
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["task"]->get_crash_counter()
)
);
if($taskPoll[$tkey]["task"]->get_crash_counter() > 5)
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
else
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
break;
case task_abstract::STATUS_TOSTOP:
if($taskPoll[$tkey]["killat"] === NULL)
$taskPoll[$tkey]["killat"] = time() + self::TASKDELAYTOQUIT;
$tpid = $taskPoll[$tkey]['task']->get_pid();
if($tpid)
{
if(($dt = $taskPoll[$tkey]["killat"] - time()) < 0)
{
if($this->method == self::METHOD_PROC_OPEN)
{
$pids = preg_split('/\s+/', `ps -o pid --no-heading --ppid $tpid`);
foreach($pids as $pid)
{
if(is_numeric($pid))
{
$this->log("Killing pid %d", $pid);
posix_kill($pid, 9);
}
}
}
elseif($this->method == self::METHOD_FORK)
{
posix_kill($tpid, 9);
}
$this->log(
sprintf(
"SIGKILL sent to task %s (pid=%s)"
, $taskPoll[$tkey]["task"]->get_task_id()
, $tpid
)
);
if($this->method == self::METHOD_PROC_OPEN)
{
proc_terminate($taskPoll[$tkey]["process"], 9);
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
proc_close($taskPoll[$tkey]["process"]);
}
unlink($lockdir . 'task_' . $taskPoll[$tkey]['task']->get_task_id() . '.lock');
$taskPoll[$tkey]["task"]->increment_crash_counter();
// $taskPoll[$tkey]["task"]->set_pid(null);
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
}
else
{
$this->log(
sprintf(
"waiting task %s to quit (kill in %d seconds)"
, $taskPoll[$tkey]["task"]->get_task_id()
, $dt
)
);
$runningtask++;
}
}
else
{
$this->log(
sprintf(
"task %s has quit"
, $taskPoll[$tkey]["task"]->get_task_id()
)
);
}
break;
case task_abstract::RETURNSTATUS_STOPPED:
case task_abstract::RETURNSTATUS_TODELETE:
if($this->method == self::METHOD_PROC_OPEN)
{
if($taskPoll[$tkey]["process"])
{
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
}
}
break;
}
}
/*
$common_status = array(
task_abstract::STATUS_STARTED
, task_abstract::RETURNSTATUS_STOPPED
);
foreach($taskPoll as $tkey => $tv)
{
// if (!in_array($taskPoll[$tkey]["task"]->get_status(), $common_status))
// {
// $this->log(
// sprintf(
// 'task %s has status %s'
// , $taskPoll[$tkey]["task"]->get_task_id()
// , $taskPoll[$tkey]["task"]->get_status()
// )
// );
// }
switch($tv['task']->get_status())
{
default:
$this->log(sprintf('Unknow status `%s`', $tv['task']->get_status()));
break;
case task_abstract::RETURNSTATUS_TORESTART:
if(!$taskPoll[$tkey]['task']->get_pid())
{
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
if($schedstatus == 'started')
{
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
// trick to start the task immediatly : DON'T break if ending with 'tostart'
// so it will continue with 'tostart' case !
}
else
{
break;
}
case task_abstract::STATUS_TOSTART:
// if scheduler is 'tostop', don't launch a new task !
if($schedstatus != 'started')
break;
$taskPoll[$tkey]["killat"] = NULL;
if(!$taskPoll[$tkey]["process"])
{
$descriptors = array(
1 => array("pipe", "w"),
2 => array("pipe", "w")
);
if($log_tasks === true)
{
$descriptors[1] = array(
"file"
, $logdir . "task_$tkey.log"
, "a+"
);
$descriptors[2] = array(
"file"
, $logdir . "task_$tkey.error.log"
, "a+"
);
}
$taskPoll[$tkey]["process"] = proc_open(
$taskPoll[$tkey]["cmd"].' '.implode(' ', $taskPoll[$tkey]["args"])
, $descriptors
, $taskPoll[$tkey]["pipes"]
, $registry->get('GV_RootPath') . "bin/"
, null
, array('bypass_shell' => true)
);
if(is_resource($taskPoll[$tkey]["process"]))
{
sleep(2); // let the process lock and write it's pid
}
if(is_resource($taskPoll[$tkey]["process"]) && $taskPoll[$tkey]['task']->get_pid() !== null)
{
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : RUNNING ? : pid=%s \n", __FILE__, __LINE__, $taskPoll[$tkey]['task']->get_pid()), FILE_APPEND);
$this->log(
sprintf(
"Task %s '%s' started (pid=%s)"
, $taskPoll[$tkey]['task']->get_task_id()
, $taskPoll[$tkey]["cmd"]
, $taskPoll[$tkey]['task']->get_pid()
)
);
$runningtask++;
}
else
{
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : NOT RUNNING ? : pid=%s \n", __FILE__, __LINE__, $taskPoll[$tkey]['task']->get_pid()), FILE_APPEND);
$taskPoll[$tkey]["task"]->increment_crash_counter();
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
$this->log(
sprintf(
"Task %s '%s' failed to start %d times"
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["cmd"]
, $taskPoll[$tkey]["task"]->get_crash_counter()
)
);
if($taskPoll[$tkey]["task"]->get_crash_counter() > 5)
{
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
}
else
{
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
}
}
@@ -398,73 +729,76 @@ class task_Scheduler
case task_abstract::STATUS_STARTED:
$crashed = false;
/**
* If no process, the task is probably manually ran
*/
if ($ttask[$tkey]["process"])
// If no process, the task is probably manually ran
if($taskPoll[$tkey]["process"])
{
$ttask[$tkey]["killat"] = NULL;
if (is_resource($ttask[$tkey]["process"]))
{
$proc_status = proc_get_status($ttask[$tkey]["process"]);
if ($proc_status['running'])
$taskPoll[$tkey]["killat"] = NULL;
// if(is_resource($taskPoll[$tkey]["process"]))
// {
// $proc_status = proc_get_status($taskPoll[$tkey]["process"]);
// if($proc_status['running'])
// $runningtask++;
// else
// $crashed = true;
// }
// else
// {
// $crashed = true;
// }
if($taskPoll[$tkey]['task']->get_pid())
$runningtask++;
else
$crashed = true;
}
else
{
$crashed = true;
}
}
if ($crashed === true && $ttask[$tkey]["task"]->get_status() === task_abstract::RETURNSTATUS_TORESTART)
if($crashed === true && $taskPoll[$tkey]["task"]->get_status() === task_abstract::RETURNSTATUS_TORESTART)
{
$crashed = false;
}
if ($crashed)
if($crashed)
{
$ttask[$tkey]["task"]->increment_crash_counter();
$taskPoll[$tkey]["task"]->increment_crash_counter();
@fclose($ttask[$tkey]["pipes"][1]);
@fclose($ttask[$tkey]["pipes"][2]);
@proc_close($ttask[$tkey]["process"]);
$ttask[$tkey]["process"] = null;
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$taskPoll[$tkey]["process"] = null;
$this->log(
sprintf(
"Task %s crashed %d times"
, $ttask[$tkey]["task"]->get_task_id()
, $ttask[$tkey]["task"]->get_crash_counter()
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["task"]->get_crash_counter()
)
);
$ttask[$tkey]["task"]->increment_crash_counter();
if ($ttask[$tkey]["task"]->get_crash_counter() > 5)
if($taskPoll[$tkey]["task"]->get_crash_counter() > 5)
{
$ttask[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
}
else
{
$ttask[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
$taskPoll[$tkey]["task"]->set_status(task_abstract::STATUS_TOSTART);
}
}
break;
case task_abstract::STATUS_TOSTOP:
if ($ttask[$tkey]["process"])
if($taskPoll[$tkey]["process"])
{
if ($ttask[$tkey]["killat"] === NULL)
$ttask[$tkey]["killat"] = time() + self::TASKDELAYTOQUIT;
if (($dt = $ttask[$tkey]["killat"] - time()) < 0)
if($taskPoll[$tkey]["killat"] === NULL)
$taskPoll[$tkey]["killat"] = time() + self::TASKDELAYTOQUIT;
if(($dt = $taskPoll[$tkey]["killat"] - time()) < 0)
{
$ppid = $ttask[$tkey]['task']->get_pid();
$ppid = $taskPoll[$tkey]['task']->get_pid();
$pids = preg_split('/\s+/', `ps -o pid --no-heading --ppid $ppid`);
foreach ($pids as $pid)
foreach($pids as $pid)
{
if (is_numeric($pid))
if(is_numeric($pid))
{
$this->log("Killing pid %d", $pid);
posix_kill($pid, 9);
@@ -474,27 +808,27 @@ class task_Scheduler
$this->log(
sprintf(
"SIGKILL sent to task %s (pid=%s)"
, $ttask[$tkey]["task"]->get_task_id()
, $ttask[$tkey]["task"]->get_pid()
, $taskPoll[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["task"]->get_pid()
)
);
proc_terminate($ttask[$tkey]["process"], 9);
@fclose($ttask[$tkey]["pipes"][1]);
@fclose($ttask[$tkey]["pipes"][2]);
proc_close($ttask[$tkey]["process"]);
unlink($lockdir . 'task_' . $ttask[$tkey]['task']->get_task_id() . '.lock');
proc_terminate($taskPoll[$tkey]["process"], 9);
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
proc_close($taskPoll[$tkey]["process"]);
unlink($lockdir . 'task_' . $taskPoll[$tkey]['task']->get_task_id() . '.lock');
$ttask[$tkey]["task"]->increment_crash_counter();
// $ttask[$tkey]["task"]->set_pid(null);
$ttask[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
$taskPoll[$tkey]["task"]->increment_crash_counter();
// $taskPoll[$tkey]["task"]->set_pid(null);
$taskPoll[$tkey]["task"]->set_status(task_abstract::RETURNSTATUS_STOPPED);
}
else
{
$this->log(
sprintf(
"waiting task %s to quit (kill in %d seconds)"
, $ttask[$tkey]["task"]->get_task_id()
, $taskPoll[$tkey]["task"]->get_task_id()
, $dt
)
);
@@ -505,27 +839,38 @@ class task_Scheduler
case task_abstract::RETURNSTATUS_STOPPED:
case task_abstract::RETURNSTATUS_TODELETE:
if ($ttask[$tkey]["process"])
if($taskPoll[$tkey]["process"])
{
@fclose($ttask[$tkey]["pipes"][1]);
@fclose($ttask[$tkey]["pipes"][2]);
@proc_close($ttask[$tkey]["process"]);
@fclose($taskPoll[$tkey]["pipes"][1]);
@fclose($taskPoll[$tkey]["pipes"][2]);
@proc_close($taskPoll[$tkey]["process"]);
$ttask[$tkey]["process"] = null;
$taskPoll[$tkey]["process"] = null;
}
break;
}
}
*/
$to_reopen = false;
if ($conn->ping())
if($conn->ping())
{
$conn->close();
unset($conn);
$to_reopen = true;
}
sleep($sleeptime);
if ($to_reopen)
if($to_reopen)
{
$conn = self::get_connection();
}

180
lib/classes/task/abstract.class.php
View File

@@ -21,6 +21,8 @@ abstract class task_abstract
const STATE_MAXRECSDONE = 'STATE_MAXRECS';
const STATE_FINISHED = 'STATE_FINISHED';
const SIGNAL_SCHEDULER_DIED = 'SIGNAL_SCHEDULER_DIED';
protected $suicidable = false;
protected $launched_by = 0;
@@ -129,7 +131,7 @@ abstract class task_abstract
, self::RETURNSTATUS_TORESTART
, self::STATUS_TOSTART
);
if (!in_array($status, $av_status))
if(!in_array($status, $av_status))
throw new Exception(sprintf('unknown status `%s`', $status));
@@ -145,18 +147,17 @@ abstract class task_abstract
return $this->task_status;
}
public function set_pid($pid)
{
$conn = connection::getPDOConnection();
$sql = 'UPDATE task2 SET pid = :pid WHERE task_id = :taskid';
$stmt = $conn->prepare($sql);
$stmt->execute(array(':pid' => $pid, ':taskid' => $this->get_task_id()));
$stmt->closeCursor();
return $this;
}
// public function set_pid($pid)
// {
// $conn = connection::getPDOConnection();
//
// $sql = 'UPDATE task2 SET pid = :pid WHERE task_id = :taskid';
// $stmt = $conn->prepare($sql);
// $stmt->execute(array(':pid' => $pid, ':taskid' => $this->get_task_id()));
// $stmt->closeCursor();
//
// return $this;
// }
// 'active' means 'auto-start when scheduler starts'
public function set_active($boolean)
{
@@ -221,6 +222,7 @@ abstract class task_abstract
{
return $this->crash_counter;
}
public function increment_crash_counter()
{
$conn = connection::getPDOConnection();
@@ -268,9 +270,9 @@ abstract class task_abstract
$this->system = system_server::get_platform();
$this->launched_by = array_key_exists("REQUEST_URI", $_SERVER) ? self::LAUCHED_BY_BROWSER : self::LAUCHED_BY_COMMANDLINE;
if ($this->system != "DARWIN" && $this->system != "WINDOWS" && $this->system != "LINUX")
if($this->system != "DARWIN" && $this->system != "WINDOWS" && $this->system != "LINUX")
{
if ($this->launched_by == self::LAUCHED_BY_COMMANDLINE)
if($this->launched_by == self::LAUCHED_BY_COMMANDLINE)
{
// printf("Desole, ce programme ne fonctionne pas sous '" . $this->system . "'.\n");
flush();
@@ -279,7 +281,7 @@ abstract class task_abstract
}
else
{
if ($this->launched_by == self::LAUCHED_BY_COMMANDLINE)
if($this->launched_by == self::LAUCHED_BY_COMMANDLINE)
{
flush();
}
@@ -289,11 +291,14 @@ abstract class task_abstract
{
$conn = connection::getPDOConnection();
}
catch (Exception $e)
catch(Exception $e)
{
$this->log($e->getMessage());
$this->log(("Warning : abox connection lost, restarting in 10 min."));
sleep(60 * 10);
for($t = 60 * 10; $t > 0; $t--) // DON'T do sleep(600) because it prevents ticks !
sleep(1);
$this->running = false;
return('');
@@ -304,7 +309,7 @@ abstract class task_abstract
$stmt->execute(array(':taskid' => $this->get_task_id()));
$row = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
if (!$row)
if(!$row)
throw new Exception('Unknown task id');
$this->title = $row['name'];
$this->crash_counter = (int) $row['crashed'];
@@ -357,7 +362,7 @@ abstract class task_abstract
$stmt->closeCursor();
$lock_file = $registry->get('GV_RootPath') . 'tmp/locks/task_' . $this->get_task_id() . '.lock';
if (is_writable($lock_file))
if(is_writable($lock_file))
unlink($lock_file);
return;
@@ -366,7 +371,7 @@ abstract class task_abstract
protected function check_memory_usage()
{
$current_memory = memory_get_usage();
if ($current_memory >> 20 >= $this->maxmegs)
if($current_memory >> 20 >= $this->maxmegs)
{
$this->log(sprintf(
"Max memory (%s M) reached (current is %s M)"
@@ -381,7 +386,7 @@ abstract class task_abstract
protected function check_records_done()
{
if ($this->records_done >= (int) ($this->maxrecs))
if($this->records_done >= (int) ($this->maxrecs))
{
$this->current_state = self::STATE_MAXRECSDONE;
}
@@ -402,26 +407,68 @@ abstract class task_abstract
public function get_pid()
{
$conn = connection::getPDOConnection();
$sql = 'SELECT pid FROM task2 WHERE task_id = :taskid';
$stmt = $conn->prepare($sql);
$stmt->execute(array(':taskid' => $this->get_task_id()));
$row = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
$pid = NULL;
$pid = $row ? $row['pid'] : null;
$taskid = $this->get_task_id();
$registry = registry::get_instance();
system_file::mkdir($lockdir = $registry->get('GV_RootPath') . 'tmp/locks/');
if(($fd = fopen(($lockfile = ($lockdir . 'task_' . $taskid . '.lock')), 'a+')))
{
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : fopen(%s) \n", __FILE__, __LINE__, $lockfile), FILE_APPEND);
if(flock($fd, LOCK_EX | LOCK_NB) === FALSE)
{
// already locked ? : task running
$pid = fgets($fd);
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : can't flock() : pid=%s \n", __FILE__, __LINE__, $pid), FILE_APPEND);
}
else
{
// can lock : not running
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : NOT RUNNING can flock() : pid=%s \n", __FILE__, __LINE__, file_get_contents($lockfile)), FILE_APPEND);
flock($fd, LOCK_UN);
}
fclose($fd);
}
return $pid;
}
/*
public function is_running()
{
$retval = false;
$registry = registry::get_instance();
$lockdir = $registry->get('GV_RootPath') . 'tmp/locks/';
$tasklock = fopen($lockfile = ($lockdir . '/task_' . $this->get_task_id() . '.lock'), 'a+');
if (flock($tasklock, LOCK_SH | LOCK_NB) != true)
{
$retval = true;
}
else
{
ftruncate($tasklock, 0);
flock($tasklock, LOCK_UN | LOCK_NB);
fclose($tasklock);
unlink($lockfile);
}
return $retval;
}
*/
protected function check_current_state()
{
switch ($this->current_state)
switch($this->current_state)
{
case self::STATE_MAXMEGSREACHED:
case self::STATE_MAXRECSDONE:
default:
if ($this->get_runner() == self::RUNNER_SCHEDULER)
if($this->get_runner() == self::RUNNER_SCHEDULER)
{
$this->task_status = self::STATUS_TOSTOP;
$this->return_value = self::RETURNSTATUS_TORESTART;
@@ -429,7 +476,7 @@ abstract class task_abstract
break;
case self::STATE_FINISHED:
$this->task_status = self::STATUS_TOSTOP;
if ($this->suicidable === true)
if($this->suicidable === true)
{
$this->return_value = self::RETURNSTATUS_TODELETE;
$this->log('will hang myself');
@@ -458,13 +505,13 @@ abstract class task_abstract
$stmt->execute(array(':taskid' => $this->get_task_id()));
$row = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
if (!$row || $row['status'] == 'tostop')
if(!$row || $row['status'] == 'tostop')
{
$this->task_status = self::STATUS_TOSTOP;
$this->return_value = self::RETURNSTATUS_STOPPED;
}
}
catch (Exception $e)
catch(Exception $e)
{
$this->return_value = self::RETURNSTATUS_STOPPED;
$this->task_status = self::STATUS_TOSTOP;
@@ -477,21 +524,26 @@ abstract class task_abstract
protected function pause($when_started=0)
{
$this->log($this->records_done . ' records done');
if ($this->running)// && $this->records_done == 0)
if($this->running)// && $this->records_done == 0)
{
$when_started = time() - $when_started;
if ($when_started < $this->period)
if($when_started < $this->period)
{
$conn = connection::getPDOConnection();
$conn->close();
unset($conn);
sleep($this->period - $when_started);
// sleep($this->period - $when_started);
for($t = $this->period - $when_started; $t > 0; $t--) // DON'T do sleep($this->period - $when_started) because it prevents ticks !
sleep(1);
}
}
}
final public function run($runner)
{
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : LAUNCHING : tid=%s \n", __FILE__, __LINE__, $this->get_task_id()), FILE_APPEND);
$taskid = $this->get_task_id();
$conn = connection::getPDOConnection();
@@ -499,8 +551,10 @@ abstract class task_abstract
system_file::mkdir($lockdir = $registry->get('GV_RootPath') . 'tmp/locks/');
$locker = true;
$tasklock = fopen(($lockfile = ($lockdir . 'task_' . $taskid . '.lock')), 'a+');
if (flock($tasklock, LOCK_EX | LOCK_NB, $locker) != true)
if(flock($tasklock, LOCK_EX | LOCK_NB, $locker) === FALSE)
{
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : LAUNCH OPENED AND CANT LOCK : pid=%s \n", __FILE__, __LINE__, getmypid()), FILE_APPEND);
printf(("runtask::ERROR : task already running.\n"), $taskid);
fclose($tasklock);
@@ -511,6 +565,8 @@ abstract class task_abstract
ftruncate($tasklock, 0);
fwrite($tasklock, '' . getmypid());
fflush($tasklock);
// ************************************************
// file_put_contents("/tmp/scheduler2.log", sprintf("%s [%d] : LAUNCH OPENED AND LOCKED : pid=%s \n", __FILE__, __LINE__, getmypid()), FILE_APPEND);
}
$this->set_runner($runner);
@@ -526,7 +582,7 @@ abstract class task_abstract
fclose($tasklock);
@unlink($lockfile);
if ($this->return_value == self::RETURNSTATUS_TODELETE)
if($this->return_value == self::RETURNSTATUS_TODELETE)
$this->delete();
else
$this->set_status($this->return_value);
@@ -536,42 +592,20 @@ abstract class task_abstract
abstract protected function run2();
public function is_running()
{
$retval = false;
$registry = registry::get_instance();
$lockdir = $registry->get('GV_RootPath') . 'tmp/locks/';
$tasklock = fopen($lockfile = ($lockdir . '/task_' . $this->get_task_id() . '.lock'), 'a+');
if (flock($tasklock, LOCK_SH | LOCK_NB) != true)
{
$retval = true;
}
else
{
ftruncate($tasklock, 0);
flock($tasklock, LOCK_UN | LOCK_NB);
fclose($tasklock);
unlink($lockfile);
}
return $retval;
}
protected function load_settings(SimpleXMLElement $sx_task_settings)
{
$this->period = (int) $sx_task_settings->period;
if ($this->period <= 0 || $this->period >= 60 * 60)
if($this->period <= 0 || $this->period >= 60 * 60)
$this->period = 60;
$this->maxrecs = (int) $sx_task_settings->maxrecs;
if ($sx_task_settings->maxrecs < 10 || $sx_task_settings->maxrecs > 1000)
if($sx_task_settings->maxrecs < 10 || $sx_task_settings->maxrecs > 1000)
$this->maxrecs = 100;
$this->maxmegs = (int) $sx_task_settings->maxmegs;
if ($sx_task_settings->maxmegs < 16 || $sx_task_settings->maxmegs > 512)
if($sx_task_settings->maxmegs < 16 || $sx_task_settings->maxmegs > 512)
$this->maxmegs = 24;
$this->record_buffer_size = (int) $sx_task_settings->flush;
if ($sx_task_settings->flush < 1 || $sx_task_settings->flush > 100)
if($sx_task_settings->flush < 1 || $sx_task_settings->flush > 100)
$this->record_buffer_size = 10;
return $this;
@@ -579,7 +613,7 @@ abstract class task_abstract
protected function increment_loops()
{
if ($this->get_runner() == self::RUNNER_SCHEDULER && $this->loop > $this->maxloops)
if($this->get_runner() == self::RUNNER_SCHEDULER && $this->loop > $this->maxloops)
{
$this->log(sprintf(('%d loops done, restarting'), $this->loop));
$this->task_status = self::STATUS_TOSTOP;
@@ -593,7 +627,7 @@ abstract class task_abstract
public function apply_task_status()
{
if ($this->task_status == self::STATUS_TOSTOP)
if($this->task_status == self::STATUS_TOSTOP)
{
$this->running = false;
}
@@ -605,7 +639,7 @@ abstract class task_abstract
{
static $lastt = null;
$t = explode(' ', ($ut = microtime()));
if ($lastt === null)
if($lastt === null)
$lastt = $t;
$dt = ($t[0] - $lastt[0]) + ($t[1] - $lastt[1]);
@@ -638,7 +672,7 @@ abstract class task_abstract
*/
public static function create(appbox $appbox, $class_name, $settings = null)
{
if (!class_exists($class_name))
if(!class_exists($class_name))
throw new Exception('Unknown task class');
$sql = 'INSERT INTO task2
@@ -649,9 +683,9 @@ abstract class task_abstract
:name, "0000/00/00 00:00:00", :class, :settings)';
if ($settings && !DOMDocument::loadXML($settings))
if($settings && !DOMDocument::loadXML($settings))
throw new Exception('settings invalide');
elseif (!$settings)
elseif(!$settings)
$settings = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<tasksettings>\n</tasksettings>";
$params = array(
@@ -673,7 +707,7 @@ abstract class task_abstract
{
global $argc, $argv;
$t = "usage: " . $argv[0] . " [options]\noptions:\n";
foreach ($this->argt as $n => $v)
foreach($this->argt as $n => $v)
$t .= "\t" . $n . $v["usage"] . "\n";
return($t);
@@ -704,7 +738,7 @@ abstract class task_abstract
$stmt->closeCursor();
$this->completed_percentage = $p;
}
catch (Exception $e)
catch(Exception $e)
{
}

37
lib/classes/task/databoxAbstract.class.php
View File

@@ -17,7 +17,6 @@
*/
abstract class task_databoxAbstract extends task_abstract
{
// abstract public function help();
//
// abstract public function getName();
@@ -34,17 +33,20 @@ abstract class task_databoxAbstract extends task_abstract
protected function run2()
{
while ($this->running)
while($this->running)
{
try
{
$conn = connection::getPDOConnection();
}
catch (Exception $e)
catch(Exception $e)
{
$this->log($e->getMessage());
$this->log(("Warning : abox connection lost, restarting in 10 min."));
sleep(60 * 10);
for($t = 60 * 10; $t; $t--) // DON'T do sleep(600) because it prevents ticks !
sleep(1);
$this->running = false;
$this->return_value = self::RETURNSTATUS_TORESTART;
@@ -63,24 +65,24 @@ abstract class task_databoxAbstract extends task_abstract
$this->records_done = 0;
$duration = time();
}
catch (Exception $e)
catch(Exception $e)
{
$this->task_status = self::STATUS_TOSTOP;
$this->return_value = self::RETURNSTATUS_STOPPED;
$rs = array();
}
foreach ($rs as $row)
foreach($rs as $row)
{
if (!$this->running)
if(!$this->running)
break;
$this->sbas_id = (int) $row['sbas_id'];
if ($this->mono_sbas_id && $this->sbas_id !== $this->mono_sbas_id)
if($this->mono_sbas_id && $this->sbas_id !== $this->mono_sbas_id)
{
continue;
}
if ($this->mono_sbas_id)
if($this->mono_sbas_id)
{
$this->log('This task works on ' . phrasea::sbas_names($this->mono_sbas_id));
}
@@ -89,7 +91,7 @@ abstract class task_databoxAbstract extends task_abstract
{
$this->load_settings(simplexml_load_string($row['settings']));
}
catch (Exception $e)
catch(Exception $e)
{
$this->log($e->getMessage());
continue;
@@ -126,7 +128,7 @@ abstract class task_databoxAbstract extends task_abstract
*/
$rs = $this->retrieve_sbas_content($databox);
}
catch (Exception $e)
catch(Exception $e)
{
$this->log('Error : ' . $e->getMessage());
$rs = array();
@@ -135,10 +137,10 @@ abstract class task_databoxAbstract extends task_abstract
$rowstodo = count($rs);
$rowsdone = 0;
if ($rowstodo > 0)
if($rowstodo > 0)
$this->setProgress(0, $rowstodo);
foreach ($rs as $row)
foreach($rs as $row)
{
try
{
@@ -148,7 +150,7 @@ abstract class task_databoxAbstract extends task_abstract
*/
$this->process_one_content($databox, $row);
}
catch (Exception $e)
catch(Exception $e)
{
$this->log("Exception : " . $e->getMessage()
. " " . basename($e->getFile()) . " " . $e->getLine());
@@ -166,7 +168,7 @@ abstract class task_databoxAbstract extends task_abstract
->check_records_done()
->check_task_status();
if (!$this->running)
if(!$this->running)
break;
}
@@ -175,17 +177,16 @@ abstract class task_databoxAbstract extends task_abstract
->check_records_done()
->check_task_status();
if ($connbas instanceof PDO)
if($connbas instanceof PDO)
{
$connbas->close();
unset($connbas);
}
if ($rowstodo > 0)
if($rowstodo > 0)
$this->setProgress(0, 0);
return $this;
}
}

49
lib/classes/task/manager.class.php
View File

@@ -90,34 +90,26 @@ class task_manager
continue;
try
{
if( ($lock = fopen( $lockdir . 'task.'.$row['task_id'].'.lock', 'a+')) )
{
if (flock($lock, LOCK_SH | LOCK_NB) === FALSE)
{
// already locked : running !
$row['pid'] = fgets($lock, 512);
}
else
{
// can lock : not running
flock($lock, LOCK_UN);
}
fclose($lock);
}
// if( ($lock = fopen( $lockdir . 'task.'.$row['task_id'].'.lock', 'a+')) )
// {
// if (flock($lock, LOCK_SH | LOCK_NB) === FALSE)
// {
// // already locked : running !
// $row['pid'] = fgets($lock, 512);
// }
// else
// {
// // can lock : not running
// flock($lock, LOCK_UN);
// }
// fclose($lock);
// }
$tasks[$row['task_id']] = new $classname($row['task_id']);
}
catch (Exception $e)
{
}
}
$this->tasks = $tasks;
@@ -175,7 +167,12 @@ class task_manager
public function get_scheduler_state2()
{
$ret = array('pid'=>NULL, 'qdelay'=>NULL);
$sql = "SELECT UNIX_TIMESTAMP()-UNIX_TIMESTAMP(schedqtime) AS qdelay, schedstatus AS status FROM sitepreff";
$stmt = $this->appbox->get_connection()->prepare($sql);
$stmt->execute();
$ret = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
$ret['pid'] = NULL;
$appbox = appbox::get_instance();
$lockdir = $appbox->get_registry()->get('GV_RootPath') . 'tmp/locks/';
@@ -184,12 +181,6 @@ class task_manager
if (flock($schedlock, LOCK_SH | LOCK_NB) === FALSE)
{
// already locked : running !
$sql = "SELECT UNIX_TIMESTAMP()-UNIX_TIMESTAMP(schedqtime) AS qdelay FROM sitepreff";
$stmt = $this->appbox->get_connection()->prepare($sql);
$stmt->execute();
$ret = $stmt->fetch(PDO::FETCH_ASSOC);
$stmt->closeCursor();
$ret['pid'] = fgets($schedlock, 512);
}
else

18
www/admin/adminFeedback.php
View File

@@ -107,9 +107,7 @@ switch ($parm['action'])
break;
case 'SETTASKSTATUS':
$parm = $request->get_parms("task_id", "status");
try
{
$task_manager = new task_manager($appbox);
@@ -120,13 +118,10 @@ switch ($parm['action'])
{
}
break;
case 'SETSCHEDSTATUS':
$parm = $request->get_parms('status');
try
{
$task_manager = new task_manager($appbox);
@@ -145,10 +140,9 @@ switch ($parm['action'])
$output = $ret->saveXML();
break;
case 'RESETTASKCRASHCOUNTER':
$parm = $request->get_parms("task_id");
try
{
$task_manager = new task_manager($appbox);
@@ -166,15 +160,10 @@ switch ($parm['action'])
$root->appendChild($ret->createCDATASection(var_export($parm, true)));
$output = $ret->saveXML();
break;
case 'CHANGETASK':
$parm = $request->get_parms('act', 'task_id', "usr");
$ret = new DOMDocument("1.0", "UTF-8");
$ret->standalone = true;
$ret->preserveWhiteSpace = false;
@@ -200,7 +189,7 @@ switch ($parm['action'])
$output = $ret->saveXML();
break;
/*
case 'PINGSCHEDULER':
$lockdir = $registry->get('GV_RootPath') . 'tmp/locks/';
@@ -267,7 +256,7 @@ switch ($parm['action'])
break;
*/
case 'PINGSCHEDULER_JS':
$ret = array();
$ret['time'] = $dat = date("H:i:s");
@@ -284,6 +273,7 @@ switch ($parm['action'])
, 'pid' =>$task->get_pid()
, 'crashed'=>$task->get_crash_counter()
, 'completed'=>$task->get_completed_percentage()
, 'status'=>$task->get_status()
);
$ret['tasks'][$_t['id']] = $_t;
}

65
www/admin/taskmanager.php
View File

@@ -107,11 +107,12 @@ $task_manager = new task_manager($appbox);
editTask(tid);
break;
case "start":
setTaskStatus(tid, 'tostart');
setTaskStatus(tid, 'tostart', true); // true : reset crash counter
break;
case "stop":
setTaskStatus(tid, 'tostop');
setTaskStatus(tid, 'tostop', false);
break;
/*
case "fix":
switch(T_task[tid])
{
@@ -135,6 +136,7 @@ $task_manager = new task_manager($appbox);
break;
}
break;
*/
case 'delete':
switch(T_task[tid])
{
@@ -222,7 +224,7 @@ foreach($tasks as $t)
$('#newTaskButton').contextMenu(
menuNewTask,
{
theme:'vista'
// theme:'vista'
}
);
@@ -358,7 +360,7 @@ foreach($tasks as $t)
);
self.setTimeout("pingScheduler();", 100);
self.setTimeout("pingScheduler(true);", 100); // true : loop forever each 2 sec
})
</script>
</head>
@@ -469,18 +471,27 @@ foreach($tasks as $t)
document.forms["task"].submit();
}
function setTaskStatus(tid, status)
function setTaskStatus(tid, status, resetCrashCounter)
{
if(resetCrashCounter)
{
var url = "/admin/adminFeedback.php";
url += "?action=SETTASKSTATUS&task_id=" + encodeURIComponent(tid);
url += "&status=" + encodeURIComponent(status);
$.ajax({
url: url,
url: "/admin/adminFeedback.php",
data : { task_id:tid, action:"RESETTASKCRASHCOUNTER" },
dataType:'xml',
success: function(ret)
{
}
});
}
$.ajax({
url: "/admin/adminFeedback.php",
data : {task_id:tid, action:"SETTASKSTATUS", status:status},
dataType:'xml',
success: function(ret)
{
pingScheduler(false); // false : just one time
}
});
}
@@ -488,15 +499,13 @@ foreach($tasks as $t)
function setSchedStatus(status)
{
var url = "/admin/adminFeedback.php";
url += "?action=SETSCHEDSTATUS&status=" + encodeURIComponent(status);
$.ajax({
url: url,
url: "/admin/adminFeedback.php",
data : { action:"SETSCHEDSTATUS", status:status },
dataType:'xml',
success: function(ret)
{
pingScheduler(false); // false : just one time
}
});
}
@@ -512,7 +521,7 @@ foreach($tasks as $t)
document.forms["taskManager"].submit();
}
}
/*
function old_pingScheduler()
{
$.ajax({
@@ -652,9 +661,10 @@ foreach($tasks as $t)
}
});
}
*/
function pingScheduler()
function pingScheduler(repeat)
{
$.ajax({
url: '/admin/adminFeedback.php?action=PINGSCHEDULER_JS',
@@ -663,22 +673,37 @@ foreach($tasks as $t)
{
retPing = ret; // global
$("#pingTime").empty().append(ret.time);
if(ret.scheduler && ret.scheduler.pid)
if(ret.scheduler)
{
if(ret.scheduler.status)
$("#STATUS_SCHED").html(ret.scheduler.status);
else
$("#STATUS_SCHED").html('');
if(ret.scheduler.pid)
$("#PID_SCHED").html(ret.scheduler.pid);
else
$("#PID_SCHED").html('-');
}
else
{
$("#STATUS_SCHED").html('');
$("#PID_SCHED").html('-');
}
if(ret.tasks)
{
for(id in ret.tasks)
{
if(ret.tasks[id].status)
$("#STATUS_"+id).html(ret.tasks[id].status);
else
$("#STATUS_"+id).html('');
if(ret.tasks[id].pid)
$("#PID_"+id).html(ret.tasks[id].pid);
else
$("#PID_"+id).html('-');
if(ret.tasks[id].completed && ret.tasks[id].completed>0 && ret.tasks[id].completed<=100)
{
$("#COMP_"+id).width(ret.tasks[id].completed + "%");
@@ -691,8 +716,8 @@ foreach($tasks as $t)
}
}
}
self.setTimeout("pingScheduler();", 3000);
if(repeat)
self.setTimeout("pingScheduler(true);", 1000);
}
});
}