Implemented supervisors, refactored some stuff, implemented closures, updated examples and added dependency for Symfony\Process

This commit is contained in:
Netkas 2023-02-05 17:24:22 -05:00
parent 84b89eaf9d
commit 1b8d2fb40a
18 changed files with 707 additions and 190 deletions

View file

@ -52,6 +52,12 @@
"version": "latest", "version": "latest",
"source_type": "remote", "source_type": "remote",
"source": "php-amqplib/php-amqplib=latest@composer" "source": "php-amqplib/php-amqplib=latest@composer"
},
{
"name": "com.symfony.process",
"version": "latest",
"source_type": "remote",
"source": "symfony/process=latest@composer"
} }
], ],
"configurations": [ "configurations": [

View file

@ -2,14 +2,30 @@
namespace Tamer\Classes; namespace Tamer\Classes;
use Exception;
use InvalidArgumentException; use InvalidArgumentException;
use OptsLib\Parse; use OptsLib\Parse;
use Symfony\Component\Process\PhpExecutableFinder;
use Tamer\Abstracts\ProtocolType; use Tamer\Abstracts\ProtocolType;
use Tamer\Interfaces\ClientProtocolInterface; use Tamer\Interfaces\ClientProtocolInterface;
use Tamer\Interfaces\WorkerProtocolInterface; use Tamer\Interfaces\WorkerProtocolInterface;
class Functions class Functions
{ {
/**
* A cache of the worker variables
*
* @var array|null
*/
private static $worker_variables;
/**
* A cache of the php binary path
*
* @var string|null
*/
private static $php_bin;
/** /**
* Attempts to get the worker id from the command line arguments or the environment variable TAMER_WORKER_ID * Attempts to get the worker id from the command line arguments or the environment variable TAMER_WORKER_ID
* If neither are set, returns null. * If neither are set, returns null.
@ -66,4 +82,49 @@
default => throw new InvalidArgumentException('Invalid protocol type'), default => throw new InvalidArgumentException('Invalid protocol type'),
}; };
} }
/**
* Returns the worker variables from the environment variables
*
* @return array
*/
public static function getWorkerVariables(): array
{
if(self::$worker_variables == null)
{
self::$worker_variables = [
'TAMER_ENABLED' => getenv('TAMER_ENABLED') === 'true',
'TAMER_PROTOCOL' => getenv('TAMER_PROTOCOL'),
'TAMER_SERVERS' => getenv('TAMER_SERVERS'),
'TAMER_USERNAME' => getenv('TAMER_USERNAME'),
'TAMER_PASSWORD' => getenv('TAMER_PASSWORD'),
'TAMER_INSTANCE_ID' => getenv('TAMER_INSTANCE_ID'),
];
if(self::$worker_variables['TAMER_SERVERS'] !== false)
self::$worker_variables['TAMER_SERVERS'] = explode(',', self::$worker_variables['TAMER_SERVERS']);
}
return self::$worker_variables;
}
/**
* Returns the path to the php binary
*
* @return string
* @throws Exception
*/
public static function findPhpBin(): string
{
if(self::$php_bin !== null)
return self::$php_bin;
$php_finder = new PhpExecutableFinder();
$php_bin = $php_finder->find();
if($php_bin === false)
throw new Exception('Unable to find the php binary');
self::$php_bin = $php_bin;
return $php_bin;
}
} }

View file

@ -0,0 +1,187 @@
<?php
/** @noinspection PhpMissingFieldTypeInspection */
namespace Tamer\Classes;
use Exception;
use LogLib\Log;
use Symfony\Component\Process\Process;
use Tamer\Objects\WorkerInstance;
class Supervisor
{
/**
* A list of all the workers that are initialized
*
* @var WorkerInstance[]
*/
private $workers;
/**
* The protocol to pass to the worker instances
*
* @var string
*/
private $protocol;
/**
* The list of servers to pass to the worker instances (eg; host:port)
*
* @var string[]
*/
private $servers;
/**
* (Optional) The username to pass to the worker instances
*
* @var string|null
*/
private $username;
/**
* (Optional) The password to pass to the worker instances
*
* @var string|null
*/
private $password;
/**
*
*/
public function __construct(string $protocol, array $servers, ?string $username = null, ?string $password = null)
{
$this->workers = [];
$this->protocol = $protocol;
$this->servers = $servers;
$this->username = $username;
$this->password = $password;
}
/**
* Adds a worker to the supervisor instance
*
* @param string $target
* @param int $instances
* @return void
* @throws Exception
*/
public function addWorker(string $target, int $instances): void
{
for ($i = 0; $i < $instances; $i++)
{
$this->workers[] = new WorkerInstance($target, $this->protocol, $this->servers, $this->username, $this->password);
}
}
/**
* Starts all the workers
*
* @return void
* @throws Exception
*/
public function start(): void
{
/** @var WorkerInstance $worker */
foreach ($this->workers as $worker)
{
$worker->start();
}
// Ensure that all the workers are running
foreach($this->workers as $worker)
{
if (!$worker->isRunning())
{
throw new Exception("Worker {$worker->getId()} is not running");
}
while(true)
{
switch($worker->getProcess()->getStatus())
{
case Process::STATUS_STARTED:
Log::debug('net.nosial.tamerlib', "worker {$worker->getId()} is running");
break 2;
case Process::STATUS_TERMINATED:
throw new Exception("Worker {$worker->getId()} has terminated");
default:
echo "Worker {$worker->getId()} is {$worker->getProcess()->getStatus()}" . PHP_EOL;
}
}
}
}
/**
* Stops all the workers
*
* @return void
* @throws Exception
*/
public function stop(): void
{
/** @var WorkerInstance $worker */
foreach ($this->workers as $worker)
{
$worker->stop();
}
}
/**
* Restarts all the workers
*
* @return void
* @throws Exception
*/
public function restart(): void
{
/** @var WorkerInstance $worker */
foreach ($this->workers as $worker)
{
$worker->stop();
$worker->start();
}
}
/**
* Monitors all the workers and restarts them if they are not running
*
* @param bool $auto_restart
* @return void
* @throws Exception
*/
public function monitor(bool $auto_restart = true): void
{
while (true)
{
/** @var WorkerInstance $worker */
foreach ($this->workers as $worker)
{
if (!$worker->isRunning())
{
if ($auto_restart)
{
$worker->start();
}
else
{
throw new Exception("Worker {$worker->getId()} is not running");
}
}
}
sleep(1);
}
}
/**
* @throws Exception
*/
public function __destruct()
{
$this->stop();
}
}

View file

@ -0,0 +1,19 @@
<?php
namespace Tamer\Exceptions;
use Exception;
use Throwable;
class UnsupervisedWorkerException extends Exception
{
/**
* @param string $message
* @param int $code
* @param Throwable|null $previous
*/
public function __construct(string $message = "", int $code = 0, ?Throwable $previous = null)
{
parent::__construct($message, $code, $previous);
}
}

View file

@ -0,0 +1,186 @@
<?php
/** @noinspection PhpMissingFieldTypeInspection */
namespace Tamer\Objects;
use Exception;
use LogLib\Log;
use Symfony\Component\Process\Process;
use Tamer\Classes\Functions;
class WorkerInstance
{
/**
* The worker's instance id
*
* @var string
*/
private $id;
/**
* The protocol to use when connecting to the server
*
* @var string
*/
private $protocol;
/**
* The servers to connect to
*
* @var array
*/
private $servers;
/**
* The username to use when connecting to the server (if applicable)
*
* @var string|null
*/
private $username;
/**
* The password to use when connecting to the server (if applicable)
*
* @var string|null
*/
private $password;
/**
* The process that is running the worker instance
*
* @var Process|null
*/
private $process;
/**
* The target to run the worker instance on (e.g. a file path)
*
* @var string
*/
private $target;
/**
* Public Constructor
*
* @param string $target
* @param string $protocol
* @param array $servers
* @param string|null $username
* @param string|null $password
* @throws Exception
*/
public function __construct(string $target, string $protocol, array $servers, ?string $username = null, ?string $password = null)
{
$this->id = uniqid();
$this->target = $target;
$this->protocol = $protocol;
$this->servers = $servers;
$this->username = $username;
$this->password = $password;
$this->process = null;
if($target !== 'closure' && file_exists($target) === false)
{
throw new Exception('The target file does not exist');
}
}
/**
* Returns the worker instance id
*
* @return string
*/
public function getId(): string
{
return $this->id;
}
/**
* Executes the worker instance in a separate process
*
* @return void
* @throws Exception
*/
public function start(): void
{
$target = $this->target;
if($target == 'closure')
{
$target = __DIR__ . DIRECTORY_SEPARATOR . 'closure';
}
$argv = $_SERVER['argv'];
array_shift($argv);
$this->process = new Process(array_merge([Functions::findPhpBin(), $target], $argv));
$this->process->setEnv([
'TAMER_ENABLED' => 'true',
'TAMER_PROTOCOL' => $this->protocol,
'TAMER_SERVERS' => implode(',', $this->servers),
'TAMER_USERNAME' => $this->username,
'TAMER_PASSWORD' => $this->password,
'TAMER_INSTANCE_ID' => $this->id
]);
Log::debug('net.nosial.tamerlib', sprintf('starting worker %s', $this->id));
// Callback for process output
$this->process->start(function ($type, $buffer)
{
// Add newline if it's missing
if(substr($buffer, -1) !== PHP_EOL)
{
$buffer .= PHP_EOL;
}
print($buffer);
});
}
/**
* Stops the worker instance
*
* @return void
*/
public function stop(): void
{
if($this->process !== null)
{
Log::debug('net.nosial.tamerlib', sprintf('Stopping worker %s', $this->id));
$this->process->stop();
}
}
/**
* Returns whether the worker instance is running
*
* @return bool
*/
public function isRunning(): bool
{
if($this->process !== null)
{
return $this->process->isRunning();
}
return false;
}
/**
* @return Process|null
*/
public function getProcess(): ?Process
{
return $this->process;
}
/**
* Destructor
*/
public function __destruct()
{
$this->stop();
}
}

16
src/Tamer/Objects/closure Normal file
View file

@ -0,0 +1,16 @@
<?php
require 'ncc';
import('net.nosial.tamerlib', 'latest');
\Tamer\Tamer::initWorker();
try
{
\Tamer\Tamer::work();
}
catch(\Exception $e)
{
\LogLib\Log::error('net.nosial.tamerlib', $e->getMessage(), $e);
exit(1);
}

View file

@ -112,7 +112,7 @@
foreach($servers as $server) foreach($servers as $server)
{ {
$server = explode(':', $server); $server = explode(':', $server);
$this->addServer($server[0], $server[1]); $this->addServer($server[0], (int)$server[1]);
} }
} }
@ -359,9 +359,7 @@
$this->preformAutoreconf(); $this->preformAutoreconf();
if(!$this->client->runTasks()) if(!$this->client->runTasks())
{
return false; return false;
}
return true; return true;
} }
@ -456,15 +454,11 @@
{ {
try try
{ {
$this->run(); $this->disconnect();
} }
catch(Exception $e) catch(Exception $e)
{ {
unset($e); unset($e);
} }
finally
{
$this->disconnect();
}
} }
} }

View file

@ -102,7 +102,7 @@
foreach($servers as $server) foreach($servers as $server)
{ {
$server = explode(':', $server); $server = explode(':', $server);
$this->addServer($server[0], $server[1]); $this->addServer($server[0], (int)$server[1]);
} }
} }
@ -120,8 +120,6 @@
$this->worker = new GearmanWorker(); $this->worker = new GearmanWorker();
$this->worker->addOptions(GEARMAN_WORKER_GRAB_UNIQ); $this->worker->addOptions(GEARMAN_WORKER_GRAB_UNIQ);
Log::debug('net.nosial.tamerlib', 'connecting to gearman server(s)');
foreach($this->defined_servers as $host => $ports) foreach($this->defined_servers as $host => $ports)
{ {
foreach($ports as $port) foreach($ports as $port)

View file

@ -5,11 +5,14 @@
namespace Tamer; namespace Tamer;
use Closure; use Closure;
use Exception;
use InvalidArgumentException; use InvalidArgumentException;
use Tamer\Abstracts\Mode; use Tamer\Abstracts\Mode;
use Tamer\Classes\Functions; use Tamer\Classes\Functions;
use Tamer\Classes\Supervisor;
use Tamer\Classes\Validate; use Tamer\Classes\Validate;
use Tamer\Exceptions\ConnectionException; use Tamer\Exceptions\ConnectionException;
use Tamer\Exceptions\UnsupervisedWorkerException;
use Tamer\Interfaces\ClientProtocolInterface; use Tamer\Interfaces\ClientProtocolInterface;
use Tamer\Interfaces\WorkerProtocolInterface; use Tamer\Interfaces\WorkerProtocolInterface;
use Tamer\Objects\Task; use Tamer\Objects\Task;
@ -53,17 +56,23 @@
private static $connected; private static $connected;
/** /**
* Connects to a server using the specified protocol and mode (client or worker) * The supervisor that is supervising the workers
*
* @var Supervisor
*/
private static $supervisor;
/**
* Initializes Tamer as a client and connects to the server
* *
* @param string $protocol * @param string $protocol
* @param string $mode
* @param array $servers * @param array $servers
* @param string|null $username * @param string|null $username
* @param string|null $password * @param string|null $password
* @return void * @return void
* @throws ConnectionException * @throws ConnectionException
*/ */
public static function connect(string $protocol, string $mode, array $servers, ?string $username=null, ?string $password=null): void public static function init(string $protocol, array $servers, ?string $username=null, ?string $password=null): void
{ {
if(self::$connected) if(self::$connected)
{ {
@ -75,31 +84,39 @@
throw new InvalidArgumentException(sprintf('Invalid protocol type: %s', $protocol)); throw new InvalidArgumentException(sprintf('Invalid protocol type: %s', $protocol));
} }
if (!Validate::mode($mode))
{
throw new InvalidArgumentException(sprintf('Invalid mode: %s', $mode));
}
self::$protocol = $protocol; self::$protocol = $protocol;
self::$mode = $mode; self::$mode = Mode::Client;
if (self::$mode === Mode::Client)
{
self::$client = Functions::createClient($protocol, $username, $password); self::$client = Functions::createClient($protocol, $username, $password);
self::$client->addServers($servers); self::$client->addServers($servers);
self::$client->connect(); self::$client->connect();
} self::$supervisor = new Supervisor($protocol, $servers, $username, $password);
elseif(self::$mode === Mode::Worker) self::$connected = true;
{
self::$worker = Functions::createWorker($protocol, $username, $password);
self::$worker->addServers($servers);
self::$worker->connect();
}
else
{
throw new InvalidArgumentException(sprintf('Invalid mode: %s', $mode));
} }
/**
* Initializes Tamer as a worker client and connects to the server
*
* @return void
* @throws ConnectionException
* @throws UnsupervisedWorkerException
*/
public static function initWorker(): void
{
if(self::$connected)
{
throw new ConnectionException('Tamer is already connected to the server');
}
if(!Functions::getWorkerVariables()['TAMER_ENABLED'])
{
throw new UnsupervisedWorkerException('Tamer is not enabled for this worker');
}
self::$protocol = Functions::getWorkerVariables()['TAMER_PROTOCOL'];
self::$mode = Mode::Worker;
self::$worker = Functions::createWorker(self::$protocol);
self::$worker->addServers(Functions::getWorkerVariables()['TAMER_SERVERS']);
self::$worker->connect();
self::$connected = true; self::$connected = true;
} }
@ -294,6 +311,80 @@
} }
} }
/**
* Adds a worker to the supervisor
*
* @param string $target
* @param int $instances
* @return void
* @throws Exception
*/
public static function addWorker(string $target, int $instances): void
{
if (self::$mode === Mode::Client)
{
self::$supervisor->addWorker($target, $instances);
}
else
{
throw new InvalidArgumentException('Tamer is not running in client mode');
}
}
/**
* Starts all workers
*
* @return void
* @throws Exception
*/
public static function startWorkers(): void
{
if (self::$mode === Mode::Client)
{
self::$supervisor->start();
}
else
{
throw new InvalidArgumentException('Tamer is not running in client mode');
}
}
/**
* Stops all workers
*
* @return void
* @throws Exception
*/
public static function stopWorkers(): void
{
if (self::$mode === Mode::Client)
{
self::$supervisor->stop();
}
else
{
throw new InvalidArgumentException('Tamer is not running in client mode');
}
}
/**
* Restarts all workers
*
* @return void
* @throws Exception
*/
public static function restartWorkers(): void
{
if (self::$mode === Mode::Client)
{
self::$supervisor->restart();
}
else
{
throw new InvalidArgumentException('Tamer is not running in client mode');
}
}
/** /**
* @return string * @return string
*/ */

View file

@ -1,26 +0,0 @@
<?php
require 'ncc';
use Tamer\Objects\JobResults;
use Tamer\Objects\Task;
import('net.nosial.tamerlib', 'latest');
$client = new \Tamer\Protocols\Gearman\Client();
$client->addServer();
$client->do(new Task('sleep', '5'));
$client->queue(new Task('sleep', '5', function(JobResults $job) {
echo "Task {$job->getId()} completed with data: {$job->getData()} \n";
}));
$client->queue(new Task('sleep', '5', function(JobResults $job) {
echo "Task {$job->getId()} completed with data: {$job->getData()} \n";
}));
$client->run();

View file

@ -1,15 +0,0 @@
<?php
require 'ncc';
import('net.nosial.tamerlib', 'latest');
$client = new \Tamer\Protocols\Gearman\Client();
$client->addServer();
$client->doClosure(function () {
require 'ncc';
import('net.nosial.loglib', 'latest');
\LogLib\Log::info('gearman_closure.php', 'closure');
});

View file

@ -1,27 +0,0 @@
<?php
require 'ncc';
use Tamer\Objects\Job;
import('net.nosial.tamerlib', 'latest');
$worker = new \Tamer\Protocols\Gearman\Worker();
$worker->addServer();
$worker->addFunction('sleep', function($job) {
/** @var Job $job */
var_dump(get_class($job));
echo "Task {$job->getId()} started with data: {$job->getData()} \n";
sleep($job->getData());
echo "Task {$job->getId()} completed with data: {$job->getData()} \n";
return $job->getData();
});
while(true)
{
echo "Waiting for job... \n";
$worker->work();
}

30
tests/no_tamer.php Normal file
View file

@ -0,0 +1,30 @@
<?php
// Pi function (closure) loop 10 times
for ($i = 0; $i < 50; $i++)
{
$start = microtime(true);
$pi = 0;
$top = 4;
$bot = 1;
$minus = true;
$iterations = 1000000;
for ($i = 0; $i < $iterations; $i++)
{
if ($minus)
{
$pi = $pi - ($top / $bot);
$minus = false;
}
else
{
$pi = $pi + ($top / $bot);
$minus = true;
}
$bot += 2;
}
}

View file

@ -1,20 +0,0 @@
<?php
use Tamer\Abstracts\TaskPriority;
use Tamer\Objects\Task;
require 'ncc';
import('net.nosial.tamerlib', 'latest');
$client = new \Tamer\Protocols\RabbitMq\Client('guest', 'guest');
$client->addServer('127.0.0.1', 5672);
// Loop through 10 tasks
for($i = 0; $i < 500; $i++)
{
$client->do(Task::create('sleep', '5')
->setPriority(TaskPriority::High)
);
}

View file

@ -1,27 +0,0 @@
<?php
require 'ncc';
use Tamer\Objects\Job;
import('net.nosial.tamerlib', 'latest');
$worker = new \Tamer\Protocols\RabbitMq\Worker('guest', 'guest');
$worker->addServer('127.0.0.1', 5672);
$worker->addFunction('sleep', function($job) {
/** @var Job $job */
var_dump(get_class($job));
echo "Task {$job->getId()} started with data: {$job->getData()} \n";
sleep($job->getData());
echo "Task {$job->getId()} completed with data: {$job->getData()} \n";
return $job->getData();
});
while(true)
{
echo "Waiting for job... \n";
$worker->work();
}

70
tests/tamer.php Normal file
View file

@ -0,0 +1,70 @@
<?php
use Tamer\Abstracts\ProtocolType;
use Tamer\Tamer;
require 'ncc';
import('net.nosial.tamerlib', 'latest');
Tamer::init(ProtocolType::Gearman,
['127.0.0.1:4730']
);
$instances = 10;
Tamer::addWorker('closure', $instances);
Tamer::startWorkers();
$a = microtime(true);
$times = [];
$jobs = 30;
// Pi function (closure) loop 10 times
for ($i = 0; $i < $jobs; $i++)
{
Tamer::queueClosure(function(){
// Full pi calculation implementation
$start = microtime(true);
$pi = 0;
$top = 4;
$bot = 1;
$minus = true;
$iterations = 1000000;
for ($i = 0; $i < $iterations; $i++)
{
if ($minus)
{
$pi = $pi - ($top / $bot);
$minus = false;
}
else
{
$pi = $pi + ($top / $bot);
$minus = true;
}
$bot += 2;
}
return json_encode([$pi, $start]);
},
function($return) use ($a, &$times)
{
$return = json_decode($return, true);
$end_time = microtime(true) - $return[1];
$times[] = $end_time;
echo "Pi is {$return[0]}, completed in " . ($end_time) . " seconds \n";
});
}
echo "Waiting for $jobs jobs to finish on $instances workers \n";
Tamer::run();
$b = microtime(true);
echo PHP_EOL;
echo "Average time: " . (array_sum($times) / count($times)) . " seconds \n";
echo "Took (with tamer)" . ($b - $a) . " seconds \n";
echo "Total time (without tamer): " . (array_sum($times)) . " seconds \n";
echo "Tamer overhead: " . (($b - $a) - array_sum($times)) . " seconds \n";

View file

@ -1,57 +1,33 @@
<?php <?php
use Tamer\Abstracts\Mode;
use Tamer\Abstracts\ProtocolType; use Tamer\Abstracts\ProtocolType;
use Tamer\Objects\JobResults;
use Tamer\Objects\Task;
use Tamer\Tamer; use Tamer\Tamer;
require 'ncc'; require 'ncc';
import('net.nosial.tamerlib', 'latest'); import('net.nosial.tamerlib', 'latest');
Tamer::connect(ProtocolType::Gearman, Mode::Client, Tamer::init(ProtocolType::Gearman,
['127.0.0.1:4730'] ['127.0.0.1:4730']
); );
// Pi calculation (closure) Tamer::addWorker(__DIR__ . '/tamer_worker.php', 10);
// Add it 10 times Tamer::startWorkers();
for($i = 0; $i < 100; $i++)
{
Tamer::queueClosure(function() {
// Do Pi calculation
$pi = 0;
$top = 4.0;
$bot = 1.0;
$minus = true;
for($i = 0; $i < 1000000; $i++)
{
if($minus)
{
$pi -= ($top / $bot);
$minus = false;
}
else
{
$pi += ($top / $bot);
$minus = true;
}
$bot += 2.0; // Sleep function (task) loop 10 times
} for ($i = 0; $i < 10; $i++)
{
\LogLib\Log::info('net.nosial.tamerlib', sprintf('Pi: %s', $pi)); Tamer::queue(Task::create('sleep', 5, function(JobResults $data)
return $pi;
});
}
// Sleep function (task)
Tamer::queue(\Tamer\Objects\Task::create('sleep', 5, function(\Tamer\Objects\JobResults $data)
{ {
echo "Slept for {$data->getData()} seconds \n"; echo "Slept for {$data->getData()} seconds \n";
})); }));
}
echo "Waiting for jobs to finish \n";
$a = microtime(true); $a = microtime(true);
Tamer::run(); Tamer::run();
$b = microtime(true); $b = microtime(true);
echo "Took " . ($b - $a) . " seconds \n"; echo "Took " . ($b - $a) . " seconds \n";

View file

@ -8,9 +8,7 @@
import('net.nosial.tamerlib', 'latest'); import('net.nosial.tamerlib', 'latest');
Tamer::connect(ProtocolType::Gearman, Mode::Worker, Tamer::initWorker();
['127.0.0.1:4730']
);
Tamer::addFunction('sleep', function(\Tamer\Objects\Job $job) { Tamer::addFunction('sleep', function(\Tamer\Objects\Job $job) {
sleep($job->getData()); sleep($job->getData());