[feature request?] Worker count scale

[mastir]

Hey @walkor, thank you for you work! I`ve been worknig with workerman for like 5 years already

Long story short: is it possible to sclae up/down worker count from child process?

Currently i'm working on a project based on workerman in kubernetes. With some minor tricks it works great in container. I use it not as a microservice, but as a multi service framework. I bootstrap different apps based on $worker->name and declare $worker->count based on avarage workloads. I have monitoring app, all other workers send usage statistics there and i hope soon i will see pretty charts in admin panel. Sometimes there is a big spikes like daily sync from external system. It dumps a lots of data at once. I can see when all of the workers busy and more data waits to be processed, so is there a simple way to scale worker count? I cant scale pod count as it produces lots of new bugs, workers share most of data only in their own container. So i'm looking for more "native" solution, like when i recieve big chunk of data, spawn 20 more workers and when most of workers in idle state to scale down and free resources. Is there a way to achive this? maybe some kind of socket in master process to comunicate with monitoring?

p.s. Workerman works great with lumen framework, but i had to rewrite whole request/response flow to implement fibers. Performance is amazing.

[walkor]

Hey @mastir

Thank you very much for your long-term use and support of Workerman! Regarding your need to dynamically adjust the number of workers, Workerman currently doesn't provide a direct interface to modify worker count from child processes. However, we can achieve similar functionality through some indirect approaches. Here's an example that might be helpful to you.

#492 (comment)

[mastir]

Thanks, this was a good starting point. Now i`m able to scale up workers using custom signal:

class Worker extends \Workerman\Worker
{

    const SIG_PATCH = 62;
    public static string $patch_file = '/tmp/workerman.patch';

    /**
     * @param array<string,int> $workers
     * @return bool
     */
    public static function modifyWorkerCount(array $workers)
    {
        if (posix_getpid() === self::$_masterPid)
            return false;
        if (file_exists(self::$patch_file)) {
            return false;
        }
        file_put_contents(self::$patch_file, serialize($workers));
        return \posix_kill(self::$_masterPid, self::SIG_PATCH);
    }

    protected static function installSignal()
    {
        parent::installSignal();
        \pcntl_signal(self::SIG_PATCH, self::onPatchSignal(...), false);
    }

    protected static function onPatchSignal(int $signal)
    {
        if (file_exists(self::$patch_file)) {
            $patch = file_get_contents(self::$patch_file, length: 2048);
            unlink(self::$patch_file);
            $patch_data = unserialize($patch);
            foreach (static::$_workers as $instance) {
                if (isset($patch_data[$instance->name])) {
                    $instance->count = (int)$patch_data[$instance->name];
                }
            }
            self::initId();
            self::forkWorkers();
        }
    }

So i run it in worker thread:

Worker::modifyWorkerCount(['http' => 4]);

And i can see 3 more workers online.

It took some time to find out that self::initId() is required to create correct $_idMap. Now i have 4 idle workers and i want to scale them back to 1. Should i recreate $_idMap and kill idle workers? I think about somethinkg like this:

Worker::modifyWorkerCount(['http' => 1]);
foreach($clients as $client){
    if ($client->info['name']==='http' && $client->info['id'] > 1){
        $client->connection->send(['command'=>'stop']);
    }
}

Worker on stop call unlisten() for main socket, wait for any active tasks (if any), close all connections and call exit(0). Or maybe i should get pids from $_pidMap and kill them with SIG_USR2?

[mastir]

So i decided to use more general solution:

class Worker extends \Workerman\Worker
{

    const SIG_PATCH = 62;
    public static string $patch_file = '/tmp/workerman.patch';
    private array $instances = [];
    private array $jobCounters = [];



    /**
     * @param array<string,int> $workers
     * @return bool
     */
    public static function modifyWorkerCount(array $workers)
    {
        if (posix_getpid() === self::$_masterPid)
            return false;
        if (file_exists(self::$patch_file)) {
            return false;
        }
        file_put_contents(self::$patch_file, serialize($workers));
        return \posix_kill(self::$_masterPid, self::SIG_PATCH);
    }

    protected static function installSignal()
    {
        parent::installSignal();
        \pcntl_signal(self::SIG_PATCH, self::onPatchSignal(...), false);
    }

    protected static function onPatchSignal(int $signal)
    {
        if (file_exists(self::$patch_file)) {
            $patch = file_get_contents(self::$patch_file, length: 2048);
            unlink(self::$patch_file);
            $patch_data = unserialize($patch);
            $kill_pids = [];
            foreach (static::$_workers as $instance) {
                if (isset($patch_data[$instance->name])) {
                    $count = (int)$patch_data[$instance->name];
                    if ($count < $instance->count) {
                        for($i=$count;$i<$instance->count;$i++) {
                            $kill_pids[] = self::$_idMap[$instance->workerId][$i];
                        }
                    }
                    $instance->count = $count;
                }
            }
            self::initId();
            self::forkWorkers();
            foreach ($kill_pids as $pid) {
                \posix_kill($pid, SIGQUIT);
            }
        }
    }

Any suggestions or possible troubles i should be aware of?

[walkor]

Everything seems fine, but you'll need to test it yourself to be sure.

[mastir]

For simple workers i see scaling works as expected. But in current apps i see worker->stopping === true, but thread never dies. I think client connection to discavery service and database connections prevents shut down. How should i redefine stop handler? I want to unlisten (or pause) main socket, wait for active jobs in fibers (success or timeout) then close database and discovery connections and exit. As i think discovery and database connections counted in ConnectionInterface::$statistics['connection_count'] so exit() is not called

[walkor]

Using SIGQUIT cannot immediately stop the corresponding process unless all client connections are completely closed (database connections do not affect the shutdown).
It is recommended to use SIGINT, as the current process will exit immediately after completing its current task.

[mastir]

Is AsyncTcpConnection counts in ConnectionInterface::$statistics['connection_count']? Even threads with 0 client requests cant stop with SIGQUIT (SIGINT works). I use AsyncTcpConnection for discovery (monitoring) service, database workers and mqtt connection ( Workerman\Mqtt\Client ). How can i check what is preventing shutdown?

[mastir]

my target is stability: if worker have active client and unprocessed request (or multiple requests) and STOP signal was recieved thread should pause (or unlisten) main socket to prevent new requests and when all active connections got response and closed, only then shut down.

Maybe somthing like this?

    protected static function reinstallSignal()
    {
        parent::reinstallSignal();
        static::$globalEvent->del(\SIGUSR2, EventInterface::EV_SIGNAL);
        static::$globalEvent->add(\SIGUSR2, EventInterface::EV_SIGNAL, self::handleShutdownSignal(...));
    }

    protected static function handleShutdownSignal(int $signal)
    {
        app()->get(Worker::class)->shutdown();
    }

    public function shutdown()
    {
        if (isset($this->shutdown_timer)) {
            Log::debug(sprintf('[worker] %s #%d PID %d got secondary shutdown call', $this->name, $this->id, \posix_getpid()));
            return;
        }
        Log::debug(sprintf('[worker] %s #%d PID %d is shutting down', $this->name, $this->id, \posix_getpid()));
        $this->unlisten();
        $this->try_shutdown();
        $this->shutdown_timer = Timer::add(1.0, $this->try_shutdown(...));
    }

    private function try_shutdown()
    {
        if ($this->getActiveJobCount() === 0) {
            self::$_gracefulStop = false;
            self::stopAll(0, '[worker] worker destroyed');
        }
    }

[mastir]

So, after a while i got some stable solution

class Worker extends \Workerman\Worker
{

    const SIG_PATCH = 62;
    public static string $patch_file = '/tmp/workerman.patch';
    private array $jobCounters = [];
    private ?int $shutdown_timer;


    /**
     * @param array<string,int> $workers
     * @return bool
     */
    public static function modifyWorkerCount(array $workers)
    {
        if (posix_getpid() === self::$_masterPid)
            return false;
        if (file_exists(self::$patch_file)) {
            return false;
        }
        file_put_contents(self::$patch_file, serialize($workers));
        return \posix_kill(self::$_masterPid, self::SIG_PATCH);
    }

    protected static function installSignal()
    {
        parent::installSignal();
        \pcntl_signal(self::SIG_PATCH, self::onPatchSignal(...), false);
    }

    protected static function reinstallSignal()
    {
        parent::reinstallSignal();
        static::$globalEvent->del(\SIGQUIT, EventInterface::EV_SIGNAL);
        static::$globalEvent->add(\SIGQUIT, EventInterface::EV_SIGNAL, self::handleShutdownSignal(...));
    }

    protected static function onPatchSignal(int $signal)
    {
        if (file_exists(self::$patch_file)) {
            $patch = file_get_contents(self::$patch_file, length: 2048);
            unlink(self::$patch_file);
            $patch_data = unserialize($patch);
            $kill_pids = [];
            foreach (static::$_workers as $instance) {
                if (isset($patch_data[$instance->name])) {
                    $count = (int)$patch_data[$instance->name];
                    if ($count < $instance->count) {
                        for($i=$count;$i<$instance->count;$i++) {
                            $kill_pids[] = self::$_idMap[$instance->workerId][$i];
                        }
                    }
                    $instance->count = $count;
                }
            }
            self::initId();
            self::forkWorkers();
            foreach ($kill_pids as $pid) {
                if ($pid === 0) continue;
                \posix_kill($pid, SIGQUIT);
            }
        }
    }

    protected static function handleShutdownSignal(int $signal)
    {
        app()->get(Worker::class)->shutdown();
    }

    public function shutdown()
    {
        if (isset($this->shutdown_timer)) {
            Log::debug(sprintf('[worker] %s #%d PID %d got secondary shutdown call', $this->name, $this->id, \posix_getpid()));
            return;
        }
        Log::debug(sprintf('[worker] %s #%d PID %d is shutting down', $this->name, $this->id, \posix_getpid()));
        $this->unlisten();
        $this->try_shutdown();
        $this->shutdown_timer = Timer::add(1.0, $this->try_shutdown(...));
    }

    private function try_shutdown()
    {
        if ($this->getActiveJobCount() === 0) {
            if ($this->onWorkerStop) {
                try {
                    \call_user_func($this->onWorkerStop, $this);
                } catch (\Exception $e) {
                    static::stopAll(250, $e);
                } catch (\Error $e) {
                    static::stopAll(250, $e);
                }
            }
            if (static::$globalEvent) {
                static::$globalEvent->destroy();
            }
            try {
                exit(0);
            } catch (\Exception $e) {

            }
        }
    }

    public function registerJobCounter(\Closure $handler)
    {
        $this->jobCounters[] = $handler;
    }

    public function getActiveJobCount(): int
    {
        $sum = 0;
        foreach ($this->jobCounters as $jobCounter) {
            $sum += $jobCounter();
        }
        return $sum;
    }

It works, but with one little bug #1115