multi-threads support about rxphp

[davidwdan]

Every time we look into creating a pthreads scheduler we come to the same conclusion as @martinsik: It would be a lot of work and not very useful.

I'd love to be proven wrong.

[martinsik]

I have some previous experience with pthreads PHP extension and I think it's hardly useful in practise. It works only in CLI environment (this means no SAPI => you can't use it behind a webserver => you can't use it for websites). Documentation is very poor and often not up to date. Support from the main extension developer is basically non-existent.

I can show you how I used it. I wanted to put this into a separate repo with some additional RxPHP operators but then I found out it's not very useful:

This example takes a directory and analyses PHP files using PHPParser library in multiple threads. I made a custom ThreadPoolOperator operator that takes tasks as PHPParserThread and runs them in multiple threads.

// demo.php
Observable::create(
    function (ObserverInterface $observer) use ($loop) {
        $start = microtime(true);
        $dirIter = new \RecursiveDirectoryIterator(__DIR__ . '/../symfony_template/vendor');
        $iter = new \RecursiveIteratorIterator($dirIter);

        while ($iter->valid()) {
            /** @var SplFileInfo $file */
            $file = $iter->current();
            if ($file->getExtension() === 'php' && $file->isReadable()) {
                $observer->onNext($file->getRealPath());
            }
            $iter->next();
        }

        return new CallbackDisposable(function() use ($loop, $start) {
            echo "duration: " . round(microtime(true) - $start, 2) . "s\n";
            $loop->stop();
        });
    })
    ->bufferWithCount(BUFFER_SIZE)
    ->map(function($filenames) {
        return new PHPParserThread($filenames);
    })
    ->lift(function() use ($scheduler) {
        return new ThreadPoolOperator(4, PHPParserWorker::class, [__DIR__ . '/../vendor/autoload.php'], $scheduler);
    })
    ->flatMap(function($result) {
        return Observable::fromArray((array)$result, Scheduler::getImmediate());
    })
    ->take(MAX_FILES)
    ->filter(function($result) {
        return count($result['results']) > 0;
    })
    ->subscribe(function($result) {
         print_r($result);
    });

$loop->run();

// PHPParserThread.php
class PHPParserThread extends AbstractRxThread
{
    private $filenames;

    public function __construct($filename)
    {
        $this->filenames = (array)(is_array($filename) ? $filename : [$filename]);
        /** @var Volatile result */
        $this->result = [];
    }

    public function run()
    {
        $last = 0;
        Observable::fromArray($this->filenames, Scheduler::getImmediate())
            ->lift(function() {
                $classes = ['AssignmentInConditionNodeVisitor'];
                return new PHPParserOperator($classes);
            })
            ->subscribe(function ($results) use (&$last) {
                $this->result[$last++] = (array)[
                    'file' => $results['file'],
                    'results' => $results['results'],
                ];
            }, null, function() {
                $this->markDone();
            });
    }
}

// AbstractRxThread.php
abstract class AbstractRxThread extends Thread
{

    private $done = false;
    protected $result;

    public function getResult()
    {
        return $this->result;
    }

    public function isDone()
    {
        return $this->done;
    }

    protected function markDone()
    {
        $this->done = true;
    }
}

// PHPParserWorker.php
class PHPParserWorker extends \Worker
{

    protected $loader;

    public function __construct($loader) {
        $this->loader = $loader;
    }

    public function run() {
        $classLoader = require_once($this->loader);
        $dir = __DIR__;
        $classLoader->addClassMap([
            'PHPParserThread' => $dir . '/PHPParserThread.php',
            'PHPParserWorker' => $dir . '/PHPParserWorker.php',
            'PHPParserOperator' => $dir . '/PHPParserOperator.php',
        ]);
    }

    public function start(int $options = PTHREADS_INHERIT_ALL) {
        return parent::start(PTHREADS_INHERIT_NONE);
    }
}

// ThreadPoolOperator.php
class ThreadPoolOperator implements OperatorInterface
{

    private $pool;
    private $scheduler;

    public function __construct($numThreads = 4, $workerClass = Worker::class, $workerArgs = [], $scheduler = null)
    {
        $this->pool = new Pool($numThreads, $workerClass, $workerArgs);
        $this->scheduler = $scheduler ?: Scheduler::getDefault();
    }

    public function __invoke(ObservableInterface $observable, ObserverInterface $observer): DisposableInterface
    {
        $callbackObserver = new CallbackObserver(function(AbstractRxThread $task) {
                /** @var AbstractRxThread $task */
                $this->pool->submit($task);
            },
            [$observer, 'onError'],
            [$observer, 'onCompleted']
        );

        $dis1 = $this->scheduler->schedulePeriodic(function() use ($observer) {
            $this->pool->collect(function(AbstractRxThread $task) use ($observer) {
                /** @var AbstractRxThread $task */
                if ($task->isDone()) {
                    $observer->onNext($task->getResult());
                    return true;
                } else {
                    return false;
                }
            });
        }, 0, 10);

        $dis2 = $observable->subscribe($callbackObserver);
        $disposable = new BinaryDisposable($dis1, $dis2);

        return $disposable;
    }
}

[Vinceveve]

The main problem is that pthreads is for PHP 7.2 only (issues on 7.0 and 7.1 with ZTS)
A ThreadedScheduler with a worker pool could work well in RX context.
Plugged after an async http server (reactphp, icicle ...) it can also be useful to run blocking code on a php web microservice

[Vinceveve]

For coding you are right it’s not that useful
But for architecture it can be really useful !
Php-fpm is a great peace of work but for microservices it’s a pain in the ass
With ThreadingScheduler we can imagine using every php extensions not only the too few async ones

[davidwdan]

Pthreads has been archived by its owner, so I'm closing this issue.