DominionEnterprises\Cronus\ProcessRegistry

Complexity

Total Complexity

23

Size/Duplication

Total Lines	193
Duplicated Lines	7.25 %

Coupling/Cohesion

Components	1
Dependencies	2

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
D	add()	7	84	17
A	remove()	0	10	1
B	reset()	7	24	3
A	_getEncodedHostname()	0	4	1

<?php
/**
 * Defines the ProcessRegistry class which uses MongoDB as a backend.
 */

namespace DominionEnterprises\Cronus;

use MongoDB\BSON\ObjectID;
use MongoDB\BSON\UTCDateTime;
use MongoDB\Collection;

/**
 * Class that adds/removes from a process registry.
 */
final class ProcessRegistry implements ProcessRegistryInterface
{
    /** example doc:
     * {
     *     '_id': 'a unique id',
     *     'hosts': {
     *         'a hostname' : {
     *             'a pid': UTCDateTime(expire time),
     *             ...
     *         },
     *         ...
     *     },
     *     'version' => ObjectID(an id),
     * }
     */

    const MONGO_INT32_MAX = 2147483647;//2147483648 can overflow in php mongo without using the MongoInt64

    /**
     * MongoDB collection containing the process information.
     *
     * @var Collection
     */
    private $collection;

    /**
     * Construct a new instance of the registry.
     *
     * @param Collection $collection The MongoDB collection containing the process information.
     */
    public function __construct(Collection $collection)
    {
        $this->collection = $collection;
    }

    /**
     * Add to process registry. Adds based on $maxGlobalProcesses and $maxHostProcesses after a process registry
     * cleaning.
     *
     * @param string  $id                 A unique id.
     * @param integer $minsBeforeExpire   Number of minutes before a process is considered expired.
     * @param integer $maxGlobalProcesses Max processes of an id allowed to run across all hosts.
     * @param integer $maxHostProcesses   Max processes of an id allowed to run across a single host.
     *
     * @return boolean true if the process was added, false if not or there is too much concurrency at the moment.
     */
    public function add(

Function's nesting level (4) exceeds 2; consider refactoring the function

Unknown type hint "string" found for $id

Unknown type hint "int" found for $minsBeforeExpire

Unknown type hint "int" found for $maxGlobalProcesses

Unknown type hint "int" found for $maxHostProcesses

        string $id,
        int $minsBeforeExpire = PHP_INT_MAX,
        int $maxGlobalProcesses = 1,
        int $maxHostProcesses = 1
    ) : bool {
        $thisHostName = self::_getEncodedHostname();
        $thisPid = getmypid();

        //loop in case the update fails its optimistic concurrency check
        for ($i = 0; $i < 5; ++$i) {
            $this->collection->findOneAndUpdate(
                ['_id' => $id],
                ['$setOnInsert' => ['hosts' => [], 'version' => new ObjectID()]],
                ['upsert' => true]
            );
            $existing = $this->collection->findOne(
                ['_id' => $id],
                ['typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array']]
            );

            $replacement = $existing;
            $replacement['version'] = new ObjectID();

            //clean $replacement based on their pids and expire times
            foreach ($existing['hosts'] as $hostname => $pids) {
                foreach ($pids as $pid => $expires) {
                    //our machine and not running
                    //the task expired
                    //our machine and pid is recycled (should rarely happen)
                    if (($hostname === $thisHostName && !file_exists("/proc/{$pid}"))
                        || time() > $expires->toDateTime()->getTimestamp()
                        || ($hostname === $thisHostName && $pid === $thisPid)
                    ) {
                        unset($replacement['hosts'][$hostname][$pid]);
                    }
                }

                if (empty($replacement['hosts'][$hostname])) {
                    unset($replacement['hosts'][$hostname]);
                }
            }

            $totalPidCount = 0;
            foreach ($replacement['hosts'] as $hostname => $pids) {
                $totalPidCount += count($pids);
            }

            $thisHostPids = array_key_exists($thisHostName, $replacement['hosts']) ? $replacement['hosts'][$thisHostName] : [];

            if ($totalPidCount >= $maxGlobalProcesses || count($thisHostPids) >= $maxHostProcesses) {
                return false;
            }

            // add our process
            $expireSecs = time() + $minsBeforeExpire * 60;
            if (!is_int($expireSecs)) {

This code seems to be duplicated across your project.

                if ($minsBeforeExpire > 0) {
                    $expireSecs = self::MONGO_INT32_MAX;
                } else {
                    $expireSecs = 0;
                }
            }

            $thisHostPids[$thisPid] = new UTCDateTime($expireSecs * 1000);
            $replacement['hosts'][$thisHostName] = $thisHostPids;

            $status = $this->collection->replaceOne(
                ['_id' => $existing['_id'], 'version' => $existing['version']],
                $replacement,
                ['writeConcern' => new \MongoDB\Driver\WriteConcern(1, 100, true)]
            );
            if ($status->getMatchedCount() === 1) {
                return true;
            }

            //@codeCoverageIgnoreStart
            //hard to test the optimistic concurrency check
        }

        //too much concurrency at the moment, return false to signify not added.
        return false;
        //@codeCoverageIgnoreEnd
    }

    /**
     * Removes from process registry. Does not do anything needed for use of the add() method. Most will only use at the
     * end of their script so the mongo collection is up to date.
     *
     * @param string $id A unique id.
     *
     * @return void
     */
    public function remove(string $id)

Unknown type hint "string" found for $id

    {
        $thisHostName = self::_getEncodedHostname();
        $thisPid = getmypid();

        $this->collection->updateOne(
            ['_id' => $id],
            ['$unset' => ["hosts.{$thisHostName}.{$thisPid}" => ''], '$set' => ['version' => new ObjectID()]]
        );
    }

    /**
     * Reset a process expire time in the registry.
     *
     * @param string  $id               A unique id.
     * @param integer $minsBeforeExpire Number of minutes before a process is considered expired.
     *
     * @return void
     */
    public function reset(string $id, int $minsBeforeExpire)

Unknown type hint "string" found for $id

Unknown type hint "int" found for $minsBeforeExpire

    {
        $expireSecs = time() + $minsBeforeExpire * 60;
        if (!is_int($expireSecs)) {

This code seems to be duplicated across your project.

            if ($minsBeforeExpire > 0) {
                $expireSecs = self::MONGO_INT32_MAX;
            } else {
                $expireSecs = 0;
            }
        }

        $thisHostName = self::_getEncodedHostname();
        $thisPid = getmypid();

        $this->collection->updateOne(
            ['_id' => $id],
            [
                '$set' => [
                    "hosts.{$thisHostName}.{$thisPid}" => new UTCDateTime($expireSecs * 1000),
                    'version' => new ObjectID(),
                ],
            ]
        );
    }

    /**
     * Encodes '.' and '$' to be used as a mongo field name.
     *
     * @return string the encoded hostname from gethostname().
     */
    private static function _getEncodedHostname() : string
    {
        return str_replace(['.', '$'], ['_DOT_', '_DOLLAR_'], gethostname());
    }
}