PHP Fatal error: Uncaught Amp\Parallel\Worker\TaskFailureError: Uncaught Error in worker with message "User-defined functions must be autoloadable"

[iamsyh]

Hi,

I'm trying to use a custom class in a worker but it doesn't seem to be working. The below class is already auto-loaded using composer. Please help me out with this issue. My code is below:

Class

namespace Jobs;

require '/home/xxx/vendor/autoload.php';

use Amp\Parallel\Worker\Environment;
use Amp\Parallel\Worker\Task;

class GetMarketJob implements Task {
    /**
     * @var callable
     */
    private $function;
    /**
     * @var mixed[]
     */
    private $args;

    public function __construct($function, ...$args) {
        $this->function = $function;
        $this->args = $args;
    }

    /**
     * {@inheritdoc}
     */
    public function run(Environment $environment)
    {
        if ($this->function instanceof \__PHP_Incomplete_Class) {
            throw new \Error('When using a class instance as a callable, the class must be autoloadable');
        }

        if (\is_array($this->function) && ($this->function[0] ?? null) instanceof \__PHP_Incomplete_Class) {
            throw new \Error('When using a class instance method as a callable, the class must be autoloadable');
        }

        if (!\is_callable($this->function)) {
            $message = 'User-defined functions must be autoloadable (that is, defined in a file autoloaded by composer)';
            if (\is_string($this->function)) {
                $message .= \sprintf("; unable to load function '%s'", $this->function);
            }

            throw new \Error($message);
        }

        return ($this->function)(...$this->args);
    }

    public function testMe($url = NULL) {
        $test = file_get_contents($url);
        return $test;
    }    
}

File calling the worker/class:

require '/home/xxxx/vendor/autoload.php';

use Jobs\GetMarketJob;
// Example async producer using promisor

use Amp\Parallel\Worker;
use Amp\Promise;
use Amp\Loop;
use Amp\Parallel\Worker\DefaultPool;
use Amp\Parallel\Worker\Task;
use Amp\Parallel\Worker\Environment;
use Amp\Parallel\Worker\TaskFailureError;
use Amp\Parallel\Worker\DefaultWorkerFactory;

Amp\Loop::run(function () {
    $factory = new DefaultWorkerFactory();

    $worker = $factory->create();

    $result = yield $worker->enqueue(new GetMarketJob('testMe', ['https://github.com']));

    print($result);

    $code = yield $worker->shutdown();
    \printf("Code: %d\n", $code);
});

[kelunik]

How does you composer.json look like?

[iamsyh]

How does you composer.json look like?

{
    "require": {
        "phpmailer/phpmailer": "^6.2",
        "phpoffice/phpspreadsheet": "^1.16",
        "verot/class.upload.php": "^2.1",
        "elasticsearch/elasticsearch": "^7.12",
        "amphp/amp": "^2.5",
        "amphp/parallel": "^1.4"
    },
    "autoload": {
        "psr-4": {
            "Jobs\\": "public_html/scripts/jobs/"
        }
    }
}

[iamsyh]

Okay for some reason, updating my class to the below code made it work!

namespace Jobs;

require '/home/xxx/vendor/autoload.php';

use Amp\Parallel\Worker\Environment;
use Amp\Parallel\Worker\Task;

class GetMarketJob implements Task {
    /**
     * @var callable
     */
    private $function;
    /**
     * @var mixed[]
     */
    private $args;

    public function __construct($function, ...$args) {
        $this->function = $function;
        $this->args = $args;
    }

    /**
     * {@inheritdoc}
     */
    public function run(Environment $environment)
    {
      $function = $this->function;

        return $this->$function($this->args);
    }

    public function testMe($url = NULL) {
        $test = file_get_contents($url);
        return $test;
    }    
}

For some reason having __construct(callable $function) (with the explicit callable type) wasn't working and also the run function having several is_callable conditions didn't work.

[kelunik]

Yeah, makes sense. If you just pass the method name, it previously tried to execute the testMe function, not the method. I guess we can close this issue now?

[iamsyh]

Yeah, makes sense. If you just pass the method name, it previously tried to execute the testMe function, not the method. I guess we can close this issue now?

yes please. although I do have another question; can you please tell me if amphp automatically handles workers in the pool? like what happens when a worker is done with a task? does amphp reuse it or does it just sit idle?

and is it safe to shutdown idle workers manually from within a loop to free up server resources?

Thank you so much for this awesome library!!

[iamsyh]

I'm also having another issue where when a script containing amp code is run via cron jobs, it fails to run properly. I retrieved the below error code from error.log:

`PHP Fatal error: Uncaught Amp\TimeoutException: Operation timed out in /home/xxxx/vendor/amphp/amp/lib/functions.php:275 Stack trace:

0 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(96): Amp\Promise{closure}('s', NULL)

1 [internal function]: Amp\Loop\EvDriver->Amp\Loop{closure}(Object(EvTimer))

2 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(234): EvLoop->run(2)

3 /home/xxxx/vendor/amphp/amp/lib/Loop/Driver.php(138): Amp\Loop\EvDriver->dispatch(true)

4 /home/xxxx/vendor/amphp/amp/lib/Loop/Driver.php(72): Amp\Loop\Driver->tick()

5 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(186): Amp\Loop\Driver->run()

6 /home/xxxx/vendor/amphp/amp/lib/Loop.php(95): Amp\Loop\EvDriver->run()

7 /home/xxxx/public_html/scripts/testme.php(41): Amp\Loop::run(Object(Closure))

8 {main}`

whereas it works fine if run from terminal. the exact same command

cd /home/xxxx/public_html/scripts;/usr/bin/php testme.php

[iamsyh]

I'm also having another issue where when a script containing amp code is run via cron jobs, it fails to run properly. I retrieved the below error code from error.log:

PHP Fatal error: Uncaught Amp\TimeoutException: Operation timed out in /home/xxxx/vendor/amphp/amp/lib/functions.php:275 Stack trace: #0 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(96): Amp\Promise\{closure}('s', NULL) #1 [internal function]: Amp\Loop\EvDriver->Amp\Loop\{closure}(Object(EvTimer)) #2 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(234): EvLoop->run(2) #3 /home/xxxx/vendor/amphp/amp/lib/Loop/Driver.php(138): Amp\Loop\EvDriver->dispatch(true) #4 /home/xxxx/vendor/amphp/amp/lib/Loop/Driver.php(72): Amp\Loop\Driver->tick() #5 /home/xxxx/vendor/amphp/amp/lib/Loop/EvDriver.php(186): Amp\Loop\Driver->run() #6 /home/xxxx/vendor/amphp/amp/lib/Loop.php(95): Amp\Loop\EvDriver->run() #7 /home/xxxx/public_html/scripts/testme.php(41): Amp\Loop::run(Object(Closure)) #8 {main}

whereas it works fine if run from terminal. the exact same command

cd /home/xxxx/public_html/scripts;/usr/bin/php testme.php

Okay I've managed to fix this issue by replacing the following:

* * * * * cd /home/xxxx/public_html/scripts;/usr/bin/php testme.php

To this:

* * * * * cd /home/xxxx/public_html/scripts;/usr/local/bin/php testme.php

Turns out, there was an error.log file being stored inside:

vendor/amphp/parallel/lib/Context/Internal

Which had the following error:

[27-May-2021 02:51:01 UTC] PHP Fatal error:  Uncaught Error: Undefined constant 'STDIN' in /home/xxxx/vendor/amphp/parallel/lib/Context/Internal/process-runner.php:58
Stack trace:
#0 /home/xxxx/vendor/amphp/parallel/lib/Context/Internal/process-runner.php(123): Amp\Parallel\Context\Internal\{closure}()
#1 {main}
  thrown in /home/xxxx/vendor/amphp/parallel/lib/Context/Internal/process-runner.php on line 58

seems like STDIN was not being defined by the instance of php i was using for some reason. hope it helps anyone with a similar issue in the future! perhaps the same issue was occuring #112 here as well?

[kelunik]

Glad you got this sorted! Regarding your other question: Yes, when using the pool API, workers are automatically managed and reused.