DominionEnterprises\Cronus\ProcessRegistry::add()

Complexity

Conditions	21
Paths	201

Size

Total Lines	100
Code Lines	54

Duplication

Lines	7
Ratio	7 %

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	ProcessRegistry::__construct()	0	6	1
B		0	26	5
A	ProcessRegistry::updateExistingDocument()	0	10	1
A	ProcessRegistry::ensureProcessDocumentExists()	0	8	1
A	ProcessRegistry::generateReplacementDocument()	0	15	3
A	ProcessRegistry::getExistingProcessDocument()	0	7	1

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

namespace TraderInteractive\Cronus;

use MongoDB\BSON\ObjectID;
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': expire timestamp,
     *             ...
     *         },
     *         ...
     *     },
     *     '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;

    /**
     * @var string
     */
    private $encodedHostName;

    /**
     * @var integer
     */
    private $pid;

    /**
     * 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;
        $this->encodedHostName = str_replace(['.', '$'], ['_DOT_', '_DOLLAR_'], gethostname());
        $this->pid = getmypid();
    }

    /**
     * Add to process registry. Adds based on $maxGlobalProcesses and $maxHostProcesses after a process registry
     * cleaning.
     *
     * @param ProcessInterface $process The process to add.
     *
     * @return boolean true if the process was added, false if not or there is too much concurrency at the moment.
     */
    public function add(ProcessInterface $process) : bool
    {
        //loop in case the update fails its optimistic concurrency check
        for ($i = 0; $i < 5; ++$i) {
            $this->ensureProcessDocumentExists($process->getName());
            $existing = $this->getExistingProcessDocument($process->getName());
            $replacement = $this->generateReplacementDocument($existing);
            $totalPidCount = $this->getTotalPidCount($replacement);
            $thisHostPids = $replacement['hosts'][$this->encodedHostName] ?? [];
            if ($totalPidCount >= $process->getMaxGlobalProcesses()
                    || count($thisHostPids) >= $process->getMaxHostProcesses()) {
                return false;
            }

            // add our process
            $thisHostPids[$this->pid] = $this->getExpiresSeconds($process->getMinsBeforeExpire());
            $replacement['hosts'][$this->encodedHostName] = $thisHostPids;

            if ($this->updateExistingDocument($existing['_id'], $existing['version'], $replacement)) {
                return true;
            }
        }

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

    private function updateExistingDocument(string $id, ObjectID $version, array $replacement) : bool
    {
        $status = $this->collection->replaceOne(
            ['_id' => $id, 'version' => $version],
            $replacement,
            ['writeConcern' => new \MongoDB\Driver\WriteConcern(1, 100, true)]
        );

        return $status->getMatchedCount() === 1;
    }

    private function ensureProcessDocumentExists(string $id)
    {
        $this->collection->findOneAndUpdate(
            ['_id' => $id],
            ['$setOnInsert' => ['hosts' => [], 'version' => new ObjectID()]],
            ['upsert' => true]
        );
    }

    private function generateReplacementDocument(array $existing) : array
    {
        $replacement = $existing;
        $replacement['version'] = new ObjectID();

        //clean $replacement based on their pids and expire times
        foreach ($existing['hosts'] as $hostname => $pids) {
            $this->removeExpiredPids($pids, $hostname, $replacement);
            if (empty($replacement['hosts'][$hostname])) {
                unset($replacement['hosts'][$hostname]);
            }
        }

        return $replacement;
    }

    private function getExistingProcessDocument(string $id) : array
    {
        return $this->collection->findOne(
            ['_id' => $id],
            ['typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array']]
        );
    }

    private function removeExpiredPids(array $pids, string $hostname, array &$replacement)
    {
        foreach ($pids as $pid => $expires) {
            if ($this->canRemovePid($hostname, $pid, $expires)) {
                unset($replacement['hosts'][$hostname][$pid]);
            }
        }
    }

    private function canRemovePid(string $hostname, int $pid, int $expires) : bool
    {
        return (($hostname === $this->encodedHostName && !file_exists("/proc/{$pid}"))
                || time() > $expires
                || ($hostname === $this->encodedHostName && $pid === $this->pid));
    }

    private function getTotalPidCount(array $document) : int
    {
        $totalPidCount = 0;
        foreach ($document['hosts'] as $hostname => $pids) {
            $totalPidCount += count($pids);
        }

        return $totalPidCount;
    }

    private function getExpiresSeconds(int $minsBeforeExpire) : int
    {
         $expireSecs = time() + $minsBeforeExpire * 60;
         //ensure expireSecs is between 0 and self::MONGO_INT32_MAX
         return max(0, min($expireSecs, self::MONGO_INT32_MAX));
    }

    /**
     * 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 ProcessInterface $process The process to remove.
     *
     * @return void
     */
    public function remove(ProcessInterface $process)
    {
        $this->collection->updateOne(
            ['_id' => $process->getName()],
            ['$unset' => ["hosts.{$this->encodedHostName}.{$this->pid}" => ''], '$set' => ['version' => new ObjectID()]]
        );
    }

    /**
     * Reset a process expire time in the registry.
     *
     * @param ProcessInterface $process The process to reset.
     *
     * @return void
     */
    public function reset(ProcessInterface $process)
    {
        $expireSecs = time() + $process->getMinsBeforeExpire() * 60;
        //ensure expireSecs is between 0 and self::MONGO_INT32_MAX
        $expireSecs = max(0, min($expireSecs, self::MONGO_INT32_MAX));

        $this->collection->updateOne(
            ['_id' => $process->getName()],
            [
                '$set' => [
                    "hosts.{$this->encodedHostName}.{$this->pid}" => $expireSecs,
                    'version' => new ObjectID(),
                ],
            ]
        );
    }
}