ProcessRegistry::reset() - Code Metrics - dominionenterprises/cronus - Measure and Improve Code Quality continuously with Scrutinizer

ProcessRegistry::reset() B
last analyzed 2016-08-03 20:09 UTC

↳ Parent: ProcessRegistry

Complexity

Conditions	5
Paths	5

Size

Total Lines	32
Code Lines	18

Duplication

Lines	7
Ratio	21.88 %

Importance

Changes	5
Bugs	0	Features	1

Metric	Value
c	5
b	0
f	1
dl	7
loc	32
rs	8.439
cc	5
eloc	18
nc	5
nop	3

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

namespace DominionEnterprises\Cronus;

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

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

    /**
     * Add to process registry. Adds based on $maxGlobalProcesses and $maxHostProcesses after a process registry cleaning.
     *
     * @param \MongoDB\Collection $collection the collection
     * @param string $id a unique id
     * @param int $minsBeforeExpire number of minutes before a process is considered expired.
     * @param int $maxGlobalProcesses max processes of an id allowed to run across all hosts.
     * @param int $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.
     *
     * @throws \InvalidArgumentException if $id was not a string
     * @throws \InvalidArgumentException if $minsBeforeExpire was not an int
     * @throws \InvalidArgumentException if $maxGlobalProcesses was not an int
     * @throws \InvalidArgumentException if $maxHostProcesses was not an int
     */
    public static function add(
        \MongoDB\Collection $collection,
        $id,
        $minsBeforeExpire = PHP_INT_MAX,
        $maxGlobalProcesses = 1,
        $maxHostProcesses = 1
    )
    {
        if (!is_string($id)) {
            throw new \InvalidArgumentException('$id was not a string');
        }

        if (!is_int($minsBeforeExpire)) {
            throw new \InvalidArgumentException('$minsBeforeExpire was not an int');
        }

        if (!is_int($maxGlobalProcesses)) {
            throw new \InvalidArgumentException('$maxGlobalProcesses was not an int');
        }

        if (!is_int($maxHostProcesses)) {
            throw new \InvalidArgumentException('$maxHostProcesses was not an int');
        }

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

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

            $replacement = $existing;
            $replacement['version'] = new \MongoDB\BSON\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)) {

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

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

            $status = $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 \MongoDB\Collection $collection the collection
     * @param string $id a unique id
     *
     * @return void
     *
     * @throws \InvalidArgumentException if $id was not a string
     */
    public static function remove(\MongoDB\Collection $collection, $id)
    {
        if (!is_string($id)) {
            throw new \InvalidArgumentException('$id was not a string');
        }

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

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

    /**
     * Reset a process expire time in the registry.
     *
     * @param \MongoDB\Collection $collection the collection
     * @param string $id a unique id
     * @param int $minsBeforeExpire number of minutes before a process is considered expired.
     *
     * @return void
     *
     * @throws \InvalidArgumentException if $id was not a string
     * @throws \InvalidArgumentException if $minsBeforeExpire was not an int
     */
    public static function reset(\MongoDB\Collection $collection, $id, $minsBeforeExpire)
    {
        if (!is_string($id)) {
            throw new \InvalidArgumentException('$id was not a string');
        }

        if (!is_int($minsBeforeExpire)) {
            throw new \InvalidArgumentException('$minsBeforeExpire was not an int');
        }

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

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

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

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

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


1		<?php
2		/**
3		* Defines the ProcessRegistry class which uses MongoDB as a backend.
4		*/
5
6		namespace DominionEnterprises\Cronus;
7
8		/**
9		* Class that adds/removes from a process registry.
10		*/
11		final class ProcessRegistry
12		{
13		/** example doc:
14		* {
15		* '_id': 'a unique id',
16		* 'hosts': {
17		* 'a hostname' : {
18		* 'a pid': \MongoDB\BSON\UTCDateTime(expire time),
19		* ...
20		* },
21		* ...
22		* },
23		* 'version' => \MongoDB\BSON\ObjectID(an id),
24		* }
25		*/
26
27		const MONGO_INT32_MAX = 2147483647;//2147483648 can overflow in php mongo without using the MongoInt64
28
29		/**
30		* Add to process registry. Adds based on $maxGlobalProcesses and $maxHostProcesses after a process registry cleaning.
31		*
32		* @param \MongoDB\Collection $collection the collection
33		* @param string $id a unique id
34		* @param int $minsBeforeExpire number of minutes before a process is considered expired.
35		* @param int $maxGlobalProcesses max processes of an id allowed to run across all hosts.
36		* @param int $maxHostProcesses max processes of an id allowed to run across a single host.
37		*
38		* @return boolean true if the process was added, false if not or there is too much concurrency at the moment.
39		*
40		* @throws \InvalidArgumentException if $id was not a string
41		* @throws \InvalidArgumentException if $minsBeforeExpire was not an int
42		* @throws \InvalidArgumentException if $maxGlobalProcesses was not an int
43		* @throws \InvalidArgumentException if $maxHostProcesses was not an int
44		*/
45		public static function add(
46		\MongoDB\Collection $collection,
47		$id,
48		$minsBeforeExpire = PHP_INT_MAX,
49		$maxGlobalProcesses = 1,
50		$maxHostProcesses = 1
51		)
52		{
53		if (!is_string($id)) {
54		throw new \InvalidArgumentException('$id was not a string');
55		}
56
57		if (!is_int($minsBeforeExpire)) {
58		throw new \InvalidArgumentException('$minsBeforeExpire was not an int');
59		}
60
61		if (!is_int($maxGlobalProcesses)) {
62		throw new \InvalidArgumentException('$maxGlobalProcesses was not an int');
63		}
64
65		if (!is_int($maxHostProcesses)) {
66		throw new \InvalidArgumentException('$maxHostProcesses was not an int');
67		}
68
69		$thisHostName = self::_getEncodedHostname();
70		$thisPid = getmypid();
71
72		//loop in case the update fails its optimistic concurrency check
73		for ($i = 0; $i < 5; ++$i) {
74		$collection->findOneAndUpdate(
75		['_id' => $id],
76		['$setOnInsert' => ['hosts' => [], 'version' => new \MongoDB\BSON\ObjectID()]],
77		['upsert' => true]
78		);
79		$existing = $collection->findOne(['_id' => $id], ['typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array']]);
80
81		$replacement = $existing;
82		$replacement['version'] = new \MongoDB\BSON\ObjectID();
83
84		//clean $replacement based on their pids and expire times
85		foreach ($existing['hosts'] as $hostname => $pids) {
86		foreach ($pids as $pid => $expires) {
87		//our machine and not running
88		//the task expired
89		//our machine and pid is recycled (should rarely happen)
90		if (
91		($hostname === $thisHostName && !file_exists("/proc/{$pid}"))
92		\|\| time() >= $expires->toDateTime()->getTimestamp()
93		\|\| ($hostname === $thisHostName && $pid === $thisPid)
94		) {
95		unset($replacement['hosts'][$hostname][$pid]);
96		}
97		}
98
99		if (empty($replacement['hosts'][$hostname])) {
100		unset($replacement['hosts'][$hostname]);
101		}
102		}
103
104		$totalPidCount = 0;
105		foreach ($replacement['hosts'] as $hostname => $pids) {
106		$totalPidCount += count($pids);
107		}
108
109		$thisHostPids = array_key_exists($thisHostName, $replacement['hosts']) ? $replacement['hosts'][$thisHostName] : [];
110
111		if ($totalPidCount >= $maxGlobalProcesses \|\| count($thisHostPids) >= $maxHostProcesses) {
112		return false;
113		}
114
115		// add our process
116		$expireSecs = time() + $minsBeforeExpire * 60;
117	View Code Duplication	if (!is_int($expireSecs)) {
		0 ignored issues – show Duplication introduced 2016-05-26 20:19 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
118		if ($minsBeforeExpire > 0) {
119		$expireSecs = self::MONGO_INT32_MAX;
120		} else {
121		$expireSecs = 0;
122		}
123		}
124
125		$thisHostPids[$thisPid] = new \MongoDB\BSON\UTCDateTime($expireSecs * 1000);
126		$replacement['hosts'][$thisHostName] = $thisHostPids;
127
128		$status = $collection->replaceOne(
129		['_id' => $existing['_id'], 'version' => $existing['version']],
130		$replacement,
131		['writeConcern' => new \MongoDB\Driver\WriteConcern(1, 100, true)]
132		);
133		if ($status->getMatchedCount() === 1) {
134		return true;
135		}
136
137		//@codeCoverageIgnoreStart
138		//hard to test the optimistic concurrency check
139		}
140
141		//too much concurrency at the moment, return false to signify not added.
142		return false;
143		//@codeCoverageIgnoreEnd
144		}
145
146		/**
147		* 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
148		* so the mongo collection is up to date.
149		*
150		* @param \MongoDB\Collection $collection the collection
151		* @param string $id a unique id
152		*
153		* @return void
154		*
155		* @throws \InvalidArgumentException if $id was not a string
156		*/
157		public static function remove(\MongoDB\Collection $collection, $id)
158		{
159		if (!is_string($id)) {
160		throw new \InvalidArgumentException('$id was not a string');
161		}
162
163		$thisHostName = self::_getEncodedHostname();
164		$thisPid = getmypid();
165
166		$collection->updateOne(
167		['_id' => $id],
168		['$unset' => ["hosts.{$thisHostName}.{$thisPid}" => ''], '$set' => ['version' => new \MongoDB\BSON\ObjectID()]]
169		);
170		}
171
172		/**
173		* Reset a process expire time in the registry.
174		*
175		* @param \MongoDB\Collection $collection the collection
176		* @param string $id a unique id
177		* @param int $minsBeforeExpire number of minutes before a process is considered expired.
178		*
179		* @return void
180		*
181		* @throws \InvalidArgumentException if $id was not a string
182		* @throws \InvalidArgumentException if $minsBeforeExpire was not an int
183		*/
184		public static function reset(\MongoDB\Collection $collection, $id, $minsBeforeExpire)
185		{
186		if (!is_string($id)) {
187		throw new \InvalidArgumentException('$id was not a string');
188		}
189
190		if (!is_int($minsBeforeExpire)) {
191		throw new \InvalidArgumentException('$minsBeforeExpire was not an int');
192		}
193
194		$expireSecs = time() + $minsBeforeExpire * 60;
195	View Code Duplication	if (!is_int($expireSecs)) {
		0 ignored issues – show Duplication introduced 2016-05-26 20:19 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
196		if ($minsBeforeExpire > 0) {
197		$expireSecs = self::MONGO_INT32_MAX;
198		} else {
199		$expireSecs = 0;
200		}
201		}
202
203		$thisHostName = self::_getEncodedHostname();
204		$thisPid = getmypid();
205
206		$collection->updateOne(
207		['_id' => $id],
208		[
209		'$set' => [
210		"hosts.{$thisHostName}.{$thisPid}" => new \MongoDB\BSON\UTCDateTime($expireSecs * 1000),
211		'version' => new \MongoDB\BSON\ObjectID(),
212		],
213		]
214		);
215		}
216
217		/**
218		* Encodes '.' and '$' to be used as a mongo field name.
219		*
220		* @return string the encoded hostname from gethostname().
221		*/
222		private static function _getEncodedHostname()
223		{
224		return str_replace(['.', '$'], ['_DOT_', '_DOLLAR_'], gethostname());
225		}
226		}
227

dominionenterprises / cronus

ProcessRegistry::reset() B last analyzed 2016-08-03 20:09 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ProcessRegistry::reset() B
last analyzed 2016-08-03 20:09 UTC