misterion / ko-process Goto Github PK

Hi!

I get this error:

[RuntimeException] 
Unable to create the semaphore (0)
/htdocs/vendor/misterion/ko-process/src/Ko/Semaphore.php:81
#0: Ko\Semaphore->__construct(integer)
        /htdocs/vendor/misterion/ko-process/src/Ko/SharedMemory.php:90
#1: Ko\SharedMemory->__construct()
        /htdocs/vendor/misterion/ko-process/src/Ko/ProcessManager.php:196
#2: Ko\ProcessManager->createProcess(object)
        /htdocs/vendor/misterion/ko-process/src/Ko/ProcessManager.php:190
#3: Ko\ProcessManager->fork(object)

What could be the problem?

Unable to create the shared memory segment 

Warning: shm_attach(): failed for key 0x61019b9c: Too many open files in /Users/Elfet/Sites/.../vendor/misterion/ko-process/src/Ko/SharedMemory.php on line 55

I got this. What to do?

Make build-in max active child feature.

Hey!

https://packagist.org/packages/misterion/ko-process

I need 0.5.3

Or use auto hook, settings -> web service (hooks) -> add service -> packagist

Thanks 🍰

Hi,
I'd like to suggest some improvement to ko-process:
In normal situation when we need to shutdown worker process we are sending to it SIGTERM signal.
The child worker with "waiting next envelope" status cann't to be closed cause need to process some message before it (and if message rate is slow, we can wait worker shutdown for a long time) . Could you send to the child SIGKILL signal in that situation?

Test code:

$manager = new Ko\ProcessManager();
for ($i = 0; $i < 300; $i++) {
    $manager->fork(function(Ko\Process $p) {
        echo 'Hello from ' . $p->getPid() . "\n";
        sleep(1);
        exit(1); //exit with non 0 exit code
    });
}
$manager->wait();

Result:

[root@host-192-168-113-41 demo]# php pcntl.php 
Hello from 9167
......
Hello from 9290

Warning: sem_get(): failed for key 0x61003ddc: No space left on device in /var/www/html/queue/vendor/misterion/ko-process/src/Ko/Semaphore.php on line 79

Fatal error: Uncaught RuntimeException: Unable to create the semaphore in /var/www/html/queue/vendor/misterion/ko-process/src/Ko/Semaphore.php:81
Stack trace:
#0 /var/www/html/queue/vendor/misterion/ko-process/src/Ko/SharedMemory.php(90): Ko\Semaphore->__construct(1627405788)
#1 /var/www/html/queue/vendor/misterion/ko-process/src/Ko/ProcessManager.php(195): Ko\SharedMemory->__construct()
#2 /var/www/html/queue/vendor/misterion/ko-process/src/Ko/ProcessManager.php(189): Ko\ProcessManager->createProcess(Object(Closure))
#3 /var/www/html/queue/demo/pcntl.php(34): Ko\ProcessManager->fork(Object(Closure))
#4 {main}
  thrown in /var/www/html/queue/vendor/misterion/ko-process/src/Ko/Semaphore.php on line 81

Motivation is make more easy working with shared memory,

Now

$m->fork(function(Process $p) {
     $sm= $p->getSharedMemory();
     $sm['field'] ='value';
});

Should be

$m->fork(function(Process $p) {
     $p['field'] ='value';
});

Hi, the composer packages suggests ext-sysvshm, but is this actually a requires? Without ext-sysvshm installed executing code from 'Usage' in README.md fails with an fatal error:

PHP Fatal error: Call to undefined function Ko\shm_attach() in /srv/movember.com/releases/20120817072736/vendor/misterion/ko-process/src/Ko/SharedMemory.php on line 83

Thanks.

I am trying to get advantage of onShutdown handler:

$processManager = new Ko\ProcessManager()
$processManager->fork(function(Process $process) {
});

$processManager->fork(function(Process $process) {
});

$processManager->onShutdown(function () {
    echo 'Shutting down';
});

$processManager->wait();

But it is never called.

When I interrupt my script with Ctrl + C SIGINT signal is sent.

SIGTERM is expected in:

protected function setupSignalHandlers()
{
    $this->signalHandler = new SignalHandler();
    $this->signalHandler->registerHandler(SIGCHLD, function() {
        $this->handleSigChild();
    });

    $this->signalHandler->registerHandler(SIGTERM, function() {
        $this->handleSigTerm();
    });
}

I have tried adding:

$this->signalHandler->registerHandler(SIGINT, function() {
    $this->handleSigTerm();
});

But then the script is never interrupted.

/**
 * Return count of currently running child process.
 *
 * @return int
 */
public function getProcessCount()
{
    return count($this);
}
/**
 * Return count of currently running child process.
 *
 * @return int
 */
public function count()
{
    return count($this->children);
}

Should be in first function?
return $this->count($this);

PHP Fatal error: Call to undefined function Ko\shm_attach() in /Paidata/www/process/vendor/misterion/ko-process/src/Ko/SharedMemory.php on line 83

Fatal error: Call to undefined function Ko\shm_attach() in /Paidata/www/process/vendor/misterion/ko-process/src/Ko/SharedMemory.php on line 83

I needed some additional parameter when starting thread.
So I added it in Process class and fork and spawn method in ProcessManager class.
I don' have enough time to PR, now. So I just leave an issue here.

$param = "whatever";

$process = $manager->fork(function(Ko\Process $p, $param) {
    // do something
}, $param);

source.zip

$manager = new \Ko\ProcessManager();
        for ($i = 0; $i < 10; $i++) {
            $manager->fork(function(\Ko\Process $p) use ($i) {
                echo 'Hello from ' . $p->getPid();
                sleep(1);
            })->onSuccess(function() {
                echo 'end';
            })->onError(function() {
                echo 'fail end';
            });
        }
$manager->wait();

OnSuccess / onError callbacks don't work if call wait() on process manager. If call wait () for each process handles call correct, but not parallel

First of all @misterion, thank you for this awesome library, I really like it.

I am playing with videlalvaro/php-amqplib.
What I wanted to do is have parallel publisher and consumer running in child processes:

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('q1');
$processManager = new ProcessManager();

$processManager->fork(function (Process $process) use ($channel) {
    echo 'Publishing from '.$process->getPid().PHP_EOL;

    $channel->basic_publish(new AMQPMessage('Hello from '.$process->getPid()), null, 'q1');
    $channel->close();
});

$processManager->fork(function (Process $process) use ($channel) {
    echo 'Consuming from '.$process->getPid().PHP_EOL;

    $channel->basic_consume('q1', '', false, false, false, false, function ($message) {
        echo 'Consuming message: '.$message->body.PHP_EOL;
    });

    while (count($channel->callbacks)) {
        $channel->wait();
    }
});

$processManager->wait();

Output:

Publishing from 9830
Consuming from 9831

Messages are published, but never consumed. Do you have any idea what can be the reason?

Full example is available on https://github.com/umpirsky/Extraload/blob/feature/queue-playground/test.php, can be easily checkout and run.