Jabe\Engine\Impl\Util\Concurrent\ProcessPoolExecutor

Complexity

Total Complexity

108

Size/Duplication

Total Lines	587
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

33 Methods

Rating	Name	Duplication	Size	Complexity
A	isTerminating()	0	4	2
A	shutdownNow()	0	14	1
B	processWorkerExit()	0	33	9
A	completedTaskCount()	0	3	1
A	reject()	0	2	1
A	compareAndIncrementWorkerCount()	0	3	1
A	remove()	0	5	1
A	runStateAtLeast()	0	3	1
A	advanceRunState()	0	9	4
A	workerCountOf()	0	3	1
A	isRunning()	0	3	1
A	shutdown()	0	12	1
A	runStateOf()	0	3	1
A	drainQueue()	0	14	4
B	execute()	0	21	9
B	tryTerminate()	0	30	8
A	isShutdown()	0	3	1
A	isTerminated()	0	3	1
A	awaitTermination()	0	16	4
A	onShutdown()	0	2	1
A	ctlOf()	0	3	1
A	runStateLessThan()	0	3	1
A	compareAndDecrementWorkerCount()	0	3	1
A	isRunningOrShutdown()	0	4	3
C	addWorker()	0	69	16
A	terminated()	0	2	1
A	__construct()	0	16	3
A	runWorker()	0	26	4
C	getTask()	0	45	14
A	decrementWorkerCount()	0	4	2
A	interruptIdleWorkers()	0	21	5
A	checkShutdownAccess()	0	2	1
A	interruptWorkers()	0	14	3

<?php

namespace Jabe\Engine\Impl\Util\Concurrent;

class ProcessPoolExecutor implements ExecutorServiceInterface
{
    private $ctl;
    private const COUNT_BITS = ( PHP_INT_SIZE * 4 ) - 3;
    private const CAPACITY   = (1 << self::COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    private const RUNNING    = -1 << self::COUNT_BITS;
    private const SHUTDOWN   =  0 << self::COUNT_BITS;
    private const STOP       =  1 << self::COUNT_BITS;
    private const TIDYING    =  2 << self::COUNT_BITS;
    private const TERMINATED =  3 << self::COUNT_BITS;

    private const ONLY_ONE = true;

    private $workQueue;

    private $mainLock;

    /**
     * Set containing all worker processes in pool
     */
    private $workers = [];

    /**
     * Wait condition to support awaitTermination
     */
    //private final Condition termination = mainLock.newCondition();

    /**
     * Tracks largest attained pool size. Accessed only under
     * mainLock.
     */
    private $largestPoolSize;

    /**
     * Counter for completed tasks
     */
    private $completedTaskCount = 0;

    /*
     * All user control parameters are declared as volatiles so that
     * ongoing actions are based on freshest values, but without need
     * for locking, since no internal invariants depend on them
     * changing synchronously with respect to other actions.
     */

    /**
     * Handler called when saturated or shutdown in execute.
     */
    private $handler;

The private property $handler is not used, and could be removed.

    /**
     * Timeout in nanoseconds for idle processes
     */
    private $keepAliveTime;

    /**
     * Maximum pool size.
     */
    private $poolSize;

    // Packing and unpacking ctl
    private static function runStateOf(int $c): int
    {
        return $c & ~self::CAPACITY;
    }

    private static function workerCountOf(int $c): int
    {
        return $c & self::CAPACITY;
    }

    private static function ctlOf(int $rs, int $wc)
    {
        return $rs | $wc;
    }

    /*
     * Bit field accessors that don't require unpacking ctl.
     * These depend on the bit layout and on workerCount being never negative.
     */
    private static function runStateLessThan(int $c, int $s): bool
    {
        return $c < $s;
    }

    private static function runStateAtLeast(int $c, int $s): bool
    {
        return $c >= $s;
    }

    private static function isRunning(int $c): bool
    {
        return $c < self::SHUTDOWN;
    }

    /**
     * Attempt to CAS-increment the workerCount field of ctl.
     */
    private function compareAndIncrementWorkerCount(int $expect): bool
    {
        return $this->ctl->cmpset($expect, $expect + 1);
    }

    /**
     * Attempt to CAS-decrement the workerCount field of ctl.
     */
    private function compareAndDecrementWorkerCount(int $expect): bool
    {
        return $this->ctl->cmpset($expect, $expect - 1);
    }

    /**
     * Decrements the workerCount field of ctl.
     */
    private function decrementWorkerCount(): void
    {
        do {
        } while (!$this->compareAndDecrementWorkerCount($this->ctl->get()));
    }

    /**
     * Transitions runState to given target, or leaves it alone if
     * already at least the given target.
     *
     * @param targetState the desired state, either SHUTDOWN or STOP

The type Jabe\Engine\Impl\Util\Concurrent\the was not found. Maybe you did not declare it correctly or list all dependencies?

     *        (but not TIDYING or TERMINATED -- use tryTerminate for that)
     */
    private function advanceRunState(int $targetState): void
    {
        for (;;) {
            $c = $this->ctl->get();
            if (
                self::runStateAtLeast($c, $targetState) ||
                $this->ctl->cmpset($c, self::ctlOf($targetState, self::workerCountOf($c)))
            ) {
                break;
            }
        }
    }

    /**
     * Transitions to TERMINATED state if either (SHUTDOWN and pool
     * and queue empty) or (STOP and pool empty).  If otherwise
     * eligible to terminate but workerCount is nonzero, interrupts an
     * idle worker to ensure that shutdown signals propagate. This
     * method must be called following any action that might make
     * termination possible -- reducing worker count or removing tasks
     * from the queue during shutdown. The method is non-private to
     * allow access from ScheduledThreadPoolExecutor.
     */
    public function tryTerminate(): void
    {
        for (;;) {
            $c = $this->ctl->get();
            if (
                self::isRunning($c) ||
                self::runStateAtLeast($c, self::TIDYING) ||
                (self::runStateOf($c) == self::SHUTDOWN && !$this->workQueue->isEmpty())
            ) {
                return;
            }
            if (self::workerCountOf($c) != 0) { // Eligible to terminate
                $this->interruptIdleWorkers(self::ONLY_ONE);
                return;
            }

            $mainLock = $this->mainLock;
            $mainLock->lock();
            try {
                if ($this->ctl->cmpset($c, self::ctlOf(self::TIDYING, 0))) {
                    try {
                        $this->terminated();
                    } finally {
                        $this->ctl->set(self::ctlOf(self::TERMINATED, 0));
                        //termination.signalAll();
                    }
                    return;
                }
            } finally {
                $mainLock->unlock();
            }
            // else retry on failed CAS
        }
    }

    private function checkShutdownAccess(): void
    {
    }

    /**
     * Interrupts all processes, even if active.
     */
    private function interruptWorkers(): void
    {
        $mainLock = $this->mainLock;
        $mainLock->lock();
        try {
            foreach ($this->workers as $w) {
                try {
                    $w->interrupt();
                } catch (\Exception $ignore) {
                    //
                }
            }
        } finally {
            $mainLock->unlock();
        }
    }

    /**
     * Interrupts processes that might be waiting for tasks
     */
    private function interruptIdleWorkers(bool $onlyOne = false): void
    {
        $mainLock = $this->mainLock;
        $mainLock->lock();
        try {
            foreach ($this->workers as $w) {
                $t = $w->process;
                if (!$t->isInterrupted()) {
                    try {
                        $t->interrupt();
                    } catch (\Exception $ignore) {

Consider adding a comment why this CATCH block is empty.

                    } finally {
                        $w->unlock();
                    }
                }
                if ($onlyOne) {
                    break;
                }
            }
        } finally {
            $mainLock->unlock();
        }
    }

    /**
     * Invokes the rejected execution handler for the given command.
     * Package-protected for use by ScheduledThreadPoolExecutor.
     */
    private function reject(RunnableInterface $command): void

The parameter $command is not used and could be removed.

    {
    }

    private function drainQueue(): array
    {
        $q = $this->workQueue;
        $taskList = [];
        $q->drainTo($taskList);
        if (!$q->isEmpty()) {
            $runnable = [];
            foreach ($q->toArray($runnable) as $r) {
                if ($q->remove($r)) {
                    $taskList[] = $r;
                }
            }
        }
        return $taskList;
    }

    private function addWorker(?RunnableInterface $firstTask)
    {
        for (;;) {
            $c = $this->ctl->get();
            $rs = self::runStateOf($c);

            // Check if queue empty only if necessary.
            if (
                $rs >= self::SHUTDOWN &&
                ! (
                    $rs == self::SHUTDOWN &&
                    $firstTask == null &&
                   !$this->workQueue->isEmpty()
                )
            ) {
                return false;
            }

            for (;;) {
                $wc = self::workerCountOf($c);
                if (
                    $wc >= self::CAPACITY ||
                    $wc >= $this->poolSize
                ) {
                    return false;
                }
                if ($this->compareAndIncrementWorkerCount($c)) {
                    break 2;
                }
                $c = $this->ctl->get();  // Re-read ctl
                if (self::runStateOf($c) != $rs) {
                    continue 2;
                }
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        $w = new Worker($firstTask, $this);
        $t = $w->process;

        $this->mainLock->lock();
        try {
            // Recheck while holding lock.
            // Back out on ThreadFactory failure or if
            // shut down before lock acquired.
            $c = $this->ctl->get();
            $rs = self::runStateOf($c);

            if (
                $t == null ||
                ($rs >= self::SHUTDOWN &&
                 !($rs == self::SHUTDOWN && $firstTask == null))
            ) {
                $this->decrementWorkerCount();
                $this->tryTerminate();
                return false;
            }

            $this->workers[] = $w;

            $s = count($this->workers);
            if ($s > $this->largestPoolSize) {
                $this->largestPoolSize = $s;
            }
        } finally {
            $this->mainLock->unlock();
        }
        $t->start();
        return true;
    }

    private function processWorkerExit(Worker $w, bool $completedAbruptly): void
    {
        if ($completedAbruptly) {// If abrupt, then workerCount wasn't adjusted
            $this->decrementWorkerCount();
        }

        $this->mainLock->lock();
        try {
            $this->completedTaskCount += $w->completedTasks;
            foreach ($this->workers as $key => $val) {
                if ($val == $w) {
                    unset($this->workers[$key]);
                    break;
                }
            }
        } finally {
            $this->mainLock->unlock();
        }

        $this->tryTerminate();

        $c = $this->ctl->get();
        if (self::runStateLessThan($c, self::STOP)) {
            if (!$completedAbruptly) {
                $min = $this->poolSize;
                if ($min == 0 && !$this->workQueue->isEmpty()) {
                    $min = 1;
                }
                if (self::workerCountOf($c) >= $min) {
                    return; // replacement not needed
                }
            }
            $this->addWorker(null);
        }
    }

    private function getTask(InterruptibleProcess $process): ?RunnableInterface
    {
        $timedOut = false;
        for (;;) {
             $c = $this->ctl->get();
            $rs = self::runStateOf($c);

            // Check if queue empty only if necessary.
            if ($rs >= self::SHUTDOWN && ($rs >= self::STOP || $this->workQueue->isEmpty())) {
                $this->decrementWorkerCount();
                return null;
            }

            $timed = false;

            for (;;) {
                $wc = self::workerCountOf($c);
                $timed = $wc > $this->poolSize;

                if ($wc <= $this->poolSize && !($timedOut && $timed)) {
                    break;
                }
                if ($this->compareAndDecrementWorkerCount($c)) {
                    return null;
                }
                $c = $this->ctl->get();  // Re-read ctl
                if (self::runStateOf($c) != $rs) {
                    continue 2;
                }
                // else CAS failed due to workerCount change; retry inner loop
            }
            try {
                $r = $timed ?
                    $this->workQueue->poll($this->keepAliveTime, TimeUnit::NANOSECONDS, $process) :
                    $this->workQueue->take($process);
                if ($r != null) {
                    return unserialize($r);
                }
                $timedOut = true;

The assignment to $timedOut is dead and can be removed.

            } catch (\Exception $retry) {
                $timedOut = false;
            }
            return null;
        }
        return null;
    }

    public function runWorker(Worker $w): void
    {
        $task = $w->firstTask;
        $w->firstTask = null;
        $completedAbruptly = true;
        try {
            while ($task != null || ($task = $this->getTask($w->process)) != null) {
                //$w->lock();
                try {
                    $thrown = null;
                    try {
                        $task->run();
                    } catch (\Exception $x) {
                        $thrown = $x;

The assignment to $thrown is dead and can be removed.

                        throw $x;
                    }
                } finally {
                    $task = null;
                    $w->firstTask = null;
                    $w->completedTasks += 1;
                    //$w->unlock();
                }
            }
            $completedAbruptly = false;
        } finally {
            $this->processWorkerExit($w, $completedAbruptly);
        }
    }

    public function __construct(
        int $poolSize,
        int $keepAliveTime,
        string $unit,
        ProcessQueueInterface $workQueue
    ) {
        $this->ctl = new \Swoole\Atomic\Long(self::ctlOf(self::RUNNING, 0));

The type Swoole\Atomic\Long was not found. Maybe you did not declare it correctly or list all dependencies?

        $this->mainLock = new \Swoole\Lock(SWOOLE_MUTEX);

The constant Jabe\Engine\Impl\Util\Concurrent\SWOOLE_MUTEX was not found. Maybe you did not declare it correctly or list all dependencies?

The type Swoole\Lock was not found. Maybe you did not declare it correctly or list all dependencies?

        if (
            $poolSize <= 0 || $keepAliveTime < 0
        ) {
            throw new \Exception("Illegal argument");
        }
        $this->poolSize = $poolSize;
        $this->workQueue = $workQueue;
        $this->keepAliveTime = TimeUnit::toNanos($keepAliveTime, $unit);
    }

    public function execute(RunnableInterface $command): void
    {
        $c = $this->ctl->get();
        if (self::workerCountOf($c) < $this->poolSize) {
            if ($this->addWorker($command)) {
                return;
            }
            $c = $this->ctl->get();
        }
        if (self::isRunning($c)) {
            $receiver = $this->workers[rand(0, count($this->workers) - 1)]->process;
            if ($this->workQueue->offer($command, $receiver)) {
                $recheck = $this->ctl->get();
                if (!self::isRunning($recheck) && $this->remove($command)) {
                    $this->reject($command);
                } elseif (self::workerCountOf($recheck) == 0) {
                    $this->addWorker(null);
                }
            }
        } elseif (!$this->addWorker($command)) {
            $this->reject($command);
        }
    }

    public function remove(RunnableInterface $task): bool
    {
        $removed = $this->workQueue->remove($task);
        $this->tryTerminate(); // In case SHUTDOWN and now empty
        return $removed;
    }

    /**
     * Performs any further cleanup following run state transition on
     * invocation of shutdown.  A no-op here, but used by
     * ScheduledThreadPoolExecutor to cancel delayed tasks.
     */
    public function onShutdown(): void
    {
    }

    /**
     * State check needed by ScheduledThreadPoolExecutor to
     * enable running tasks during shutdown.
     *
     * @param shutdownOK true if should return true if SHUTDOWN
     */
    public function isRunningOrShutdown(bool $shutdownOK): bool
    {
        $rs = self::runStateOf($this->ctl->get());
        return $rs == self::RUNNING || ($rs == self::SHUTDOWN && $shutdownOK);
    }

    public function shutdown(): void
    {
        $this->mainLock->lock();
        try {
            $this->checkShutdownAccess();
            $this->advanceRunState(self::SHUTDOWN);
            $this->interruptIdleWorkers();
            $this->onShutdown();
        } finally {
            $this->mainLock->unlock();
        }
        $this->tryTerminate();
    }

    public function shutdownNow(): array
    {
        $tasks = [];
        $this->mainLock->lock();
        try {
            $this->checkShutdownAccess();
            $this->advanceRunState(self::STOP);
            $this->interruptWorkers();
            $tasks = $this->drainQueue();
        } finally {
            $this->mainLock->unlock();
        }
        $this->tryTerminate();
        return $tasks;
    }

    public function isShutdown(): bool
    {
        return !self::isRunning($this->ctl->get());
    }

    public function isTerminating(): bool
    {
        $c = $this->ctl->get();
        return !self::isRunning($c) && self::runStateLessThan($c, self::TERMINATED);
    }

    public function isTerminated(): bool
    {
        return self::runStateAtLeast($this->ctl->get(), self::TERMINATED);
    }

    public function awaitTermination(int $timeout, string $unit)
    {
        $nanos = TimeUnit::toNanos($timeout, $unit);
        $this->mainLock->lock();
        try {
            for (;;) {
                if (self::runStateAtLeast($this->ctl->get(), self::TERMINATED)) {
                    return true;
                }
                if ($nanos <= 0) {
                    return false;
                }
                time_nanosleep(0, $nanos);
            }
        } finally {
            $this->mainLock->unlock();
        }
    }

    protected function terminated(): void
    {
    }

    public function completedTaskCount(): int
    {
        return $this->completedTaskCount;
    }
}