Worker.php

Go to the documentation of this file.

<?php
 
namespace Rsi\Fred;
 
class Worker extends Component{
 
  public $classNameAlias = []; //!< Job class name aliasses (key = alias, value = class name).
  public $parallelCommand = null; //!< Command for executing a single job as a seperate process (asterisk is replaced with job
    // ID, dollar sign with PHP binary path; empty = no parallel processing).
  public $timeLimit = 60; //!< Default time limit for a job (seconds).
  public $minTime = 10; //!< Keep processing for at least this amount of time (seconds; even when there are no more jobs
    // scheduled).
  public $sleepTime = 0.5; //!< Time to sleep when there are no more jobs scheduled (but still within minTime; seconds).
  public $maxTime = 300; //!< Halt the processig of jobs after this time (seconds; even when there are more jobs scheduled).
  public $maxJobs = 100; //!< Halt the processig of jobs after this amount of jobs processed (even when there are more jobs
    // scheduled).
  public $defaultOverduePeriod = '+1 hour'; //!< Default overdue period for jobs that do not have a rescheduling interval.
  public $logTimerPrio = null; //!< Log the begin and end times of a job with this (custom) prio.
 
  /**
   * Translate a job class name.
   * @param string $class_name  Class name for the job, or an alias.
   * @return string  Class name.
   */
  protected function transClassName($class_name){
    return $this->classNameAlias[$class_name] ?? $class_name;
  }
  /**
   * Add a job to the queue.
   * @param string|array $class_name  Class name for the job (or array with group code and class name).
   * @param array $config  Configuration options (key-value pairs) for the job.
   * @param int $time  Time to run the job (null for as soon as possible).
   * @param string $interval  String that is understood by PHP's strtotime() to reschedule the job after it is finished. Prefix
   *   with an '@' to make it relative to the time of finishing (normally it is relative to the planned time).
   * @param int $max_time  Maximum time for executing job (seconds).
   * @return string  Job ID when successful.
   */
  public function schedule($class_name,$config = null,$time = null,$interval = null,$max_time = null){
    return false;
  }
  /**
   * Delete a scheduled job from the queue.
   * @param string $id  Job ID.
   * @return bool  True when successful.
   */
  public function delete($id){
    return false;
  }
  /**
   * Create a job object.
   * @param string $class_name  Class name for the job.
   * @param string $id  Unique ID for the job.
   * @param array $config  Configuration options (key-value pairs) for the job.
   * @param int $time  Time to run the job (null for as soon as possible).
   * @param string $interval  String that is understood by PHP's strtotime() to reschedule the job after it is finished. Prefix
   *   with an '@' to make it relative to the time of finishing (normally it is relative to the planned time).
   * @param int $max_time  Maximum time for executing job (seconds).
   * @return \\Rsi\\Fred\\Worker\\Job
   */
  protected function create($class_name,$id,$config = null,$time = null,$interval = null,$max_time = null){
    return new $class_name($this,$id,$config,$time,$interval,$max_time);
  }
  /**
   * Run a job instantly.
   * @param string $class_name  Class name for the job.
   * @param array $config  Configuration options (key-value pairs) for the job.
   * @param string $timer_id  Timer ID for the job to log to.
   */
  public function instant($class_name,$config = null,$timer_id = null){
    $this->create($class_name,null,$config)->execute($timer_id);
  }
  /**
   * Next scheduled job to execute.
   * @param string $id  Only check this job ID.
   * @return \\Rsi\\Fred\\Worker\\Job (empty if there is no job to be executed).
   */
  protected function next($id = null){
    return null;
  }
  /**
   * Lock a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to lock.
   * @return bool  True if the job has been locked.
   */
  protected function lock($job){
    return false;
  }
  /**
   * Unlock a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to unlock.
   * @return bool  True if the job has been unlocked.
   */
  protected function unlock($job){
    return true;
  }
  /**
   * Reschedule a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to reschedule.
   * @param int $time  Time to reschedule at (null for as soon as possible).
   * @return bool  True if the job has been rescheduled.
   */
  protected function reschedule($job,$time = null){
    return false;
  }
  /**
   * Mark a job as done.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job.
   */
  protected function ready($job){
    $this->delete($job->id);
  }
  /**
   * Number of active jobs in a group.
   * @param int $group_id  Group ID.
   * @return int  Number of active (locked) jobs.
   */
  protected function groupCount($group_id){
    return 0;
  }
  /**
   * Check if the group (if any) is within its limit.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job.
   * @return bool  True when OK.
   */
  protected function checkGroup($job){
    return !$job->groupId || !$job->groupMaxCount || ($this->groupCount($job->groupId) < $job->groupMaxCount);
  }
  /**
   * Overdue jobs.
   * Jobs that have been locked longer than twice their interval or the default overdue interval.
   * @return array  Of \\Rsi\\Fred\\Worker\\Job items.
   */
  protected function overdue(){
    return [];
  }
  /**
   * Execute the worker / process the scheduled jobs.
   * Call his function as often as possible.
   * @param string $id  Only check this job ID.
   */
  public function execute($id = null){
    $log = $this->component('log');
    $count = 0;
    $start = time();
    if($this->parallelCommand && !$id){
      $done = [];
      while(time() - $start <= $this->minTime){
        if(($job = $this->next()) && (($done[$job->id] ?? -1) != $job->time) && $this->checkGroup($job)){
          $command = strtr($this->parallelCommand,['*' => $job->id,'$' => escapeshellarg(PHP_BINARY)]);
          $log->info("Executing job '{$job->id}' in parallel",__FILE__,__LINE__,compact('command'));
          \Rsi::executeParallel($command);
          if((++$count >= $this->maxJobs) || (time() - $start >= $this->maxTime)) break;
          $done[$job->id] = $job->time;
        }
        else usleep($this->sleepTime * 1000000);
      }
    }
    else do{
      while(($job = $this->next($id)) && $this->checkGroup($job) && $this->lock($job)) try{
        $this->_fred->timeLimit = $this->timeLimit;
        $timer_id = $this->logTimerPrio ? $log->start($this->logTimerPrio,"Executing job '{$job->id}'",__FILE__,__LINE__) : null;
        $job->execute($timer_id);
        if($timer_id) $log->stop($timer_id);
        if($job->interval){
          $diff = ($time = $job->nextTime) - time();
          if(!$time) $log->alert("Invalid interval for job '{$job->id}'",__FILE__,__LINE__);
          elseif($diff < 0) $log->warning("Job '{$job->id}' rescheduled in past",['diff' => $diff]);
          $this->reschedule($job,$time);
        }
        else $this->ready($job);
        $this->unlock($job);
        if((++$count >= $this->maxJobs) || (time() - $start >= $this->maxTime)) break;
      }
      catch(\Exception $e){
        $log->critical($e);
      }
      usleep($this->sleepTime * 1000000);
    }
    while(!$id && (time() - $start <= $this->minTime));
    foreach($this->overdue() as $job) $log->critical("Job '{$job->id}' is overdue",['time' => $job->time,'interval' => $job->interval]);
  }
 
}