Queues. "memory" parameter is not working as expected.

[guram-vashakidze]

Laravel Version

10.48.28

PHP Version

8.2

Database Driver & Version

No response

Description

Hello,

I had to implement my own internal queue handler to replace queue:work. In my handler, I use ini_set to set a memory limit. However, I encountered an issue where some tasks stopped executing due to memory restrictions. At the same time, the configured memory limit did not change. For queue:work I used the --memory=512 parameter.
As a result, I decided to check the logs in Kibana (where I also record memory_get_peak_usage), and it turned out that many tasks exceeded the specified limits by a significant margin.
I would be interested to know the purpose of the --memory parameter in Laravel's queue handler. I assumed it was a memory limit for the tasks, wasn't it?

Steps To Reproduce

1. Setup simple job, with memory overloading. For example:

while(true) {
    $arr[] = mt_rand();
}

2. Set this job into queue
3. Run queue handler with memory parameter. For example: php artisan queue:work --queue=<your-queue-name> --memory=50
4. See how your memory overloaded

[macropay-solutions]

\Illuminate\Queue\Console\WorkCommand

{--memory=128 : The memory limit in megabytes}

\Illuminate\Queue\WorkerOptions

    /**
     * The maximum amount of RAM the worker may consume.
     *
     * @var int
     */
    public $memory;

\Illuminate\Queue\Worker::stopIfNecessary

    protected function stopIfNecessary(
        WorkerOptions $options,
        $lastRestart,
        $startTime = 0,
        $jobsProcessed = 0,
        $job = null
    ) {
        return match (true) {
            $this->shouldQuit => static::EXIT_SUCCESS,
            $this->memoryExceeded($options->memory) => static::EXIT_MEMORY_LIMIT, // this
            $this->queueShouldRestart($lastRestart) => static::EXIT_SUCCESS,
            $options->stopWhenEmpty && is_null($job) => static::EXIT_SUCCESS,
            $options->maxTime && hrtime(true) / 1e9 - $startTime >= $options->maxTime => static::EXIT_SUCCESS,
            $options->maxJobs && $jobsProcessed >= $options->maxJobs => static::EXIT_SUCCESS,
            default => null
        };
    }

    /**
     * Determine if the memory limit has been exceeded.
     *
     * @param int $memoryLimit
     * @return bool
     */
    public function memoryExceeded($memoryLimit)
    {
        return (memory_get_usage(true) / 1024 / 1024) >= $memoryLimit;
    }

/**
 * Returns the amount of memory allocated to PHP
 * @link https://php.net/manual/en/function.memory-get-usage.php
 * @param bool $real_usage [optional] <p>
 * Set this to true to get the real size of memory allocated from
 * system. If not set or false only the memory used by
 * emalloc() is reported.
 * </p>
 * @return int the memory amount in bytes.
 */
#[Pure(true)]
function memory_get_usage(bool $real_usage = false): int {}

[macropay-solutions]

Pls update to 10.48.29 to fix a vulnerability with the validator for files.

[guram-vashakidze]

I don’t quite understand... what is the purpose of this parameter? Its definition says, "The memory limit in megabytes," yet the tasks executed within the worker process are not restricted by memory at all. In other words, the worker can consume an unlimited (by php.ini config) amount of memory while processing tasks.

[macropay-solutions]

@guram-vashakidze stopIfnecessary is called only after the job is processed.

    /**
     * Listen to the given queue in a loop.
     *
     * @param string $connectionName
     * @param string $queue
     * @param \Illuminate\Queue\WorkerOptions $options
     * @return int
     */
    public function daemon($connectionName, $queue, WorkerOptions $options)
    {
        if ($supportsAsyncSignals = $this->supportsAsyncSignals()) {
            $this->listenForSignals();
        }

        $lastRestart = $this->getTimestampOfLastQueueRestart();

        [$startTime, $jobsProcessed] = [hrtime(true) / 1e9, 0];

        while (true) {
            // Before reserving any jobs, we will make sure this queue is not paused and
            // if it is we will just pause this worker for a given amount of time and
            // make sure we do not need to kill this worker process off completely.
            if (!$this->daemonShouldRun($options, $connectionName, $queue)) {
                $status = $this->pauseWorker($options, $lastRestart);

                if (!is_null($status)) {
                    return $this->stop($status, $options);
                }

                continue;
            }

            if (isset($this->resetScope)) {
                ($this->resetScope)();
            }

            // First, we will attempt to get the next job off of the queue. We will also
            // register the timeout handler and reset the alarm for this job so it is
            // not stuck in a frozen state forever. Then, we can fire off this job.
            $job = $this->getNextJob(
                $this->manager->connection($connectionName),
                $queue
            );

            if ($supportsAsyncSignals) {
                $this->registerTimeoutHandler($job, $options);
            }

            // If the daemon should run (not in maintenance mode, etc.), then we can run
            // fire off this job for processing. Otherwise, we will need to sleep the
            // worker so no more jobs are processed until they should be processed.
            if ($job) {
                $jobsProcessed++;

                $this->runJob($job, $connectionName, $options); // ---------here the job is ran.

                if ($options->rest > 0) {
                    $this->sleep($options->rest);
                }
            } else {
                $this->sleep($options->sleep);
            }

            if ($supportsAsyncSignals) {
                $this->resetTimeoutHandler();
            }

            // Finally, we will check to see if we have exceeded our memory limits or if
            // the queue should restart based on other indications. If so, we'll stop
            // this worker and let whatever is "monitoring" it restart the process.
            $status = $this->stopIfNecessary( // ----------------------------------------------------------here
                $options,
                $lastRestart,
                $startTime,
                $jobsProcessed,
                $job
            );

            if (!is_null($status)) {
                return $this->stop($status, $options);
            }
        }
    }

[macropay-solutions]

@guram-vashakidze maybe

ini_set('memory_limit', '128M'); should be used instead and leave php throw the error.
This should be done before runNextJob and daemon function call:

    protected function runWorker($connection, $queue)
    {
        $workerOptions = $this->gatherWorkerOptions();
        \ini_set('memory_limit', $workerOptions->memory . 'M');

        return $this->worker
            ->setName($workerOptions->name)
            ->setCache($this->cache)
            ->{$this->option('once') ? 'runNextJob' : 'daemon'}($connection, $queue, $workerOptions);
    }

[guram-vashakidze]

@macropay-solutions
Yep, I see that it works between jobs, and that was my question, what the purpose of this "limit" if it never catch reall memory "overusing"?
Yes, I did the same thing with ini_set and faced issue that some jobs stopped working as memory limit in Laravel workers has a bit different meaning that in parameter description, and didn't catch memory overloading🤷
Thank you for answering.

[macropay-solutions]

Thank you for posting this. We will fix this also for us.

[guram-vashakidze]

Just sharing ideas: using ini_set gives ability to setup custom memory limit for specific jobs.

[macropay-solutions]

@guram-vashakidze
with this fix you should call the command like:
php artisan queue:work || echo "error..."
This will prevent the parent script that initiated the command to exit with error.

[guram-vashakidze]

@macropay-solutions thank you, i already using my own queue handler. I decided to figure out in this memory limit usage.