ProcessManager::getStdOut() - Code Metrics - silverstripe/doorman - Measure and Improve Code Quality continuously with Scrutinizer

ProcessManager::getStdOut() A
last analyzed 2021-05-30 20:53 UTC

↳ Parent: ProcessManager

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	8
c	0
b	0
f	0
rs	10
cc	2
nc	2
nop	0

<?php

namespace AsyncPHP\Doorman\Manager;

use AsyncPHP\Doorman\Cancellable;
use AsyncPHP\Doorman\Expires;
use AsyncPHP\Doorman\Manager;
use AsyncPHP\Doorman\Process;
use AsyncPHP\Doorman\Profile;
use AsyncPHP\Doorman\Profile\InMemoryProfile;
use AsyncPHP\Doorman\Rule;
use AsyncPHP\Doorman\Rules;
use AsyncPHP\Doorman\Rules\InMemoryRules;
use AsyncPHP\Doorman\Shell;
use AsyncPHP\Doorman\Shell\BashShell;
use AsyncPHP\Doorman\Task;
use SplObjectStorage;

class ProcessManager implements Manager
{
    /**
     * @var Task[]
     */
    protected $waiting = array();

    /**
     * @var Task[]
     */
    protected $running = array();

    /**
     * @var null|SplObjectStorage
     */
    protected $timings = null;

    /**
     * @var null|string
     */
    protected $logPath;

    /**
     * @var null|Rules
     */
    protected $rules;

    /**
     * @var null|Shell
     */
    protected $shell;

    /**
     * @var null|string
     */
    protected $binary;

    /**
     * @var null|string
     */
    protected $worker;

    /**
     * @inheritdoc
     *
     * @param Task $task
     *
     * @return $this
     */
    public function addTask(Task $task)
    {
        $this->waiting[] = $task;

        return $this;
    }

    /**
     * @inheritdoc
     *
     * @return bool
     */
    public function tick()
    {
        if (!$this->timings instanceof SplObjectStorage) {
            $this->timings = new SplObjectStorage();
        }

        $waiting = array();
        $running = array();

        foreach ($this->waiting as $task) {
            if ($this->isTaskCancelled($task)) {
                continue;
            }
            
            if (!$this->canRunTask($task)) {
                $waiting[] = $task;
                continue;
            }

            if ($task->stopsSiblings()) {
                $this->stopSiblingTasks($task);
            }

            $binary = $this->getBinary();
            $worker = $this->getWorker();
            $stdout = $this->getStdOut();
            $stderr = $this->getStdErr();

            if ($task instanceof Expires) {
                $this->timings[$task] = time();
            }

            $output = $this->getShell()->exec("{$binary} {$worker} %s {$stdout} {$stderr} & echo $!", array(
                $this->getTaskString($task),
            ));

            if ($task instanceof Process) {
                $task->setId($output[0]);
            }

            $this->running[] = $task;
        }

        foreach ($this->running as $task) {
            if (!$this->canRemoveTask($task)) {
                $running[] = $task;
            }
        }

        $this->waiting = $waiting;
        $this->running = $running;

        return !empty($waiting) || !empty($running);
    }

    /**
     * Stops sibling processes of a task.
     *
     * @param Task $task
     *
     * @return $this
     */
    protected function stopSiblingTasks(Task $task)
    {
        $handler = $task->getHandler();

        foreach ($this->running as $task) {
            if ($task->getHandler() === $handler && $task instanceof Process) {
                $this->getShell()->exec("kill -9 %s", array(
                    $task->getId(),
                ));
            }
        }

        return $this;
    }

    /**
     * Checks whether a new task can be run.
     *
     * @param Task $task
     *
     * @return bool
     */
    protected function canRunTask(Task $task)
    {
        if(!$task->canRunTask()) {
            return false;
        }

        if ($task->ignoresRules()) {
            return true;
        }

        $processes = array_filter($this->running, function (Task $task) {
            return $task instanceof Process;
        });

        if (count($processes) < 1) {
            return true;
        }

        $profile = $this->getProfileForProcesses($task, $processes);

        return $this->getRules()->canRunTask($task, $profile);
    }

    /**
     * Gets the load profile related to a task.
     *
     * @param Task  $task
     * @param array $processes
     *
     * @return Profile
     */
    protected function getProfileForProcesses(Task $task, array $processes)
    {
        $stats = $this->getStatsForProcesses($processes);

        $siblingProcesses = array_filter($processes, function (Task $next) use ($task) {
            return $next->getHandler() === $task->getHandler();
        });

        $siblingStats = $this->getStatsForProcesses($siblingProcesses);

        $profile = $this->newProfile();

        $profile->setProcesses($processes);
        $profile->setProcessorLoad(min(100, array_sum(array_column($stats, 1))));
        $profile->setMemoryLoad(min(100, array_sum(array_column($stats, 2))));

        $profile->setSiblingProcesses($siblingProcesses);
        $profile->setSiblingProcessorLoad(min(100, array_sum(array_column($siblingStats, 1))));
        $profile->setSiblingMemoryLoad(min(100, array_sum(array_column($siblingStats, 2))));

        return $profile;
    }

    /**
     * Gets processor and memory stats for a list of processes.
     *
     * @param Process[] $processes
     *
     * @return array
     */
    protected function getStatsForProcesses(array $processes)
    {
        $stats = array();

        foreach ($processes as $process) {
            $output = $this->getShell()->exec("ps -o pid,%%cpu,%%mem,state,start -p %s | sed 1d", array(
                $process->getId(),
            ));

            if (count($output) < 1) {
                continue;
            }

            $last = $output[count($output) - 1];

            if (trim($last) === "") {
                continue;
            }

            $parts = preg_split("/\s+/", trim($last));

            $pid = intval($parts[0]);

            if ("{$pid}" !== $parts[0]) {
                continue;
            }

            $stats[] = $parts;
        }

        return $stats;
    }

    /**
     * Gets or creates a Shell instance.
     *
     * @return Shell
     */
    public function getShell()
    {
        if ($this->shell === null) {
            $this->shell = $this->newShell();
        }

        return $this->shell;
    }

    /**
     * @param Shell $shell
     *
     * @return $this
     */
    public function setShell(Shell $shell)
    {
        $this->shell = $shell;

        return $this;
    }

    /**
     * Creates a new Shell instance.
     *
     * @return Shell
     */
    protected function newShell()
    {
        return new BashShell();
    }

    /**
     * Creates a new Profile instance.
     *
     * @return Profile
     */
    protected function newProfile()
    {
        return new InMemoryProfile();
    }

    /**
     * Gets or creates a new Rules instance.
     *
     * @return Rules
     */
    public function getRules()
    {
        if ($this->rules === null) {
            $this->rules = $this->newRules();
        }

        return $this->rules;
    }

    /**
     * @param Rules $rules
     *
     * @return $this
     */
    public function setRules(Rules $rules)
    {
        $this->rules = $rules;

        return $this;
    }

    /**
     * Creates a new Rules instance.
     *
     * @return Rules
     */
    protected function newRules()
    {
        return new InMemoryRules();
    }

    /**
     * @param string $binary
     *
     * @return $this
     */
    public function setBinary($binary)
    {
        $this->binary = $binary;

        return $this;
    }

    /**
     * Gets the path of the PHP runtime.
     *
     * @return string
     */
    public function getBinary()
    {
        if ($this->binary === null) {
            $this->binary = PHP_BINDIR."/php";
        }

        return $this->binary;
    }

    /**
     * @param string $worker
     *
     * @return $this
     */
    public function setWorker($worker)
    {
        $this->worker = $worker;

        return $this;
    }

    /**
     * Gets the path of the worker script.
     *
     * @return string
     */
    public function getWorker()
    {
        if ($this->worker === null) {
            $this->worker = realpath(__DIR__."/../../bin/worker.php");
        }

        return $this->worker;
    }

    /**
     * Gets the path to write stdout to.
     *
     * @return string
     */
    protected function getStdOut()
    {
        if ($this->getLogPath() !== null) {
            return ">> ".$this->getLogPath()."/stdout.log";
        }

        return "> /dev/null";
    }

    /**
     * @return null|string
     */
    public function getLogPath()
    {
        return $this->logPath;
    }

    /**
     * @param string $logPath
     *
     * @return $this
     */
    public function setLogPath($logPath)
    {
        $this->logPath = $logPath;

        return $this;
    }

    /**
     * Gets the path to write stderr to.
     *
     * @return string
     */
    protected function getStdErr()
    {
        if ($this->getLogPath() !== null) {
            return "2>> ".$this->getLogPath()."/stderr.log";
        }

        return "2> /dev/null";
    }

    /**
     * Gets a string representation of a task, to pass to the worker script.
     *
     * @param Task $task
     *
     * @return string
     */
    protected function getTaskString(Task $task)
    {
        return base64_encode(serialize($task));
    }

    /**
     * Checks whether a task can be removed from the list of running processes.
     *
     * @param Task $task
     *
     * @return bool
     */
    protected function canRemoveTask(Task $task)
    {
        if (!$task instanceof Process) {
            return true;
        }
        
        if ($this->isTaskExpired($task) || $this->isTaskCancelled($task)) {
            $this->killTask($task);
            return true;
        }

        $processes = array_filter($this->running, function (Task $task) {
            return $task instanceof Process;
        });

        if (count($processes) < 1) {
            return true;
        }

        $found = false;
        $stats = $this->getStatsForProcesses($processes);

        foreach ($stats as $stat) {
            if ($stat[0] === $task->getId()) {
                $found = true;
            }
        }

        return !$found;
    }

    /**
     * Check if the given task is expired
     *
     * @param Task $task
     * @return boolean
     */
    protected function isTaskExpired(Task $task) {
        if ($task instanceof Expires) {
            $expiresIn = $task->getExpiresIn();
            $startedAt = $this->timings[$task];

            if($expiresIn > 0 && (time() - $startedAt) >= $expiresIn) {
                return $task->shouldExpire($startedAt);
            }
        }

        return false;
    }

    /**
     * Check if the given task is cancelled.
     * 
     * @param Task $task
     *
     * @return bool
     */
    protected function isTaskCancelled(Task $task)
    {
        if ($task instanceof Cancellable) {
            return $task->isCancelled();
        }

        return false;
    }

    /**
     * Revoke any background processes attached to this task.
     *
     * @param Task $task
     *
     * @return bool
     */
    protected function killTask(Task $task)
    {
        if ($task instanceof Process) {
            $this->getShell()->exec("kill -9 %s", array(
                $task->getId(),
            ));

            return true;
        }

        return false;
    }

    /**
     * @param Rule $rule
     *
     * @return $this
     */
    public function addRule(Rule $rule)
    {
        $this->getRules()->addRule($rule);

        return $this;
    }

    /**
     * @param Rule $rule
     *
     * @return $this
     */
    public function removeRule(Rule $rule)
    {
        $this->getRules()->removeRule($rule);

        return $this;
    }
}


1			<?php
2
3			namespace AsyncPHP\Doorman\Manager;
4
5			use AsyncPHP\Doorman\Cancellable;
6			use AsyncPHP\Doorman\Expires;
7			use AsyncPHP\Doorman\Manager;
8			use AsyncPHP\Doorman\Process;
9			use AsyncPHP\Doorman\Profile;
10			use AsyncPHP\Doorman\Profile\InMemoryProfile;
11			use AsyncPHP\Doorman\Rule;
12			use AsyncPHP\Doorman\Rules;
13			use AsyncPHP\Doorman\Rules\InMemoryRules;
14			use AsyncPHP\Doorman\Shell;
15			use AsyncPHP\Doorman\Shell\BashShell;
16			use AsyncPHP\Doorman\Task;
17			use SplObjectStorage;
18
19			class ProcessManager implements Manager
20			{
21			/**
22			* @var Task[]
23			*/
24			protected $waiting = array();
25
26			/**
27			* @var Task[]
28			*/
29			protected $running = array();
30
31			/**
32			* @var null\|SplObjectStorage
33			*/
34			protected $timings = null;
35
36			/**
37			* @var null\|string
38			*/
39			protected $logPath;
40
41			/**
42			* @var null\|Rules
43			*/
44			protected $rules;
45
46			/**
47			* @var null\|Shell
48			*/
49			protected $shell;
50
51			/**
52			* @var null\|string
53			*/
54			protected $binary;
55
56			/**
57			* @var null\|string
58			*/
59			protected $worker;
60
61			/**
62			* @inheritdoc
63			*
64			* @param Task $task
65			*
66			* @return $this
67			*/
68			public function addTask(Task $task)
69			{
70			$this->waiting[] = $task;
71
72			return $this;
73			}
74
75			/**
76			* @inheritdoc
77			*
78			* @return bool
79			*/
80			public function tick()
81			{
82			if (!$this->timings instanceof SplObjectStorage) {
83			$this->timings = new SplObjectStorage();
84			}
85
86			$waiting = array();
87			$running = array();
88
89			foreach ($this->waiting as $task) {
90			if ($this->isTaskCancelled($task)) {
91			continue;
92			}
93
94			if (!$this->canRunTask($task)) {
95			$waiting[] = $task;
96			continue;
97			}
98
99			if ($task->stopsSiblings()) {
100			$this->stopSiblingTasks($task);
101			}
102
103			$binary = $this->getBinary();
104			$worker = $this->getWorker();
105			$stdout = $this->getStdOut();
106			$stderr = $this->getStdErr();
107
108			if ($task instanceof Expires) {
109			$this->timings[$task] = time();
110			}
111
112			$output = $this->getShell()->exec("{$binary} {$worker} %s {$stdout} {$stderr} & echo $!", array(
113			$this->getTaskString($task),
114			));
115
116			if ($task instanceof Process) {
117			$task->setId($output[0]);
118			}
119
120			$this->running[] = $task;
121			}
122
123			foreach ($this->running as $task) {
124			if (!$this->canRemoveTask($task)) {
125			$running[] = $task;
126			}
127			}
128
129			$this->waiting = $waiting;
130			$this->running = $running;
131
132			return !empty($waiting) \|\| !empty($running);
133			}
134
135			/**
136			* Stops sibling processes of a task.
137			*
138			* @param Task $task
139			*
140			* @return $this
141			*/
142			protected function stopSiblingTasks(Task $task)
143			{
144			$handler = $task->getHandler();
145
146			foreach ($this->running as $task) {
147			if ($task->getHandler() === $handler && $task instanceof Process) {
148			$this->getShell()->exec("kill -9 %s", array(
149			$task->getId(),
150			));
151			}
152			}
153
154			return $this;
155			}
156
157			/**
158			* Checks whether a new task can be run.
159			*
160			* @param Task $task
161			*
162			* @return bool
163			*/
164			protected function canRunTask(Task $task)
165			{
166			if(!$task->canRunTask()) {
167			return false;
168			}
169
170			if ($task->ignoresRules()) {
171			return true;
172			}
173
174			$processes = array_filter($this->running, function (Task $task) {
175			return $task instanceof Process;
176			});
177
178			if (count($processes) < 1) {
179			return true;
180			}
181
182			$profile = $this->getProfileForProcesses($task, $processes);
183
184			return $this->getRules()->canRunTask($task, $profile);
185			}
186
187			/**
188			* Gets the load profile related to a task.
189			*
190			* @param Task $task
191			* @param array $processes
192			*
193			* @return Profile
194			*/
195			protected function getProfileForProcesses(Task $task, array $processes)
196			{
197			$stats = $this->getStatsForProcesses($processes);
198
199			$siblingProcesses = array_filter($processes, function (Task $next) use ($task) {
200			return $next->getHandler() === $task->getHandler();
201			});
202
203			$siblingStats = $this->getStatsForProcesses($siblingProcesses);
204
205			$profile = $this->newProfile();
206
207			$profile->setProcesses($processes);
208			$profile->setProcessorLoad(min(100, array_sum(array_column($stats, 1))));
209			$profile->setMemoryLoad(min(100, array_sum(array_column($stats, 2))));
210
211			$profile->setSiblingProcesses($siblingProcesses);
212			$profile->setSiblingProcessorLoad(min(100, array_sum(array_column($siblingStats, 1))));
213			$profile->setSiblingMemoryLoad(min(100, array_sum(array_column($siblingStats, 2))));
214
215			return $profile;
216			}
217
218			/**
219			* Gets processor and memory stats for a list of processes.
220			*
221			* @param Process[] $processes
222			*
223			* @return array
224			*/
225			protected function getStatsForProcesses(array $processes)
226			{
227			$stats = array();
228
229			foreach ($processes as $process) {
230			$output = $this->getShell()->exec("ps -o pid,%%cpu,%%mem,state,start -p %s \| sed 1d", array(
231			$process->getId(),
232			));
233
234			if (count($output) < 1) {
235			continue;
236			}
237
238			$last = $output[count($output) - 1];
239
240			if (trim($last) === "") {
241			continue;
242			}
243
244			$parts = preg_split("/\s+/", trim($last));
245
246			$pid = intval($parts[0]);
247
248			if ("{$pid}" !== $parts[0]) {
249			continue;
250			}
251
252			$stats[] = $parts;
253			}
254
255			return $stats;
256			}
257
258			/**
259			* Gets or creates a Shell instance.
260			*
261			* @return Shell
262			*/
263			public function getShell()
264			{
265			if ($this->shell === null) {
266			$this->shell = $this->newShell();
267			}
268
269			return $this->shell;
270			}
271
272			/**
273			* @param Shell $shell
274			*
275			* @return $this
276			*/
277			public function setShell(Shell $shell)
278			{
279			$this->shell = $shell;
280
281			return $this;
282			}
283
284			/**
285			* Creates a new Shell instance.
286			*
287			* @return Shell
288			*/
289			protected function newShell()
290			{
291			return new BashShell();
292			}
293
294			/**
295			* Creates a new Profile instance.
296			*
297			* @return Profile
298			*/
299			protected function newProfile()
300			{
301			return new InMemoryProfile();
302			}
303
304			/**
305			* Gets or creates a new Rules instance.
306			*
307			* @return Rules
308			*/
309			public function getRules()
310			{
311			if ($this->rules === null) {
312			$this->rules = $this->newRules();
313			}
314
315			return $this->rules;
316			}
317
318			/**
319			* @param Rules $rules
320			*
321			* @return $this
322			*/
323			public function setRules(Rules $rules)
324			{
325			$this->rules = $rules;
326
327			return $this;
328			}
329
330			/**
331			* Creates a new Rules instance.
332			*
333			* @return Rules
334			*/
335			protected function newRules()
336			{
337			return new InMemoryRules();
338			}
339
340			/**
341			* @param string $binary
342			*
343			* @return $this
344			*/
345			public function setBinary($binary)
346			{
347			$this->binary = $binary;
348
349			return $this;
350			}
351
352			/**
353			* Gets the path of the PHP runtime.
354			*
355			* @return string
356			*/
357			public function getBinary()
358			{
359			if ($this->binary === null) {
360			$this->binary = PHP_BINDIR."/php";
361			}
362
363			return $this->binary;
364			}
365
366			/**
367			* @param string $worker
368			*
369			* @return $this
370			*/
371			public function setWorker($worker)
372			{
373			$this->worker = $worker;
374
375			return $this;
376			}
377
378			/**
379			* Gets the path of the worker script.
380			*
381			* @return string
382			*/
383			public function getWorker()
384			{
385			if ($this->worker === null) {
386			$this->worker = realpath(__DIR__."/../../bin/worker.php");
387			}
388
389			return $this->worker;
390			}
391
392			/**
393			* Gets the path to write stdout to.
394			*
395			* @return string
396			*/
397			protected function getStdOut()
398			{
399			if ($this->getLogPath() !== null) {
400			return ">> ".$this->getLogPath()."/stdout.log";
401			}
402
403			return "> /dev/null";
404			}
405
406			/**
407			* @return null\|string
408			*/
409			public function getLogPath()
410			{
411			return $this->logPath;
412			}
413
414			/**
415			* @param string $logPath
416			*
417			* @return $this
418			*/
419			public function setLogPath($logPath)
420			{
421			$this->logPath = $logPath;
422
423			return $this;
424			}
425
426			/**
427			* Gets the path to write stderr to.
428			*
429			* @return string
430			*/
431			protected function getStdErr()
432			{
433			if ($this->getLogPath() !== null) {
434			return "2>> ".$this->getLogPath()."/stderr.log";
435			}
436
437			return "2> /dev/null";
438			}
439
440			/**
441			* Gets a string representation of a task, to pass to the worker script.
442			*
443			* @param Task $task
444			*
445			* @return string
446			*/
447			protected function getTaskString(Task $task)
448			{
449			return base64_encode(serialize($task));
450			}
451
452			/**
453			* Checks whether a task can be removed from the list of running processes.
454			*
455			* @param Task $task
456			*
457			* @return bool
458			*/
459			protected function canRemoveTask(Task $task)
460			{
461			if (!$task instanceof Process) {
462			return true;
463			}
464
465			if ($this->isTaskExpired($task) \|\| $this->isTaskCancelled($task)) {
466			$this->killTask($task);
467			return true;
468			}
469
470			$processes = array_filter($this->running, function (Task $task) {
471			return $task instanceof Process;
472			});
473
474			if (count($processes) < 1) {
475			return true;
476			}
477
478			$found = false;
479			$stats = $this->getStatsForProcesses($processes);
480
481			foreach ($stats as $stat) {
482			if ($stat[0] === $task->getId()) {
483			$found = true;
484			}
485			}
486
487			return !$found;
488			}
489
490			/**
491			* Check if the given task is expired
492			*
493			* @param Task $task
494			* @return boolean
495			*/
496			protected function isTaskExpired(Task $task) {
497			if ($task instanceof Expires) {
498			$expiresIn = $task->getExpiresIn();
499			$startedAt = $this->timings[$task];
500
501			if($expiresIn > 0 && (time() - $startedAt) >= $expiresIn) {
502			return $task->shouldExpire($startedAt);
503			}
504			}
505
506			return false;
507			}
508
509			/**
510			* Check if the given task is cancelled.
511			*
512			* @param Task $task
513			*
514			* @return bool
515			*/
516			protected function isTaskCancelled(Task $task)
517			{
518			if ($task instanceof Cancellable) {
519			return $task->isCancelled();
520			}
521
522			return false;
523			}
524
525			/**
526			* Revoke any background processes attached to this task.
527			*
528			* @param Task $task
529			*
530			* @return bool
531			*/
532			protected function killTask(Task $task)
533			{
534			if ($task instanceof Process) {
535			$this->getShell()->exec("kill -9 %s", array(
536			$task->getId(),
537			));
538
539			return true;
540			}
541
542			return false;
543			}
544
545			/**
546			* @param Rule $rule
547			*
548			* @return $this
549			*/
550			public function addRule(Rule $rule)
551			{
552			$this->getRules()->addRule($rule);
553
554			return $this;
555			}
556
557			/**
558			* @param Rule $rule
559			*
560			* @return $this
561			*/
562			public function removeRule(Rule $rule)
563			{
564			$this->getRules()->removeRule($rule);
565
566			return $this;
567			}
568			}
569

silverstripe / doorman

ProcessManager::getStdOut() A last analyzed 2021-05-30 20:53 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ProcessManager::getStdOut() A
last analyzed 2021-05-30 20:53 UTC