<commit_before><?php

/**
 * Redis Queue Extension
 *
 * Copyright 2016-2019 秋水之冰 <<EMAIL>>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

namespace ext;

use core\parser\data;

use core\handler\platform;

class redis_queue extends redis
{
    //Queue type
    const TYPE_REALTIME = 'realtime';
    const TYPE_UNIQUE   = 'unique';
    const TYPE_DELAY    = 'delay';

    //Wait properties
    const WAIT_IDLE = 3;
    const WAIT_SCAN = 60;

    //Queue keys
    const KEY_LISTEN = 'RQ:listen';
    const KEY_FAILED = 'RQ:failed';

    //Delay keys
    const KEY_DELAY_LOCK = 'RQ:delay:lock';
    const KEY_DELAY_JOBS = 'RQ:delay:jobs';

    //Queue key prefix
    const PREFIX_CMD    = 'RQ:cmd:';
    const PREFIX_JOBS   = 'RQ:jobs:';
    const PREFIX_WATCH  = 'RQ:watch:';
    const PREFIX_WORKER = 'RQ:worker:';

    //Process properties
    protected $max_fork = 10;
    protected $max_exec = 200;

    //Unit resources
    private $unit = '';

    /**
     * Add job
     * Caution: Do NOT expose "add" to TrustZone directly
     *
     * @param string $cmd
     * @param array  $data
     * @param string $group
     * @param string $type default realtime
     * @param int    $time in seconds
     *
     * @return int -1: add unique job failed; 0: add job failed; other: succeeded with job length
     */
    public function add(string $cmd, array $data = [], string $group = 'main', string $type = self::TYPE_REALTIME, int $time = 0): int
    {
        //Add command
        $data['cmd'] = &$cmd;

        //Sort data by key
        ksort($data);

        //Build queue data
        $queue = json_encode($data, JSON_FORMAT);

        //Check group key
        if ('' === $group) {
            $group = 'main';
        }

        //Redirect to REALTIME
        if (0 === $time) {
            $type = self::TYPE_REALTIME;
        }

        //Add job
        switch ($type) {
            case self::TYPE_UNIQUE:
                $result = $this->add_unique($group, $queue, $time);
                break;

            case self::TYPE_DELAY:
                $result = $this->add_delay($group, $queue, $time);
                break;

            default:
                $result = $this->add_realtime($group, $queue);
                break;
        }

        unset($cmd, $data, $group, $type, $time, $queue);
        return $result;
    }

    /**
     * Close worker process
     * Caution: Do NOT expose "close" to TrustZone directly
     *
     * @param string $key
     *
     * @return int
     */
    public function close(string $key = ''): int
    {
        //Get process list
        $process = '' === $key ? array_keys($this->show_process()) : [self::PREFIX_WORKER . $key];

        if (empty($process)) {
            return 0;
        }

        //Remove worker from process list
        $result = call_user_func_array([$this->instance, 'del'], $process);

        //Remove worker keys
        array_unshift($process, $this->get_watch_key());
        call_user_func_array([$this->instance, 'hDel'], $process);

        unset($key, $process);
        return $result;
    }

    /**
     * Show fail list
     *
     * @param int $start
     * @param int $end
     *
     * @return array
     */
    public function show_fail(int $start = 0, int $end = -1): array
    {
        $list = [
            'len'  => $this->instance->lLen(self::KEY_FAILED),
            'data' => $this->instance->lRange(self::KEY_FAILED, $start, $end)
        ];

        unset($start, $end);
        return $list;
    }

    /**
     * Show queue length
     *
     * @param string $queue_key
     *
     * @return int
     */
    public function show_length(string $queue_key): int
    {
        return (int)$this->instance->lLen($queue_key);
    }

    /**
     * Show queue list
     *
     * @return array
     */
    public function show_queue(): array
    {
        return $this->get_keys(self::KEY_LISTEN);
    }

    /**
     * Show process list
     *
     * @return array
     */
    public function show_process(): array
    {
        return $this->get_keys($this->get_watch_key());
    }

    /**
     * Start master process
     *
     * @param int $max_fork
     * @param int $max_exec
     *
     * @throws \Exception
     */
    public function start(int $max_fork = 10, int $max_exec = 200): void
    {
        //Detect running mode
        if (!parent::$is_CLI) {
            throw new \Exception('Redis queue only supports CLI!', E_USER_ERROR);
        }

        //Set max forks
        if (0 < $max_fork) {
            $this->max_fork = &$max_fork;
        }

        //Set max executes
        if (0 < $max_exec) {
            $this->max_exec = &$max_exec;
        }

        unset($max_fork, $max_exec);

        //Get idle time
        $idle_time = $this->get_idle_time();

        //Build master hash and key
        $master_hash = hash('crc32b', uniqid(mt_rand(), true));
        $master_key  = self::PREFIX_WORKER . ($_SERVER['HOSTNAME'] ?? 'master');

        //Exit when master process exists
        if (!$this->instance->setnx($master_key, $master_hash)) {
            exit('Already running!');
        }

        //Set process life and add to watch list
        $this->instance->expire($master_key, self::WAIT_SCAN);
        $this->instance->hSet($this->get_watch_key(), $master_key, time());

        //Close on shutdown
        register_shutdown_function([$this, 'close']);

        //Build unit command
        $this->unit = platform::cmd_bg(
            '"' . platform::php_path() . '" '
            . '"' . ENTRY_SCRIPT . '" --ret '
            . '--cmd="' . parent::get_app_cmd(strtr(get_class($this), '\\', '/')) . '-unit"'
        );

        do {
            //Call realtime unit
            $this->call_unit_delay();

            //Get process status
            $valid   = $this->instance->get($master_key) === $master_hash;
            $running = $this->instance->expire($master_key, self::WAIT_SCAN);

            //Idle wait on no job or unit process running
            if (empty($list = $this->show_queue()) || 1 < count($this->show_process())) {
                sleep(self::WAIT_IDLE);
                continue;
            }

            //Idle wait on no job
            if (empty($queue = $this->instance->brPop(array_keys($list), $idle_time))) {
                sleep(self::WAIT_IDLE);
                continue;
            }

            //Re-add queue job
            $this->instance->rPush($queue[0], $queue[1]);

            //Call realtime unit
            $this->call_unit_realtime();
        } while ($valid && $running);

        //On exit
        self::close();
        unset($idle_time, $master_hash, $master_key, $valid, $running, $list, $queue);
    }

    /**
     * Start unit process
     *
     * @param string $type
     *
     * @throws \Exception
     */
    public function unit(string $type): void
    {
        switch ($type) {
            case 'delay':
                //Delay unit on going
                if (!$this->instance->setnx(self::KEY_DELAY_LOCK, time())) {
                    return;
                }

                do {
                    $delay_job = $this->instance->rPop(self::KEY_DELAY_JOBS);

                    if (empty($delay_job)) {
                        $this->instance->del(self::KEY_DELAY_LOCK);
                        break;
                    }

                    $job_data = json_decode($delay_job, true);

                    if ($job_data['time'] > time()) {
                        $this->instance->rPush(self::KEY_DELAY_JOBS, $delay_job);
                        $this->instance->del(self::KEY_DELAY_LOCK);
                        break;
                    }

                    $this->add_realtime($job_data['group'], $job_data['job']);
                } while (true);

                break;

            default:
                //Detect running mode
                if (!parent::$is_CLI) {
                    throw new \Exception('Redis queue only supports CLI!', E_USER_ERROR);
                }

                //Get idle time
                $idle_time = $this->get_idle_time();

                //Build unit hash and key
                $unit_hash = hash('crc32b', uniqid(mt_rand(), true));
                $unit_key  = self::PREFIX_WORKER . $unit_hash;

                //Add to watch list
                $this->instance->set($unit_key, '', self::WAIT_SCAN);
                $this->instance->hSet($this->get_watch_key(), $unit_key, time());

                //Close on exit
                register_shutdown_function([$this, 'close'], $unit_hash);

                $execute = 0;

                do {
                    //Exit on no job
                    if (empty($list = $this->show_queue())) {
                        break;
                    }

                    //Execute job
                    if (!empty($queue = $this->instance->brPop(array_keys($list), $idle_time))) {
                        self::exec_job($queue[1]);
                    }
                } while (0 < $this->instance->exists($unit_key) && $this->instance->expire($unit_key, self::WAIT_SCAN) && ++$execute < $this->max_exec);

                //On exit
                self::stop($unit_hash);
                unset($idle_time, $unit_hash, $unit_key, $execute, $list, $queue);
                break;
        }
    }

    /**
     * Add realtime job
     *
     * @param string $group
     * @param string $data
     *
     * @return int
     */
    private function add_realtime(string $group, string $data): int
    {
        //Add watch list
        $this->instance->hSet(self::KEY_LISTEN, $key = self::PREFIX_JOBS . $group, time());

        //Add job
        $result = (int)$this->instance->lPush($key, $data);

        unset($group, $data, $key);
        return $result;
    }

    /**
     * Add unique job
     *
     * @param string $group
     * @param string $data
     * @param int    $time
     *
     * @return int
     */
    private function add_unique(string $group, string $data, int $time): int
    {
        //Check job duration
        if (!$this->instance->setnx($key = self::PREFIX_CMD . hash('crc32b', $data), time())) {
            return -1;
        }

        //Set duration life
        $this->instance->expire($key, $time);

        //Add realtime job
        $result = $this->add_realtime($group, $data);

        unset($group, $data, $time, $key);
        return $result;
    }

    /**
     * Add delay job
     *
     * @param string $group
     * @param string $data
     * @param int    $time
     *
     * @return int
     */
    private function add_delay(string $group, string $data, int $time): int
    {
        $result = $this->instance->lPush(self::KEY_DELAY_JOBS, json_encode([
            'group' => &$group,
            'time'  => time() + $time,
            'job'   => &$data
        ], JSON_FORMAT));

        unset($group, $data, $time);
        return $result;
    }

    /**
     * Add delay jobs
     *
     * @param int $time
     * @param int $idle_time
     */
    private function add_delay_jobs(int $time, int $idle_time): void
    {
        //Get read from & read till
        $read_till = $this->instance->incrBy(self::KEY_DELAY_LOCK, $idle_time - 1);
        $read_from = $read_till - $idle_time + 1;

        //Read time NOT arrive
        if ($read_from > $time) {
            $this->instance->decrBy(self::KEY_DELAY_LOCK, $idle_time - 1);
            return;
        }

        //Read time over now
        if ($read_till > $time) {
            $read_till = $this->instance->decrBy(self::KEY_DELAY_LOCK, $read_till - $time + 1);
        }

        //Read jobs from range
        $delay_jobs = $this->instance->zRangeByScore(self::KEY_DELAY_JOBS, $read_from, $read_till);

        //No delay job found
        if (empty($delay_jobs)) {
            return;
        }

        //Parse & add job
        foreach ($delay_jobs as $item) {
            $data = json_decode($item, true);

            //Delay job error
            if (!is_array($data) || !isset($data['group']) || !isset($data['job'])) {
                $this->instance->lPush(self::KEY_FAILED, json_encode(['data' => &$item, 'return' => 'Delay Job ERROR!'], JSON_FORMAT));
                continue;
            }

            $this->add_realtime($data['group'], $data['job']);
        }

        //Remove added jobs
        $this->instance->zRemRangeByScore(self::KEY_DELAY_JOBS, $read_from, $read_till);
        unset($time, $idle_time, $read_till, $read_from, $delay_jobs, $item, $data);
    }

    /**
     * Get active keys
     *
     * @param string $key
     *
     * @return array
     */
    private function get_keys(string $key): array
    {
        if (0 === $this->instance->exists($key)) {
            return [];
        }

        if (empty($keys = $this->instance->hGetAll($key))) {
            return [];
        }

        foreach ($keys as $k => $v) {
            if (0 === $this->instance->exists($k)) {
                $this->instance->hDel($key, $k);
                unset($keys[$k]);
            }
        }

        unset($key, $k, $v);
        return $keys;
    }

    /**
     * Get watch key
     *
     * @return string
     */
    private function get_watch_key(): string
    {
        return self::PREFIX_WATCH . ($_SERVER['HOSTNAME'] ?? 'worker');
    }

    /**
     * Get idle time
     *
     * @return int
     */
    private function get_idle_time(): int
    {
        return (int)(self::WAIT_SCAN / 2);
    }

    /**
     * Delay unit process
     */
    private function call_unit_delay(): void
    {
        //Call unit processes
        pclose(popen($this->unit . ' --data="type=delay"', 'r'));
    }

    /**
     * Realtime unit process
     */
    private function call_unit_realtime(): void
    {
        //Count running processes
        $running = count($this->show_process());

        if (0 >= $left = $this->max_fork - $running + 1) {<commit_msg>delay logic changed (still has timeline mistake when adding within long/short delay)<commit_after><?php

/**
 * Redis Queue Extension
 *
 * Copyright 2016-2019 秋水之冰 <<EMAIL>>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

namespace ext;

use core\parser\data;

use core\handler\platform;

class redis_queue extends redis
{
    //Queue type
    const TYPE_REALTIME = 'realtime';
    const TYPE_UNIQUE   = 'unique';
    const TYPE_DELAY    = 'delay';

    //Wait properties
    const WAIT_IDLE = 3;
    const WAIT_SCAN = 60;

    //Queue keys
    const KEY_LISTEN = 'RQ:listen';
    const KEY_FAILED = 'RQ:failed';

    //Delay keys
    const KEY_DELAY_LOCK = 'RQ:delay:lock';
    const KEY_DELAY_JOBS = 'RQ:delay:jobs';

    //Queue key prefix
    const PREFIX_CMD    = 'RQ:cmd:';
    const PREFIX_JOBS   = 'RQ:jobs:';
    const PREFIX_WATCH  = 'RQ:watch:';
    const PREFIX_WORKER = 'RQ:worker:';

    //Process properties
    protected $max_fork = 10;
    protected $max_exec = 200;

    //Unit resources
    private $unit = '';

    /**
     * Add job
     * Caution: Do NOT expose "add" to TrustZone directly
     *
     * @param string $cmd
     * @param array  $data
     * @param string $group
     * @param string $type default realtime
     * @param int    $time in seconds
     *
     * @return int -1: add unique job failed; 0: add job failed; other: succeeded with job length
     */
    public function add(string $cmd, array $data = [], string $group = 'main', string $type = self::TYPE_REALTIME, int $time = 0): int
    {
        //Add command
        $data['cmd'] = &$cmd;

        //Sort data by key
        ksort($data);

        //Build queue data
        $queue = json_encode($data, JSON_FORMAT);

        //Check group key
        if ('' === $group) {
            $group = 'main';
        }

        //Redirect to REALTIME
        if (0 === $time) {
            $type = self::TYPE_REALTIME;
        }

        //Add job
        switch ($type) {
            case self::TYPE_UNIQUE:
                $result = $this->add_unique($group, $queue, $time);
                break;

            case self::TYPE_DELAY:
                $result = $this->add_delay($group, $queue, $time);
                break;

            default:
                $result = $this->add_realtime($group, $queue);
                break;
        }

        unset($cmd, $data, $group, $type, $time, $queue);
        return $result;
    }

    /**
     * Close worker process
     * Caution: Do NOT expose "close" to TrustZone directly
     *
     * @param string $key
     *
     * @return int
     */
    public function close(string $key = ''): int
    {
        //Get process list
        $process = '' === $key ? array_keys($this->show_process()) : [self::PREFIX_WORKER . $key];

        if (empty($process)) {
            return 0;
        }

        //Remove worker from process list
        $result = call_user_func_array([$this->instance, 'del'], $process);

        //Remove worker keys
        array_unshift($process, $this->get_watch_key());
        call_user_func_array([$this->instance, 'hDel'], $process);

        unset($key, $process);
        return $result;
    }

    /**
     * Show fail list
     *
     * @param int $start
     * @param int $end
     *
     * @return array
     */
    public function show_fail(int $start = 0, int $end = -1): array
    {
        $list = [
            'len'  => $this->instance->lLen(self::KEY_FAILED),
            'data' => $this->instance->lRange(self::KEY_FAILED, $start, $end)
        ];

        unset($start, $end);
        return $list;
    }

    /**
     * Show queue length
     *
     * @param string $queue_key
     *
     * @return int
     */
    public function show_length(string $queue_key): int
    {
        return (int)$this->instance->lLen($queue_key);
    }

    /**
     * Show queue list
     *
     * @return array
     */
    public function show_queue(): array
    {
        return $this->get_keys(self::KEY_LISTEN);
    }

    /**
     * Show process list
     *
     * @return array
     */
    public function show_process(): array
    {
        return $this->get_keys($this->get_watch_key());
    }

    /**
     * Start master process
     *
     * @param int $max_fork
     * @param int $max_exec
     *
     * @throws \Exception
     */
    public function start(int $max_fork = 10, int $max_exec = 200): void
    {
        //Detect env
        if (!parent::$is_CLI) {
            throw new \Exception('Redis queue only supports CLI!', E_USER_ERROR);
        }

        //Set max forks
        if (0 < $max_fork) {
            $this->max_fork = &$max_fork;
        }

        //Set max executes
        if (0 < $max_exec) {
            $this->max_exec = &$max_exec;
        }

        unset($max_fork, $max_exec);

        //Get idle time
        $idle_time = $this->get_idle_time();

        //Build master hash and key
        $master_hash = hash('crc32b', uniqid(mt_rand(), true));
        $master_key  = self::PREFIX_WORKER . ($_SERVER['HOSTNAME'] ?? 'master');

        //Exit when master process exists
        if (!$this->instance->setnx($master_key, $master_hash)) {
            exit('Already running!');
        }

        //Set process life and add to watch list
        $this->instance->expire($master_key, self::WAIT_SCAN);
        $this->instance->hSet($this->get_watch_key(), $master_key, time());

        //Close on shutdown
        register_shutdown_function([$this, 'close']);

        //Build unit command
        $this->unit = platform::cmd_bg(
            '"' . platform::php_path() . '" '
            . '"' . ENTRY_SCRIPT . '" --ret '
            . '--cmd="' . parent::get_app_cmd(strtr(get_class($this), '\\', '/')) . '-unit"'
        );

        do {
            //Call delay unit
            $this->call_unit_delay();

            //Get process status
            $valid   = $this->instance->get($master_key) === $master_hash;
            $running = $this->instance->expire($master_key, self::WAIT_SCAN);

            //Idle wait on no job or unit process running
            if (empty($list = $this->show_queue()) || 1 < count($this->show_process())) {
                sleep(self::WAIT_IDLE);
                continue;
            }

            //Idle wait on no job
            if (empty($queue = $this->instance->brPop(array_keys($list), $idle_time))) {
                sleep(self::WAIT_IDLE);
                continue;
            }

            //Re-add queue job
            $this->instance->rPush($queue[0], $queue[1]);

            //Call realtime unit
            $this->call_unit_realtime();
        } while ($valid && $running);

        //On exit
        self::close();
        unset($idle_time, $master_hash, $master_key, $valid, $running, $list, $queue);
    }

    /**
     * Start unit process
     *
     * @param string $type
     *
     * @throws \Exception
     */
    public function unit(string $type): void
    {
        //Detect env
        if (!parent::$is_CLI) {
            throw new \Exception('Redis queue only supports CLI!', E_USER_ERROR);
        }

        $execute = 0;

        switch ($type) {
            case 'delay':
                //Delay unit on going
                if (!$this->instance->setnx(self::KEY_DELAY_LOCK, time())) {
                    break;
                }

                do {
                    if (false === $delay_job = $this->instance->rPop(self::KEY_DELAY_JOBS)) {
                        $this->instance->del(self::KEY_DELAY_LOCK);
                        break;
                    }

                    $job_data = json_decode($delay_job, true);

                    if ($job_data['time'] > time()) {
                        $this->instance->rPush(self::KEY_DELAY_JOBS, $delay_job);
                        $this->instance->del(self::KEY_DELAY_LOCK);
                        break;
                    }

                    //Add realtime job
                    $this->add_realtime($job_data['group'], $job_data['job']);
                } while ($this->unit_is_alive(self::KEY_DELAY_LOCK, $execute));

                $this->instance->del(self::KEY_DELAY_LOCK);

                unset($delay_job, $job_data);
                break;

            default:
                //Get idle time
                $idle_time = $this->get_idle_time();

                //Build unit hash and key
                $unit_hash = hash('crc32b', uniqid(mt_rand(), true));
                $unit_key  = self::PREFIX_WORKER . $unit_hash;

                //Add to watch list
                $this->instance->set($unit_key, '', self::WAIT_SCAN);
                $this->instance->hSet($this->get_watch_key(), $unit_key, time());

                //Close on exit
                register_shutdown_function([$this, 'close'], $unit_hash);

                do {
                    //Exit on no job
                    if (empty($list = $this->show_queue())) {
                        break;
                    }

                    //Execute job
                    if (!empty($queue = $this->instance->brPop(array_keys($list), $idle_time))) {
                        self::exec_job($queue[1]);
                    }
                } while ($this->unit_is_alive($unit_key, $execute));

                //On exit
                self::stop($unit_hash);

                unset($idle_time, $unit_hash, $unit_key, $list, $queue);
                break;
        }

        unset($type, $execute);
    }

    /**
     * Get unit alive status
     *
     * @param string $unit_key
     * @param int    $unit_exec
     *
     * @return bool
     */
    private function unit_is_alive(string $unit_key, int &$unit_exec): bool
    {
        return 0 < $this->instance->exists($unit_key) && $this->instance->expire($unit_key, self::WAIT_SCAN) && ++$unit_exec < $this->max_exec;
    }

    /**
     * Add realtime job
     *
     * @param string $group
     * @param string $data
     *
     * @return int
     */
    private function add_realtime(string $group, string $data): int
    {
        //Add watch list
        $this->instance->hSet(self::KEY_LISTEN, $key = self::PREFIX_JOBS . $group, time());

        //Add job
        $result = (int)$this->instance->lPush($key, $data);

        unset($group, $data, $key);
        return $result;
    }

    /**
     * Add unique job
     *
     * @param string $group
     * @param string $data
     * @param int    $time
     *
     * @return int
     */
    private function add_unique(string $group, string $data, int $time): int
    {
        //Check job duration
        if (!$this->instance->setnx($key = self::PREFIX_CMD . hash('crc32b', $data), time())) {
            return -1;
        }

        //Set duration life
        $this->instance->expire($key, $time);

        //Add realtime job
        $result = $this->add_realtime($group, $data);

        unset($group, $data, $time, $key);
        return $result;
    }

    /**
     * Add delay job
     *
     * @param string $group
     * @param string $data
     * @param int    $time
     *
     * @return int
     */
    private function add_delay(string $group, string $data, int $time): int
    {
        $result = $this->instance->lPush(self::KEY_DELAY_JOBS, json_encode([
            'group' => &$group,
            'time'  => time() + $time,
            'job'   => &$data
        ], JSON_FORMAT));

        unset($group, $data, $time);
        return $result;
    }

    /**
     * Get active keys
     *
     * @param string $key
     *
     * @return array
     */
    private function get_keys(string $key): array
    {
        if (0 === $this->instance->exists($key)) {
            return [];
        }

        if (empty($keys = $this->instance->hGetAll($key))) {
            return [];
        }

        foreach ($keys as $k => $v) {
            if (0 === $this->instance->exists($k)) {
                $this->instance->hDel($key, $k);
                unset($keys[$k]);
            }
        }

        unset($key, $k, $v);
        return $keys;
    }

    /**
     * Get watch key
     *
     * @return string
     */
    private function get_watch_key(): string
    {
        return self::PREFIX_WATCH . ($_SERVER['HOSTNAME'] ?? 'worker');
    }

    /**
     * Get idle time
     *
     * @return int
     */
    private function get_idle_time(): int
    {
        return (int)(self::WAIT_SCAN / 2);
    }

    /**
     * Delay unit process
     */
    private function call_unit_delay(): void
    {
        pclose(popen($this->unit . ' --data="type=delay"', 'r'));
    }

    /**
     * Realtime unit process
     */
    private function call_unit_realtime(): void
    {
        //Count running processes
        $running = count($this->show_process());

        if (0 >= $left = $this->max_fork - $running + 1) {
{'fraction_non_alphanumeric': 0.14338778895740922, 'fraction_numerical': 0.0038948393378773127, 'mean_word_length': 1.6911878881987579, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 216, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1635151', 'n_tokens_mistral': 8578, 'n_tokens_neox': 8205, 'n_words': 1984}