Nosial/tamerlib
2
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions

Removed all unused files

Browse source
This commit is contained in:
Netkas 2023-06-16 02:17:34 -04:00
parent 2b39a9df9e
commit 44b554d08c
No known key found for this signature in database
GPG key ID: 5DAF58535614062B
6 changed files with 0 additions and 452 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
functions.txt
View file

@ -1,66 +0,0 @@
== GLOBAL ==
Both the client and the worker have access to the following functions:
initialize($mode, $server_config): void - Initializes TamerLib in the given mode, if no server_config is provided then
Tamer will initialize it's own server and use its own configuration. When initializing as a worker, the
$server_config is ignored as the configuration is provided by the parent process.
shutdown(): void - Shuts down the TamerLib by resetting the global variables and releasing any resources it may have
allocated.
== CLIENT ==
The client is the master process of TamerLib that is used to push tasks to the worker processes, optionally a client may
also be responsible for supervising workers and or managing its own server process.
createWorker($count, $path): void - Creates workers and starts them, the count parameter specifies how many workers to create
and the path parameter being optional, if provided, specifies the path to the worker executable. If the path
parameter is not provided, TamerLib will use it's own subprocess to create the workers which can only handle closures
do($callable, $channel=0): string - Pushes a task to the worker pool and returns the job id, this adds the job to its own watchlist
so that it can be monitored by Tamer.
call($function, $args, $channel=0): string - Pushes a function call to the worker pool and returns the job id, this adds the job to its own watchlist
so that it can be monitored by Tamer.
dof($callable, $channel=0): void - Pushes a task to the worker pool and forgets about it, this does not add the job to the watchlist
callf($function, $args, $channel=0): void - Pushes a function call to the worker pool and forgets about it,
this does not add the job to the watchlist
wait($callback($job_id, $return)): void - Waits for all jobs dispatched with do() to complete, the callback is called
for each job that completes with the job id and the return value of the job. If one or more jobs fail, the function
will throw an replicated exception.
waitFor($job_id): mixed - Waits for a specific job to complete and returns the return value of the job, if the job
fails, the function will throw an replicated exception.
clear(): void - Clears the watchlist of all jobs.
NOTE: wait & waitFor will execute supervisory tasks while waiting for jobs to complete if the client is configured to
supervise workers.
== WORKER ==
The worker is the slave process of TamerLib that is used to execute tasks pushed to it by the client process.
setFunction($function, $callable): void - Sets the function that will be called when the worker receives a task with
the given function name.
removeFunction($function): void - Removes the function that will be called when the worker receives a task with
the given function name.
getFunctions(): array - Returns an array of all the functions that the worker has registered.
run($channels, $timeout=0): void - Runs the worker process, this function will block until the worker is shutdown or
until the timeout is reached. If the timeout is reached, the worker will shutdown. The channels parameter is an
int or an array of ints that specifies which channels the worker will listen on. If the channels parameter is 0
then the worker will listen only on channel 0.
return($job_id, $return): void - Returns the return value of the job to the client process, this function is called
automatically when the worker completes a job.
throw($job_id, $exception): void - Throws an exception to the client process, this function is called automatically
when the worker completes a job and the job throws an exception.
reject($job_id, $exception): void - Rejects a job, this silently rejects the job and pushes it back onto the queue
for another worker to pick up.

261
old
View file

@ -1,261 +0,0 @@
/**
* Configures the Redis server.
*
* @param string $cmd
* @param string $host
* @param int|null $port
* @return void
* @throws NoAvailablePortException
*/
public static function configureRedisServer(string $cmd='redis-server', string $host='127.0.0.1', ?int $port=null): void
{
if(self::$redis_server instanceof RedisServer)
{
throw new RuntimeException('Redis server already configured.');
}
self::$redis_server = new RedisServer($cmd, $host, $port);
}
/**
* Returns the Redis server
*
* @return RedisServer|null
*/
private static function getRedisServer(): ?RedisServer
{
if(is_null(self::$redis_server))
{
self::$redis_server = new RedisServer();
}
return self::$redis_server;
}
/**
* Adds the job ID to the watch list.
*
* @param int $job_id
* @return void
*/
private static function addToWatch(int $job_id): void
{
self::$watching_jobs[] = $job_id;
}
/**
* Cleans up the watch list by removing jobs that no longer exist.
*
* @return void
*/
public static function cleanWatchList(): void
{
try
{
foreach(self::$watching_jobs as $job_id)
{
if(!self::getRedisClient()->exists($job_id))
{
unset(self::$watching_jobs[$job_id]);
}
}
}
catch(Exception $e)
{
throw new RuntimeException('Failed to clean watch list.', 0, $e);
}
}
/**
* Returns the Redis client.
*
* @return Redis
* @throws RedisException
*/
private static function getRedisClient(): Redis
{
if(is_null(self::$redis_server))
{
self::$redis_server = new RedisServer();
}
if(!self::$redis_server->isRunning())
{
self::$redis_server->start();
}
if(is_null(self::$redis_client))
{
self::$redis_client = new Redis();
/** @noinspection NullPointerExceptionInspection */
var_dump(self::getRedisServer()->getHost(), self::getRedisServer()->getPort());
self::$redis_client->connect(self::getRedisServer()->getHost(), self::getRedisServer()->getPort());
}
return self::$redis_client;
}
/**
* Returns the worker supervisor.
*
* @return WorkerSupervisor
*/
private static function getSupervisor(): WorkerSupervisor
{
if(is_null(self::$supervisor))
{
self::$supervisor = new WorkerSupervisor();
}
return self::$supervisor;
}
/**
* Spawns a closure worker.
*
* @param int $count
* @return void
*/
public static function spawnClosure(int $count=8): void
{
self::getSupervisor()->spawnClosure(self::getRedisServer(), $count);
}
/**
* Spawns a worker of a specific php file.
*
* @param array $cmd
* @param int $count
* @return void
*/
public static function spawn(array $cmd, int $count=8): void
{
self::getSupervisor()->spawnWorker($cmd, self::getRedisServer(), $count);
}
/**
* Sends a closure to be executed by a worker, returns the job ID for the closure.
*
* @param callable $closure
* @return string
*/
public static function do(callable $closure): string
{
try
{
$closure = serialize(new SerializableClosure($closure));
}
catch(Exception $e)
{
throw new RuntimeException('Failed to serialize closure.', 2000, $e);
}
$job_packet = new JobPacket();
$job_packet->setJobType(JobType::CLOSURE);
$job_packet->setEncodingType(EncodingType::SERIALIZED);
$job_packet->setPayload($closure);
// Push as hash using toArray()
try
{
self::getRedisClient()->hMSet($job_packet->getId(), $job_packet->toArray());
self::getRedisClient()->expire($job_packet->getId(), 60);
self::getRedisClient()->rPush(sprintf('ch%s', $job_packet->getChannel()), $job_packet->getId());
}
catch(Exception $e)
{
throw new RuntimeException('Failed to push job to Redis.', 2000, $e);
}
self::addToWatch($job_packet->getId());
return $job_packet->getId();
}
/**
* Waits for all jobs to complete.
*
* @param callable|null $callback
* @return void
* @throws RedisException
*/
public static function waitAll(?callable $callback=null): void
{
while(count(self::$watching_jobs) > 0)
{
foreach(self::$watching_jobs as $job_id)
{
if(!self::getRedisClient()->exists($job_id))
{
unset(self::$watching_jobs[$job_id]);
continue;
}
if(!in_array(self::getRedisClient()->hGet($job_id, 'status'), JobStatus::PROCESSING_STATES))
{
if(!is_null($callback))
{
$callback(self::getRedisClient()->hGet($job_id, 'return_value'));
}
self::getRedisClient()->del($job_id);
unset(self::$watching_jobs[$job_id]);
}
}
}
}
/**
* Waits for a job to complete, returns the result.
*
* @param string $job_id
* @param int $timeout
* @return mixed
* @throws RedisException
*/
public function waitFor(string $job_id, int $timeout=0)
{
$timeout_count = time();
while(in_array(self::getRedisClient()->hGet($job_id, 'status'), JobStatus::PROCESSING_STATES, true))
{
if($timeout > 0 && time() - $timeout_count > $timeout)
{
throw new RuntimeException(sprintf('waitFor %s timed out.', $job_id));
}
usleep(100);
}
$return = self::getRedisClient()->hGet($job_id, 'result');
self::getRedisClient()->del($job_id);
return $return;
}
/**
* Gets a key from the shared memory.
*
* @param string $key
* @param null $default
* @return mixed
*/
public static function getKey(string $key, $default=null): mixed
{
if(isset(self::$shared_memory[$key]))
{
return self::$shared_memory[$key];
}
return $default;
}
/**
* Sets a key in the shared memory.
*
* @param string $key
* @param mixed $value
* @return void
*/
public static function setKey(string $key, mixed $value): void
{
self::$shared_memory[$key] = $value;
}

77
src/TamerLib/Classes/AdaptiveSleep.php
View file

@ -1,77 +0,0 @@
<?php
/** @noinspection PhpMissingFieldTypeInspection */
namespace TamerLib\Classes;
class AdaptiveSleep
{
/**
* @var int int
*/
private $max_sleep_time;
/**
* @var array
*/
private $busy_buffer;
/**
* @var int
*/
private $buffer_size;
/**
* @var int
*/
private $buffer_index;
/**
* AdaptiveSleep constructor.
*
* @param int $max_sleep_time
* @param int $buffer_size
*/
public function __construct(int $max_sleep_time=500, int $buffer_size=30)
{
$this->max_sleep_time = $max_sleep_time;
$this->busy_buffer = array_fill(0, $buffer_size, false); // Fill the buffer with false values (not busy
$this->buffer_size = $buffer_size;
$this->buffer_index = 0;
}
/**
* Preforms an adaptive sleep.
*
* @param bool $busy
* @return int
*/
public function sleep(bool $busy): int
{
// Add the busy state to the buffer
$this->busy_buffer[$this->buffer_index] = $busy;
$this->buffer_index = ($this->buffer_index + 1) % $this->buffer_size; // Circular buffer
// Calculate the average busy state
$busy_count = 0;
foreach($this->busy_buffer as $busy_state)
{
if($busy_state)
{
$busy_count++;
}
}
$busy_average = $busy_count / $this->buffer_size;
// Calculate the sleep time
$sleep_time = $this->max_sleep_time * (1 - $busy_average);
// Sleep
if($sleep_time > 0)
{
usleep($sleep_time);
}
return $sleep_time;
}
}

10
src/TamerLib/Enums/JobType.php
View file

@ -1,10 +0,0 @@
<?php
namespace TamerLib\Enums;
final class JobType
{
public const CLOSURE = 100;
public const FUNCTION = 200;
}

9
src/TamerLib/Enums/WorkerType.php
View file

@ -1,9 +0,0 @@
<?php
namespace TamerLib\Enums;
final class WorkerType
{
public const SCRIPT = 10;
public const CLOSURE = 20;
}

29
tests/adaptive_sleep.php
View file

@ -1,29 +0,0 @@
<?php
use TamerLib\Classes\AdaptiveSleep;
require 'ncc';
import('net.nosial.tamerlib');
$adaptive_sleep = new AdaptiveSleep(10);
$time = time();
while(true)
{
if(time() - $time > 3)
{
// Simulate traffic every 3 seconds
$sleep = $adaptive_sleep->sleep(random_int(0, 100) < 90);
if(time() - $time > 5)
{
// Stop the simulation after 5 seconds
$time = time();
}
}
else
{
// No traffic
$sleep = $adaptive_sleep->sleep(false);
}
}