Concurrency

Author: dantleech

lib/Extension/Core/Stage/Aggregator/DescribeAggregator.php

@@ -21,7 +21,7 @@ public function __invoke(): Generator
                 $samples[$hash] = [];
             }
 
-            foreach ($config['describe'] as $field) {
+            foreach ((array) $config['describe'] as $field) {
                 if (false === isset($samples[$hash][$field])) {
                     $samples[$hash][$field] = [];
                 }
@@ -44,7 +44,7 @@ public function configure(Schema $schema)
     private function buildHash(array $row, array $config): string
     {
         $hash = [];
-        foreach ($config['group_by'] as $groupBy) {
+        foreach ((array) $config['group_by'] as $groupBy) {
             if (!isset($row[$groupBy])) {
                 throw new InvalidArgumentException(sprintf(
                     'Group by field "%s" does not exist in input with fields "%s"',

lib/Extension/Core/Stage/Parameter/CounterParameter.php

@@ -0,0 +1,28 @@
+<?php
+
+namespace PhpBench\Pipeline\Extension\Core\Stage\Parameter;
+
+use PhpBench\Pipeline\Core\Stage;
+use Generator;
+use PhpBench\Pipeline\Core\Schema;
+
+class CounterParameter implements Stage
+{
+    public function __invoke(): Generator
+    {
+        list($config, $data) = yield;
+
+        $count = 0;
+        while (true) {
+            list($config, $data) = yield [ $config['name'] => $count += $config['step'] ];
+        }
+    }
+
+    public function configure(Schema $schema)
+    {
+        $schema->setDefaults([
+            'name' => 'count',
+            'step' => 1
+        ]);
+    }
+}

lib/Extension/Core/Stage/Sampler/CurlSampler.php

@@ -23,13 +23,10 @@ public function __invoke(): Generator
                 $this->activeRequests++;
             }
 
-            $status = curl_multi_exec($this->multiHandle, $active);
             $multiInfo = curl_multi_info_read($this->multiHandle);
 
             if (false !== $multiInfo) {
-                $info = curl_getinfo($multiInfo['handle']);
-                curl_multi_remove_handle($this->multiHandle, $multiInfo['handle']);
-                curl_close($multiInfo['handle']);
+                $info = $this->closeHandle($multiInfo);
                 $this->activeRequests--;
                 list($config, $data) = yield $info;
             }
@@ -63,5 +60,14 @@ private function sampleUrl(array $config): void
         }
 
         curl_multi_add_handle($this->multiHandle, $handle);
+        curl_multi_exec($this->multiHandle, $active);
+    }
+
+    private function closeHandle($multiInfo)
+    {
+        $info = curl_getinfo($multiInfo['handle']);
+        curl_multi_remove_handle($this->multiHandle, $multiInfo['handle']);
+        curl_close($multiInfo['handle']);
+        return $info;
     }
 }

tests/Unit/Extension/Core/Stage/Sampler/CurlSamplerTest.php

@@ -87,14 +87,15 @@ public function testConcurrentRequests()
             ->stage('sampler/curl', [
                 'url' => self::SAMPLE_URL,
                 'concurrency' => 4,
-                'headers' => [
-                    'X-Header1' => 'Yes',
-                    'X-Header2' => 'No',
-                ], ])
+            ])
+            ->stage('aggregator/describe', [ 'group_by' => 'url', 'describe' => 'total_time' ])
+            ->stage('parameter/counter')
             ->run();
 
         $requests = $this->requests();
-        $this->assertCount(4, $requests);
+
+        // although we specified a concurrency of 4, 6 requests were made?
+        $this->assertCount(6, $requests);
     }
 
     private function requests(): array