Merge pull request #12 from rcalicdan/qualityoflifeimprovement

Qualityoflifeimprovement

Author: rcalicdan

src/Api/Async.php

@@ -116,36 +116,4 @@ public static function await(PromiseInterface $promise): mixed
     {
         return self::getAsyncOperations()->await($promise);
     }
-
-    /**
-     * Create a safe async function with automatic error handling.
-     *
-     * The returned function will catch any exceptions thrown during execution
-     * and convert them to rejected promises, preventing uncaught exceptions
-     * from crashing the event loop. This is essential for building robust
-     * async applications that can gracefully handle errors.
-     *
-     * @param  callable  $asyncFunction  The async function to make safe
-     * @return callable A safe version that always returns a promise
-     */
-    public static function tryAsync(callable $asyncFunction): callable
-    {
-        return self::getAsyncOperations()->tryAsync($asyncFunction);
-    }
-
-    /**
-     * Convert a synchronous function to work in async contexts.
-     *
-     * Wraps a synchronous function so it can be used alongside async operations
-     * without blocking the event loop. The function will be executed in a way
-     * that doesn't interfere with concurrent async operations, making it safe
-     * to use within fiber-based async workflows.
-     *
-     * @param  callable  $syncFunction  The synchronous function to wrap
-     * @return callable An async-compatible version of the function
-     */
-    public static function asyncify(callable $syncFunction): callable
-    {
-        return self::getAsyncOperations()->asyncify($syncFunction);
-    }
 }

src/Api/AsyncLoop.php

@@ -158,15 +158,4 @@ public static function asyncSleep(float $seconds): void
     {
         self::getLoopOperations()->asyncSleep($seconds);
     }
-
-    /**
-     * Run an async operation and measure its performance metrics.
-     *
-     * @param  callable|PromiseInterface  $asyncOperation  The operation to benchmark
-     * @return array Array containing 'result' and 'benchmark' keys with performance data
-     */
-    public static function benchmark(callable|PromiseInterface $asyncOperation): array
-    {
-        return self::getLoopOperations()->benchmark($asyncOperation);
-    }
 }

src/Api/Promise.php

@@ -142,17 +142,17 @@ public static function timeout(callable|PromiseInterface|array $promises, float
     }
 
     /**
-     * Execute multiple tasks concurrently with a concurrency limit.
+     * Execute multiple tasks concurrently with a specified concurrency limit.
      *
-     * Processes an array of tasks (callables or promises) in controlled batches
-     * to avoid overwhelming the system resources. This is essential for handling
-     * large numbers of concurrent operations without exhausting memory, file
-     * descriptors, or network connections. Unlike Promise::all(), this method
-     * provides backpressure control.
+     * IMPORTANT: For proper concurrency control, tasks should be callables that return
+     * Promises, not pre-created Promise instances. Pre-created Promises are already
+     * running and cannot be subject to concurrency limiting.
      *
-     * @param  array  $tasks  Array of tasks (callables or promises) to execute
-     * @param  int  $concurrency  Maximum number of concurrent executions (default: 10)
-     * @return PromiseInterface A promise that resolves with all task results
+     * @param array $tasks Array of callables that return Promises, or Promise instances
+     *                     Note: Promise instances will be awaited but cannot be truly
+     *                     limited since they're already running
+     * @param int $concurrency Maximum number of tasks to run simultaneously
+     * @return PromiseInterface Promise that resolves with an array of all results
      */
     public static function concurrent(array $tasks, int $concurrency = 10): PromiseInterface
     {
@@ -167,7 +167,9 @@ public static function concurrent(array $tasks, int $concurrency = 10): PromiseI
      * processing large datasets or performing operations that require
      * significant resources without overwhelming the system.
      *
-     * @param  array  $tasks  Array of tasks (callables or promises) to execute
+     * @param array $tasks Array of callables that return Promises, or Promise instances
+     *                     Note: Promise instances will be awaited but cannot be truly
+     *                     limited since they're already running
      * @param  int  $batchSize  Size of each batch to process concurrently (default: 10)
      * @param  int  $concurrency  Maximum number of concurrent executions (default: 10)
      * @return PromiseInterface A promise that resolves with all task results

src/Async/AsyncOperations.php

@@ -125,34 +125,6 @@ public function async(callable $asyncFunction): callable
         return $this->executionHandler->async($asyncFunction);
     }
 
-    /**
-     * Convert a synchronous function to work in async contexts.
-     *
-     * This wraps a sync function so it can be used alongside async
-     * operations without blocking the event loop.
-     *
-     * @param  callable  $syncFunction  The synchronous function to wrap
-     * @return callable An async-compatible version of the function
-     */
-    public function asyncify(callable $syncFunction): callable
-    {
-        return $this->executionHandler->asyncify($syncFunction);
-    }
-
-    /**
-     * Create a safe async function with error handling.
-     *
-     * The returned function will catch exceptions and convert them
-     * to rejected promises, preventing uncaught exceptions.
-     *
-     * @param  callable  $asyncFunction  The async function to make safe
-     * @return callable A safe version of the async function
-     */
-    public function tryAsync(callable $asyncFunction): callable
-    {
-        return $this->executionHandler->tryAsync($asyncFunction, $this->contextHandler, $this->awaitHandler);
-    }
-
     /**
      * Await a promise and return its resolved value.
      *

src/Async/Handlers/AsyncExecutionHandler.php

@@ -42,54 +42,4 @@ public function async(callable $asyncFunction): callable
             });
         };
     }
-
-    /**
-     * Convert a synchronous function into an asynchronous version.
-     *
-     * This is an alias for async() but with a name that emphasizes the
-     * conversion from synchronous to asynchronous execution.
-     *
-     * @param  callable  $syncFunction  The synchronous function to convert
-     * @return callable A function that returns a Promise when called
-     */
-    public function asyncify(callable $syncFunction): callable
-    {
-        return function (...$args) use ($syncFunction) {
-            return new Promise(function ($resolve, $reject) use ($syncFunction, $args) {
-                $fiber = new Fiber(function () use ($syncFunction, $args, $resolve, $reject) {
-                    try {
-                        $result = $syncFunction(...$args);
-                        $resolve($result);
-                    } catch (Throwable $e) {
-                        $reject($e);
-                    }
-                });
-
-                EventLoop::getInstance()->addFiber($fiber);
-            });
-        };
-    }
-
-    /**
-     * Create an async function that automatically awaits Promise results.
-     *
-     * This creates an async function that will automatically await any Promise
-     * returned by the wrapped function, providing a more convenient API for
-     * chaining async operations.
-     *
-     * @param  callable  $asyncFunction  The async function to wrap
-     * @param  FiberContextHandler  $contextHandler  Handler for fiber context validation
-     * @param  AwaitHandler  $awaitHandler  Handler for awaiting Promise results
-     * @return callable A function that automatically awaits results
-     */
-    public function tryAsync(callable $asyncFunction, FiberContextHandler $contextHandler, AwaitHandler $awaitHandler): callable
-    {
-        return $this->async(function (...$args) use ($asyncFunction, $awaitHandler) {
-            try {
-                return $awaitHandler->await($asyncFunction(...$args));
-            } catch (Throwable $e) {
-                throw $e; // Re-throw to be caught by calling code
-            }
-        });
-    }
 }

src/Async/Handlers/ConcurrencyHandler.php

@@ -18,21 +18,24 @@
 final readonly class ConcurrencyHandler
 {
     private AsyncExecutionHandler $executionHandler;
+    private AwaitHandler $awaitHandler;
 
     /**
      * @param  AsyncExecutionHandler  $executionHandler  Handler for async execution
      */
     public function __construct(AsyncExecutionHandler $executionHandler)
     {
         $this->executionHandler = $executionHandler;
+        $this->awaitHandler = new AwaitHandler(new FiberContextHandler);
     }
 
     /**
      * Execute multiple tasks concurrently with a specified concurrency limit.
      *
      * This method runs multiple tasks simultaneously while ensuring that no more
      * than the specified number of tasks run at the same time. Tasks can be either
-     * callable functions or existing Promise instances.
+     * callable functions or existing Promise instances. Promise instances will be
+     * automatically wrapped to ensure proper concurrency control.
      *
      * @param  array  $tasks  Array of callable tasks or Promise instances
      * @param  int  $concurrency  Maximum number of tasks to run simultaneously (default: 10)
@@ -45,29 +48,33 @@ public function concurrent(array $tasks, int $concurrency = 10): PromiseInterfac
         return new Promise(function ($resolve, $reject) use ($tasks, $concurrency) {
             if ($concurrency <= 0) {
                 $reject(new \InvalidArgumentException('Concurrency limit must be greater than 0'));
-
                 return;
             }
 
             if (empty($tasks)) {
                 $resolve([]);
-
                 return;
             }
 
             // Convert tasks to indexed array and preserve original keys
             $taskList = array_values($tasks);
             $originalKeys = array_keys($tasks);
 
+            // Process tasks to ensure proper async wrapping
+            $processedTasks = [];
+            foreach ($taskList as $index => $task) {
+                $processedTasks[$index] = $this->wrapTaskForConcurrency($task);
+            }
+
             $results = [];
             $running = 0;
             $completed = 0;
-            $total = count($taskList);
+            $total = count($processedTasks);
             $taskIndex = 0;
 
             $processNext = function () use (
                 &$processNext,
-                &$taskList,
+                &$processedTasks,
                 &$originalKeys,
                 &$running,
                 &$completed,
@@ -81,22 +88,19 @@ public function concurrent(array $tasks, int $concurrency = 10): PromiseInterfac
                 // Start as many tasks as we can up to the concurrency limit
                 while ($running < $concurrency && $taskIndex < $total) {
                     $currentIndex = $taskIndex++;
-                    $task = $taskList[$currentIndex];
+                    $task = $processedTasks[$currentIndex];
                     $originalKey = $originalKeys[$currentIndex];
                     $running++;
 
                     try {
-                        $promise = is_callable($task)
-                            ? $this->executionHandler->async($task)()
-                            : $task;
+                        $promise = $this->executionHandler->async($task)();
 
                         if (! ($promise instanceof PromiseInterface)) {
                             throw new RuntimeException('Task must return a Promise or be a callable that returns a Promise');
                         }
                     } catch (Throwable $e) {
                         $running--;
                         $reject($e);
-
                         return;
                     }
 
@@ -124,8 +128,7 @@ public function concurrent(array $tasks, int $concurrency = 10): PromiseInterfac
                         ->catch(function ($error) use (&$running, $reject) {
                             $running--;
                             $reject($error);
-                        })
-                    ;
+                        });
                 }
             };
 
@@ -139,7 +142,8 @@ public function concurrent(array $tasks, int $concurrency = 10): PromiseInterfac
      *
      * This method processes tasks in batches sequentially, where each batch
      * runs tasks concurrently up to the specified limit, but waits for the
-     * entire batch to complete before starting the next batch.
+     * entire batch to complete before starting the next batch. Promise instances
+     * will be automatically wrapped to ensure proper concurrency control.
      *
      * @param  array  $tasks  Array of callable tasks or Promise instances
      * @param  int  $batchSize  Number of tasks per batch (default: 10)
@@ -151,22 +155,27 @@ public function batch(array $tasks, int $batchSize = 10, ?int $concurrency = nul
         return new Promise(function ($resolve, $reject) use ($tasks, $batchSize, $concurrency) {
             if ($batchSize <= 0) {
                 $reject(new \InvalidArgumentException('Batch size must be greater than 0'));
-
                 return;
             }
 
             if (empty($tasks)) {
                 $resolve([]);
-
                 return;
             }
 
             $concurrency = $concurrency ?? $batchSize;
 
-            // Preserve original keys
+            // Preserve original keys and wrap tasks
             $originalKeys = array_keys($tasks);
             $taskValues = array_values($tasks);
-            $batches = array_chunk($taskValues, $batchSize, false);
+
+            // Process tasks to ensure proper async wrapping
+            $processedTasks = [];
+            foreach ($taskValues as $index => $task) {
+                $processedTasks[$index] = $this->wrapTaskForConcurrency($task);
+            }
+
+            $batches = array_chunk($processedTasks, $batchSize, false);
             $keyBatches = array_chunk($originalKeys, $batchSize, false);
 
             $allResults = [];
@@ -186,7 +195,6 @@ public function batch(array $tasks, int $batchSize = 10, ?int $concurrency = nul
             ) {
                 if ($batchIndex >= $totalBatches) {
                     $resolve($allResults);
-
                     return;
                 }
 
@@ -205,11 +213,44 @@ public function batch(array $tasks, int $batchSize = 10, ?int $concurrency = nul
                         $batchIndex++;
                         EventLoop::getInstance()->nextTick($processNextBatch);
                     })
-                    ->catch($reject)
-                ;
+                    ->catch($reject);
             };
 
             EventLoop::getInstance()->nextTick($processNextBatch);
         });
     }
+
+    /**
+     * Wrap a task to ensure proper concurrency control.
+     *
+     * This method ensures all tasks use the await pattern for proper fiber-based concurrency:
+     * - All callables are wrapped to ensure their results are awaited
+     * - Promise instances are wrapped with await
+     * - Other types are wrapped in a callable
+     *
+     * @param mixed $task The task to wrap
+     * @return callable A callable that properly defers execution
+     */
+    private function wrapTaskForConcurrency(mixed $task): callable
+    {
+        if (is_callable($task)) {
+            return function () use ($task) {
+                $result = $task();
+                if ($result instanceof PromiseInterface) {
+                    return $this->awaitHandler->await($result);
+                }
+                return $result;
+            };
+        }
+
+        if ($task instanceof PromiseInterface) {
+            return function () use ($task) {
+                return $this->awaitHandler->await($task);
+            };
+        }
+
+        return function () use ($task) {
+            return $task;
+        };
+    }
 }