strk/friendica
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Refactor Process for new paradigm

Browse source
This commit is contained in:
Philipp Holzer 2021-10-24 20:43:59 +02:00
parent 0e2e488521
commit 38f70cc55a
Signed by: nupplaPhil
GPG key ID: 24A7501396EB5432
14 changed files with 456 additions and 491 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
bin/daemon.php
View file

@ -196,7 +196,7 @@ while (true) {
$do_cron = true; $do_cron = true;
} }
if ($do_cron || (!DI::process()->isMaxLoadReached() && Worker::entriesExists() && Worker::isReady())) { if ($do_cron || (!DI::system()->isMaxLoadReached() && Worker::entriesExists() && Worker::isReady())) {
Worker::spawnWorker($do_cron); Worker::spawnWorker($do_cron);
} else { } else {
Logger::info('Cool down for 5 seconds', ['pid' => $pid]); Logger::info('Cool down for 5 seconds', ['pid' => $pid]);

8
bin/worker.php
View file

@ -81,8 +81,10 @@ if ($spawn) {
$run_cron = !array_key_exists('n', $options) && !array_key_exists('no_cron', $options); $run_cron = !array_key_exists('n', $options) && !array_key_exists('no_cron', $options);
Worker::processQueue($run_cron); $process = DI::process()->create(getmypid());
Worker::unclaimProcess(); Worker::processQueue($run_cron, $process);
DI::process()->end(); Worker::unclaimProcess($process);
DI::process()->delete($process);

12
src/App.php
View file

@ -118,9 +118,9 @@ class App
private $args; private $args;
/** /**
* @var Core\Process The process methods * @var Core\System The system methods
*/ */
private $process; private $system;
/** /**
* @var IManagePersonalConfigValues * @var IManagePersonalConfigValues
@ -327,10 +327,10 @@ class App
* @param Profiler $profiler The profiler of this application * @param Profiler $profiler The profiler of this application
* @param L10n $l10n The translator instance * @param L10n $l10n The translator instance
* @param App\Arguments $args The Friendica Arguments of the call * @param App\Arguments $args The Friendica Arguments of the call
* @param Core\Process $process The process methods * @param Core\System $system The system methods
* @param IManagePersonalConfigValues $pConfig Personal configuration * @param IManagePersonalConfigValues $pConfig Personal configuration
*/ */
public function __construct(Database $database, IManageConfigValues $config, App\Mode $mode, BaseURL $baseURL, LoggerInterface $logger, Profiler $profiler, L10n $l10n, Arguments $args, Core\Process $process, IManagePersonalConfigValues $pConfig) public function __construct(Database $database, IManageConfigValues $config, App\Mode $mode, BaseURL $baseURL, LoggerInterface $logger, Profiler $profiler, L10n $l10n, Arguments $args, Core\System $system, IManagePersonalConfigValues $pConfig)
{ {
$this->database = $database; $this->database = $database;
$this->config = $config; $this->config = $config;
@ -340,7 +340,7 @@ class App
$this->logger = $logger; $this->logger = $logger;
$this->l10n = $l10n; $this->l10n = $l10n;
$this->args = $args; $this->args = $args;
$this->process = $process; $this->system = $system;
$this->pConfig = $pConfig; $this->pConfig = $pConfig;
$this->load(); $this->load();
@ -589,7 +589,7 @@ class App
} }
// Max Load Average reached: ERROR // Max Load Average reached: ERROR
if ($this->process->isMaxProcessesReached() || $this->process->isMaxLoadReached()) { if ($this->system->isMaxProcessesReached() || $this->system->isMaxLoadReached()) {
header('Retry-After: 120'); header('Retry-After: 120');
header('Refresh: 120; url=' . $this->baseURL->get() . "/" . $this->args->getQueryString()); header('Refresh: 120; url=' . $this->baseURL->get() . "/" . $this->args->getQueryString());

274
src/Core/Process.php
View file

@ -1,274 +0,0 @@
<?php
/**
* @copyright Copyright (C) 2010-2021, the Friendica project
*
* @license GNU AGPL version 3 or any later version
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
*/
namespace Friendica\Core;
use Friendica\App;
use Friendica\Core\Config\Capability\IManageConfigValues;
use Friendica\Model;
use Psr\Log\LoggerInterface;
/**
* Methods for interacting with the current process or create new process
*
* @todo 2019.12 Next release, this class holds all process relevant methods based on the big Worker class
* - Starting new processes (including checks)
* - Enabling multi-node processing (e.g. for docker service)
* - Using an process-id per node
* - Using memory locks for multi-node locking (redis, memcached, ..)
*/
class Process
{
/**
* @var LoggerInterface
*/
private $logger;
/**
* @var App\Mode
*/
private $mode;
/**
* @var IManageConfigValues
*/
private $config;
/**
* @var string
*/
private $basePath;
/** @var Model\Process */
private $processModel;
/**
* The Process ID of this process
*
* @var int
*/
private $pid;
public function __construct(LoggerInterface $logger, App\Mode $mode, IManageConfigValues $config, Model\Process $processModel, string $basepath, int $pid)
{
$this->logger = $logger;
$this->mode = $mode;
$this->config = $config;
$this->basePath = $basepath;
$this->processModel = $processModel;
$this->pid = $pid;
}
/**
* Set the process id
*
* @param integer $pid
* @return void
*/
public function setPid(int $pid)
{
$this->pid = $pid;
}
/**
* Log active processes into the "process" table
*/
public function start()
{
$trace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS, 1);
$command = basename($trace[0]['file']);
$this->processModel->deleteInactive();
$this->processModel->insert($command, $this->pid);
}
/**
* Remove the active process from the "process" table
*
* @return bool
* @throws \Exception
*/
public function end()
{
return $this->processModel->deleteByPid($this->pid);
}
/**
* Checks if the maximum number of database processes is reached
*
* @return bool Is the limit reached?
*/
public function isMaxProcessesReached()
{
// Deactivated, needs more investigating if this check really makes sense
return false;
/*
* Commented out to suppress static analyzer issues
*
if ($this->mode->isBackend()) {
$process = 'backend';
$max_processes = $this->config->get('system', 'max_processes_backend');
if (intval($max_processes) == 0) {
$max_processes = 5;
}
} else {
$process = 'frontend';
$max_processes = $this->config->get('system', 'max_processes_frontend');
if (intval($max_processes) == 0) {
$max_processes = 20;
}
}
$processlist = DBA::processlist();
if ($processlist['list'] != '') {
$this->logger->debug('Processcheck: Processes: ' . $processlist['amount'] . ' - Processlist: ' . $processlist['list']);
if ($processlist['amount'] > $max_processes) {
$this->logger->debug('Processcheck: Maximum number of processes for ' . $process . ' tasks (' . $max_processes . ') reached.');
return true;
}
}
return false;
*/
}
/**
* Checks if the minimal memory is reached
*
* @return bool Is the memory limit reached?
*/
public function isMinMemoryReached()
{
$min_memory = $this->config->get('system', 'min_memory', 0);
if ($min_memory == 0) {
return false;
}
if (!is_readable('/proc/meminfo')) {
return false;
}
$memdata = explode("\n", file_get_contents('/proc/meminfo'));
$meminfo = [];
foreach ($memdata as $line) {
$data = explode(':', $line);
if (count($data) != 2) {
continue;
}
[$key, $val] = $data;
$meminfo[$key] = (int)trim(str_replace('kB', '', $val));
$meminfo[$key] = (int)($meminfo[$key] / 1024);
}
if (!isset($meminfo['MemFree'])) {
return false;
}
$free = $meminfo['MemFree'];
$reached = ($free < $min_memory);
if ($reached) {
$this->logger->warning('Minimal memory reached.', ['free' => $free, 'memtotal' => $meminfo['MemTotal'], 'limit' => $min_memory]);
}
return $reached;
}
/**
* Checks if the maximum load is reached
*
* @return bool Is the load reached?
*/
public function isMaxLoadReached()
{
if ($this->mode->isBackend()) {
$process = 'backend';
$maxsysload = intval($this->config->get('system', 'maxloadavg'));
if ($maxsysload < 1) {
$maxsysload = 50;
}
} else {
$process = 'frontend';
$maxsysload = intval($this->config->get('system', 'maxloadavg_frontend'));
if ($maxsysload < 1) {
$maxsysload = 50;
}
}
$load = System::currentLoad();
if ($load) {
if (intval($load) > $maxsysload) {
$this->logger->warning('system load for process too high.', ['load' => $load, 'process' => $process, 'maxsysload' => $maxsysload]);
return true;
}
}
return false;
}
/**
* Executes a child process with 'proc_open'
*
* @param string $command The command to execute
* @param array $args Arguments to pass to the command ( [ 'key' => value, 'key2' => value2, ... ]
*/
public function run($command, $args)
{
if (!function_exists('proc_open')) {
$this->logger->warning('"proc_open" not available - quitting');
return;
}
$cmdline = $this->config->get('config', 'php_path', 'php') . ' ' . escapeshellarg($command);
foreach ($args as $key => $value) {
if (!is_null($value) && is_bool($value) && !$value) {
continue;
}
$cmdline .= ' --' . $key;
if (!is_null($value) && !is_bool($value)) {
$cmdline .= ' ' . $value;
}
}
if ($this->isMinMemoryReached()) {
$this->logger->warning('Memory limit reached - quitting');
return;
}
if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN') {
$resource = proc_open('cmd /c start /b ' . $cmdline, [], $foo, $this->basePath);
} else {
$resource = proc_open($cmdline . ' &', [], $foo, $this->basePath);
}
if (!is_resource($resource)) {
$this->logger->warning('We got no resource for command.', ['command' => $cmdline]);
return;
}
proc_close($resource);
$this->logger->info('Executed "proc_open"', ['command' => $cmdline, 'callstack' => System::callstack(10)]);
}
}

193
src/Core/System.php
View file

@ -22,18 +22,209 @@
namespace Friendica\Core; namespace Friendica\Core;
use Exception; use Exception;
use Friendica\App;
use Friendica\Core\Config\Capability\IManageConfigValues;
use Friendica\DI; use Friendica\DI;
use Friendica\Network\HTTPException\FoundException; use Friendica\Network\HTTPException\FoundException;
use Friendica\Network\HTTPException\MovedPermanentlyException; use Friendica\Network\HTTPException\MovedPermanentlyException;
use Friendica\Network\HTTPException\TemporaryRedirectException; use Friendica\Network\HTTPException\TemporaryRedirectException;
use Friendica\Util\BasePath; use Friendica\Util\BasePath;
use Friendica\Util\XML; use Friendica\Util\XML;
use Psr\Log\LoggerInterface;
/** /**
* Contains the class with system relevant stuff * Contains the class with system relevant stuff
*/ */
class System class System
{ {
/**
* @var LoggerInterface
*/
private $logger;
/**
* @var App\Mode
*/
private $mode;
/**
* @var IManageConfigValues
*/
private $config;
/**
* @var string
*/
private $basePath;
public function __construct(LoggerInterface $logger, App\Mode $mode, IManageConfigValues $config, string $basepath)
{
$this->logger = $logger;
$this->mode = $mode;
$this->config = $config;
$this->basePath = $basepath;
}
/**
* Checks if the maximum number of database processes is reached
*
* @return bool Is the limit reached?
*/
public function isMaxProcessesReached(): bool
{
// Deactivated, needs more investigating if this check really makes sense
return false;
/*
* Commented out to suppress static analyzer issues
*
if ($this->mode->isBackend()) {
$process = 'backend';
$max_processes = $this->config->get('system', 'max_processes_backend');
if (intval($max_processes) == 0) {
$max_processes = 5;
}
} else {
$process = 'frontend';
$max_processes = $this->config->get('system', 'max_processes_frontend');
if (intval($max_processes) == 0) {
$max_processes = 20;
}
}
$processlist = DBA::processlist();
if ($processlist['list'] != '') {
$this->logger->debug('Processcheck: Processes: ' . $processlist['amount'] . ' - Processlist: ' . $processlist['list']);
if ($processlist['amount'] > $max_processes) {
$this->logger->debug('Processcheck: Maximum number of processes for ' . $process . ' tasks (' . $max_processes . ') reached.');
return true;
}
}
return false;
*/
}
/**
* Checks if the minimal memory is reached
*
* @return bool Is the memory limit reached?
*/
public function isMinMemoryReached(): bool
{
$min_memory = $this->config->get('system', 'min_memory', 0);
if ($min_memory == 0) {
return false;
}
if (!is_readable('/proc/meminfo')) {
return false;
}
$memdata = explode("\n", file_get_contents('/proc/meminfo'));
$meminfo = [];
foreach ($memdata as $line) {
$data = explode(':', $line);
if (count($data) != 2) {
continue;
}
[$key, $val] = $data;
$meminfo[$key] = (int)trim(str_replace('kB', '', $val));
$meminfo[$key] = (int)($meminfo[$key] / 1024);
}
if (!isset($meminfo['MemFree'])) {
return false;
}
$free = $meminfo['MemFree'];
$reached = ($free < $min_memory);
if ($reached) {
$this->logger->warning('Minimal memory reached.', ['free' => $free, 'memtotal' => $meminfo['MemTotal'], 'limit' => $min_memory]);
}
return $reached;
}
/**
* Checks if the maximum load is reached
*
* @return bool Is the load reached?
*/
public function isMaxLoadReached(): bool
{
if ($this->mode->isBackend()) {
$process = 'backend';
$maxsysload = intval($this->config->get('system', 'maxloadavg'));
if ($maxsysload < 1) {
$maxsysload = 50;
}
} else {
$process = 'frontend';
$maxsysload = intval($this->config->get('system', 'maxloadavg_frontend'));
if ($maxsysload < 1) {
$maxsysload = 50;
}
}
$load = System::currentLoad();
if ($load) {
if (intval($load) > $maxsysload) {
$this->logger->warning('system load for process too high.', ['load' => $load, 'process' => $process, 'maxsysload' => $maxsysload]);
return true;
}
}
return false;
}
/**
* Executes a child process with 'proc_open'
*
* @param string $command The command to execute
* @param array $args Arguments to pass to the command ( [ 'key' => value, 'key2' => value2, ... ]
*/
public function run(string $command, array $args)
{
if (!function_exists('proc_open')) {
$this->logger->warning('"proc_open" not available - quitting');
return;
}
$cmdline = $this->config->get('config', 'php_path', 'php') . ' ' . escapeshellarg($command);
foreach ($args as $key => $value) {
if (!is_null($value) && is_bool($value) && !$value) {
continue;
}
$cmdline .= ' --' . $key;
if (!is_null($value) && !is_bool($value)) {
$cmdline .= ' ' . $value;
}
}
if ($this->isMinMemoryReached()) {
$this->logger->warning('Memory limit reached - quitting');
return;
}
if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN') {
$resource = proc_open('cmd /c start /b ' . $cmdline, [], $foo, $this->basePath);
} else {
$resource = proc_open($cmdline . ' &', [], $foo, $this->basePath);
}
if (!is_resource($resource)) {
$this->logger->warning('We got no resource for command.', ['command' => $cmdline]);
return;
}
proc_close($resource);
$this->logger->info('Executed "proc_open"', ['command' => $cmdline, 'callstack' => System::callstack(10)]);
}
/** /**
* Returns a string with a callstack. Can be used for logging. * Returns a string with a callstack. Can be used for logging.
* *
@ -42,7 +233,7 @@ class System
* this is called from a centralized method that isn't relevant to the callstack * this is called from a centralized method that isn't relevant to the callstack
* @return string * @return string
*/ */
public static function callstack(int $depth = 4, int $offset = 0) public static function callstack(int $depth = 4, int $offset = 0): string
{ {
$trace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS); $trace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS);

48
src/Core/Worker.php
View file

@ -23,6 +23,7 @@ namespace Friendica\Core;
use Friendica\App\Mode; use Friendica\App\Mode;
use Friendica\Core; use Friendica\Core;
use Friendica\Core\Worker\Entity\Process;
use Friendica\Database\DBA; use Friendica\Database\DBA;
use Friendica\DI; use Friendica\DI;
use Friendica\Util\DateTimeFormat; use Friendica\Util\DateTimeFormat;
@ -51,26 +52,29 @@ class Worker
private static $last_update; private static $last_update;
private static $state; private static $state;
private static $daemon_mode = null; private static $daemon_mode = null;
/** @var Worker\Entity\Process */
private static $process;
/** /**
* Processes the tasks that are in the workerqueue table * Processes the tasks that are in the workerqueue table
* *
* @param boolean $run_cron Should the cron processes be executed? * @param boolean $run_cron Should the cron processes be executed?
* @param Process $process The current running process
* @return void * @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException * @throws \Friendica\Network\HTTPException\InternalServerErrorException
*/ */
public static function processQueue($run_cron = true) public static function processQueue($run_cron, Process $process)
{ {
self::$up_start = microtime(true); self::$up_start = microtime(true);
// At first check the maximum load. We shouldn't continue with a high load // At first check the maximum load. We shouldn't continue with a high load
if (DI::process()->isMaxLoadReached()) { if (DI::system()->isMaxLoadReached()) {
Logger::notice('Pre check: maximum load reached, quitting.'); Logger::notice('Pre check: maximum load reached, quitting.');
return; return;
} }
// We now start the process. This is done after the load check since this could increase the load. // We now start the process. This is done after the load check since this could increase the load.
DI::process()->start(); self::$process = $process;
// Kill stale processes every 5 minutes // Kill stale processes every 5 minutes
$last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0); $last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0);
@ -134,7 +138,7 @@ class Worker
} }
// Check free memory // Check free memory
if (DI::process()->isMinMemoryReached()) { if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.'); Logger::warning('Memory limit reached, quitting.');
DI::lock()->release(self::LOCK_WORKER); DI::lock()->release(self::LOCK_WORKER);
return; return;
@ -147,7 +151,7 @@ class Worker
// Quit the worker once every cron interval // Quit the worker once every cron interval
if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) { if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
Logger::info('Process lifetime reached, respawning.'); Logger::info('Process lifetime reached, respawning.');
self::unclaimProcess(); self::unclaimProcess($process);
if (self::isDaemonMode()) { if (self::isDaemonMode()) {
self::IPCSetJobState(true); self::IPCSetJobState(true);
} else { } else {
@ -180,7 +184,7 @@ class Worker
} }
// Do we have too few memory? // Do we have too few memory?
if (DI::process()->isMinMemoryReached()) { if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.'); Logger::warning('Memory limit reached, quitting.');
return false; return false;
} }
@ -192,7 +196,7 @@ class Worker
} }
// Possibly there are too much database processes that block the system // Possibly there are too much database processes that block the system
if (DI::process()->isMaxProcessesReached()) { if (DI::system()->isMaxProcessesReached()) {
Logger::warning('Maximum processes reached, quitting.'); Logger::warning('Maximum processes reached, quitting.');
return false; return false;
} }
@ -334,7 +338,7 @@ class Worker
} }
// Constantly check the number of parallel database processes // Constantly check the number of parallel database processes
if (DI::process()->isMaxProcessesReached()) { if (DI::system()->isMaxProcessesReached()) {
Logger::warning("Max processes reached for process", ['pid' => $mypid]); Logger::warning("Max processes reached for process", ['pid' => $mypid]);
return false; return false;
} }
@ -1105,15 +1109,15 @@ class Worker
/** /**
* Removes a workerqueue entry from the current process * Removes a workerqueue entry from the current process
* *
* @param Process $process the process behind the workerqueue
*
* @return void * @return void
* @throws \Exception * @throws \Exception
*/ */
public static function unclaimProcess() public static function unclaimProcess(Process $process)
{ {
$mypid = getmypid();
$stamp = (float)microtime(true); $stamp = (float)microtime(true);
DBA::update('workerqueue', ['executed' => DBA::NULL_DATETIME, 'pid' => 0], ['pid' => $mypid, 'done' => false]); DBA::update('workerqueue', ['executed' => DBA::NULL_DATETIME, 'pid' => 0], ['pid' => $process->pid, 'done' => false]);
self::$db_duration += (microtime(true) - $stamp); self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_write += (microtime(true) - $stamp); self::$db_duration_write += (microtime(true) - $stamp);
} }
@ -1146,7 +1150,7 @@ class Worker
*/ */
private static function forkProcess(bool $do_cron) private static function forkProcess(bool $do_cron)
{ {
if (DI::process()->isMinMemoryReached()) { if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached - quitting'); Logger::warning('Memory limit reached - quitting');
return; return;
} }
@ -1176,11 +1180,10 @@ class Worker
} }
// We now are in the new worker // We now are in the new worker
$pid = getmypid();
DBA::connect(); DBA::connect();
/// @todo Reinitialize the logger to set a new process_id and uid /// @todo Reinitialize the logger to set a new process_id and uid
DI::process()->setPid($pid); $process = DI::process()->create($pid);
$cycles = 0; $cycles = 0;
while (!self::IPCJobsExists($pid) && (++$cycles < 100)) { while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
@ -1189,12 +1192,12 @@ class Worker
Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]); Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
self::processQueue($do_cron); self::processQueue($do_cron, $process);
self::unclaimProcess(); self::unclaimProcess($process);
self::IPCSetJobState(false, $pid); self::IPCSetJobState(false, $pid);
DI::process()->end(); DI::process()->delete($process);
Logger::info('Worker ended', ['pid' => $pid]); Logger::info('Worker ended', ['pid' => $pid]);
exit(); exit();
} }
@ -1211,9 +1214,7 @@ class Worker
if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) { if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) {
self::forkProcess($do_cron); self::forkProcess($do_cron);
} else { } else {
$process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::system()->run('bin/worker.php', ['no_cron' => !$do_cron]);
DI::modelProcess(), DI::app()->getBasePath(), getmypid());
$process->run('bin/worker.php', ['no_cron' => !$do_cron]);
} }
if (self::isDaemonMode()) { if (self::isDaemonMode()) {
self::IPCSetJobState(false); self::IPCSetJobState(false);
@ -1571,8 +1572,7 @@ class Worker
Logger::notice('Starting new daemon process'); Logger::notice('Starting new daemon process');
$command = 'bin/daemon.php'; $command = 'bin/daemon.php';
$a = DI::app(); $a = DI::app();
$process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::modelProcess(), $a->getBasePath(), getmypid()); DI::system()->run($command, ['start']);
$process->run($command, ['start']);
Logger::notice('New daemon process started'); Logger::notice('New daemon process started');
} }

46
src/Core/Worker/Entity/Process.php Normal file
View file

@ -0,0 +1,46 @@
<?php
namespace Friendica\Core\Worker\Entity;
use DateTime;
use Friendica\BaseEntity;
/**
* @property-read int $pid
* @property-read string $command
* @property-read DateTime $created
*/
class Process extends BaseEntity
{
/** @var int */
protected $pid;
/** @var string */
protected $command;
/** @var DateTime */
protected $created;
/**
* @param int $pid
* @param string $command
* @param DateTime $created
*/
public function __construct(int $pid, string $command, DateTime $created)
{
$this->pid = $pid;
$this->command = $command;
$this->created = $created;
}
/**
* Returns a new Process with the given PID
*
* @param int $pid
*
* @return $this
* @throws \Exception
*/
public function withPid(int $pid): Process
{
return new static($pid, $this->command, new DateTime('now', new \DateTimeZone('URC')));
}
}

13
src/Core/Worker/Exception/ProcessPersistenceException.php Normal file
View file

@ -0,0 +1,13 @@
<?php
namespace Friendica\Core\Worker\Exception;
use Throwable;
class ProcessPersistenceException extends \RuntimeException
{
public function __construct($message = "", Throwable $previous = null)
{
parent::__construct($message, 500, $previous);
}
}

38
src/Core/Worker/Factory/Process.php Normal file
View file

@ -0,0 +1,38 @@
<?php
namespace Friendica\Core\Worker\Factory;
use Friendica\BaseFactory;
use Friendica\Capabilities\ICanCreateFromTableRow;
use Friendica\Core\Worker\Entity;
class Process extends BaseFactory implements ICanCreateFromTableRow
{
public function createFromTableRow(array $row): Entity\Process
{
return new Entity\Process(
$row['pid'],
$row['command'],
new \DateTime($row['created'] ?? 'now', new \DateTimeZone('UTC'))
);
}
/**
* Creates a new process entry for a given PID
*
* @param int $pid
*
* @return Entity\Process
*/
public function create(int $pid): Entity\Process
{
$trace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS, 1);
$command = basename($trace[0]['file']);
return $this->createFromTableRow([
'pid' => $pid,
'command' => $command,
]);
}
}

118
src/Core/Worker/Repository/Process.php Normal file
View file

@ -0,0 +1,118 @@
<?php
/**
* @copyright Copyright (C) 2010-2021, the Friendica project
*
* @license GNU AGPL version 3 or any later version
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
*/
namespace Friendica\Core\Worker\Repository;
use Friendica\BaseRepository;
use Friendica\Core\Worker\Exception\ProcessPersistenceException;
use Friendica\Database\Database;
use Friendica\Util\DateTimeFormat;
use Friendica\Core\Worker\Factory;
use Friendica\Core\Worker\Entity;
use Psr\Log\LoggerInterface;
/**
* functions for interacting with a process
*/
class Process extends BaseRepository
{
protected static $table_name = 'process';
/** @var Factory\Process */
protected $factory;
public function __construct(Database $database, LoggerInterface $logger, Factory\Process $factory)
{
parent::__construct($database, $logger, $factory);
}
/**
* Starts and Returns the process for a given PID
*
* @param int $pid
*
* @return Entity\Process
*/
public function create(int $pid): Entity\Process
{
// Cleanup inactive process
$this->deleteInactive();
try {
$this->db->transaction();
$newProcess = $this->factory->create($pid);
if (!$this->db->exists('process', ['pid' => $pid])) {
if (!$this->db->insert(static::$table_name, [
'pid' => $newProcess->pid,
'command' => $newProcess->command,
'created' => $newProcess->created->format(DateTimeFormat::MYSQL)
])) {
throw new ProcessPersistenceException(sprintf('The process with PID %s already exists.', $pid));
}
}
$result = $this->_selectOne(['pid' => $pid]);
$this->db->commit();
return $result;
} catch (\Exception $exception) {
throw new ProcessPersistenceException(sprintf('Cannot save process with PID %s.', $pid), $exception);
}
}
public function delete(Entity\Process $process)
{
try {
if (!$this->db->delete(static::$table_name, [
'pid' => $process->pid
])) {
throw new ProcessPersistenceException(sprintf('The process with PID %s doesn\'t exists.', $process->pi));
}
} catch (\Exception $exception) {
throw new ProcessPersistenceException(sprintf('Cannot delete process with PID %s.', $process->pid), $exception);
}
}
/**
* Clean the process table of inactive physical processes
*/
private function deleteInactive()
{
$this->db->transaction();
try {
$processes = $this->db->select('process', ['pid']);
while ($process = $this->db->fetch($processes)) {
if (!posix_kill($process['pid'], 0)) {
$this->db->delete('process', ['pid' => $process['pid']]);
}
}
$this->db->close($processes);
} catch (\Exception $exception) {
throw new ProcessPersistenceException('Cannot delete inactive process', $exception);
} finally {
$this->db->commit();
}
}
}

16
src/DI.php
View file

@ -195,11 +195,11 @@ abstract class DI
} }
/** /**
* @return Core\Process * @return Core\Worker\Repository\Process
*/ */
public static function process() public static function process()
{ {
return self::$dice->create(Core\Process::class); return self::$dice->create(Core\Worker\Repository\Process::class);
} }
/** /**
@ -218,6 +218,14 @@ abstract class DI
return self::$dice->create(Core\Storage\Repository\StorageManager::class); return self::$dice->create(Core\Storage\Repository\StorageManager::class);
} }
/**
* @return \Friendica\Core\System
*/
public static function system()
{
return self::$dice->create(Core\System::class);
}
// //
// "LoggerInterface" instances // "LoggerInterface" instances
// //
@ -379,11 +387,11 @@ abstract class DI
// "Model" namespace instances // "Model" namespace instances
// //
/** /**
* @return Model\Process * @return \Friendica\Core\Worker\Repository\Process
*/ */
public static function modelProcess() public static function modelProcess()
{ {
return self::$dice->create(Model\Process::class); return self::$dice->create(Core\Worker\Repository\Process::class);
} }
/** /**

91
src/Model/Process.php
View file

@ -1,91 +0,0 @@
<?php
/**
* @copyright Copyright (C) 2010-2021, the Friendica project
*
* @license GNU AGPL version 3 or any later version
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
*/
namespace Friendica\Model;
use Friendica\Database\Database;
use Friendica\Util\DateTimeFormat;
/**
* functions for interacting with a process
*/
class Process
{
/** @var Database */
private $dba;
public function __construct(Database $dba)
{
$this->dba = $dba;
}
/**
* Insert a new process row. If the pid parameter is omitted, we use the current pid
*
* @param string $command
* @param int $pid The process id to insert
* @return bool
* @throws \Exception
*/
public function insert(string $command, int $pid)
{
$return = true;
$this->dba->transaction();
if (!$this->dba->exists('process', ['pid' => $pid])) {
$return = $this->dba->insert('process', ['pid' => $pid, 'command' => $command, 'created' => DateTimeFormat::utcNow()]);
}
$this->dba->commit();
return $return;
}
/**
* Remove a process row by pid. If the pid parameter is omitted, we use the current pid
*
* @param int $pid The pid to delete
* @return bool
* @throws \Exception
*/
public function deleteByPid(int $pid)
{
return $this->dba->delete('process', ['pid' => $pid]);
}
/**
* Clean the process table of inactive physical processes
*/
public function deleteInactive()
{
$this->dba->transaction();
$processes = $this->dba->select('process', ['pid']);
while($process = $this->dba->fetch($processes)) {
if (!posix_kill($process['pid'], 0)) {
$this->deleteByPid($process['pid']);
}
}
$this->dba->close($processes);
$this->dba->commit();
}
}

3
static/dependencies.config.php
View file

@ -188,10 +188,9 @@ return [
['determineModule', [], Dice::CHAIN_CALL], ['determineModule', [], Dice::CHAIN_CALL],
], ],
], ],
Process::class => [ \Friendica\Core\System::class => [
'constructParams' => [ 'constructParams' => [
[Dice::INSTANCE => '$basepath'], [Dice::INSTANCE => '$basepath'],
getmypid(),
], ],
], ],
App\Router::class => [ App\Router::class => [

85
tests/src/Model/ProcessTest.php
View file

@ -1,85 +0,0 @@
<?php
namespace Friendica\Test\src\Model;
use Friendica\Core\Config\Factory\Config;
use Friendica\Model\Process;
use Friendica\Test\DatabaseTest;
use Friendica\Test\Util\Database\StaticDatabase;
use Friendica\Test\Util\VFSTrait;
use Friendica\Util\Profiler;
use Psr\Log\NullLogger;
class ProcessTest extends DatabaseTest
{
use VFSTrait;
/** @var StaticDatabase */
private $dba;
protected function setUp(): void
{
parent::setUp();
$this->setUpVfsDir();
$logger = new NullLogger();
$profiler = \Mockery::mock(Profiler::class);
$profiler->shouldReceive('startRecording');
$profiler->shouldReceive('stopRecording');
$profiler->shouldReceive('saveTimestamp')->withAnyArgs()->andReturn(true);
// load real config to avoid mocking every config-entry which is related to the Database class
$configFactory = new Config();
$loader = (new Config())->createConfigFileLoader($this->root->url(), []);
$configCache = $configFactory->createCache($loader);
$this->dba = new StaticDatabase($configCache, $profiler, $logger);
}
public function testInsertDelete()
{
$process = new Process($this->dba);
self::assertEquals(0, $this->dba->count('process'));
$process->insert('test', 1);
$process->insert('test2', 2);
$process->insert('test3', 3);
self::assertEquals(3, $this->dba->count('process'));
self::assertEquals([
['command' => 'test']
], $this->dba->selectToArray('process', ['command'], ['pid' => 1]));
$process->deleteByPid(1);
self::assertEmpty($this->dba->selectToArray('process', ['command'], ['pid' => 1]));
self::assertEquals(2, $this->dba->count('process'));
}
public function testDoubleInsert()
{
$process = new Process($this->dba);
$process->insert('test', 1);
// double insert doesn't work
$process->insert('test23', 1);
self::assertEquals([['command' => 'test']], $this->dba->selectToArray('process', ['command'], ['pid' => 1]));
}
/**
* @doesNotPerformAssertions
*/
public function testWrongDelete()
{
$process = new Process($this->dba);
// Just ignore wrong deletes, no execution is thrown
$process->deleteByPid(-1);
}
}