mexon/friendica
1
0
Fork 0
Code Pull requests Activity

Merge pull request #10941 from nupplaphil/feat/worker_paradigm

Browse source 
Paradigm Restructuring Part 4 - Process
This commit is contained in:
Hypolite Petovan 2021-11-05 17:38:19 -04:00 committed by GitHub
commit 19d62621d5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 738 additions and 775 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;
}
if ($do_cron || (!DI::process()->isMaxLoadReached() && Worker::entriesExists() && Worker::isReady())) {
if ($do_cron || (!DI::system()->isMaxLoadReached() && Worker::entriesExists() && Worker::isReady())) {
Worker::spawnWorker($do_cron);
} else {
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);
Worker::processQueue($run_cron);
$process = DI::process()->create(getmypid(), basename(__FILE__));
Worker::unclaimProcess();
Worker::processQueue($run_cron, $process);
DI::process()->end();
Worker::unclaimProcess($process);
DI::process()->delete($process);

18
src/App.php
View file

@ -118,9 +118,9 @@ class App
private $args;
/**
* @var Core\Process The process methods
* @var Core\System The system methods
*/
private $process;
private $system;
/**
* @var IManagePersonalConfigValues
@ -327,10 +327,10 @@ class App
* @param Profiler $profiler The profiler of this application
* @param L10n $l10n The translator instance
* @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
*/
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->config = $config;
@ -340,7 +340,7 @@ class App
$this->logger = $logger;
$this->l10n = $l10n;
$this->args = $args;
$this->process = $process;
$this->system = $system;
$this->pConfig = $pConfig;
$this->load();
@ -588,14 +588,6 @@ class App
throw new HTTPException\InternalServerErrorException('Apologies but the website is unavailable at the moment.');
}
// Max Load Average reached: ERROR
if ($this->process->isMaxProcessesReached() || $this->process->isMaxLoadReached()) {
header('Retry-After: 120');
header('Refresh: 120; url=' . $this->baseURL->get() . "/" . $this->args->getQueryString());
throw new HTTPException\ServiceUnavailableException('The node is currently overloaded. Please try again later.');
}
if (!$this->mode->isInstall()) {
// Force SSL redirection
if ($this->baseURL->checkRedirectHttps()) {

17
src/Core/Logger/Type/WorkerLogger.php
View file

@ -31,6 +31,9 @@ use Psr\Log\LoggerInterface;
*/
class WorkerLogger implements LoggerInterface
{
/** @var int Length of the unique worker id */
const WORKER_ID_LENGTH = 7;
/**
* @var LoggerInterface The original Logger instance
*/
@ -48,17 +51,14 @@ class WorkerLogger implements LoggerInterface
/**
* @param LoggerInterface $logger The logger for worker entries
* @param string $functionName The current function name of the worker
* @param int $idLength The length of the generated worker ID
*
* @throws LoggerException
*/
public function __construct(LoggerInterface $logger, string $functionName = '', int $idLength = 7)
public function __construct(LoggerInterface $logger)
{
$this->logger = $logger;
$this->functionName = $functionName;
try {
$this->workerId = Strings::getRandomHex($idLength);
$this->workerId = Strings::getRandomHex(self::WORKER_ID_LENGTH);
} catch (\Exception $exception) {
throw new LoggerException('Cannot generate random Hex.', $exception);
}
@ -68,10 +68,17 @@ class WorkerLogger implements LoggerInterface
* Sets the function name for additional logging
*
* @param string $functionName
*
* @throws LoggerException
*/
public function setFunctionName(string $functionName)
{
$this->functionName = $functionName;
try {
$this->workerId = Strings::getRandomHex(self::WORKER_ID_LENGTH);
} catch (\Exception $exception) {
throw new LoggerException('Cannot generate random Hex.', $exception);
}
}
/**

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)]);
}
}

187
src/Core/System.php
View file

@ -22,18 +22,203 @@
namespace Friendica\Core;
use Exception;
use Friendica\App;
use Friendica\Core\Config\Capability\IManageConfigValues;
use Friendica\DI;
use Friendica\Network\HTTPException\FoundException;
use Friendica\Network\HTTPException\MovedPermanentlyException;
use Friendica\Network\HTTPException\TemporaryRedirectException;
use Friendica\Util\BasePath;
use Friendica\Util\XML;
use Psr\Log\LoggerInterface;
/**
* Contains the class with system relevant stuff
*/
class System
{
/**
* @var LoggerInterface
*/
private $logger;
/**
* @var IManageConfigValues
*/
private $config;
/**
* @var string
*/
private $basePath;
public function __construct(LoggerInterface $logger, IManageConfigValues $config, string $basepath)
{
$this->logger = $logger;
$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
{
// Deactivated, needs more investigating if this check really makes sense
return false;
/*
* Commented out to suppress static analyzer issues
*
$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
{
$maxsysload = intval($this->config->get('system', 'maxloadavg'));
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' => 'backend', '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.
*
@ -42,7 +227,7 @@ class System
* this is called from a centralized method that isn't relevant to the callstack
* @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);

60
src/Core/Worker.php
View file

@ -23,6 +23,7 @@ namespace Friendica\Core;
use Friendica\App\Mode;
use Friendica\Core;
use Friendica\Core\Worker\Entity\Process;
use Friendica\Database\DBA;
use Friendica\DI;
use Friendica\Util\DateTimeFormat;
@ -51,26 +52,29 @@ class Worker
private static $last_update;
private static $state;
private static $daemon_mode = null;
/** @var Process */
private static $process;
/**
* Processes the tasks that are in the workerqueue table
*
* @param boolean $run_cron Should the cron processes be executed?
* @param Process $process The current running process
* @return void
* @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);
// 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.');
return;
}
// 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
$last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0);
@ -134,7 +138,7 @@ class Worker
}
// Check free memory
if (DI::process()->isMinMemoryReached()) {
if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.');
DI::lock()->release(self::LOCK_WORKER);
return;
@ -147,7 +151,7 @@ class Worker
// Quit the worker once every cron interval
if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
Logger::info('Process lifetime reached, respawning.');
self::unclaimProcess();
self::unclaimProcess($process);
if (self::isDaemonMode()) {
self::IPCSetJobState(true);
} else {
@ -180,7 +184,7 @@ class Worker
}
// Do we have too few memory?
if (DI::process()->isMinMemoryReached()) {
if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.');
return false;
}
@ -192,7 +196,7 @@ class Worker
}
// 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.');
return false;
}
@ -334,7 +338,7 @@ class Worker
}
// 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]);
return false;
}
@ -842,7 +846,7 @@ class Worker
private static function activeWorkers()
{
$stamp = (float)microtime(true);
$count = DBA::count('process', ['command' => 'Worker.php']);
$count = DI::process()->countCommand('Worker.php');
self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_count += (microtime(true) - $stamp);
return $count;
@ -1105,15 +1109,15 @@ class Worker
/**
* Removes a workerqueue entry from the current process
*
* @param Process $process the process behind the workerqueue
*
* @return void
* @throws \Exception
*/
public static function unclaimProcess()
public static function unclaimProcess(Process $process)
{
$mypid = getmypid();
$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_write += (microtime(true) - $stamp);
}
@ -1146,7 +1150,7 @@ class Worker
*/
private static function forkProcess(bool $do_cron)
{
if (DI::process()->isMinMemoryReached()) {
if (DI::system()->isMinMemoryReached()) {
Logger::warning('Memory limit reached - quitting');
return;
}
@ -1176,26 +1180,25 @@ class Worker
}
// We now are in the new worker
$pid = getmypid();
DBA::connect();
/// @todo Reinitialize the logger to set a new process_id and uid
DI::process()->setPid($pid);
DI::flushLogger();
$process = DI::process()->create(getmypid(), basename(__FILE__));
$cycles = 0;
while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
while (!self::IPCJobsExists($process->pid) && (++$cycles < 100)) {
usleep(10000);
}
Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
Logger::info('Worker spawned', ['pid' => $process->pid, 'wait_cycles' => $cycles]);
self::processQueue($do_cron);
self::processQueue($do_cron, $process);
self::unclaimProcess();
self::unclaimProcess($process);
self::IPCSetJobState(false, $pid);
DI::process()->end();
Logger::info('Worker ended', ['pid' => $pid]);
self::IPCSetJobState(false, $process->pid);
DI::process()->delete($process);
Logger::info('Worker ended', ['pid' => $process->pid]);
exit();
}
@ -1211,9 +1214,7 @@ class Worker
if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) {
self::forkProcess($do_cron);
} else {
$process = new Core\Process(DI::logger(), DI::mode(), DI::config(),
DI::modelProcess(), DI::app()->getBasePath(), getmypid());
$process->run('bin/worker.php', ['no_cron' => !$do_cron]);
DI::system()->run('bin/worker.php', ['no_cron' => !$do_cron]);
}
if (self::isDaemonMode()) {
self::IPCSetJobState(false);
@ -1571,8 +1572,7 @@ class Worker
Logger::notice('Starting new daemon process');
$command = 'bin/daemon.php';
$a = DI::app();
$process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::modelProcess(), $a->getBasePath(), getmypid());
$process->run($command, ['start']);
DI::system()->run($command, ['start']);
Logger::notice('New daemon process started');
}

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

@ -0,0 +1,33 @@
<?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;
}
}

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);
}
}

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

@ -0,0 +1,35 @@
<?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
* @param string $command
*
* @return Entity\Process
*/
public function create(int $pid, string $command): Entity\Process
{
return $this->createFromTableRow([
'pid' => $pid,
'command' => $command,
]);
}
}

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

@ -0,0 +1,137 @@
<?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
* @param string $command
*
* @return Entity\Process
*/
public function create(int $pid, string $command): Entity\Process
{
// Cleanup inactive process
$this->deleteInactive();
try {
$this->db->transaction();
$newProcess = $this->factory->create($pid, $command);
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);
$this->db->commit();
} catch (\Exception $exception) {
$this->db->rollback();
throw new ProcessPersistenceException('Cannot delete inactive process', $exception);
}
}
/**
* Returns the number of processes with a given command
*
* @param string $command
*
* @return int Number of processes
*
* @throws ProcessPersistenceException
*/
public function countCommand(string $command): int
{
try {
return $this->count(['command' => strtolower($command)]);
} catch (\Exception $exception) {
throw new ProcessPersistenceException('Cannot count ', $exception);
}
}
}

28
src/DI.php
View file

@ -195,11 +195,11 @@ abstract class DI
}
/**
* @return Core\Process
* @return Core\Worker\Repository\Process
*/
public static function process()
{
return self::$dice->create(Core\Process::class);
return self::$dice->create(Core\Worker\Repository\Process::class);
}
/**
@ -218,10 +218,30 @@ abstract class DI
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
//
/**
* Flushes the Logger instance, so the factory is called again
* (creates a new id and retrieves the current PID)
*/
public static function flushLogger()
{
$flushDice = self::$dice
->addRule(LoggerInterface::class, self::$dice->getRule(LoggerInterface::class))
->addRule('$devLogger', self::$dice->getRule('$devLogger'));
static::init($flushDice);
}
/**
* @return LoggerInterface
*/
@ -379,11 +399,11 @@ abstract class DI
// "Model" namespace instances
//
/**
* @return Model\Process
* @return \Friendica\Core\Worker\Repository\Process
*/
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
src/Module/Admin/Site.php
View file

@ -177,7 +177,6 @@ class Site extends BaseAdmin
$proxy = (!empty($_POST['proxy']) ? Strings::escapeTags(trim($_POST['proxy'])) : '');
$timeout = (!empty($_POST['timeout']) ? intval(trim($_POST['timeout'])) : 60);
$maxloadavg = (!empty($_POST['maxloadavg']) ? intval(trim($_POST['maxloadavg'])) : 20);
$maxloadavg_frontend = (!empty($_POST['maxloadavg_frontend']) ? intval(trim($_POST['maxloadavg_frontend'])) : 50);
$min_memory = (!empty($_POST['min_memory']) ? intval(trim($_POST['min_memory'])) : 0);
$optimize_tables = (!empty($_POST['optimize_tables']) ? intval(trim($_POST['optimize_tables'])) : false);
$contact_discovery = (!empty($_POST['contact_discovery']) ? intval(trim($_POST['contact_discovery'])) : Contact\Relation::DISCOVERY_NONE);
@ -264,7 +263,6 @@ class Site extends BaseAdmin
}
DI::config()->set('system', 'ssl_policy' , $ssl_policy);
DI::config()->set('system', 'maxloadavg' , $maxloadavg);
DI::config()->set('system', 'maxloadavg_frontend' , $maxloadavg_frontend);
DI::config()->set('system', 'min_memory' , $min_memory);
DI::config()->set('system', 'optimize_tables' , $optimize_tables);
DI::config()->set('system', 'contact_discovery' , $contact_discovery);
@ -576,7 +574,6 @@ class Site extends BaseAdmin
'$proxy' => ['proxy', DI::l10n()->t('Proxy URL'), DI::config()->get('system', 'proxy'), ''],
'$timeout' => ['timeout', DI::l10n()->t('Network timeout'), DI::config()->get('system', 'curl_timeout'), DI::l10n()->t('Value is in seconds. Set to 0 for unlimited (not recommended).')],
'$maxloadavg' => ['maxloadavg', DI::l10n()->t('Maximum Load Average'), DI::config()->get('system', 'maxloadavg'), DI::l10n()->t('Maximum system load before delivery and poll processes are deferred - default %d.', 20)],
'$maxloadavg_frontend' => ['maxloadavg_frontend', DI::l10n()->t('Maximum Load Average (Frontend)'), DI::config()->get('system', 'maxloadavg_frontend'), DI::l10n()->t('Maximum system load before the frontend quits service - default 50.')],
'$min_memory' => ['min_memory', DI::l10n()->t('Minimal Memory'), DI::config()->get('system', 'min_memory'), DI::l10n()->t('Minimal free memory in MB for the worker. Needs access to /proc/meminfo - default 0 (deactivated).')],
'$optimize_tables' => ['optimize_tables', DI::l10n()->t('Periodically optimize tables'), DI::config()->get('system', 'optimize_tables'), DI::l10n()->t('Periodically optimize tables like the cache and the workerqueue')],

0
src/Network/HTTPException/MovedPermanently.php → src/Network/HTTPException/MovedPermanentlyException.php
View file

4
static/dependencies.config.php
View file

@ -40,7 +40,6 @@ use Friendica\Core\Config;
use Friendica\Core\PConfig;
use Friendica\Core\L10n;
use Friendica\Core\Lock;
use Friendica\Core\Process;
use Friendica\Core\Session\Capability\IHandleSessions;
use Friendica\Core\Storage\Repository\StorageManager;
use Friendica\Database\Database;
@ -188,10 +187,9 @@ return [
['determineModule', [], Dice::CHAIN_CALL],
],
],
Process::class => [
\Friendica\Core\System::class => [
'constructParams' => [
[Dice::INSTANCE => '$basepath'],
getmypid(),
],
],
App\Router::class => [

4
static/settings.config.php
View file

@ -148,10 +148,6 @@ return [
// Maximum system load before delivery and poll processes are deferred.
'maxloadavg' => 20,
// maxloadavg_frontend (Integer)
// Maximum system load before the frontend quits service - default 50.
'maxloadavg_frontend' => 50,
// min_memory (Integer)
// Minimal free memory in MB for the worker. Needs access to /proc/meminfo - default 0 (deactivated).
'min_memory' => 0,

70
tests/src/Core/Logger/WorkerLoggerTest.php
View file

@ -27,34 +27,9 @@ use Psr\Log\LoggerInterface;
class WorkerLoggerTest extends MockedTest
{
private function assertUid($uid, $length = 7)
private function assertUid($uid)
{
self::assertRegExp('/^[a-zA-Z0-9]{' . $length . '}+$/', $uid);
}
/**
* Test the a id with length zero
*
*/
public function testGetWorkerIdZero()
{
$this->expectException(\Error::class);
$logger = \Mockery::mock(LoggerInterface::class);
new WorkerLogger($logger, 'test', 0);
}
/**
* Test the generated Uid
*/
public function testGetWorkerId()
{
$logger = \Mockery::mock(LoggerInterface::class);
for ($i = 1; $i < 14; $i++) {
$workLogger = new WorkerLogger($logger, 'test', $i);
$uid = $workLogger->getWorkerId();
self::assertUid($uid, $i);
}
self::assertRegExp('/^[a-zA-Z0-9]{' . WorkerLogger::WORKER_ID_LENGTH . '}+$/', $uid);
}
public function dataTest()
@ -105,10 +80,10 @@ class WorkerLoggerTest extends MockedTest
public function testEmergency($func, $msg, $context = [])
{
$logger = \Mockery::mock(LoggerInterface::class);
$workLogger = new WorkerLogger($logger, 'test');
$workLogger = new WorkerLogger($logger);
$testContext = $context;
$testContext['worker_id'] = $workLogger->getWorkerId();
$testContext['worker_cmd'] = 'test';
$testContext['worker_cmd'] = '';
self::assertUid($testContext['worker_id']);
$logger
->shouldReceive($func)
@ -123,10 +98,43 @@ class WorkerLoggerTest extends MockedTest
public function testLog()
{
$logger = \Mockery::mock(LoggerInterface::class);
$workLogger = new WorkerLogger($logger, 'test');
$workLogger = new WorkerLogger($logger);
$context = $testContext = ['test' => 'it'];
$testContext['worker_id'] = $workLogger->getWorkerId();
$testContext['worker_cmd'] = 'test';
$testContext['worker_cmd'] = '';
self::assertUid($testContext['worker_id']);
$logger
->shouldReceive('log')
->with('debug', 'a test', $testContext)
->once();
$workLogger->log('debug', 'a test', $context);
}
/**
* Test the WorkerLogger after setting a worker function
*/
public function testChangedId()
{
$logger = \Mockery::mock(LoggerInterface::class);
$workLogger = new WorkerLogger($logger);
$context = $testContext = ['test' => 'it'];
$testContext['worker_id'] = $workLogger->getWorkerId();
$testContext['worker_cmd'] = '';
self::assertUid($testContext['worker_id']);
$logger
->shouldReceive('log')
->with('debug', 'a test', $testContext)
->once();
$workLogger->log('debug', 'a test', $context);
$workLogger->setFunctionName('testFunc');
self::assertNotEquals($testContext['worker_id'], $workLogger->getWorkerId());
$context = $testContext = ['test' => 'it'];
$testContext['worker_id'] = $workLogger->getWorkerId();
$testContext['worker_cmd'] = 'testFunc';
self::assertUid($testContext['worker_id']);
$logger
->shouldReceive('log')

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);
}
}

400
view/lang/C/messages.po
View file

File diff suppressed because it is too large Load diff

1
view/templates/admin/site.tpl
View file

@ -91,7 +91,6 @@
{{include file="field_input.tpl" field=$proxy}}
{{include file="field_input.tpl" field=$proxyuser}}
{{include file="field_input.tpl" field=$timeout}}
{{include file="field_input.tpl" field=$maxloadavg_frontend}}
{{include file="field_input.tpl" field=$abandon_days}}
{{include file="field_input.tpl" field=$temppath}}
{{include file="field_checkbox.tpl" field=$suppress_tags}}

1
view/theme/frio/templates/admin/site.tpl
View file

@ -190,7 +190,6 @@
{{include file="field_input.tpl" field=$proxy}}
{{include file="field_input.tpl" field=$proxyuser}}
{{include file="field_input.tpl" field=$timeout}}
{{include file="field_input.tpl" field=$maxloadavg_frontend}}
{{include file="field_input.tpl" field=$abandon_days}}
{{include file="field_input.tpl" field=$temppath}}
{{include file="field_checkbox.tpl" field=$suppress_tags}}