amphp / parallel Goto Github PK

An advanced parallelization library for PHP, enabling efficient multitasking, optimizing resource use, and application responsiveness through multiple CPU threads.

License: MIT License

PHP 100.00%

concurrency multiprocessing php parallel-processing amphp parallel revolt

parallel's Issues

Improve error message if task is not autoloadable

See #26 (comment).

Parallel call on user defined function

Is there any way to call user defined function in parallel .

<?php
require \dirname(__DIR__) . '/vendor/autoload.php';
use Amp\Parallel\Worker;
use Amp\Promise;
$urls = [
    'https://secure.php.net',
    'https://amphp.org',
    'https://github.com',
];
$promises = [];
foreach ($urls as $url) {
    $promises[$url] = Worker\enqueueCallable('file_get_contents', $url);
}
$responses = Promise\wait(Promise\all($promises));
foreach ($responses as $url => $response) {
    \printf("Read %d bytes from %s\n", \strlen($response), $url);
}

Like below

<?php

require 'vendor/autoload.php';

use Amp\Parallel\Worker;
use Amp\Promise;
$urls = [
    'https://secure.php.net',
    'https://amphp.org',
    'https://github.com',
];

$promises = [];
foreach ($urls as $url) {

    $promises[$url] = Worker\enqueueCallable('call_user_func','_subFunction');
}

$responses = Promise\wait(Promise\all($promises));

foreach ($responses as $url => $response) {
    // Cumulated response
}

function _subFunction()
{
	// User defined function
}

Custom Environment

I would like to share non-serialisable objects across workers/threads. What is the recommended way to do so? Do I need to create my own /bin/worker, WorkerProcess, WorkerThread and WorkerFactory or is there a more simple way of creating a new (or populating existing) environment?

PosixSemaphore spend too many time on loop

parallel/lib/Sync/PosixSemaphore.php

Lines 131 to 145 in 0ef7441

    
           do { 
        
               // Attempt to acquire a lock from the semaphore. 
        
               if (@\msg_receive($this->queue, 0, $type, 1, $chr, false, \MSG_IPC_NOWAIT, $errno)) { 
        
                   // A free lock was found, so resolve with a lock object that can 
        
                   // be used to release the lock. 
        
                   return new Lock(function (Lock $lock) { 
        
                       $this->release(); 
        
                   }); 
        
               } 
        
               // Check for unusual errors. 
        
               if ($errno !== \MSG_ENOMSG) { 
        
                   throw new SemaphoreException('Failed to acquire a lock.'); 
        
               } 
        
           } while (yield new Delayed(self::LATENCY_TIMEOUT, true));

Those code is trying to acquire lock at loop.

If we have 1000 coroutine try to acquire, there are 1000 coroutine scheduled by Amp\Loop.

For example(with uv extension):

Loop::run(function () {
    $client = new DefaultClient();

    for ($i = 0; $i < 1000; $i++) {
        $resp = yield $client->request('http://localhost:4444/');
        unset($resp);
    }
});

real	0m1.443s
user	0m0.971s
sys	0m0.056s

Loop::run(function () {
    $s = new PosixSemaphore(2);
    $client = new DefaultClient();

    $coroutines = [];
    for ($i = 0; $i < 1000; $i++) {
        $coroutines = new Coroutine(request($client, $s));
    }
    yield $coroutines;
});

function request($client, &$s)
{
    $lock = yield $s->acquire();
    yield $client->request('http://localhost:4444/');
    $lock->release();
}

real	0m19.484s
user	0m14.544s
sys	0m0.354s

Don't forward exception codes

This is a tracking issue for v0.2, so we don't forget about it. Exception codes shouldn't be passed to the new exception's constructor, but only be in the previous exception.

See #19.

Multicore support

Currently I am developing something like master -> workers.

Why in aerys you use proc_open for spawn workers, and not Thread ?
Is Thread using multicore on spawn ?
Why aerys don't use amphp/process or amphp/parallel for [master -> workers]
Why from amphp/parallel > v0.2 was removed fork() support ?
Is there, may be, advantages to use proc_open vs fork ?

How to cancel task by timeout

Hi. How can i cancel task by timeout ?
Currently i use this way:

			$worker = $this->pool->getWorker();
			$promise = $worker->enqueue(new Executor($job));
			$token = new TimeoutCancellationToken(60000);
			$token->subscribe([$worker, 'cancel']);

Will it work as exoected ?

Context apparently died but still running?

Hi,

I'm using amp and parallel from about 1 year ago with no issues at all. Recently, I have updated both using Composer and now, from time to time, I'm getting this error:

[2018-05-04 08:30:04] app.CRITICAL: Error sending ID 4833. Error: The channel closed unexpectedly. Did the context die? {"script":"Watcher.php"} []

The weird thing is that according to my log's the thread is still running:

[2018-05-04 08:30:04] app.INFO: Processing ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:30:04] app.CRITICAL: Error sending ID 4833. Error: The channel closed unexpectedly. Did the context die? {"script":"Watcher.php"} []
----- AT THIS POINT THE THREAD SHOULD BE DEAD, BUT I'M GETTING LOGS FROM IT ----
[2018-05-04 08:32:45] app.INFO: Sending batch 0 to 1000 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:32:50] app.INFO: Batch 0 sent ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:33:51] app.INFO: Sending batch 1 to 1000 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:33:56] app.INFO: Batch sent 1 ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:19] app.INFO: Sending batch 2 to 1000 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:26] app.INFO: Batch sent 2 ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:29] app.INFO: Sending batch 3 to 1000 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:38] app.INFO: Batch sent 3 ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:40] app.INFO: Sending batch 4 to 1000 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:48] app.INFO: Batch sent 4 ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:50] app.INFO: Sending batch 5 to 278 emails ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:52] app.INFO: Batch sent 5 ID 4833 {"script":"Watcher.php"} []
[2018-05-04 08:34:52] app.INFO: Invalid emails ID 4833: 0 | Array ( )  {"script":"Watcher.php"} []
[2018-05-04 08:34:52] app.INFO:  Done ID 4833 | Emails sent: 5278 | Final state: sent {"script":"Watcher.php"} []

Hope you can help me, thanks!

Possible mistake that keeps loop running

parallel/bin/worker

Line 45 in be0d786

return; // Parent context died, simply exit.

@trowski, seems like that should actually be exit; or Loop::stop()?

Type error

Got this error recently:

Argument 3 passed to Amp\Parallel\Worker\TaskException::__construct() 
must be of the type integer, string given, 
called in .../vendor/amphp/parallel/lib/Worker/Internal/TaskFailure.php on line 54

This happens because PDOException::getCode() can return a string.

Error using $output from Symfony Command

Hi, thanks for your effort and sharing this library.

I have optimized some of our Symfony CLI apps with this parallels extension. So far all works, but I can't use the Symfony Output anymore.

Simple example for a task:

<?php

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

class MyTask implements Task {

  private $output;

  public function __construct($output) {
    $this->output = $output;
    $this->output->writeln("OUTPUT INSIDE CONSTRUCT WORKS");
  }

  public function run(Environment $environment) {
    echo "THIS WORKS\n";
    $this->output->writeln("OUTPUT INSIDE RUN BREAKS");
    $sleep = rand(1, 3);
    sleep($sleep); // Blocking call in thread.
    return 0
  }

}

And my Symfony Command looks like (I tried to simplify as much as possible):

<?php

use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Formatter\OutputFormatterStyle;

use Amp\Loop;
use Amp\Parallel\Worker\DefaultPool;

use MyTask;

class MyCommand extends Command {

  protected function configure(){
    $this->setName("Process:Amp")
         ->setDescription("This will run some parallel processes.")
         ->addArgument('count', InputArgument::REQUIRED, 'Number of concurrent pool processes.');
  }

  protected function execute(InputInterface $input, OutputInterface $output){

    $output->writeln('START COMMAND ...');

    // We can first define tasks and then run them
    $tasks = [
      new MyTask($output),
      new MyTask($output),
      new MyTask($output),
      new MyTask($output),
      new MyTask($output),
    ];

    try {

      // get parameters
      $cnt_parallels = $input->getArgument('count');

      $pool = new DefaultPool($cnt_parallels);

      foreach ($tasks as $task) {
        $pool->enqueue($task);
      }

      // Event loop for parallel tasks
      Loop::run(function () use (&$pool) {
        return $pool->shutdown();
      });

    }
    finally {

      $output->writeln('WELL, COMMAND Finally reached!');

    }

  }

}

The above will immediately stop but not shown any exception.

If changing the loop code with Coroutines then following exception is shown and formatted via Symfony:

In TaskFailure.php line 55:

Uncaught Symfony\Component\Console\Exception\RuntimeException in worker with message "Unable to write output." and code "0"

Thanks for some hints or solution in advance
Cheers, Tom

Amp\Parallel\Test\Forking\ForkTest::testExitingContextOnSend hanging

Amp\Parallel\Test\Forking\ForkTest::testExitingContextOnSend hangs with 100% CPU when UV is enabled.

Support delayed port access (e.g. due to firewalls)

@kelunik told me to open a ticket here.

Please open an issue at https://github.com/amphp/parallel for the infinite loop and https://github.com/amphp/process for the ports if you want.

its quite a while but as the problem with box was easy solvable by using the --no-parallel option i simply forgot until i hit the problem again with fink.

i am not sure where to write which info, it feels i would post the same text in parallel and process as i haven't used amphp directly but only tools which use it, i don't know the internals.

if i run fink it simply gets stuck forever on a windows system with a firewall.

php fink.phar https://getcomposer.org

if my firewall is in learning mode, it ask me to allow ampXXXX.tmp a local connection.

after allowing it, the connection monitor show the following:

this 3 php and amp processes stay there forever in an infinite loop and are doing nothing (no more as the initial 22 bytes are sent or received).

if i run the firewall in allow all mode (which sadly makes no sense at all), it works without a problem.

so it looks like amphp has a problem if the port is not instantly accessible but delayed by a firewall learning mode dialog.
as the amp process names are randomly generated it is not possible to precreate the firewall rules.
having some kind of check if ports can be accessed and if not some kind of fallback could be a solution. a slow synchronous processing is still faster as a infinite do nothing processing :)

Development tags

Would there be any chance this library (and its amp dependencies) could be tagged following semver guidelines? The only way I can safely add this library without worrying about BC breaking changes is to manually specify a commit, and an minimum-stability of dev.

    "minimum-stability": "dev",
    "prefer-stable": true,
    "require" : {
        "amphp/loop": "dev-master#302a5c714c2700d32ca9444be09f21b2835b213f",
        "amphp/amp": "2.0.x-dev#72378e2b82d9f1dee6b93de2bee8aa138e7f67e1",
        "amphp/parallel": "0.1.x-dev#9cc70c1f86bf8ffa8ca744e40702502410b733f0",
        "async-interop/promise": "^0.4"
    }

This is the only configuration I could find which works for me and is safe. Any suggestions as to improve this would be welcome.

Watch memory and restart workers

Consider to implement the "watch and restart workers" feature described in https://github.com/amphp/aerys/issues/135.

Remove current changelog

The current changelog seems to be still from Icicle and is now intermixed, probably because search & replace.

Class not found

I run codes like this:

use Amp\Parallel\Worker;
use Amp\Promise;
use common\services\internal\TigerApi;

        $work = Worker\enqueueCallable([TigerApi::class, 'getPoliciesTitle']);
        $promises = ['policy' => $work];
        $result = Promise\wait(Promise\all($promises));

The output: Uncaught Error in worker with message \"Class 'common\\services\\internal\\TigerApi' not found\" and code \"0\"

How can I add the class to autoload?
I used Yii2.0 and I tried the method: #55, but it appeared yii is not found.

I know that yii has it's owner autoload class, how can I let this package use the autoloader of yii?

Cannot acquire lock?

There is only release method, haven't acquire method?

Make classes autoloadable

Hi there,

When implementing the a Task I get this error message:

Uncaught Error in worker with message "Classes implementing Amp\Parallel\Worker\Task must be autoloadable by the Composer autoloader" and code "0"

<?php

namespace Jobs;

use Phalcon\Mvc\Controller;
use Amp\Parallel\Worker\Environment;
use Amp\Parallel\Worker\Task;

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

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

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

    public function getArgs() {
        return $this->args;
    }    
}

How do I solve this?

Debug mode

Moving my question from box-project/box#53 (comment) to here.

I think it would be good if it would be possible to easily switch from parallelMap() to a plain old array_map(), be it for debugging purposes or simply because in a special case parallelMap() is not working e.g. when the passed argument is not serializable.

Maybe an env variable would work, the inconvenient being it's gonna rely on global state, the benefit is would be easier and won't require much changes.

On the same token, the major downside I have with this library is the inability to use break points for the paralellised work. Would make it work?

If you have some insight regarding this that would be greatly appreciated, otherwise I'll try to give it a stab in the coming months :)

Using object and its dependencies vs problem with serialization

I was trying out worker pool example. It works very well when using simple things, like php function or simple anonymous function without binding them with any existing object. Could you please tell if while using parallel it's possible to work with objects and its dependencies?

There is always problem with serialization. I have read the note about opis/closure library and restrictions about serializing. I was trying to wrap my function call in new SerializableClosure(), but then I get exceptions as following:

[Amp\Parallel\Context\ContextException]                                           
  The context stopped responding, potentially due to a fatal error or calling exit  
                                                                                    
  [Amp\Parallel\Sync\ChannelException]                   
  The channel closed unexpectedly. Did the context die?

Here there is an example of what I am trying to execute:

class Something {

public function test($item)
    {
        var_dump($item);
    }


public function execute() {

            $tasks = [
                 new BlockingTask(new SerializableClosure(function ($item) {
                    $this->test($item);
                } ), 5)
            ]
           

            Loop::run(function () use (&$results, &$tasks) {
            $timer = Loop::repeat(200, function () {
                printf(".");
            });
            Loop::unreference($timer);
            $pool = new DefaultPool;
            $coroutines = [];
            foreach ($tasks as $task) {
                $coroutines[] = function () use ($pool, $task, &$results) {
                    $result = yield $pool->enqueue($task);
                    $url = $task->getArgs()[0];
                    //printf("\nRead from %s: %d bytes\n", $url, strlen($result));
                    $results[$url] = $result;
                };
            }
            $coroutines = array_map(function (callable $coroutine): Coroutine {
                return new Coroutine($coroutine());
            }, $coroutines);
            yield \Amp\Promise\all($coroutines);
            return yield $pool->shutdown();
        });
        echo "\nResult array keys:\n";
        echo var_export(array_keys($results), true);
    }
}

Of course tricky part is that with building new BlockingTask(). Is it generally possible - I mean pass as callable to parallel concrete class method and run it successfully?

BasicEnvironment leaks watcher

This is due to a missing __destruct + the cyclic reference in the loop watcher. I have fixed this in my local ext-async branch, but not in master yet.

PooledWorker::__destroy cause fatal error

How to reproduce it:

getWorker from pool
Kill or shutdown it.
Get all other workers from pool.
Trigger gc to work.

Because worker is not running amp will detach it from SplObjectHash. And if polledworker return worker to hash then:

PHP Fatal error:  Uncaught UnexpectedValueException: Object not found in /app/vendor/amphp/parallel/lib/Worker/DefaultPool.php:73
Stack trace:
#0 /app/vendor/amphp/parallel/lib/Worker/DefaultPool.php(73): SplObjectStorage->offsetGet(Object(Amp\Parallel\Worker\WorkerProcess))
#1 /app/vendor/amphp/parallel/lib/Worker/Internal/PooledWorker.php(33): Amp\Parallel\Worker\DefaultPool::Amp\Parallel\Worker\{closure}(Object(Amp\Parallel\Worker\WorkerProcess))
#2 [internal function]: Amp\Parallel\Worker\Internal\PooledWorker->__destruct()
#3 {main}
  thrown in /app/vendor/amphp/parallel/lib/Worker/DefaultPool.php on line 73

Multiple errors encountered {"exception":"[object] (Amp\\MultiReasonException(code: 0)

I don't know what's the problem. This is the full log:

[2019-02-25 16:07:17] local.ERROR: Multiple errors encountered {"exception":"[object] (Amp\\MultiReasonException(code: 0): Multiple errors encountered at /var/www/vendor/amphp/parallel-functions/src/functions.php:56)
[stacktrace]
#0 [internal function]: Amp\\ParallelFunctions\\{closure}()
#1 /var/www/vendor/amphp/amp/lib/Coroutine.php(76): Generator->send(Array)
#2 /var/www/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, Array)
#3 /var/www/vendor/amphp/amp/lib/Deferred.php(45): class@anonymous->resolve(Array)
#4 /var/www/vendor/amphp/amp/lib/functions.php(466): Amp\\Deferred->resolve(Array)
#5 /var/www/vendor/amphp/amp/lib/Internal/ResolutionQueue.php(55): Amp\\Promise\\{closure}(Object(Amp\\Parallel\\Worker\\TaskError), NULL)
#6 /var/www/vendor/amphp/amp/lib/Failure.php(29): Amp\\Internal\\ResolutionQueue->__invoke(Object(Amp\\Parallel\\Worker\\TaskError), NULL)
#7 /var/www/vendor/amphp/amp/lib/Internal/Placeholder.php(125): Amp\\Failure->onResolve(Object(Amp\\Internal\\ResolutionQueue))
#8 /var/www/vendor/amphp/amp/lib/Coroutine.php(81): Amp\\Coroutine->resolve(Object(Amp\\Failure))
#9 /var/www/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))
#10 /var/www/vendor/amphp/amp/lib/Coroutine.php(81): Amp\\Coroutine->resolve(Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))
#11 /var/www/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))
#12 /var/www/vendor/amphp/amp/lib/Coroutine.php(81): Amp\\Coroutine->resolve(Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))
#13 /var/www/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, '\\x00\\x03\\x13\\x00\\x00O:40:\"Amp\\\\...')
#14 /var/www/vendor/amphp/amp/lib/Deferred.php(45): class@anonymous->resolve('\\x00\\x03\\x13\\x00\\x00O:40:\"Amp\\\\...')
#15 /var/www/vendor/amphp/byte-stream/lib/ResourceInputStream.php(99): Amp\\Deferred->resolve('\\x00\\x03\\x13\\x00\\x00O:40:\"Amp\\\\...')
#16 /var/www/vendor/amphp/amp/lib/Loop/NativeDriver.php(206): Amp\\ByteStream\\ResourceInputStream::Amp\\ByteStream\\{closure}('bg', Resource id #546, NULL)
#17 /var/www/vendor/amphp/amp/lib/Loop/NativeDriver.php(97): Amp\\Loop\\NativeDriver->selectStreams(Array, Array, -0.001)
#18 /var/www/vendor/amphp/amp/lib/Loop/Driver.php(134): Amp\\Loop\\NativeDriver->dispatch(true)
#19 /var/www/vendor/amphp/amp/lib/Loop/Driver.php(72): Amp\\Loop\\Driver->tick()
#20 /var/www/vendor/amphp/amp/lib/Loop.php(84): Amp\\Loop\\Driver->run()
#21 /var/www/vendor/amphp/amp/lib/functions.php(170): Amp\\Loop::run(Object(Closure))
#22 /var/www/vendor/spatie/laravel-collection-macros/src/macros/parallelMap.php(31): Amp\\Promise\\wait(Object(Amp\\Coroutine))
#23 [internal function]: Illuminate\\Database\\Eloquent\\Collection->{closure}(Object(Closure))
#24 /var/www/vendor/laravel/framework/src/Illuminate/Support/Traits/Macroable.php(107): call_user_func_array(Object(Closure), Array)
#25 /var/www/app/Http/Controllers/EdgesController.php(204): Illuminate\\Support\\Collection->__call('parallelMap', Array)
#26 [internal function]: App\\Http\\Controllers\\EdgesController->path(Object(App\\Http\\Requests\\PathRequest))
#27 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Controller.php(54): call_user_func_array(Array, Array)
#28 /var/www/vendor/laravel/framework/src/Illuminate/Routing/ControllerDispatcher.php(45): Illuminate\\Routing\\Controller->callAction('path', Array)
#29 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Route.php(212): Illuminate\\Routing\\ControllerDispatcher->dispatch(Object(Illuminate\\Routing\\Route), Object(App\\Http\\Controllers\\EdgesController), 'path')
#30 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Route.php(169): Illuminate\\Routing\\Route->runController()
#31 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Router.php(665): Illuminate\\Routing\\Route->run()
#32 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(30): Illuminate\\Routing\\Router->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#33 /var/www/vendor/barryvdh/laravel-cors/src/HandleCors.php(36): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#34 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Barryvdh\\Cors\\HandleCors->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#35 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#36 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Middleware/SubstituteBindings.php(41): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#37 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Illuminate\\Routing\\Middleware\\SubstituteBindings->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#38 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#39 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(104): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#40 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Router.php(667): Illuminate\\Pipeline\\Pipeline->then(Object(Closure))
#41 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Router.php(642): Illuminate\\Routing\\Router->runRouteWithinStack(Object(Illuminate\\Routing\\Route), Object(Illuminate\\Http\\Request))
#42 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Router.php(608): Illuminate\\Routing\\Router->runRoute(Object(Illuminate\\Http\\Request), Object(Illuminate\\Routing\\Route))
#43 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Router.php(597): Illuminate\\Routing\\Router->dispatchToRoute(Object(Illuminate\\Http\\Request))
#44 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Kernel.php(176): Illuminate\\Routing\\Router->dispatch(Object(Illuminate\\Http\\Request))
#45 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(30): Illuminate\\Foundation\\Http\\Kernel->Illuminate\\Foundation\\Http\\{closure}(Object(Illuminate\\Http\\Request))
#46 /var/www/vendor/fideloper/proxy/src/TrustProxies.php(57): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#47 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Fideloper\\Proxy\\TrustProxies->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#48 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#49 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Middleware/TransformsRequest.php(31): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#50 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Illuminate\\Foundation\\Http\\Middleware\\TransformsRequest->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#51 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#52 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Middleware/TransformsRequest.php(31): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#53 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Illuminate\\Foundation\\Http\\Middleware\\TransformsRequest->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#54 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#55 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Middleware/ValidatePostSize.php(27): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#56 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Illuminate\\Foundation\\Http\\Middleware\\ValidatePostSize->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#57 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#58 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Middleware/CheckForMaintenanceMode.php(62): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#59 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Illuminate\\Foundation\\Http\\Middleware\\CheckForMaintenanceMode->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#60 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#61 /var/www/vendor/barryvdh/laravel-cors/src/HandlePreflight.php(29): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#62 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(151): Barryvdh\\Cors\\HandlePreflight->handle(Object(Illuminate\\Http\\Request), Object(Closure))
#63 /var/www/vendor/laravel/framework/src/Illuminate/Routing/Pipeline.php(53): Illuminate\\Pipeline\\Pipeline->Illuminate\\Pipeline\\{closure}(Object(Illuminate\\Http\\Request))
#64 /var/www/vendor/laravel/framework/src/Illuminate/Pipeline/Pipeline.php(104): Illuminate\\Routing\\Pipeline->Illuminate\\Routing\\{closure}(Object(Illuminate\\Http\\Request))
#65 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Kernel.php(151): Illuminate\\Pipeline\\Pipeline->then(Object(Closure))
#66 /var/www/vendor/laravel/framework/src/Illuminate/Foundation/Http/Kernel.php(116): Illuminate\\Foundation\\Http\\Kernel->sendRequestThroughRouter(Object(Illuminate\\Http\\Request))
#67 /var/www/public/index.php(55): Illuminate\\Foundation\\Http\\Kernel->handle(Object(Illuminate\\Http\\Request))
#68 {main}
"}

Amp\Parallel\ChannelException: The socket unexpectedly closed

This excpetion is throw ± 1 out of 10 times when running this test. The test itself will take a path of an image, scales it down and saves it about 10 times. Each downscaling is a separate Fork.

This is the testcode.

    public function test_async() {
        $url = 'img/image.jpeg';
        $factory = new ResponsiveFactory(new DefaultConfigurator([
            'publicPath'   => $this->publicPath,
            'engine'       => 'gd',
            'stepModifier' => 0.5,
            'scaler'       => 'filesize',
            'enableCache'  => false,
            'async'        => true,
        ]));

        $responsiveImage = $factory->create($url);

        $this->assertTrue(count($responsiveImage->getSrcset()) > 1);
        $this->assertEquals("/{$url}", $responsiveImage->src());

        $testCase = $this;
        $responsiveImage->onSaved(function () use ($testCase, $responsiveImage) {
            $fs = new Filesystem();

            foreach ($responsiveImage->getSrcset() as $src) {
                $src = trim($src, '/');

                $testCase->assertTrue($fs->exists("{$testCase->publicPath}/{$src}"));
            }
        });

        \Amp\wait($responsiveImage->getMainPromise());
    }

And this is the exception trace.

PHPUnit 5.7.14 by Sebastian Bergmann and contributors.

E                                                                   1 / 1 (100%)

Time: 673 ms, Memory: 30.00MB

There was 1 error:

1) Brendt\Image\Tests\Phpunit\ResponsiveFactoryTest::test_async
Amp\Parallel\ChannelException: The socket unexpectedly closed

/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/parallel/lib/Sync/ChannelledSocket.php:91
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/loop/lib/NativeLoop.php:113
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/loop/lib/NativeLoop.php:42
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/loop/lib/Loop.php:96
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/loop/lib/Loop.php:47
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/async-interop/event-loop/src/Loop.php:78
/Users/brent1/Documents/sites/brendt/responsive-image/vendor/amphp/amp/lib/functions.php:116
/Users/brent1/Documents/sites/brendt/responsive-image/tests/phpunit/ResponsiveFactoryTest.php:164

The code for downscaling images:

$factory = $this;

$promise = Fork::spawn(function () use ($factory, $imageObject, $width, $height, $scaledFilePath) {
    $scaledImage = $imageObject->resize((int) $width, (int) $height)->encode($imageObject->extension);
    $factory->saveImageFile($scaledFilePath, $scaledImage);
})->join();

return $promise;

All promises are combined into one using \Amp\all().

PHP Fatal error: Could not send result to parent

Hi,

While experimenting a bit I am seeing the following error in my logs:

 PHP Fatal error:  Could not send result to parent in /socket/vendor/amphp/parallel/lib/Context/Internal/process-runner.php on line 105

I've done some debugging and the root cause is because the child process process-runner.php is not stopped gracefully. When the loop ends it's forcefully killed via Runner->kill() which causes an Exception and then the error the triggered.

It's reproducible by using the simple http client example script. A simple setup to reproduce using Docker:

Run:

docker run --rm -it php:7.2 bash -l

And then just paste:

   apt-get update \
&& apt-get install --no-install-recommends -y git unzip \
&& curl -sS https://getcomposer.org/installer | php -- --install-dir=/usr/local/bin/ --filename=composer \
&& git clone https://github.com/amphp/socket.git \
&& cd socket \
&& composer install --no-dev \
&& echo '[PHP]
error_log=/proc/1/fd/2
' > /usr/local/etc/php/php.ini \
&& (php -S localhost:80 &) \
&& touch index.php \
&& php examples/simple-http-client.php http://localhost/

That just creates a clone of this repository, sets the PHP error log to the stdout of PID 1 so the error becomes visible and then launches the example against a local web server started via PHP.

Any idea why the child process isn't stopped properly?

The channel is has been closed When we use more that 1 parallel process

Hello. I tried to use this code for getting some data from url array

Loop::run(function ()  {
    $uris = [
        'http://local.test/sleep.php?sec=1',
        'http://local.test/sleep.php?sec=5',
    ];

    $threads = [];
    foreach ($uris as $key => $uri) {

        $threads[] = Thread::spawn(function ($uri) {
            $result = file_get_contents($uri);
            return $result;
        }, $uri);
    }

    foreach ($threads as $thread) {
        $result[] = yield $thread->join();
    }
});

Code from sleep.php : <?php sleep($_GET['sec']);

And it is works correctly, but if i change the line order of urls from
'http://local.test/sleep.php?sec=1',
'http://local.test/sleep.php?sec=5',
to
'http://local.test/sleep.php?sec=5',
'http://local.test/sleep.php?sec=1',

Every time I've error:

Fatal error: Uncaught Amp\Parallel\Sync\ChannelException: The channel is has been closed in /home/anton/www/test/php/parallel/vendor/amphp/parallel/lib/Sync/ChannelledSocket.php:251

Could you tell me how I can solve such problem?

Call to a member function connection() on null with PHP 7.1

I was using this effectively until I upgraded to 7.1. I dont think thats an issue as I havent been able to find anyone else with it, but the error message is not descriptive so im not sure whats going on. With the basic code example:

$responses = wait(parallelMap([
      'https://google.com/',
      'https://github.com/',
      'https://stackoverflow.com/',
    ], function ($url) {
      return file_get_contents($url);
    }));

I get this error:

An uncaught Exception was encountered
Type: Amp\MultiReasonException

Message: Multiple errors encountered

Filename: /home/orgname/public_html/vendor/amphp/parallel-functions/src/functions.php

Line Number: 56

Backtrace:

File: /home/orgname/public_html/vendor/amphp/amp/lib/Coroutine.php
Line: 76
Function: send

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 130
Function: Amp\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Deferred.php
Line: 45
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/functions.php
Line: 466
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/ResolutionQueue.php
Line: 55
Function: Amp\Promise\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Failure.php
Line: 29
Function: __invoke

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 125
Function: onResolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 155
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Coroutine.php
Line: 95
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/Failure.php
Line: 29
Function: Amp\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 125
Function: onResolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 155
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Coroutine.php
Line: 95
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/Failure.php
Line: 29
Function: Amp\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 125
Function: onResolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 155
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Coroutine.php
Line: 95
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/Failure.php
Line: 29
Function: Amp\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 125
Function: onResolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 155
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Coroutine.php
Line: 95
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/Failure.php
Line: 29
Function: Amp\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 125
Function: onResolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Internal/Placeholder.php
Line: 155
Function: resolve

File: /home/orgname/public_html/vendor/amphp/amp/lib/Deferred.php
Line: 55
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/functions.php
Line: 214
Function: fail

File: /home/orgname/public_html/vendor/amphp/amp/lib/Loop/NativeDriver.php
Line: 130
Function: Amp\Promise\{closure}

File: /home/orgname/public_html/vendor/amphp/amp/lib/Loop/Driver.php
Line: 134
Function: dispatch

File: /home/orgname/public_html/vendor/amphp/amp/lib/Loop/Driver.php
Line: 72
Function: tick

File: /home/orgname/public_html/vendor/amphp/amp/lib/Loop.php
Line: 84
Function: run

File: /home/orgname/public_html/vendor/amphp/amp/lib/functions.php
Line: 170
Function: run

File: /home/orgname/public_html/application/controllers/Lists.php
Line: 63
Function: Amp\Promise\wait

File: /home/orgname/public_html/index.php
Line: 315
Function: require_once

if i print out out the error on line 56 of /home/orgname/public_html/vendor/amphp/parallel-functions/src/functions.php
i see a very long error, here's some of it:

any help would be appreciated!

Worker kill not worked

Hi. I wrote simple test code:

		$task = new \application\Test();
		\Amp\Loop::repeat(100, function($reference) use($task){
			\Amp\Loop::disable($reference);
			$pool = \Amp\Parallel\Worker\pool();
			echo "Get worker";
			$worker = $pool->getWorker();
			try {
				echo 'enqueue task';
				$promise = $worker->enqueue($task);
				$canceller = new \Amp\TimeoutCancellationToken(1000);
				$id = $canceller->subscribe(function() use($worker){
					echo "Killing";
					$worker->kill();
				});
				echo "Waiting";
				$result = yield $promise;
				echo "done";
				$canceller->unsubscribe($id);
			} catch(Throwable $exception) {
				echo $exception->getMessage();
			}
			echo 'Result is:' . $result;
			if(!$worker->isRunning()) {
				echo "Worker not running";
			}
			\Amp\Loop::enable($reference);
		});
		\Amp\Loop::run();

and Test is:

class Test implements Task
{
	/**
	 * @inheritDoc
	 */
	public function run(Environment $environment)
	{
		sleep(30);
		return 'ok';
	}

}

I expect what worker will be killed after 1 second and "enqueue" promise will fail. But instead of that promise succeeds after 30 seconds.

Running slower with amphp/parallel then when running code synchrounously

When running code in parallel the results say its running much slower, about 3 seconds slower than without. System is Ubuntu 16, PHP 7 FPM/FastCGI, amphp/parallel "v0.2.3" (reference 11a3e27), 4 cores, 16 GB of RAM.

Code samples are below, unfortunately to post what is inside these methods is just too much for a code snippet, but Foo() is mainly just waiting on an HTTP GET request and doing some light pre/post processing. Bar() is just waiting on a SOAP request and doing some light pre/post processing as well.

Without parallel about 8 seconds

if (!empty($fooProperties)) {
    $Foo = new Foo();
    $results['foo'] = $Foo->properties($fooProperties, $request);
}

if (!empty($barProperties)) {
    $Bar = new Bar();
    $results['bar'] = $Bar->properties($barProperties, $request);
}

foreach ($results as $r) {
    foreach ($r as $property_id => $p) {
        if (isset($properties[$property_id])) {
            $properties[$property_id] = $p;
        }
    }
}

With parallel about 11 seconds

if (!empty($fooProperties)) {
    $Foo = new Foo();
    $promises['foo'] = parallel(function () use($Foo, $fooProperties, $request) {
        return $Foo->properties($fooProperties, $request);
    })();
}

if (!empty($barProperties)) {
    $Bar = new Bar();
    $promises['bar'] = parallel(function () use($Bar, $barProperties, $request) {
        return $Bar->properties($barProperties, $request);
    })();
}


if (!empty($promises)) {
    $results = \Amp\Promise\wait(\Amp\Promise\all($promises));
    foreach ($results as $r) {
        foreach ($r as $property_id => $p) {
            if (isset($properties[$property_id])) {
                $properties[$property_id] = $p;
            }
        }
    }
}

Any ideas here?

Using worker in php fpm

To reply to a user request faster, i want to start multiple parallel jobs, wait for all jobs to finish, collect all responses of the jobs and return them to the user. For this requirement i'd like to use this library.

As a start I tried the worker and worker-pool example within my php-fpm process and got the following error: "Undefined constant 'STDERR' in /var/www/vendor/amphp/parallel/lib/Process/ChannelledProcess.php on line 82"

When i define STDERR myself, i get the next error:
"Uncaught Amp\Parallel\Sync\ChannelException: Invalid header received: [...] in /var/www/vendor/amphp/parallel/lib/Worker/AbstractWorker.php on line 172"

I'm running the code in a docker image php:7.1-fpm-alpine with only a few additional extensions. Non of the suggested, but apparently not required, extensions for threads or forks have been installed.

Is this library restricted to cli use? The readme doesn't state anything like this.

Issues with pthreads

Using test-script: https://github.com/amphp/parallel/blob/master/examples/worker.php

Environment

PHP: 7.2.5
pthreads: compiled from master (today)
OS: Ubuntu 16.04

Add ability to restart pool and workers

This feature can be used for stop long running tasks

Fails on Windows if path to this lib contains non-ascii chars

On fresh git clone && composer install:

Správca@GADELAT-PC C:\
$ php parallel-functions\examples\1-simple-function.php
Took 3.409206867218 milliseconds.

Správca@GADELAT-PC C:\
$ mv parallel-functions á

Správca@GADELAT-PC C:\
$ php á\examples\1-simple-function.php
PHP Fatal error:  Uncaught Amp\MultiReasonException: Multiple errors encountered in C:\á\src\functions.php:59
Stack trace:
#0 [internal function]: Amp\ParallelFunctions\{closure}()
#1 C:\á\vendor\amphp\amp\lib\Coroutine.php(74): Generator->send(Array)
#2 C:\á\vendor\amphp\amp\lib\Internal\Placeholder.php(127): Amp\Coroutine->Amp\{closure}(NULL, Array)
#3 C:\á\vendor\amphp\amp\lib\Deferred.php(41): class@anonymous->resolve(Array)
#4 C:\á\vendor\amphp\amp\lib\functions.php(415): Amp\Deferred->resolve(Array)
#5 C:\á\vendor\amphp\amp\lib\Internal\ResolutionQueue.php(51): Amp\Promise\{closure}(Object(Amp\Parallel\Context\ContextException), NULL)
#6 C:\á\vendor\amphp\amp\lib\Failure.php(26): Amp\Internal\ResolutionQueue->__invoke(Object(Amp\Parallel\Context\ContextException), NULL)
#7 C:\á\vendor\amphp\amp\lib\Internal\Placeholder.php(122): Amp\Failure->onResolve(Object(Amp\Internal\ResolutionQueue))
#8 C:\á\vendor\amphp\amp\lib\Internal\Placeholder.php(151): Amp\Coroutine->resolve(Object(Amp\Failure))
#9 C:\á\vendor\amphp in C:\á\src\functions.php on line 59

Fatal error: Uncaught Amp\MultiReasonException: Multiple errors encountered in C:\á\src\functions.php on line 59

Amp\MultiReasonException: Multiple errors encountered in C:\á\src\functions.php on line 59

Call Stack:
    0.2040     381376   1. {main}() C:\á\examples\1-simple-function.php:0
    0.2550    1547464   2. Amp\Promise\wait() C:\á\examples\1-simple-function.php:14


Správca@GADELAT-PC C:\

I streamlined bug to this library thanks to different error message I receive in other project where I use AMP

In Process.php line 177:

  [Amp\Parallel\Context\ContextException]
  The context stopped responding, potentially due to a fatal error or calling exit


Exception trace:
 Amp\Parallel\Context\Process->Amp\Parallel\Context\{closure}() at n/a:n/a
 Generator->throw() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Coroutine.php:71
 Amp\Coroutine->Amp\{closure}() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Failure.php:26
 Amp\Failure->onResolve() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Internal\Placeholder.php:122
 Amp\Coroutine->resolve() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Internal\Placeholder.php:151
 Amp\Coroutine->fail() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Coroutine.php:93
 Amp\Coroutine->Amp\{closure}() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Internal\Placeholder.php:127
 class@anonymous\C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Deferred.php004402D3->resolve() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Deferred.php:41
 Amp\Deferred->resolve() at C:\Users\Správca\src\php\Project\vendor\amphp\byte-stream\lib\ResourceInputStream.php:72
 Amp\ByteStream\ResourceInputStream::Amp\ByteStream\{closure}() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\NativeDriver.php:172
 Amp\Loop\NativeDriver->selectStreams() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\NativeDriver.php:68
 Amp\Loop\NativeDriver->dispatch() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\Driver.php:130
 Amp\Loop\Driver->tick() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\Driver.php:70
 Amp\Loop\Driver->run() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop.php:76
 Amp\Loop::run() at C:\Users\Správca\src\php\Project\bin\project:53
 Closure->{closure}() at n/a:n/a
 call_user_func() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Command\Command.php:250
 Symfony\Component\Console\Command\Command->run() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:865
 Symfony\Component\Console\Application->doRunCommand() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:241
 Symfony\Component\Console\Application->doRun() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:143
 Symfony\Component\Console\Application->run() at C:\Users\Správca\src\php\Project\bin\project:59

In ChannelParser.php line 61:

  [Amp\Parallel\Sync\ChannelException]
  Invalid packet received: Could not open input file: C:\Users\Spr\xc4\x82\xcb\x87vca\src\php\Project\vendor\amphp\parallel\lib\Context/Internal/proces
  s-runner.php\xa


Exception trace:
 Amp\Parallel\Sync\ChannelParser::parser() at n/a:n/a
 Generator->send() at C:\Users\Správca\src\php\Project\vendor\amphp\parser\lib\Parser.php:102
 Amp\Parser\Parser->push() at C:\Users\Správca\src\php\Project\vendor\amphp\parallel\lib\Sync\ChannelledStream.php:71
 Amp\Parallel\Sync\ChannelledStream->Amp\Parallel\Sync\{closure}() at n/a:n/a
 Generator->send() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Coroutine.php:74
 Amp\Coroutine->Amp\{closure}() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Internal\Placeholder.php:127
 class@anonymous\C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Deferred.php004402D3->resolve() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Deferred.php:41
 Amp\Deferred->resolve() at C:\Users\Správca\src\php\Project\vendor\amphp\byte-stream\lib\ResourceInputStream.php:72
 Amp\ByteStream\ResourceInputStream::Amp\ByteStream\{closure}() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\NativeDriver.php:172
 Amp\Loop\NativeDriver->selectStreams() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\NativeDriver.php:68
 Amp\Loop\NativeDriver->dispatch() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\Driver.php:130
 Amp\Loop\Driver->tick() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop\Driver.php:70
 Amp\Loop\Driver->run() at C:\Users\Správca\src\php\Project\vendor\amphp\amp\lib\Loop.php:76
 Amp\Loop::run() at C:\Users\Správca\src\php\Project\bin\project:53
 Closure->{closure}() at n/a:n/a
 call_user_func() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Command\Command.php:250
 Symfony\Component\Console\Command\Command->run() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:865
 Symfony\Component\Console\Application->doRunCommand() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:241
 Symfony\Component\Console\Application->doRun() at C:\Users\Správca\Downloads\symfony\src\Symfony\Component\Console\Application.php:143
 Symfony\Component\Console\Application->run() at C:\Users\Správca\src\php\Project\bin\project:59

Tested on Windows 7

Add documentation

There should be at least a rough overview.

Worker examples did not work

Hi,

I want try the worker examples but got an error:

Uncaught Error: Call to undefined function Amp\Promise\capture() in ... lib/Process/ChannelledProcess.php:100

Executed with "PHP 7.1.4". Let me hear if you need more informations.

Undefined Constant STDERR in parallel/lib/Context/Process

I'm using php-fpm7.0.22 + nginx
when I tried to execute this piece of code demo in artax documentation, it showed the error message as title.

        foreach ($requests as $request) {
            $promises[] = call(function () use ($defaultClient, $request) {
                $response = yield $defaultClient->request($request);
                $body = yield $response->getBody();
                return $body;
            });
        }
        $response = wait(all($promises));

I think this issue is relate to #26 . and I'm using artax 3.0.13 and parallel ^0.2. Is this issue not resolved completely?

Documentation/stable release

What's the status of this library? Is it still going to replace amphp/thread? If so, will there be a stable release and possibly documentation?

Hi guys

A little documentation or examples of usage could be help full.

Not able to locate PHP Binary on macOS

I was trying this library today with brew installed PHP packages and I ended up with this exception:

Could not locate PHP executable binary

Turns out, the getenv('PATH') returns false.

I tried Symfony's PhpExecutableFinder and it works fine and is able to locate the binary.

Would it be possible to either use Symfony's PHPExecutable Finder or make changes to the existing code to do what Symfony does to locate the binary?

Thanks!

Can't use objects instances inside Thread::spawn

I am using the following sample about threads [https://github.com/amphp/parallel/blob/master/examples/thread.php]

And works fine. However, when I try to use an object inside spawn, is not working.

Without object instance

<?php

require(dirname(__DIR__) . '/autoload.php');

use Amp\Loop;
use Amp\Parallel\Threading\Thread;
use Amp\Delayed;

Loop::run(function () {
    $timer = Loop::repeat(1000, function () {
        static $i;
        $i = $i ? ++$i : 1;
        print "Demonstrating how alive the parent is for the {$i}th time.\n";
    });
    try {
        // Create a new child thread that does some blocking stuff.
        $context1 = Thread::spawn(function () {
            echo "inside spawn \n";
            sleep(1);
            echo "end spawn \n";
        });

        yield new Delayed(2000);
        yield $context1->send([1]);
    } finally {
        Loop::cancel($timer);
    }
});

The output is :

$ zts-php modules/Jobs/scripts/test.php
inside spawn 
Demonstrating how alive the parent is for the 1th time.
end spawn

With object instance

require(dirname(__DIR__) . '/autoload.php');

use Amp\Loop;
use Amp\Parallel\Threading\Thread;
use Amp\Delayed;

class A {
    public function get() {
        return "a value";
    }
}
Loop::run(function () {
    $timer = Loop::repeat(1000, function () {
        static $i;
        $i = $i ? ++$i : 1;
        print "Demonstrating how alive the parent is for the {$i}th time.\n";
    });
    try {
        // Create a new child thread that does some blocking stuff.
        $context1 = Thread::spawn(function () {
            echo "inside spawn \n";
            $a = new A();
            echo $a->get();
            echo "end spawn \n";
        });

        yield new Delayed(2000);
        yield $context1->send([1]);
    } finally {
        Loop::cancel($timer);
    }
});

The output is :

$ zts-php modules/Jobs/scripts/test.php
inside spawn 
Demonstrating how alive the parent is for the 1th time.

Some additional information:

Working with centos 7.3
Working with PHP 7.2
Working with Zend Framework
Working with pthreads.so extension

Any suggestion? I think that my class A must have some interface or extends from other class build on the Amp package but I can not find any documentation.

How to stop a loop and waiting gracefully for end of current spawned sub-processes

Hi Niklas @kelunik

one more but different question.

I can't figure out how to stop a loop and wait until the current active sub-processes are finished.

My code so far:

        // define worker pool
        $pool = new DefaultPool($cnt_threads);

        // loop through files
        foreach ($files as $file) {
          // create the task and push to pool
          $task = new FileTask($file);
          $pool->enqueue($task);
        }

        // event loop for parallel tasks
        Loop::run(function () use (&$pool) {

          // What to do when a signal is received
          foreach (array(SIGINT, SIGHUP, SIGTERM) as $signal) {
            Loop::unreference(
              Loop::onSignal($signal,
                function () use ($signal) {
                  echo "\nABORT: caught SIGNAL [" . $signal . "]! Trying gracefully shutdown ...\n";
                }
              )
            );
          }

          return $pool->shutdown();
        });

If I press now CTRL-C the message comes up each time pressing keys and the Loop waits until current tasks are finished and stops then.

BUT

If I send kill -2 process-id from console, just the message comes up but the Loop runs until all tasks are done.

Wondering what to do to gracefully break the Loop::run(). Do you have implement a special handler so that CTRL-C works?

If I add Loop::stop(); in the onSignal function, the Loop immediately stops and the main-app ends. BUT all current sub-processes are still running and visible as processes (ps -ef).

Thanks for your help
Tom

Use custom error handler

We should use a custom error handler to transfer errors back to the parent process. We need to figure out how to deal with startup errors. I think we should use -d to only log them to STDERR. Another alternative would be simply parsing them. See https://travis-ci.org/amphp/file/jobs/245354788#L1645 which fails because UV can't be loaded, as compiling it fails.

Fork alternative on 0.2?

Since 0.2:

The forking context has been completely removed. Forking the entire process state proved to be too error-prone to be useful as a general-purpose parallel execution context.

What is the alternative to that if I want to keep using pcntl? I can't install pthread.

Regards.

Logging

It's worth mentioning I'm mostly using parallelMap rather than parallel directly and this may be purely a doc issue.

My main "complain" and by that I mean pain to check, monitor or debug what's going on when using Amp is that there is absolutely nothing to see. I see a list of workers with their CPU usages, and that's it. There is no logs about how many workers, which command is used to start them, what is their workload or status.

Did I miss something or this is completely missing? If so it would be really interesting to investigate how this could be done. I really like for example how XdebugHandler does it: just plug your logger there and you have everything you need

Append exceptions properly to make root cause analysis easier

See http://chat.stackoverflow.com/transcript/message/37677416#37677416. This should forward the exception when cancelling the pending jobs inside kill().

* Error in `php': malloc(): 0x000055c853bafde0 * and e.t.c

Hello. When I use code:

for ($i = 0; $i < 1000; $i++) {
    Loop::run(function ()  {
        $threads = [];
        for ($i = 0; $i < 100; $i++) {

            $threads[] = Thread::spawn(function () {
                return rand(1,100);
            });
        }

        foreach ($threads as $thread) {
            $thread->join();
        }
    });
    echo $i . "\n";
}

Everytime I'v get system error ex:

*** Error in `php': malloc(): memory corruption (fast): 0x0000000002209ee2 ***
Aborted (core dumped)

Segmentation fault (core dumped)

*** Error in `php': free(): invalid pointer: 0x00005579a7f2157d ***

*** Error in `php': corrupted double-linked list: 0x000055ca863cc7e0 ***

This can happen after the third iteration and can be after 1000 iterations. I can not find the relationship between the number of iterations and the receipt of this error.

Could you tell me how can I avoid that problem?

Better error reporting if worker dies

4546ef8 reports the header now, I'm not sure whether it's safe to just use ": " . $header . yield there to provide more debugging info.

Referenced watchers with only idle workers

Idle workers shouldn't keep the loop running, but the stderr piping keeps the loop running with only idle workers.

Not working inside a PHAR

After debugging this for a while I couldn't find any solution or pinpoint the real issue so attempting it here...

Basically for Box, when you rename it's PHAR from box.phar to box, i.e. renaming it to a name without the .phar extension, it stops being usable with Amp Parallel.

Indeed when Amp spawns Workers, it seems to be requiring the Composer Autoloader but that's it. In the case of Box as it needs more in the spawners, I'm registering a bootstrap closure which I call in the worker.

Now there is two issues when the PHAR is renamed to box:

It cannot find the Composer autoloader => actually it's not an issue as Amp already does it, so even if it did it would end up in a fatal error because trying to declare again the Composer classes. So the fix here is to remove it as done in https://github.com/humbug/box/pull/160/files
It cannot find the other function class vendor/humbug/php-scoper/src/functions.php

That's where I'm not sure what Amp does with the temporary files to try to execute the Workers when it's inside a PHAR. But it seems this part is not execute from inside the PHAR so I can't use the phar://box-alias.phar/vendor/humbug/php-scoper/src/functions.php to point to functions.php (the Box PHAR does not have an alias yet but I tried with it).

Any idea?

	do {
	// Attempt to acquire a lock from the semaphore.
	if (@\msg_receive($this->queue, 0, $type, 1, $chr, false, \MSG_IPC_NOWAIT, $errno)) {
	// A free lock was found, so resolve with a lock object that can
	// be used to release the lock.
	return new Lock(function (Lock $lock) {
	$this->release();
	});
	}

	// Check for unusual errors.
	if ($errno !== \MSG_ENOMSG) {
	throw new SemaphoreException('Failed to acquire a lock.');
	}
	} while (yield new Delayed(self::LATENCY_TIMEOUT, true));

amphp / parallel Goto Github PK

parallel's Issues

Without object instance

With object instance

Recommend Projects

Recommend Topics

Recommend Org