public function actionIndex($queueIds = '0', $currentTime = null, $forceNoParallel = 0)
    {
        $currentTime = $currentTime ? intval($currentTime) : time();
        $this->forceNoParallel = intval($forceNoParallel) === 1;

        // acquire lock for all queue
        if ($this->getMutex()->acquire('DeferredQueueSelect') === false) {
            // another process is fetching deferred queue
            // that means your machine is too slow or SQL server is overloaded
            return 0;
        }

        $this->stdout("Getting queue\n", Console::FG_GREEN);

        // get scheduled queue
        /** @var DeferredQueue[] $queue */
        if (intval($queueIds) === 0) {
            $queueIds = null;
        } else {
            $queueIds = explode(',', $queueIds);
            array_walk($queueIds, 'intval');
        }

        $queue = DeferredQueue::getNextTasks($currentTime, $queueIds);
        if (count($queue) === 0) {
            $this->getMutex()->release('DeferredQueueSelect');
            $this->stdout("No tasks to run\n", Console::FG_GREEN);
            return 0;
        }

        // group queue
        $grouppedQueue = [];
        $ids = [];
        foreach ($queue as $item) {
            $ids[] = $item->id;

            $itemCanBeAdded = true;

            if (isset($grouppedQueue[$item->deferred_group_id]) === false) {
                // lock group so no new tasks will be run before this batch is ended
                // not applied to zero group(no group)
                if ($item->deferred_group_id > 0) {
                    $itemCanBeAdded = $this->getMutex()->acquire('DeferredQueueGroup:' . $item->deferred_group_id);
                    $grouppedQueue[$item->deferred_group_id] = [];
                }
            }
            if ($itemCanBeAdded === true) {
                $grouppedQueue[$item->deferred_group_id][] = $item;
            }


        }
        $queue = null;

        // lock queue elements
        Yii::$app->db->createCommand()
            ->update(
                DeferredQueue::tableName(),
                ['status' => DeferredQueue::STATUS_RUNNING],
                ['id' => $ids]
            )
            ->execute();

        // release DeferredQueueSelect lock
        $this->getMutex()->release('DeferredQueueSelect');

        // ok, now we can process groups
        if ($this->canRunInParallel() && $this->forceNoParallel === false) {
            $fork = new Fork;

            foreach ($grouppedQueue as $groupId => $items) {
                $fork->call(function () use ($groupId, $items) {
                    $this->processGroup($groupId, $items);
                }, [$groupId, $items]);
            }
            $fork->wait();

        } else {
            foreach ($grouppedQueue as $groupId => $items) {
                $this->processGroup($groupId, $items);
            }
        }
        $this->stdout("All tasks finished\n", Console::FG_GREEN);
        return 0;
    }
public function actionIndex($queueIds = '0', $currentTime = null, $forceNoParallel = 0)
    {
        $currentTime = $currentTime ? intval($currentTime) : time();
        $this->forceNoParallel = intval($forceNoParallel) === 1;

        // acquire lock for all queue
        if ($this->getMutex()->acquire('DeferredQueueSelect') === false) {
            // another process is fetching deferred queue
            // that means your machine is too slow or SQL server is overloaded
            return 0;
        }

        $this->stdout("Getting queue\n", Console::FG_GREEN);

        // get scheduled queue
        /** @var DeferredQueue[] $queue */
        if (intval($queueIds) === 0) {
            $queueIds = null;
        } else {
            $queueIds = explode(',', $queueIds);
            array_walk($queueIds, 'intval');
        }

        $queue = DeferredQueue::getNextTasks($currentTime, $queueIds);
        if (count($queue) === 0) {
            $this->getMutex()->release('DeferredQueueSelect');
            $this->stdout("No tasks to run\n", Console::FG_GREEN);
            return 0;
        }

        // group queue
        $grouppedQueue = [];
        $ids = [];
        foreach ($queue as $item) {
            $ids[] = $item->id;

            $itemCanBeAdded = true;

            if (isset($grouppedQueue[$item->deferred_group_id]) === false) {
                // lock group so no new tasks will be run before this batch is ended
                // not applied to zero group(no group)
                if ($item->deferred_group_id > 0) {
                    $itemCanBeAdded = $this->getMutex()->acquire('DeferredQueueGroup:' . $item->deferred_group_id);
                    $grouppedQueue[$item->deferred_group_id] = [];
                }
            }
            if ($itemCanBeAdded === true) {
                $grouppedQueue[$item->deferred_group_id][] = $item;
            }


        }
        $queue = null;

        // lock queue elements
        Yii::$app->db->createCommand()
            ->update(
                DeferredQueue::tableName(),
                ['status' => DeferredQueue::STATUS_RUNNING],
                ['id' => $ids]
            )
            ->execute();

        // release DeferredQueueSelect lock
        $this->getMutex()->release('DeferredQueueSelect');

        // ok, now we can process groups
        if ($this->canRunInParallel() && $this->forceNoParallel === false) {
            $fork = new Fork;

            foreach ($grouppedQueue as $groupId => $items) {
                $fork->call(function () use ($groupId, $items) {
                    $this->processGroup($groupId, $items);
                }, [$groupId, $items]);
            }
            $fork->wait();

        } else {
            foreach ($grouppedQueue as $groupId => $items) {
                $this->processGroup($groupId, $items);
            }
        }
        $this->stdout("All tasks finished\n", Console::FG_GREEN);
        return 0;
    }
public function actionIndex ( $ queueIds = '0' , $ currentTime = null , $ forceNoParallel = 0 ) { $ currentTime = $ currentTime ? intval ( $ currentTime ) : time ( ) ; $ this -> forceNoParallel = intval ( $ forceNoParallel ) === 1 ; // acquire lock for all queue if ( $ this -> getMutex ( ) -> acquire ( 'DeferredQueueSelect' ) === false ) { // another process is fetching deferred queue // that means your machine is too slow or SQL server is overloaded return 0 ; } $ this -> stdout ( "Getting queue\n" , Console :: FG_GREEN ) ; // get scheduled queue /** @var DeferredQueue[] $queue */ if ( intval ( $ queueIds ) === 0 ) { $ queueIds = null ; } else { $ queueIds = explode ( ',' , $ queueIds ) ; array_walk ( $ queueIds , 'intval' ) ; } $ queue = DeferredQueue :: getNextTasks ( $ currentTime , $ queueIds ) ; if ( count ( $ queue ) === 0 ) { $ this -> getMutex ( ) -> release ( 'DeferredQueueSelect' ) ; $ this -> stdout ( "No tasks to run\n" , Console :: FG_GREEN ) ; return 0 ; } // group queue $ grouppedQueue = [ ] ; $ ids = [ ] ; foreach ( $ queue as $ item ) { $ ids [ ] = $ item -> id ; $ itemCanBeAdded = true ; if ( isset ( $ grouppedQueue [ $ item -> deferred_group_id ] ) === false ) { // lock group so no new tasks will be run before this batch is ended // not applied to zero group(no group) if ( $ item -> deferred_group_id > 0 ) { $ itemCanBeAdded = $ this -> getMutex ( ) -> acquire ( 'DeferredQueueGroup:' . $ item -> deferred_group_id ) ; $ grouppedQueue [ $ item -> deferred_group_id ] = [ ] ; } } if ( $ itemCanBeAdded === true ) { $ grouppedQueue [ $ item -> deferred_group_id ] [ ] = $ item ; } } $ queue = null ; // lock queue elements Yii :: $ app -> db -> createCommand ( ) -> update ( DeferredQueue :: tableName ( ) , [ 'status' => DeferredQueue :: STATUS_RUNNING ] , [ 'id' => $ ids ] ) -> execute ( ) ; // release DeferredQueueSelect lock $ this -> getMutex ( ) -> release ( 'DeferredQueueSelect' ) ; // ok, now we can process groups if ( $ this -> canRunInParallel ( ) && $ this -> forceNoParallel === false ) { $ fork = new Fork ; foreach ( $ grouppedQueue as $ groupId => $ items ) { $ fork -> call ( function ( ) use ( $ groupId , $ items ) { $ this -> processGroup ( $ groupId , $ items ) ; } , [ $ groupId , $ items ] ) ; } $ fork -> wait ( ) ; } else { foreach ( $ grouppedQueue as $ groupId => $ items ) { $ this -> processGroup ( $ groupId , $ items ) ; } } $ this -> stdout ( "All tasks finished\n" , Console :: FG_GREEN ) ; return 0 ; }
Runs all deferred commands

@param string  $queueIds        Run exact queue ID, multiple tasks should be split by comma
@param integer $currentTime     Current timestamp mock
@param integer $forceNoParallel 1 to force running tasks one-by-one(no parallel run)

@return integer Status code
Runs all deferred commands