Parallel.php

Go to the documentation of this file.

<?php
 
namespace Rsi\Fred;
 
/**
 *  Component for parallel execution.
 */
class Parallel extends Component{
 
  public $timeout = 600; //!< Time after which the results may be removed by the garbage collector (seconds).
  public $garbageChance = 10000; //!< Run garbage collector once every ... times a call is not ready.
  public $lastError = [];
 
  protected $_path = null; //!< Path for storing the result files.
  protected $_ext = '.dat';
  protected $_initCode = null; //!< Initialization code (PHP).
  protected $_fredVar = '$fred'; //!< Name for the variable that holds the \\Rsi\\Fred object after initalization.
 
  protected function filename($id){
    if(!$this->_path) throw new \Exception('No path set');
    if(!is_scalar($id) || !preg_match('/^\\w+$/',$id)) throw new \Exception('Invalid ID');
    return $this->_path . $id . $this->_ext;
  }
 
  protected function writeResult($id,$result = null){
    file_put_contents($this->filename($id),func_num_args() > 1 ? serialize($result) : null);
  }
 
  protected function readResult($id){
    $result = unserialize(file_get_contents($filename = $this->filename($id)));
    \Rsi\File::unlink($filename);
    return $result;
  }
 
  public function worker($id,$component_name,$func_name,$params = null){
    $result = [];
    if(is_string($component_name) && is_string($func_name)) try{
      $result['result'] = call_user_func_array([$this->component($component_name),$func_name],is_array($params) ? $params : []);
    }
    catch(\Exception $e){
      $result += [
        'error' => $e->getMessage(),
        'filename' => $e->getFile(),
        'lineNo' => $e->getLine(),
        'trace' => $e->getTrace()
      ];
    }
    $this->writeResult($id,$result);
    $this->_fred->halt();
  }
  /**
   * Execute a command in the background.
   * @param string $component_name  Component to run the command on.
   * @param string $func_name  Function to call.
   * @param array $params  Parameters for the call.
   * @return string  Reference ID for the call (false on error).
   */
  public function execute($component_name,$func_name,$params = null){
    if(!is_string($component_name) || !is_string($func_name)) return false;
    $this->writeResult($id = \Rsi\Str::random());
    \Rsi::executeParallel(escapeshellarg(PHP_BINARY) . ' -r ' . escapeshellarg(
      $this->_initCode .
      $this->_fredVar . '->component(' . var_export($this->_name,true) . ')->worker(' .
        var_export($id,true) . ',' . var_export($component_name,true) . ',' . var_export($func_name,true) .
        (is_array($params) ? ',' . var_export(array_values($params),true) : '') .
      ');'
    ));
    return $id;
  }
 
  protected function garbage(){
    $time = time() - $this->timeout;
    if($this->_path) foreach((new \GlobIterator($this->_path . '*' . $this->_ext)) as $filename => $file) try{
      if($file->getMTime() < $time) unlink($filename);
    }
    catch(\Exception $e){
      $this->component('log')->info($e);
      error_clear_last();
    }
  }
  /**
   * Check if the calls are ready.
   * @param string $ids  Reference ID's for the calls.
   * @return bool  True if all calls are ready.
   */
  public function ready(...$ids){
    foreach($ids as $id){
      clearstatcache(false,$filename = $this->filename($id));
      if(!filesize($filename)){
        if(!rand(0,$this->garbageChance)) $this->garbage();
        return false;
      }
    }
    return true;
  }
  /**
   * Get the result from the parallel call.
   * @param string $id  Reference ID for the call.
   * @return mixed
   */
  public function result($id){
    if(array_key_exists('result',$result = $this->readResult($id))) return $result['result'];
    $this->lastError = $result;
    throw new \Exception($result['error'] ?? 'Empty result');
  }
 
  protected function getAvailable(){
    return $this->_path && $this->_initCode;
  }
 
}