Experimental library for forking PHP

I ran into a limit when the FIFO pipe would get full (> ~62k of data), and cause the child processes to block until the pipe was cleared (e.g., by running the cat command in the appropriate FIFO handle on the file system). The use of the shmop_ functions seemed like a good alternative.

In the PR, I added a unit test which re-creates the problem that kept the original version from working on a typical Linux system.

One thing I will note is that while I did all the due-diligence I could with the shmop_ functions, I had to end up using @shmop_open with the error suppressor operator twice, to test for the presence for an existing memory segment. It's not ideal, but seemed like the only way.

A decent way to test that there are no left-over open memory segments is to run ipcs -m on the Linux system. You can then pass any offending shmid value over to the ipcrm binary. My testing as of yet indicated no issues of this kind, but it's theoretically possible it can happen in some edge-cases.

The extension should be checked only for this library development.

Example: Developer cannot identify that user use Windows/or/unix based

However, this package have to be added as "require" package so, it will always failed to install on Windows.

I think it better to let developer who use this library decide what is the behavior more then fail to install the application because extension is missing.

Because developer may let Unix user with extension can run with parallel processing while windows still be able to run as serial processing.

This allows for cleaner code in someplaces where you would otherwise have to use a closure, which can be hard to test.

My use case.

I have a class with a method where i want to fork out, that method takes one argument like

<?php

function doSomethingExciting($myLuckyNumber) {
    // do something here
}

Currently i would have to do something like

<?

$lucky = 42;

$manager = new Spork\ProcessManager;
$manager->fork(function ($fifo) use ($lucky) {
    doSomethingExiting($lucky);
});

Instead of

<?php

function doSomethingExciting($lucky, $optionalFifo) {
}

$manager = new Sprok\ProcessManager;
$manager->fork('doSomethingExciting', $lucky);

Fixed potential wrong $limit parameter in the child cursor if a $limit is provided to the parent.

A problem occured when i passed a cursor with a defined limit.

// Parent cursor with a limit
$items = $documentManager->getRepository('...')->findBy(array())->limit(50);

$self = $this;
$this->spork->process(
    $items,
    function($item, $index, $batch, $fifo) use ($self) {
        $self->process($item);
    },
    new DoctrineMongoStrategy(2)
);

And i would like to have the following forks: [0-25[ [25-50[

Thanks for this very useful library !

The example code doesn't pass in a new EventDispatcher but the tests do. I'm confused.

https://github.com/kriswallsmith/spork#spork-php-on-a-fork https://github.com/kriswallsmith/spork/blob/master/tests/Spork/Test/ProcessManagerTest.php#L23

I have config

<phpdox xmlns="http://phpdox.de/config">
    <project name=pr" source="src" workdir="build/phpdox">
        <collector publiconly="false">
            <include mask="*.php" />
            <exclude mask="*Autoload.php" />
            <exclude mask="vendor" />
        </collector>
        <generator output="build">
            <build engine="html" enabled="true" output="api"/>
        </generator>
    </project>
</phpdox>

I need exclude vendor, but it don't work. And i'm have error in some vendor package.

[04.10.2013 - 06:42:55]  - src/application/classes/mkv.php (iconv(): Wrong charset, conversion from `binary' to `UTF-8//TRANSLIT' is not allowed)
[04.10.2013 - 06:42:55] Saving results to directory 'build/phpdox'
[04.10.2013 - 06:42:57] Removed 901 vanished file(s) from project

Oups... phpDox encountered a problem and has terminated!
It most likely means you've found a bug, so please file a report for this and paste the stacktrace (if given) along!

Exception: TheSeer\phpDox\Project\ProjectException
Location: /usr/share/php/TheSeer/phpDox/project/Project.php (Line 268)

Internal Error: Unit 'Spork\Test\Util\ThrottleIteratorTest' not found in index (ns: Spork\Test\Util, n: ThrottleIteratorTest).

#0 /usr/share/php/TheSeer/phpDox/CLI.php(132): TheSeer\phpDox\Application->runCollector()
#1 /usr/bin/phpdox(60): TheSeer\phpDox\CLI->run()

Spork packet in vendor/spork directory.

Currently it is unclear to me how queueing of several processes/callbacks should be implemented the correct way. I used the attached example (a simple symfony2 command) where some file should be added to one archive. This workflow does not work as the file additions are overwriting themselves and at the end not every file will be placed in the archive. So this feels wrong and i think it should be handled in a fork->then->fork->then workflow, but i do not get it running in a nice way.

Could you maybe add an example how this stuff should be handled? And please ignore the used example here, i think it is quite useless but i could not think of another simple example :)

class SporkDebugCommand extends ContainerAwareCommand
{

    protected function configure()
    {
        $this
            ->setName('spork:debug')
            ->addOption('loops', null, InputOption::VALUE_REQUIRED, 'Loops', 10);
    }

    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $zipPath = $this->getContainer()->getParameter('kernel.cache_dir') . '/archive_' . uniqid() . '.zip';
        $output->writeln('archive: '.$zipPath);
        $output->writeln('process: '.posix_getpid());

        $zip = new \ZipArchive();
        if ($zip->open($zipPath, \ZIPARCHIVE::CREATE | \ZIPARCHIVE::OVERWRITE) !== true) {
            throw new \Exception('could not open archive');
        }

        $manager = new ProcessManager(new DeferredFactory());
        $file = __FILE__;

        // adding something to a zip archive but in different processes
        for ($i = 0; $i <= $input->getOption('loops'); $i++) {
            $manager->fork(function() use ($zip, $file, $i, $output) {
                echo "$i";

                $output->writeln(sprintf('loop process [%s]: [%s]', $i, posix_getpid()));
                $zip->addFile($file, $i.'_'.basename($file));
            })->then(function($forkOutput, $forkStatus) use ($output) {
                // do something in the parent process when it's done!
                $output->writeln('then process: ' .  posix_getpid(). " output: $forkOutput status: $forkStatus");
            });

        }

        $zip->close();
    }
}

it reads

$images = new RecursiveDirectoryIterator('/path/to/images');
$images = new RecursiveIteratorIterator($it);

$manager->process($images, function(SplFileInfo $file) {
    // upload this file
});

but should read

$it = new RecursiveDirectoryIterator('/path/to/images');
$images = new RecursiveIteratorIterator($it);

$manager->process($images, function(SplFileInfo $file) {
    // upload this file
});

notice the name of the first var, $it instead of $images.

I just did a composer update and got the following messages:

  - Updating symfony/event-dispatcher (v4.2.1 => v5.0.7): Downloading (100%)

Nothing special, just a major version update, but it seems something changed under the hood:

TypeError: Argument 1 passed to Symfony\Component\EventDispatcher\EventDispatcher::dispatch() must be an object, string given, called in <my project>/vendor/kriswallsmith/spork/src/Spork/ProcessManager.php on line 94 and defined in <my project>/vendor/symfony/event-dispatcher/EventDispatcher.php:48
Stack trace:
#0 <my project>/vendor/kriswallsmith/spork/src/Spork/ProcessManager.php(94): Symfony\Component\EventDispatcher\EventDispatcher->dispatch('spork.pre_fork')
#1 <my project>/class.php(268): Spork\ProcessManager->fork(Array)
#2 <my project>/download.php(7): class_Downloader->start()
#3 {main}[2020-04-05 08:34:17] !! Exception encountered:

TypeError: Argument 1 passed to Symfony\Component\EventDispatcher\EventDispatcher::dispatch() must be an object, string given, called in <my project>/vendor/kriswallsmith/spork/src/Spork/ProcessManager.php on line 94 and defined in <my project>/vendor/symfony/event-dispatcher/EventDispatcher.php:48
Stack trace:
#0 <my project>/vendor/kriswallsmith/spork/src/Spork/ProcessManager.php(94): Symfony\Component\EventDispatcher\EventDispatcher->dispatch('spork.pre_fork')
#1 <my project>/class.php(268): Spork\ProcessManager->fork(Array)
#2 <my project>/download.php(7): class_Downloader->start()
#3 {main}

Hello, not an issue, just a question I have in order to figure out few things with that library.

if I have the following code:

<?php
declare(strict_types=1);

require_once dirname(__DIR__) . '/vendor/autoload.php';

$manager = new \Spork\ProcessManager();

$processes = [
    function () {
        sleep(3);
        return "I am slow";
    },
    function () {
        sleep(5);
        return "Nope, I'm slower!";
    },
    function () {
        return "I'm really fast!";
    },
    function () {
        sleep(1);
        return "I have to train more!";
    }
];

foreach ($processes as $process) {
    $manager->fork($process)->done(function (\Spork\Fork $fork) {
        echo "PID: [{$fork->getPid()}] | {$fork->getResult()}\n";
    });
}

echo 'I am the flash!' . PHP_EOL;

when I run $ php fork/test.php, I get the following output:

I am the flash!
PID: [21444] | I am slow
PID: [21445] | Nope, I'm slower!
PID: [21446] | I'm really fast!
PID: [21447] | I have to train more!

shouldn't instead of getting the following output?

I am the flash!
PID: [21446] | I'm really fast!
PID: [21447] | I have to train more!
PID: [21444] | I am slow
PID: [21445] | Nope, I'm slower!

Am I missing something? should I use the $manager->wait() method instead of sleep instead?

Thanks!

when running with --forks=4 i get a fatal error. I suspect spork is not compatible with some of the newer versions of symfony components.

Fatal error: Class Spork\EventDispatcher\WrappedEventDispatcher contains 1 abstract method and must therefore be declared abstract or implement the remaining methods (Symfony\Component\EventDispatcher\EventDispatcherInterface::getListenerPriority) in ..../sf/vendor/kriswallsmith/spork/src/Spork/EventDispatcher/WrappedEventDispatcher.php on line 79

Call Stack:
    0.0005     224584   1. {main}() ..../sf/bin/console:0
    0.2655    1345056   2. Symfony\Component\Console\Application->run(class Symfony\Component\Console\Input\ArgvInput, ???) ..../sf/bin/console:29
    0.3133    1432272   3. Symfony\Bundle\FrameworkBundle\Console\Application->doRun(class Symfony\Component\Console\Input\ArgvInput, class Symfony\Component\Console\Output\ConsoleOutput) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:117
    1.7341    4333360   4. Symfony\Component\Console\Application->doRun(class Symfony\Component\Console\Input\ArgvInput, class Symfony\Component\Console\Output\ConsoleOutput) ..../sf/vendor/symfony/symfony/src/Symfony/Bundle/FrameworkBundle/Console/Application.php:80
    1.7344    4334392   5. Symfony\Component\Console\Application->doRunCommand(class Symfony\Bundle\AsseticBundle\Command\DumpCommand, class Symfony\Component\Console\Input\ArgvInput, class Symfony\Component\Console\Output\ConsoleOutput) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:186
    1.7951    4425144   6. Symfony\Component\Console\Command\Command->run(class Symfony\Component\Console\Input\ArgvInput, class Symfony\Component\Console\Output\ConsoleOutput) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:807
    1.7955    4426592   7. Symfony\Bundle\AsseticBundle\Command\DumpCommand->initialize(class Symfony\Component\Console\Input\ArgvInput, class Symfony\Component\Console\Output\ConsoleOutput) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Command/Command.php:228
    1.8020    4435568   8. spl_autoload_call(string(44)) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Command/Command.php:58
    1.8021    4435632   9. Symfony\Component\Debug\DebugClassLoader->loadClass(string(44)) ..../sf/vendor/symfony/symfony/src/Symfony/Component/Console/Command/Command.php:58
    1.8065    4442592  10. require_once('..../sf/vendor/kriswallsmith/spork/src/Spork/EventDispatcher/WrappedEventDispatcher.php') ..../sf/vendor/symfony/symfony/src/Symfony/Component/Debug/DebugClassLoader.php:142

installed symfony components:

symfony/assetic-bundle               v2.8.0             Integrates Assetic into Symfony2
symfony/monolog-bundle               v2.10.0            Symfony MonologBundle
symfony/phpunit-bridge               v2.8.4             Symfony PHPUnit Bridge
symfony/polyfill-intl-icu            v1.1.1             Symfony polyfill for intl's ICU-related data and classes
symfony/polyfill-mbstring            v1.1.1             Symfony polyfill for the Mbstring extension
symfony/polyfill-php56               v1.1.1             Symfony polyfill backporting some PHP 5.6+ features to lower PHP versions
symfony/polyfill-php70               v1.1.1             Symfony polyfill backporting some PHP 7.0+ features to lower PHP versions
symfony/polyfill-util                v1.1.1             Symfony utilities for portability of PHP codes
symfony/security-acl                 v3.0.0             Symfony Security Component - ACL (Access Control List)
symfony/swiftmailer-bundle           v2.3.11            Symfony SwiftmailerBundle
symfony/symfony                      v3.0.4             The Symfony PHP framework
kriswallsmith/assetic                v1.3.2             Asset Management for PHP
kriswallsmith/spork                  v0.3               Asynchronous PHP

Hi, I have found next in MongoStrategy

 If you use this strategy you MUST close your Mongo connection on the spork.pre_fork event.
     $mongo = new Mongo();
     $manager->addListener(Events::PRE_FORK, array($mongo, 'close'));

I can not understand why I need to do this?

Also my code looks like, I mean that I open connection out of logic closure that runs inside fork.

$mongo  = new \MongoClient();
$db     = $mongo->selectDB(DB_NAME);
$coll   = $db->selectCollection(COLL_NAME);
$cursor = $coll->find([/*some query here*/]);

$dispatcher = new SporkDispatcher\EventDispatcher();
$dispatcher->addListener(SporkDispatcher\Events::PRE_FORK, function() use($mongo) {
    $mongo->close();
});
$manager  = new Spork\ProcessManager($dispatcher);
$strategy = new Spork\Batch\Strategy\MongoStrategy(5);

$manager->process($cursor, function($document) {
    $document; // some work here
}, $strategy);

There are few more questions:

• if I close connection BEFORE any forking operation, what will be with cursor?
• will cursor close as well as connection?
• will forked process open new connection inside it-self, or what?

So, looks like @kriswallsmith will be helpful here.

Two issues fixed:

1. The Fork was losing received messages. When waiting it does a ::receive and does nothing with the result. The fix was to add a messages collection to the form and have the ::receive populate that collection. This way messages can be retrieved at any time using ::getMessages.
2. Shared memory was creating imbricated arrays with each ::send making the Fork::receive end up with array in array in array, etc. instead of a flat array of messages.

| Q | A | | --- | --- | | Bug fix? | yes | | New feature? | np | | BC breaks? | no | | Deprecations? | no | | Fixed tickets | |

When updating to symfony 3, the interface Symfony\Component\EventDispatcher\EventDispatcherInterface has a new method implemented.

Therefore we need to implement it to.