Implemented changes requested by Wettennet.
Some checks failed
continuous-integration/drone/push Build is failing

Should still be tested!
This commit is contained in:
Abel Hoogeveen 2020-06-13 11:50:32 +02:00
parent 3e0a312c80
commit 4758f4154b
No known key found for this signature in database
GPG Key ID: 96C2234920BF4292
11 changed files with 361 additions and 9 deletions

View File

@ -44,9 +44,10 @@ interface Constraint
* Return true ONLY when this constraint changes the execution of the Task. Otherwise return false.
*
* @param Task $task
* @param Task[] $tasks
* @return bool
*/
public function intervene(Task $task): bool;
public function intervene(Task $task, array $tasks): bool;
/**
* When intervene() returns true, this method should return the new status of the task.

View File

@ -84,12 +84,12 @@ class DependencyConstraint implements Constraint
/**
* @inheritDoc
*/
public function intervene(Task $task): bool
public function intervene(Task $task, array $tasks): bool
{
// Fetch taskStorage
try {
$tasks = $this->loadTasksLib();
$taskStorage = $tasks->getTaskStorage();
$tasksLib = $this->loadTasksLib();
$taskStorage = $tasksLib->getTaskStorage();
// Is any dependency unresolved?
$hasUnresolved = false;

View File

@ -70,7 +70,7 @@ class FixedTimeConstraint implements Constraint
/**
* @inheritDoc
*/
public function intervene(Task $task): bool
public function intervene(Task $task, array $tasks): bool
{
if ($task->getStatus() === Task::PENDING && time() < $this->timestamp)
return true;

View File

@ -0,0 +1,119 @@
<?php
/**
* FuzeWorks Component.
*
* 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\Constraint;
use FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Task;
/**
* Class GroupConstraint
*
* Use this constraint to maximize the amount of concurrent tasks within a certain group.
* If the amount of running tasks equals the maximum concurrent tasks, this constraint will issue a delay.
* Useful when multiple tasks are not dependent on each other, but should be run separately.
*
* @package FuzeWorks\Async\Constraint
*/
class GroupConstraint implements Constraint
{
/**
* @var string
*/
protected $groupName;
/**
* @var int
*/
protected $maxConcurrent = 3;
protected $returnStatus = Task::DELAYED;
/**
* GroupConstraint constructor.
*
* To add a group, simply provide the $groupName and the maximum of concurrent tasks within this group.
*
* @param string $groupName
* @param int $maxConcurrent
*/
public function __construct(string $groupName, int $maxConcurrent = 3)
{
$this->groupName = $groupName;
$this->maxConcurrent = $maxConcurrent;
}
public function getGroupName(): string
{
return $this->groupName;
}
public function intervene(Task $task, array $tasks): bool
{
// Find all tasks within this group and that are running
$runningTasks = 0;
foreach ($tasks as $t)
{
foreach ($t->getConstraints() as $constraint)
{
if ($constraint instanceof GroupConstraint && $this->groupName === $constraint->getGroupName())
{
if ($t->getStatus() === Task::RUNNING || $t->getStatus() === Task::POST)
$runningTasks++;
}
}
// If the running tasks exceeds the maximum, block this task from running
if ($runningTasks >= $this->maxConcurrent)
{
$this->returnStatus = Task::DELAYED;
return true;
}
}
return false;
}
public function blockCode(): int
{
return $this->returnStatus;
}
public function delayTime(): int
{
return time() + 1;
}
}

View File

@ -0,0 +1,118 @@
<?php
/**
* FuzeWorks Component.
*
* 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\Constraint;
use FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Task;
use FuzeWorks\Factory;
class GroupDependencyConstraint implements Constraint
{
/**
* @var string
*/
protected $groupName;
protected $returnStatus = Task::DELAYED;
public function __construct(string $groupName)
{
$this->groupName = $groupName;
}
/**
* @inheritDoc
*/
public function intervene(Task $task, array $tasks): bool
{
// Find whether any task is not completed
$hasFailed = false;
$hasUnresolved = false;
foreach ($tasks as $t)
{
foreach ($t->getConstraints() as $constraint)
{
// Check whether the constraint is a GroupConstraint
if ($constraint instanceof GroupConstraint && $this->groupName === $constraint->getGroupName())
{
// Check whether GroupConstraint has failed
if ($t->getStatus() === Task::CANCELLED)
$hasFailed = true;
// If the task is not cancelled and not completed, there are unresolved tasks
elseif ($t->getStatus() !== Task::COMPLETED)
$hasUnresolved = true;
}
}
// If there are failed tasks, cancel this dependent
if ($hasFailed)
{
$this->returnStatus = Task::CANCELLED;
$task->setOutput('', 'Task cancelled due to failed group dependency.');
return true;
}
elseif ($hasUnresolved)
{
$this->returnStatus = Task::DELAYED;
return true;
}
}
// Return the taskList
//$task->setOutput(serialize($tasks), '');
//$this->returnStatus = Task::CANCELLED;
//return true;
return false;
}
/**
* @inheritDoc
*/
public function blockCode(): int
{
return $this->returnStatus;
}
/**
* @inheritDoc
*/
public function delayTime(): int
{
return time() + 3;
}
}

View File

@ -59,13 +59,23 @@ class ShellExecutor implements Executor
if (!isset($parameters['workerFile']) || !isset($parameters['bootstrapFile']))
throw new TasksException("Could not construct ShellExecutor. Parameter failure.");
// Determine alternative worker file
$alternative = dirname(__DIR__, 4) . DIRECTORY_SEPARATOR . 'bin' . DIRECTORY_SEPARATOR . 'worker';
// Fetch workerFile
$this->worker = $parameters['workerFile'];
if (!file_exists($this->worker))
$workerFile = null;
if (file_exists($parameters['workerFile']))
$workerFile = $parameters['workerFile'];
elseif (file_exists($alternative))
$workerFile = $alternative;
// If no workerFile is found, throw an exception
if (is_null($workerFile))
throw new TasksException("Could not construct ShellExecutor. ShellWorker script does not exist.");
// First determine the PHP binary
$this->binary = PHP_BINDIR . DS . 'php';
$this->worker = $workerFile;
$this->bootstrapFile = $parameters['bootstrapFile'];
if (!file_exists($this->bootstrapFile))
throw new TasksException("Could not construct ShellExecutor. No bootstrap file found.");

View File

@ -149,6 +149,9 @@ class ControllerHandler implements Handler
*/
public function getOutput(): string
{
if (!is_string($this->output))
return '';
return $this->output;
}
@ -189,6 +192,9 @@ class ControllerHandler implements Handler
*/
public function getPostOutput(): string
{
if (!is_string($this->postOutput))
return '';
return $this->postOutput;
}

View File

@ -62,6 +62,7 @@ class ParallelSuperVisor implements SuperVisor
{
$this->taskStorage = $taskStorage;
$this->executor = $executor;
gc_enable();
}
/**
@ -88,7 +89,10 @@ class ParallelSuperVisor implements SuperVisor
// If the task changed status, task is no longer pending and should be processed by another statement
if ($task->getStatus() !== Task::PENDING)
{
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
continue;
}
// Start the process using the executor service
$task = $this->executor->startTask($task);
@ -101,6 +105,7 @@ class ParallelSuperVisor implements SuperVisor
}
// DELAYED: If task is delayed, and enough time has passed, change the status back to pending
// @todo Don't set to pending if still constrained
elseif ($task->getStatus() === Task::DELAYED && time() > $task->getDelayTime())
{
$task->setStatus(Task::PENDING);
@ -262,12 +267,19 @@ class ParallelSuperVisor implements SuperVisor
return SuperVisor::RUNNING;
}
/**
* @todo Here it saves the constraint code regardless of whether there is a change. Fix!
*
* @param Task $task
* @return Task
* @throws TasksException
*/
private function testConstraints(Task $task): Task
{
$constraints = $task->getConstraints();
foreach ($constraints as $constraint)
{
if ($constraint->intervene($task) && $constraint->blockCode() != 0)
if ($constraint->intervene($task, $this->tasks) && $constraint->blockCode() != 0)
{
$task->setStatus($constraint->blockCode());
if ($constraint->blockCode() === Task::DELAYED)
@ -275,7 +287,6 @@ class ParallelSuperVisor implements SuperVisor
// Save changes to TaskStorage
$this->taskStorage->modifyTask($task);
fwrite(STDOUT, "\nChanged status of task '".$task->getId()."' to status " . Task::getStatusType($task->getStatus()));
}
}

View File

@ -262,6 +262,9 @@ class Task
*/
public function addConstraint(Constraint $constraint)
{
if (method_exists($constraint, 'init'))
$constraint->init($this);
$this->constraints[] = $constraint;
}
@ -319,6 +322,16 @@ class Task
/* ---------------------------------- Attributes setters and getters ------------------ */
/**
* Retrieve every attribute of this Task
*
* @return array
*/
public function getAttributes(): array
{
return $this->attributes;
}
/**
* Fetch an attribute of this task
*

View File

@ -74,6 +74,15 @@ interface TaskStorage
*/
public function getTaskById(string $identifier): Task;
/**
* Retrieve a list of all tasks that have a certain attribute
*
* @param string $attributeKey
* @param string $attributeValue
* @return array
*/
public function getTasksByAttribute(string $attributeKey, string $attributeValue): array;
/**
* Modifies a task
*

View File

@ -54,6 +54,8 @@ class RedisTaskStorage implements TaskStorage
protected $indexSet = 'async_index';
protected $unfinishedSet = 'async_index_unfinished';
protected $attributesSet = 'async_attributes';
protected $attributesSetPrefix = 'async_attributes_';
protected $key_prefix = 'async_task_';
/**
@ -115,6 +117,22 @@ class RedisTaskStorage implements TaskStorage
if ($this->conn->hSet($this->key_prefix . $taskId, 'data', $taskData) === FALSE)
return false;
// And create the attributes sets
foreach ($task->getAttributes() as $key => $val)
{
// If the attribute value is not a string it can't be used as a searchable attribute
if (!is_string($val))
continue;
// And finally add it to the set
$setKey = $this->attributesSetPrefix . $key . '_' . $val;
$this->conn->sAdd($setKey, $taskId);
// And add the attribute combo to the attributesSet
if (!$this->conn->sIsMember($this->attributesSet, $key . '_' . $val))
$this->conn->sAdd($this->attributesSet, $key . '_' . $val);
}
return true;
}
@ -155,6 +173,22 @@ class RedisTaskStorage implements TaskStorage
return $task;
}
/**
* @inheritDoc
*/
public function getTasksByAttribute(string $attributeKey, string $attributeValue): array
{
// Fetch the taskList by the attribute
$setKey = $this->attributesSetPrefix . $attributeKey . '_' . $attributeValue;
$taskList = $this->conn->sMembers($setKey);
$tasks = [];
foreach ($taskList as $taskId)
$tasks[] = unserialize($this->conn->hGet($this->key_prefix . $taskId, 'data'));
return $tasks;
}
/**
* @inheritDoc
*/
@ -182,6 +216,29 @@ class RedisTaskStorage implements TaskStorage
if ($this->conn->hSet($this->key_prefix . $taskId, 'data', $taskData) === FALSE)
return false;
// Delete all the attributes
$allLoggedAttributes = $this->conn->sMembers($this->attributesSet);
foreach ($allLoggedAttributes as $attribute)
{
if ($this->conn->sIsMember($this->attributesSetPrefix . $attribute, $taskId))
$this->conn->sRem($this->attributesSetPrefix . $attribute, $taskId);
}
// And write all the attributes
foreach ($task->getAttributes() as $key => $val)
{
// If the attribute value is not a string it can't be used as a searchable attribute
if (!is_string($val))
continue;
// And finally add it to the set
$setKey = $this->attributesSetPrefix . $key . '_' . $val;
$this->conn->sAdd($setKey, $taskId);
// And add the attribute combo to the attributesSet
$this->conn->sAdd($this->attributesSet, $key . '_' . $val);
}
// Modify the unfinished set
if ($this->conn->sIsMember($this->unfinishedSet, $taskId) && ($task->getStatus() === Task::COMPLETED || $task->getStatus() === Task::CANCELLED ))
$this->conn->sRem($this->unfinishedSet, $taskId);
@ -212,6 +269,14 @@ class RedisTaskStorage implements TaskStorage
if ($this->conn->sIsMember($this->unfinishedSet, $taskId))
$this->conn->sRem($this->unfinishedSet, $taskId);
// Delete all the attributes
$allLoggedAttributes = $this->conn->sMembers($this->attributesSet);
foreach ($allLoggedAttributes as $attribute)
{
if ($this->conn->sIsMember($this->attributesSetPrefix . $attribute, $taskId))
$this->conn->sRem($this->attributesSetPrefix . $attribute, $taskId);
}
// Delete the task itself
if ($this->conn->del($this->key_prefix . $taskId) > 0)
return true;