FuzeWorks
/
Async
1
0
Fork 0
Code Issues 5 Pull Requests Releases 1 Activity

Release of RC1 (#7)
continuous-integration/drone/push Build is passing Details 

Finished ControllerHandler into a working state.

Merge remote-tracking branch 'origin/master' into 3-features

# Conflicts:
#	Dockerfile
#	bin/supervisor
#	bin/worker
#	composer.json
#	src/FuzeWorks/Async/Executors/ShellExecutor.php
#	src/FuzeWorks/Async/ShellWorker.php
#	src/FuzeWorks/Async/Supervisors/ParallelSuperVisor.php
#	src/FuzeWorks/Async/TaskStorage/RedisTaskStorage.php
#	src/FuzeWorks/Async/Tasks.php
#	test/bootstrap.php
#	test/mock/Handlers/EmptyHandler.php
#	test/mock/Handlers/TestStopTaskHandler.php

Started work on making tasks forcefully quit after a maximum time has expired.

Finished tasks are no longer loaded in SuperVisor.

By adding a parameter in TaskStorage, it is now possible to distinguish between finished and unfinished tasks.
Finished tasks are those tasks that have a status of Task::COMPLETED or Task::CANCELLED. Unfinished tasks are all others.

This allows the SuperVisor to not bother with the mountain of tasks that will be saved during large projects.

Implemented proper dependencies.

Dependencies can now pass output to each other using the DependentTaskHandler.
Also fixed some general problems in the Tasks class, for instance the Executor not starting correctly because of problematic parameters.
Also, SuperVisor now sets the output of a Task using the last output of the task, and not the first.

Return all output when providing attempt = 0.

When providing $attempt = 0 at readTaskOutput and readPostOutput, all output shall be returned.
This is the default return. Hence, a lot of tests had to be altered slightly.

Changed the way task output and post output is saved.

Redis now saves all output for a task within a hash. This hash contains individual tasks and all its output.
Attempts also start at 1, since that makes most sense for this context. When output is written, the TaskStorage must figure out at which attempt the Task is.

Implemented Parent Handlers.

Parent Handlers can be stacked to run in succession. Output is transfered as input into the child handler which can continue with it. If the parent Handler fails, all Child handlers also fail.

Made Handlers an object instead of a string reference.

Handlers should now be added as objects, adding some flexibility to the developer.
Developers are still cautioned to take great care that Handlers work approriately. Handlers can potentially crash the SuperVisor if not taken good care of.

Try with only Redis.

Made many changes. Fixed race-conditions in test code.

Now try while flushing a selected database.

Try again in the new environment.

Maybe Events are the problem?

Fixed DummyTaskStorage persisting outside of the storage.

Awkward how that could go wrong...

Added TaskModifyEvent.

Event gets fired when an Event is modified by sending it to TaskStorage::modifyEvent. This allows components to observe changes and report these to the user. Might also be useful to cancel unwanted changes.

Made the Docker image Alpine-based. Should work better when running Async in a Cron environment.
Also removed compatibility with PHP 7.2.

Implemented many unit tests.

Now with coverage

And remove the Redis debug again

Now?

Temporarily check if Redis works

Added separate environments for DummyTaskStorage and RedisTaskStorage.

System now uses environment variables imported through Docker. See test/config.tasks.php for all environment options available.

Now try with an added service

Attempt to run a PHPUnit batch

Try with a modified environment.

Try again

Started implementing Drone

ControllerHandler now works. Next up is a distinction between Task and Process status.

Started implementing ControllerHandler.

ControllerHandler is a standard utility handler for MVCR Controllers. This allows the user to create a task which is handled by a standardized controller.
Not finished yet! Needs some love.

Add 'addTasks' method to `Tasks` class

Implemented basic RedisTaskStorage.

- Fixed bug where worker is not provided with bootstrap by ShellExecutor.
- Added composer and Redis to Dockerfile
- Added more output to ParallelSuperVisor

Updated config format.

Implemented changes to binaries. Binaries now accept a 'bootstrap' argument, allowing the developer to load a custom bootstrap from the project they're working on.

This allows Async to run in the same environment as the project it's part of.

Co-authored-by: Abel Hoogeveen <abel@techfuze.net>
Reviewed-on: #7
master 1.1.0-RC1
Abel Hoogeveen 2020-06-07 15:54:19 +02:00
parent db962e96e1
commit 3e0a312c80
39 changed files with 4696 additions and 363 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
.drone.yml Normal file
View File

@ -0,0 +1,22 @@
kind: pipeline
type: docker
name: test
services:
- name: cache
image: redis
steps:
- name: composer
image: composer:latest
commands:
- composer install
- name: redistest
image: phpunit:7.3
commands:
- vendor/bin/phpunit -c test/phpunit.xml --coverage-php test/temp/covredis.cov
environment:
SUPERVISOR_TYPE: ParallelSuperVisor
TASKSTORAGE_TYPE: RedisTaskStorage
TASKSTORAGE_REDIS_HOST: cache

2
.gitignore vendored
View File

@ -3,3 +3,5 @@ composer.lock
.idea/
log/
vendor/
build/
test/temp/

19
Dockerfile
View File

@ -1,11 +1,18 @@
FROM php:7.3-cli-buster
FROM php:7.3-alpine
RUN apt-get update &&\
apt-get install --no-install-recommends --assume-yes --quiet procps ca-certificates curl git &&\
rm -rf /var/lib/apt/lists/*
# FOR ALPINE
# Install git and bash and procps
RUN apk add git bash procps
RUN apk add --no-cache --update --virtual .phpize-deps $PHPIZE_DEPS
# Install Redis
RUN pecl install redis-5.1.1 && docker-php-ext-enable redis
# FOR DEBIAN/UBUNTU
#RUN apt-get update &&\
# apt-get install --no-install-recommends --assume-yes --quiet procps ca-certificates curl git unzip &&\
# rm -rf /var/lib/apt/lists/*
# Install Redis and XDebug
RUN pecl install redis && docker-php-ext-enable redis
RUN pecl install xdebug && docker-php-ext-enable xdebug
# Install Composer
RUN curl -sS https://getcomposer.org/installer | php -- --install-dir=/usr/local/bin --filename=composer

6
bin/supervisor
View File

@ -43,11 +43,11 @@ use FuzeWorks\Exception\LibraryException;
use FuzeWorks\Factory;
// First perform a PHP version check
if (version_compare('7.1.0', PHP_VERSION, '>')) {
if (version_compare('7.3.0', PHP_VERSION, '>')) {
fwrite(
STDERR,
sprintf(
'FuzeWorks Async requires PHP 7.1 or higher.' . PHP_EOL .
'FuzeWorks Async requires PHP 7.3 or higher.' . PHP_EOL .
'You are using PHP %s (%s).' . PHP_EOL,
PHP_VERSION,
PHP_BINARY
@ -109,7 +109,7 @@ try {
// And finally, run the supervisor
try {
$supervisor = $lib->getSuperVisor($bootstrap);
while ($supervisor->cycle() === SuperVisor::RUNNING) {
while ($supervisor->cycle() !== SuperVisor::RUNNING) {
usleep(250000);
}

6
bin/worker
View File

@ -42,11 +42,11 @@ use FuzeWorks\Factory;
// First perform a PHP version check
if (version_compare('7.1.0', PHP_VERSION, '>')) {
if (version_compare('7.3.0', PHP_VERSION, '>')) {
fwrite(
STDERR,
sprintf(
'FuzeWorks Async requires PHP 7.1 or higher.' . PHP_EOL .
'FuzeWorks Async requires PHP 7.3 or higher.' . PHP_EOL .
'You are using PHP %s (%s).' . PHP_EOL,
PHP_VERSION,
PHP_BINARY
@ -119,7 +119,7 @@ $post = isset($arguments['p']);
// RUN THE APP
$worker = $lib->getWorker();
$worker->run($taskID, $post);
$worker->runTaskById($taskID, $post);
fwrite(STDOUT,'Finished task \'' . $taskID . "'");
?>

11
composer.json
View File

@ -13,13 +13,20 @@
}
],
"require": {
"php": ">=7.2.0",
"php": ">=7.3.0",
"fuzeworks/core": "~1.2.0",
"ext-json": "*",
"ext-redis": "*"
},
"require-dev": {
"fuzeworks/tracycomponent": "~1.2.0"
"phpunit/phpunit": "^9",
"phpunit/phpcov": "^7",
"fuzeworks/mvcr": "~1.2.0"
},
"config": {
"platform": {
"ext-redis": "1"
}
},
"autoload": {
"psr-4": {

13
src/FuzeWorks/Async/Constraint/DependencyConstraint.php
View File

@ -56,7 +56,7 @@ use FuzeWorks\Logger;
class DependencyConstraint implements Constraint
{
public $dependencies = [];
protected $dependencies = [];
protected $delayTimes = 3;
@ -100,7 +100,6 @@ class DependencyConstraint implements Constraint
// Get dependency
$dependencyTask = $taskStorage->getTaskById($dependency);
// If the dependency task is completed, ignore it and continue to next dependency
if ($dependencyTask->getStatus() === Task::COMPLETED)
continue;
@ -146,6 +145,16 @@ class DependencyConstraint implements Constraint
return time() + $this->delayTimes;
}
/**
* Return a list of dependencies
*
* @return array
*/
public function getDependencies()
{
return $this->dependencies;
}
/**
* Load the tasks library, so that dependencies can get scanned later
*

64
src/FuzeWorks/Async/Events/TaskModifyEvent.php Normal file
View File

@ -0,0 +1,64 @@
<?php
/**
* FuzeWorks CLIComponent.
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2019 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.2.0
*
* @version Version 1.2.0
*/
namespace FuzeWorks\Async\Events;
use FuzeWorks\Async\Task;
use FuzeWorks\Event;
class TaskModifyEvent extends Event
{
/**
* @var Task
*/
protected $task;
public function init(Task $task)
{
$this->task = $task;
}
public function getTask(): Task
{
return $this->task;
}
public function updateTask(Task $task)
{
$this->task = $task;
}
}

26
src/FuzeWorks/Async/Executor.php
View File

@ -36,9 +36,25 @@
namespace FuzeWorks\Async;
/**
* Interface Executor
*
*
* @todo Implement ListRunningTasks
* @package FuzeWorks\Async
*/
interface Executor
{
/**
* Executor constructor.
*
* Parameters are a unique array which can differ for each Executor.
*
* @param array $parameters
*/
public function __construct(array $parameters);
// Control methods
/**
* Start executing a task.
@ -57,16 +73,12 @@ interface Executor
* Returns the task since it makes modifications. Has to be modified in TaskStorage by SuperVisor.
*
* @param Task $task
* @return Task
* @param bool $harshly True to KILL a process
* @return Task|null Returns modified Task on success, or null if no PID is found
*/
public function stopTask(Task $task): Task;
public function stopTask(Task $task, bool $harshly = false): ?Task;
// Task info
public function getTaskRunning(Task $task): bool;
public function getTaskStats(Task $task): ?array;
public function getTaskExitCode(Task $task): ?int;
// All tasks info
public function getRunningTasks(): array;
}

72
src/FuzeWorks/Async/Executors/ShellExecutor.php
View File

@ -36,7 +36,6 @@
namespace FuzeWorks\Async\Executors;
use FuzeWorks\Async\Executor;
use FuzeWorks\Async\Process;
use FuzeWorks\Async\Task;
use FuzeWorks\Async\TasksException;
@ -52,51 +51,56 @@ class ShellExecutor implements Executor
/**
* ShellExecutor constructor.
*
* @param string $bootstrapFile
* @param array $parameters
* @throws TasksException
*/
public function __construct(string $bootstrapFile, array $parameters)
public function __construct(array $parameters)
{
if (!isset($parameters['workerFile']) || !isset($parameters['bootstrapFile']))
throw new TasksException("Could not construct ShellExecutor. Parameter failure.");
// Fetch workerFile
$workerFile = $parameters['workerFile'];
$this->worker = $parameters['workerFile'];
if (!file_exists($this->worker))
throw new TasksException("Could not construct ShellExecutor. ShellWorker script does not exist.");
// First determine the PHP binary
$this->binary = PHP_BINDIR . DS . 'php';
$this->bootstrapFile = $bootstrapFile;
if (!file_exists($workerFile))
throw new TasksException("Could not construct ShellExecutor. ShellWorker script does not exist.");
$this->worker = $workerFile;
}
private function shellExec($format, array $parameters = [])
{
$parameters = array_map("escapeshellarg", $parameters);
array_unshift($parameters, $format);
$command = call_user_func_array("sprintf", $parameters);
exec($command, $output);
return $output;
$this->bootstrapFile = $parameters['bootstrapFile'];
if (!file_exists($this->bootstrapFile))
throw new TasksException("Could not construct ShellExecutor. No bootstrap file found.");
}
public function startTask(Task $task, bool $post = false): Task
{
// First prepare the command used to spawn workers
$commandString = "$this->binary $this->worker --bootstrap=".$this->bootstrapFile." -t %s ".($post ? 'p' : '')." $this->stdout $this->stderr & echo $!";
$commandString = "$this->binary $this->worker --bootstrap=".$this->bootstrapFile." -t %s ".($post ? '-p' : '')." $this->stdout $this->stderr & echo $!";
// Then execute the command using the base64_encoded string of the taskID
$output = $this->shellExec($commandString, [base64_encode($task->getId())]);
$pid = intval($output[0]);
$task->setProcess(new Process($pid));
$task->addAttribute('pid', $pid);
// And finally return the task
return $task;
}
public function stopTask(Task $task): Task
public function stopTask(Task $task, bool $harshly = false): ?Task
{
// TODO: Implement stopTask() method.
// First prepare the kill command
$commandString = "kill " . ($harshly ? '-9 ' : '') . "%s";
// Fetch the process ID from the task
$pid = $task->attribute('pid');
if (is_null($pid))
return null;
// Then execute the command
$this->shellExec($commandString, [$pid]);
if (!$this->getTaskRunning($task))
$task->removeAttribute('pid');
return $task;
}
public function getTaskRunning(Task $task): bool
@ -111,13 +115,10 @@ class ShellExecutor implements Executor
$commandString = "ps -o pid,%%cpu,%%mem,state,start -p %s | sed 1d";
// First we must determine what process is used.
$process = $task->getProcess();
if (is_null($process))
$pid = $task->attribute('pid');
if (is_null($pid))
return null;
// Then using that process we determine the ProcessID
$pid = $process->getPid();
// And we execute the commandString to fetch info on the process
$output = $this->shellExec($commandString, [$pid]);
@ -141,16 +142,15 @@ class ShellExecutor implements Executor
return null;
// Finally, return the Task information
return $parts;
return ['pid' => (int) $parts[0], 'cpu' => (float) $parts[1], 'mem' => (float) $parts[2], 'state' => $parts[3], 'start' => $parts[4]];
}
public function getTaskExitCode(Task $task): int
protected function shellExec($format, array $parameters = [])
{
// TODO: Implement getTaskExitCode() method.
}
public function getRunningTasks(): array
{
// TODO: Implement getRunningTasks() method.
$parameters = array_map("escapeshellarg", $parameters);
array_unshift($parameters, $format);
$command = call_user_func_array("sprintf", $parameters);
exec($command, $output);
return $output;
}
}

38
src/FuzeWorks/Async/Handler.php
View File

@ -38,6 +38,36 @@ namespace FuzeWorks\Async;
interface Handler
{
/**
* Gets invoked upon being added to the Task
*
* @param Task $task
* @return mixed
* @throws TasksException
*/
public function init(Task $task);
/**
* Retrieve the parent handler that will first handle this task, before this child Handler
*
* @return Handler|null
*/
public function getParentHandler(): ?Handler;
/**
* Set the parent handler that will fire before this Handler
*
* @param Handler $parentHandler
*/
public function setParentHandler(Handler $parentHandler): void;
/**
* Import the parent output into the child
*
* @param string $input
*/
public function setParentInput(string $input): void;
/**
* The handler method used to handle this task.
* This handler will execute the actual task.
@ -52,9 +82,9 @@ interface Handler
/**
* Any output generated by primaryHandler should be returned here.
*
* @return mixed
* @return string
*/
public function getOutput();
public function getOutput(): string;
/**
* The handler method used after the primaryHandler if so requested
@ -69,8 +99,8 @@ interface Handler
/**
* Any output generated by postHandler should be returned here
*
* @return mixed
* @return string
*/
public function getPostOutput();
public function getPostOutput(): string;
}

245
src/FuzeWorks/Async/Handler/ControllerHandler.php Normal file
View File

@ -0,0 +1,245 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async\Handler;
use FuzeWorks\Async\Handler;
use FuzeWorks\Async\Task;
use FuzeWorks\Async\TasksException;
use FuzeWorks\Controller;
use FuzeWorks\Controllers;
use FuzeWorks\Exception\ControllerException;
use FuzeWorks\Exception\FactoryException;
use FuzeWorks\Exception\NotFoundException;
use FuzeWorks\Factory;
class ControllerHandler implements Handler
{
/**
* @var string Name of the controller used to handle the task
*/
protected $controllerName;
/**
* @var string The specific method to handle the task
*/
protected $controllerMethod;
/**
* The namespace to use to load the controller
*
* @var string
*/
protected $controllerNamespace;
/**
* @var string|null The method used to handle the post phase; if requested
*/
protected $postMethod = null;
/**
* @var string The output to be returned to ShellWorker
*/
protected $output;
/**
* @var string The postOutput to be returned to ShellWorker
*/
protected $postOutput;
/**
* Input imported from the parent handler
*
* @var string|null
*/
protected $parentInput;
/**
* ControllerHandler constructor.
*
* Provides all information of which controller to use. Requests get redirected to that controller.
*
* @param string $controllerName The name of the controller to use
* @param string $controllerMethod The method to use for the task execution
* @param string|null $postMethod The method to use for the post execution
* @param string $controllerNamespace A potential custom namespace for the controller
*/
public function __construct(string $controllerName, string $controllerMethod, string $postMethod = null, string $controllerNamespace = '\Application\Controller\\')
{
$this->controllerName = $controllerName;
$this->controllerMethod = $controllerMethod;
$this->postMethod = $postMethod;
$this->controllerNamespace = $controllerNamespace;
}
/**
* @inheritDoc
*/
public function init(Task $task)
{
}
/**
* @inheritDoc
* @throws TasksException
*/
public function primaryHandler(Task $task): bool
{
// Set the arguments
$args = $task->getArguments();
array_unshift($args, $task);
// First we fetch the controller
$controller = $this->getController();
// Check if method exists
if (!method_exists($controller, $this->controllerMethod))
throw new TasksException("Could not handle task. Method '$this->controllerMethod' not found on controller.");
if (!method_exists($controller, 'getTaskStatus'))
throw new TasksException("Could not handle task. Method 'getTaskStatus()' not found on controller, which is required.");
if ($this->parentInput !== null && method_exists($controller, 'setInput'))
$controller->setInput($this->parentInput);
// Call method and collect output
$this->output = call_user_func_array([$controller, $this->controllerMethod], $args);
$success = $controller->getTaskStatus();
if (!is_bool($success))
throw new TasksException("Could not determine whether task has succeeded. getTaskStatus() returned non-bool.");
return $success;
}
/**
* @inheritDoc
*/
public function getOutput(): string
{
return $this->output;
}
/**
* @inheritDoc
* @throws TasksException
*/
public function postHandler(Task $task)
{
// Abort if no postMethod exists
if (is_null($this->postMethod))
throw new TasksException("Could not handle task. No post method provided.");
// First we fetch the controller
$controller = $this->getController();
// Check if method exists
if (!method_exists($controller, $this->postMethod))
throw new TasksException("Could not handle task. Post method '$this->postMethod' not found on controller.");
if (!method_exists($controller, 'getTaskStatus'))
throw new TasksException("Could not handle task. Method 'getTaskStatus()' not found on controller, which is required.");
if ($this->parentInput !== null && method_exists($controller, 'setInput'))
$controller->setInput($this->parentInput);
// Call method and collect output
$this->postOutput = call_user_func_array([$controller, $this->postMethod], [$task]);
$success = $controller->getTaskStatus();
if (!is_bool($success))
throw new TasksException("Could not determine whether task has succeeded. getTaskStatus() returned non-bool.");
return $success;
}
/**
* @inheritDoc
*/
public function getPostOutput(): string
{
return $this->postOutput;
}
/**
* @return Controller
* @throws TasksException
*/
private function getController(): Controller
{
// First load the controllers component
try {
/** @var Controllers $controllers */
$controllers = Factory::getInstance('controllers');
// Load the requested controller
return $controllers->get($this->controllerName, [], $this->controllerNamespace);
} catch (FactoryException $e) {
throw new TasksException("Could not get controller. FuzeWorks\MVCR is not installed!");
} catch (ControllerException $e) {
throw new TasksException("Could not get controller. Controller threw exception: '" . $e->getMessage() . "'");
} catch (NotFoundException $e) {
throw new TasksException("Could not get controller. Controller was not found.");
}
}
/**
* @var Handler
*/
private $parentHandler;
/**
* @inheritDoc
*/
public function getParentHandler(): ?Handler
{
return $this->parentHandler;
}
/**
* @inheritDoc
*/
public function setParentHandler(Handler $parentHandler): void
{
$this->parentHandler = $parentHandler;
}
/**
* @inheritDoc
*/
public function setParentInput(string $input): void
{
$this->parentInput = $input;
}
}

250
src/FuzeWorks/Async/Handler/DependentTaskHandler.php Normal file
View File

@ -0,0 +1,250 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async\Handler;
use FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Constraint\DependencyConstraint;
use FuzeWorks\Async\Handler;
use FuzeWorks\Async\Task;
use FuzeWorks\Async\Tasks;
use FuzeWorks\Async\TasksException;
use FuzeWorks\Exception\FactoryException;
use FuzeWorks\Exception\LibraryException;
use FuzeWorks\Factory;
use FuzeWorks\Libraries;
class DependentTaskHandler implements Handler
{
/**
* @var Handler
*/
protected $parentHandler;
/**
* @var string
*/
protected $output;
/**
* @var array
*/
private $dependencyList;
/**
* @var int
*/
private $delayTimes;
/**
* DependentTaskHandler constructor.
*
* To add dependencies, the following array should be supplied:
* $dependencyList: array(string 'taskId', string 'taskId', string 'taskId')
*
* @param array $dependencyList
* @param int $delayTimes Time that a task should be delayed before retrying
*/
public function __construct(array $dependencyList = [], int $delayTimes = 3)
{
$this->dependencyList = $dependencyList;
$this->delayTimes = $delayTimes;
}
/**
* @inheritDoc
*/
public function init(Task $task)
{
if (!empty($this->dependencyList))
$task->addConstraint(new DependencyConstraint($this->dependencyList, $this->delayTimes));
}
/**
* @inheritDoc
*/
public function primaryHandler(Task $task): bool
{
// First find all the dependencies
try {
$dependencies = $this->fetchDependencies($task);
$this->output = json_encode($dependencies);
return true;
} catch (TasksException $e) {
$this->output = 'Failed to fetch dependencies. TasksException: ' . $e->getMessage();
return false;
}
}
/**
* @inheritDoc
*/
public function getOutput(): string
{
return $this->output;
}
/**
* @inheritDoc
*/
public function postHandler(Task $task)
{
// First find all the dependencies
try {
$dependencies = $this->fetchDependencies($task);
$this->output = json_encode($dependencies);
return true;
} catch (TasksException $e) {
$this->output = 'Failed to fetch dependencies. TasksException: ' . $e->getMessage();
return false;
}
}
/**
* @inheritDoc
*/
public function getPostOutput(): string
{
return $this->output;
}
/**
* @inheritDoc
*/
public function getParentHandler(): ?Handler
{
return $this->parentHandler;
}
/**
* @inheritDoc
*/
public function setParentHandler(Handler $parentHandler): void
{
$this->parentHandler = $parentHandler;
}
/**
* @inheritDoc
*/
public function setParentInput(string $input): void
{
// Parent output gets set at this handler's output.
// Only if this class has something to intervene it will override the parent output
// Which should be always... but alas.
$this->output = $input;
}
/**
* @param Task $task
* @return array
* @throws TasksException
*/
protected function fetchDependencies(Task $task): array
{
// When it receives the task, all dependencies should already be handled
// the primary handler will therefore connect with the DependencyConstraint and fetch dependencies
$constraints = $task->getConstraints();
// First prepare a list of dependencies
$dependencies = [];
$dependencyConstraints = [];
foreach ($constraints as $constraint) {
if ($constraint instanceof Constraint)
$dependencyConstraints[] = $constraint;
}
// If no dependencies found, throw exception
if (empty($dependencyConstraints))
return $dependencies;
// Afterwards build a list of dependencies
/** @var DependencyConstraint $constraint */
foreach ($dependencyConstraints as $constraint) {
foreach ($constraint->getDependencies() as $dependency) {
if (!isset($dependencies[$dependency]))
$dependencies[$dependency] = [];
}
}
// Now that all dependencies are determined, fetch all the output
$tasks = $this->loadTasksLib();
$taskStorage = $tasks->getTaskStorage();
foreach ($dependencies as $dependency => $data)
{
// Fetch the task
try {
$dependencyTask = $taskStorage->getTaskById($dependency);
// Then fetch all output
$dependencies[$dependency]['status'] = $dependencyTask->getStatus();
$dependencies[$dependency]['output'] = $dependencyTask->getOutput();
$dependencies[$dependency]['errors'] = $dependencyTask->getErrors();
$dependencies[$dependency]['post'] = $dependencyTask->getPostOutput();
$dependencies[$dependency]['postErrors'] = $dependencyTask->getPostErrors();
} catch (TasksException $e) {
$dependencies[$dependency]['status'] = Task::FAILED;
$dependencies[$dependency]['output'] = null;
$dependencies[$dependency]['errors'] = 'Task not found.';
$dependencies[$dependency]['post'] = null;
$dependencies[$dependency]['postErrors'] = null;
}
}
return $dependencies;
}
/**
* Load the tasks library, so that dependencies can get scanned later
*
* @return Tasks
* @throws TasksException
*/
private function loadTasksLib(): Tasks
{
try {
/** @var Libraries $libraries */
$libraries = Factory::getInstance('libraries');
/** @var Tasks $tasks */
$tasks = $libraries->get('async');
return $tasks;
} catch (FactoryException | LibraryException $e) {
throw new TasksException("Could not constrain task. Async library could not be loaded.");
}
}
}

110
src/FuzeWorks/Async/ShellWorker.php
View File

@ -67,88 +67,84 @@ class ShellWorker
}
/**
* @param string $taskID
* Run a task by finding its ID
*
* @param string $taskId
* @param bool $post
* @throws EventException
* @throws TasksException
*/
public function run(string $taskID, bool $post = false)
public function runTaskById(string $taskId, bool $post = false)
{
// First fetch the task
try {
$task = $this->taskStorage->getTaskById($taskID);
$task = $this->taskStorage->getTaskById($taskId);
} catch (TasksException $e) {
throw new TasksException("Could not run worker. Task not found.");
}
$this->run($task, $post);
}
/**
* @param Task $task
* @param bool $post
* @throws EventException
* @throws TasksException
*/
public function run(Task $task, bool $post = false)
{
// Fire a taskHandleEvent
/** @var TaskHandleEvent $event */
$event = Events::fireEvent(new TaskHandleEvent(), $task);
$task = $event->getTask();
// Set task to this worker
$this->task = $task;
$this->task = $event->getTask();
$this->post = $post;
// Fetch the callable
$class = $this->task->getHandlerClass();
if (!class_exists($class, true))
{
$errors = 'Could not run task. HandlerClass \'' . $class . '\' not found.';
if (!$post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
$handler = $this->task->getHandler();
throw new TasksException("Could not run task. '$class' not found.");
}
// Execute the handler and all its parent handlers
$success = $this->executeHandler($this->task, $handler, $post);
// Create the handler
/** @var Handler $object */
$object = new $class();
if (!$object instanceof Handler)
{
$errors = "Could not run task. '$class' is not instance of Handler.";
if (!$post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
throw new TasksException("Could not run task. '$class' is not instance of Handler.");
}
// Run postHandler if post mode is requested
if ($post)
{
$postSuccess = $object->postHandler($this->task);
$postOutput = $object->getPostOutput();
$postOutput = is_null($postOutput) ? '' : (string) $postOutput;
$postErrors = $this->getErrors();
if (!$postSuccess)
$this->taskStorage->writePostOutput($this->task, $postOutput, $postErrors, Task::FAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, $postOutput, $postErrors, Task::SUCCESS, $this->task->getRetries());
$this->output($postOutput, $postErrors);
return;
}
// Run primaryHandler if requested
$success = $object->primaryHandler($this->task);
$output = $object->getOutput();
$output = is_null($output) ? '' : (string) $output;
// Fetch the output and errors
$output = $post ? $handler->getPostOutput() : $handler->getOutput();
$output = is_null($output) ? '' : $output;
$errors = $this->getErrors();
// And afterwards write the results to the TaskStorage
if (!$success)
$this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::FAILED, $this->task->getRetries());
// If the task failed, write so to task storage, based on whether this is a post request or not
if (!$success && $post)
$this->taskStorage->writePostOutput($this->task, $output, $errors, Task::FAILED);
elseif (!$success && !$post)
$this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::FAILED);
elseif ($success && $post)
$this->taskStorage->writePostOutput($this->task, $output, $errors, Task::SUCCESS);
else
$this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::SUCCESS, $this->task->getRetries());
$this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::SUCCESS);
$this->output($output, $errors);
// And write the final output
$this->output((string) $output, $errors);
}
protected function executeHandler(Task $task, Handler $handler, bool $usePost = false): bool
{
// First check to see if there is a parent handler
$parent = $handler->getParentHandler();
if (!is_null($parent)) {
// Execute the parent
if ($this->executeHandler($task, $parent, $usePost) === false)
return false;
// Fetch the output of the parent
$output = $usePost ? $parent->getPostOutput() : $parent->getOutput();
// And insert it as input into the child handler
$handler->setParentInput($output);
}
return $usePost ? $handler->postHandler($task) : $handler->primaryHandler($task);
}
/**
* In case a fatal error or exception occurs, the errors shall be redirected to stderr
*
@ -170,9 +166,9 @@ class ShellWorker
try {
// Write to TaskStorage
if (!$this->post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::FAILED, $this->task->getRetries());
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::FAILED);
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::FAILED, $this->task->getRetries());
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::FAILED);
} catch (TasksException $e) {
// Ignore
}

31
src/FuzeWorks/Async/Supervisors/ParallelSuperVisor.php
View File

@ -70,8 +70,7 @@ class ParallelSuperVisor implements SuperVisor
public function cycle(): int
{
// First: if there are no tasks, load them
$this->taskStorage->refreshTasks();
$this->tasks = $this->taskStorage->readTasks();
$this->tasks = $this->taskStorage->readTasks(true);
// If there are still no tasks, nothing is queued, so this cycle can end.
if (empty($this->tasks))
@ -94,6 +93,7 @@ class ParallelSuperVisor implements SuperVisor
// Start the process using the executor service
$task = $this->executor->startTask($task);
$task->setStatus(Task::RUNNING);
$task->startTaskTime();
// Modify the task in TaskStorage
$this->taskStorage->modifyTask($task);
@ -108,16 +108,11 @@ class ParallelSuperVisor implements SuperVisor
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
}
// CANCELLED/COMPLETED: remove the task if requested to do so
elseif ($task->getStatus() === Task::COMPLETED || $task->getStatus() === Task::CANCELLED)
{
}
// RUNNING: check if task is still running. If not, set result based on output
elseif ($task->getStatus() === Task::RUNNING)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readTaskOutput($task, $task->getRetries());
$output = $this->taskStorage->readTaskOutput($task);
$hasOutput = !is_null($output);
// If nothing is found, the process has crashed and status PFAILED should be set
@ -140,6 +135,7 @@ class ParallelSuperVisor implements SuperVisor
continue;
// If any changes have been made, they should be written to TaskStorage
$task->endTaskTime();
$this->taskStorage->modifyTask($task);
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
}
@ -148,7 +144,7 @@ class ParallelSuperVisor implements SuperVisor
elseif ($task->getStatus() === Task::PFAILED || $task->getStatus() === Task::FAILED)
{
// First fetch retry conditions
$settings = $task->getRetrySettings();
$settings = $task->getSettings();
// First test if any retries should be tried at all
if ($settings['retryOnFail'] === true && $task->getRetries() < $settings['maxRetries'])
@ -163,6 +159,7 @@ class ParallelSuperVisor implements SuperVisor
$task->addRetry();
$task = $this->executor->startTask($task);
$task->setStatus(Task::RUNNING);
$task->startTaskTime();
$this->taskStorage->modifyTask($task);
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
continue;
@ -174,6 +171,7 @@ class ParallelSuperVisor implements SuperVisor
{
$task->resetRetries();
$task = $this->executor->startTask($task, true);
$task->startPostTime();
$task->setStatus(Task::POST);
}
else
@ -190,6 +188,7 @@ class ParallelSuperVisor implements SuperVisor
{
$task->resetRetries();
$task = $this->executor->startTask($task, true);
$task->startPostTime();
$task->setStatus(Task::POST);
}
else
@ -203,20 +202,22 @@ class ParallelSuperVisor implements SuperVisor
elseif ($task->getStatus() === Task::POST)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readPostOutput($task, $task->getRetries());
$output = $this->taskStorage->readPostOutput($task);
$hasOutput = !is_null($output);
// If a task is not running and has no output, an error has occurred
if (!$isRunning && !$hasOutput)
{
// Test if a retry should be attempted
$settings = $task->getRetrySettings();
$settings = $task->getSettings();
if ($settings['retryOnFail'] === true && $settings['retryPostFailures'] === true && $settings['maxRetries'] > $task->getRetries())
{
$task->addRetry();
$task = $this->executor->startTask($task, true);
}
elseif ($settings['maxRetries'] <= $task->getRetries())
elseif ($settings['retryOnFail'] === true && $settings['retryPostFailures'] === true && $settings['maxRetries'] <= $task->getRetries())
$task->setStatus(Task::CANCELLED);
else
$task->setStatus(Task::CANCELLED);
}
// @todo Retry after $max_Time
@ -224,13 +225,17 @@ class ParallelSuperVisor implements SuperVisor
elseif (!$isRunning && $hasOutput)
{
$task->setPostOutput($output['output'], $output['errors']);
$task->setStatus(Task::COMPLETED);
if ($output['statusCode'] === Task::SUCCESS)
$task->setStatus(Task::COMPLETED);
else
$task->setStatus(Task::CANCELLED);
}
// If the task is still running, leave it be
else
continue;
// If any changes have been made, they should be written to TaskStorage
$task->endPostTime();
$this->taskStorage->modifyTask($task);
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
}

183
src/FuzeWorks/Async/Task.php
View File

@ -116,14 +116,9 @@ class Task
protected $taskId;
/**
* @var string
* @var int
*/
protected $handlerClass;
/**
* @var bool
*/
protected $usePostHandler = false;
protected $status = Task::PENDING;
/**
* @var array
@ -131,9 +126,14 @@ class Task
protected $arguments;
/**
* @var int
* @var Handler
*/
protected $status = Task::PENDING;
protected $handler;
/**
* @var bool
*/
protected $usePostHandler = false;
/**
* @var Constraint[]
@ -170,11 +170,6 @@ class Task
*/
protected $attributes = [];
/**
* @var Process
*/
protected $process;
/* -------- Some settings ------------ */
protected $retryOnFail = false;
@ -183,22 +178,27 @@ class Task
protected $retryRFailures = true;
protected $retryPostFailures = true;
protected $retries = 0;
protected $maxTime = 30;
/**
* Task constructor.
*
* Creates a Task object, which can be added to the TaskQueue.
*
* @param string $identifier The unique identifier of this task. Make sure it is always unique!
* @param string $handlerClass The class that shall handle this task
* @param bool $usePostHandler Whether the postHandler on handlerClass should also be used
* @param string $identifier The unique identifier of this task. Make sure it is always unique!
* @param Handler $handler The Handler object which will run the Task in the Worker
* @param bool $usePostHandler Whether the postHandler on Handler should also be used
* @param mixed $parameters,... The arguments provided to the method that shall handle this class
* @throws TasksException
*/
public function __construct(string $identifier, string $handlerClass, bool $usePostHandler = false)
public function __construct(string $identifier, Handler $handler, bool $usePostHandler = false)
{
// Check if the provided Handler is serializable
if (!$this->isSerializable($handler))
throw new TasksException("Could not create Task. Provided Handler is not serializable.");
$this->taskId = $identifier;
$this->handlerClass = $handlerClass;
$this->handler = $handler;
$this->usePostHandler = $usePostHandler;
if (func_num_args() > 3)
$args = array_slice(func_get_args(), 3);
@ -210,6 +210,9 @@ class Task
throw new TasksException("Could not create Task. Provided arguments are not serializable.");
$this->arguments = $args;
// Init the handler
$this->handler->init($this);
}
/**
@ -223,17 +226,17 @@ class Task
}
/**
* Gets the name of the class that shall process this task
* Gets the Handler that shall process this task
*
* @return string
* @return Handler
*/
public function getHandlerClass(): string
public function getHandler(): Handler
{
return $this->handlerClass;
return $this->handler;
}
/**
* Whether the postHandler on the handlerClass should be invoked after processing the initial task.
* Whether the postHandler on the Handler should be invoked after processing the initial task.
*
* @return bool
*/
@ -314,6 +317,8 @@ class Task
return $this->delayTime;
}
/* ---------------------------------- Attributes setters and getters ------------------ */
/**
* Fetch an attribute of this task
*
@ -343,6 +348,22 @@ class Task
$this->attributes[$key] = $value;
}
/**
* Remove an attribute from a Task
*
* @param string $key
* @throws TasksException
*/
public function removeAttribute(string $key)
{
if (!isset($this->attributes[$key]))
throw new TasksException("Could not remove Task '$this->taskId' attribute '$key'. Not found.");
unset($this->attributes[$key]);
}
/* ---------------------------------- Output setters and getters ---------------------- */
/**
* Return the output of this task execution
*
@ -373,11 +394,10 @@ class Task
return $this->postErrors;
}
/**
* @todo Handle output from multiple attempts
* @param string $output
* @param string $errors
* @todo Handle output from multiple attempts
*/
public function setOutput(string $output, string $errors)
{
@ -386,9 +406,9 @@ class Task
}
/**
* @todo Handle output from multiple attempts
* @param string $output
* @