Benchmark runner

benchmark/bufferWithCount/bufferWithCount.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->bufferWithCount(5);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/bufferWithCount/bufferWithCount_eventloop.php

@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+$source = Observable::range(0, 25, $scheduler)
+    ->bufferWithCount(5);
+
+$factory = function() use ($source, $scheduler) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/distinct/distinct.php

@@ -0,0 +1,14 @@
+<?php
+
+use Rx\Observable;
+
+$range = array_map(function($val) {
+    return $val % 3;
+}, range(0, 25));
+
+$source = Observable::fromArray($range)
+    ->distinct();
+
+return function() use ($source) {
+    return $source;
+};

benchmark/distinct/distinct_eventloop.php

@@ -0,0 +1,22 @@
+<?php
+
+use Rx\Observable;
+use Rx\Observable\ArrayObservable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$range = array_map(function($val) {
+    return $val % 3;
+}, range(0, 25));
+
+$source = (new ArrayObservable($range, $scheduler))
+    ->distinct();
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/filter/filter.php

@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 50)
+    ->filter(function($value) {
+        return $value % 2 == 0;
+    })
+    ->filter(function($value) {
+        return $value % 10 == 0;
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/filter/filter_eventloop.php

@@ -0,0 +1,22 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 50, $scheduler)
+    ->filter(function($value) {
+        return $value % 2 == 0;
+    })
+    ->filter(function($value) {
+        return $value % 10 == 0;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/run.php

@@ -0,0 +1,154 @@
+<?php
+
+if (file_exists($file = __DIR__.'/../vendor/autoload.php')) {
+    $autoload = require_once $file;
+    $autoload->addPsr4('Vendor\\Rx\\Operator\\', __DIR__ . '/custom-operator');
+} else {
+    throw new RuntimeException('Install dependencies to run benchmark suite.');
+}
+
+use Rx\Observable;
+use Rx\Observer\CallbackObserver;
+use React\EventLoop\LoopInterface;
+
+// Check whether XDebug is enabled
+if (in_array('Xdebug', get_loaded_extensions(true))) {
+    printf("Please, disable Xdebug extension before running RxPHP benchmarks.\n");
+    exit(1);
+}
+
+define('MIN_TOTAL_DURATION', 5);
+$start = microtime(true);
+
+if ($_SERVER['argc'] === 1) {
+    $files = glob(__DIR__ . '/**/*.php');
+} else {
+    $files = [];
+    foreach (array_slice($_SERVER['argv'], 1) as $fileOrDir) {
+        if (is_dir($fileOrDir)) {
+            $files = array_merge($files, glob($fileOrDir . '/*.php'));
+        } else {
+            // Force absolute path
+            $files[] = $fileOrDir[0] === DIRECTORY_SEPARATOR ? $fileOrDir : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $fileOrDir;
+        }
+    }
+}
+
+
+Observable::just($files)
+    ->doOnNext(function(array $files) {
+        printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
+        printf("script_name - total_runs (single_run_mean ±standard_deviation) - mem_start [mem_100_iter] mem_end\n");
+        printf("==============================================================\n");
+    })
+    ->concatMap(function($files) { // Flatten the array
+        return Observable::fromArray($files);
+    })
+    ->doOnNext(function($file) {
+        printf('%s', pathinfo($file, PATHINFO_FILENAME));
+    })
+    ->map(function($file) { // Run benchmark
+        $durations = [];
+        /** @var Observable $observable */
+        $observable = null;
+        /** @var LoopInterface $loop */
+        $loop = null;
+        /** @var callable(): Observable $sourceFactory */
+        $sourceFactory = null;
+
+        ob_start();
+
+        $testDef = @include $file;
+
+        if (is_array($testDef)) {
+            $sourceFactory = $testDef[0];
+            $loop = $testDef[1];
+        } elseif (is_callable($testDef)) {
+            $sourceFactory = $testDef;
+        } else {
+            throw new Exception("File \"$file\" doesn't contain a valid benchmark");
+        }
+
+        $memoryUsage = [memory_get_usage()];
+
+        $benchmarkLoop = function(Observable $observable) use (&$durations, &$memoryUsage) {
+            $dummyObserver = new Rx\Observer\CallbackObserver(
+                function ($value) { },
+                function ($error) { },
+                function () use (&$start, &$durations) {
+                    $durations[] = (microtime(true) - $start) * 1000;
+                }
+            );
+
+            $start = microtime(true);
+            $observable->subscribe($dummyObserver);
+
+            if (count($durations) === 100) {
+                $memoryUsage[] = memory_get_usage();
+            }
+        };
+
+        $stopStartTime = microtime(true) + MIN_TOTAL_DURATION;
+
+        if ($loop) {
+            $reschedule = function() use (&$reschedule, $benchmarkLoop, $sourceFactory, $loop, $stopStartTime) {
+                $loop->futureTick(function () use (&$reschedule, $benchmarkLoop, $stopStartTime, $sourceFactory) {
+                    $benchmarkLoop($sourceFactory());
+                    if ($stopStartTime > microtime(true)) {
+                        $reschedule();
+                    }
+                });
+            };
+
+            $reschedule();
+            $loop->run();
+        } else {
+            while ($stopStartTime > microtime(true)) {
+                $benchmarkLoop($sourceFactory());
+            }
+        }
+
+        $memoryUsage[] = memory_get_usage();
+
+        ob_end_clean();
+
+        return [
+            'file' => $file,
+            'durations' => $durations,
+            'memory_usage' => $memoryUsage,
+        ];
+    })
+    ->doOnNext(function(array $result) { // Print the number of successful runs
+        printf(' - %d', count($result['durations']));
+    })
+    ->map(function(array $result) { // Calculate the standard deviation
+        $count = count($result['durations']);
+        $mean = array_sum($result['durations']) / $count;
+
+        $variance = array_sum(array_map(function($duration) use ($mean) {
+            return pow($mean - $duration, 2);
+        }, $result['durations']));
+
+        return [
+            'file' => $result['file'],
+            'memory_usage' => $result['memory_usage'],
+            'mean' => $mean,
+            'standard_deviation' => pow($variance / $count, 0.5),
+        ];
+    })
+    ->subscribe(new CallbackObserver(
+        function(array $result) {
+            printf(" (%.2fms ±%.2fms) - ", $result['mean'], $result['standard_deviation']);
+            foreach ($result['memory_usage'] as $memory) {
+                printf("%.2fMB ", $memory / pow(10, 6));
+            }
+            printf("\n");
+        },
+        function(\Exception $error) {
+            printf("\nError: %s\n", $error->getMessage());
+        },
+        function() use ($start) {
+            printf("============================================================\n");
+            printf("total duration: %.2fs\n", microtime(true) - $start);
+        }
+    ));

benchmark/skipLast/skipLast.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 500)
+    ->skipLast(50);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/skipLast/skipLast_eventloop.php

@@ -0,0 +1,17 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 500, $scheduler)
+    ->skipLast(50);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/takeLast/takeLast.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 500)
+    ->takeLast(50);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/takeLast/takeLast_eventloop.php

@@ -0,0 +1,17 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 500, $scheduler)
+    ->takeLast(50);
+
+$factory = function() use ($source, $scheduler) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/zip/zip.php

@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->zip([Observable::range(0, 25)], function ($a, $b) {
+        return $a + $b;
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/zip/zip_eventloop.php

@@ -0,0 +1,19 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->zip([Observable::range(0, 25, $scheduler)], function($a, $b) {
+        return $a + $b;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];