diff --git a/README.md b/README.md index a38ea4a..67a0073 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,9 @@ $handle = $heavyComputation - Type-safe IO operations - Composable effect chains, sequential with `flatMap` and `mapN`, parallel with `parMapN` - Error handling as values with `attempt()`, or recovery with `handleError()`, `handleErrorWith()` and `recover()` -- Resource management +- Resource management with `bracket`, `bracketCase`, `bracketE`, and `Resource` as a composable value +- `sequence` and `traverse` over lists of effects, `productR`, `delay`, `both` +- `race`, `either` and `timeout` on effects that run where they can be cancelled - **Async execution with `start()`** - Fork computations to background fibers - **Custom execution contexts** - Control how effects are executed diff --git a/docs/combinators.md b/docs/combinators.md index de8ddb4..9084243 100644 --- a/docs/combinators.md +++ b/docs/combinators.md @@ -37,88 +37,73 @@ $result = $empty->unsafeRun(); // throws NoSuchElementException ## Sequence and Traverse -These combinators help you work with collections of effects. +Both live in `Phunkie\Effect\Functions\io`, next to `io` and `bracket`. ### Sequence -`sequence` transforms a collection of effects into an effect of a collection: +`sequence` turns a list of effects into one effect of their results, run in order when the effect runs, keys kept: ```php -use function Phunkie\Effect\Functions\io\io; +use function Phunkie\Effect\Functions\io\{io, sequence}; -$ios = [ - io(function() { return 1; }), - io(function() { return 2; }), - io(function() { return 3; }) -]; +$ios = [io(fn() => 1), io(fn() => 2), io(fn() => 3)]; $result = sequence($ios)->unsafeRun(); // [1, 2, 3] ``` +A blocking effect in the list is awaited, so `sequence` yields values, never handles. + ### Traverse -`traverse` is like `sequence` but allows you to transform each value before sequencing: +`traverse` maps each value to an effect and sequences the results: ```php -use function Phunkie\Effect\Functions\io\io; +use function Phunkie\Effect\Functions\io\traverse; -$numbers = [1, 2, 3]; - -$result = traverse($numbers, fn($n) => - io(function() use ($n) { - return $n * 2; - }) -)->unsafeRun(); // [2, 4, 6] +$result = traverse([1, 2, 3], fn($n) => io(fn() => $n * 2))->unsafeRun(); // [2, 4, 6] ``` ## Race and Timeout -These combinators help you handle concurrent operations and timeouts. +`race`, `either` and `timeout` run an effect where it can be polled and cancelled. A blocking effect runs in the context it was given; any other effect runs in the context passed as the last argument, `ParallelExecutionContext` by default, which needs `ext-parallel`. A context whose handles cannot be polled, such as the fiber context, is refused with a `RuntimeException` that says so. ### Race -`race` runs two effects concurrently and returns the result of the first one to complete: +`race` returns the result of whichever effect finishes first and cancels the other: ```php -use function Phunkie\Effect\Functions\io\io; +use Phunkie\Effect\Concurrent\ParallelExecutionContext; + use function Phunkie\Effect\Functions\blocking\blocking; -$fast = io(function() { - return blocking(function() { - usleep(100000); // 0.1 seconds - return "fast"; - }); -}); +$fast = blocking(function() { + usleep(100000); -$slow = io(function() { - return blocking(function() { - usleep(200000); // 0.2 seconds - return "slow"; - }); -}); + return "fast"; +}, new ParallelExecutionContext()); + +$slow = blocking(function() { + usleep(200000); + + return "slow"; +}, new ParallelExecutionContext()); $result = $fast->race($slow)->unsafeRun(); // "fast" ``` +`either` is the same race that says who won: `Left` for the effect it was called on, `Right` for the other. + ### Timeout -`timeout` adds a timeout to an effect, failing if it takes too long: +`timeout` gives an effect a number of milliseconds; when they run out the effect is cancelled and a `Phunkie\Effect\Concurrent\TimeoutException` is raised: ```php -use function Phunkie\Effect\Functions\io\io; -use function Phunkie\Effect\Functions\blocking\blocking; - -$slow = io(function() { - return blocking(function() { - sleep(2); - return "done"; - }); -}); +use Phunkie\Effect\Concurrent\TimeoutException; try { - $result = $slow->timeout(1000)->unsafeRun(); // throws TimeoutException -} catch (\Exception $e) { - // Handle timeout + $result = $slow->timeout(1000)->unsafeRun(); +} catch (TimeoutException $e) { + // Timed out after 1000 ms. } ``` @@ -195,67 +180,3 @@ $message = $result->fold(fn($error) => $error->getMessage())(fn($value) => "ok: $result = $value * 2; ``` -2. **Use Sequence for Collections** - ```php - // Good: Using sequence for collections - $results = sequence($ios)->unsafeRun(); - - // Bad: Manual sequencing - $results = []; - foreach ($ios as $io) { - $results[] = $io->unsafeRun(); - } - ``` - -3. **Use Race for Concurrent Operations** - ```php - // Good: Using race for concurrent operations - $result = $operation1->race($operation2)->unsafeRun(); - - // Bad: Manual concurrent execution - $handle1 = $operation1->unsafeRun(); - $handle2 = $operation2->unsafeRun(); - $result = $handle1->await(); // Might wait unnecessarily - ``` - -4. **Use Timeout for Long-Running Operations** - ```php - // Good: Using timeout for long-running operations - $result = $operation->timeout(1000)->unsafeRun(); - - // Bad: No timeout - $result = $operation->unsafeRun(); // Might hang indefinitely - ``` - -5. **Use Error Handling Combinators** - ```php - // Good: Using error handling combinators - $result = $risky->handleError(fn($e) => "fallback")->unsafeRun(); - - // Bad: Try-catch blocks - try { - $result = $risky->unsafeRun(); - } catch (\Exception $e) { - $result = "fallback"; - } - ``` - -Remember that combinators are designed to work together. You can chain them to create complex effect compositions: - -```php -$result = $io - ->map(fn($x) => $x * 2) - ->filter(fn($x) => $x > 10) - ->timeout(1000) - ->handleError(fn($e) => 0) - ->unsafeRun(); -``` - -This composition: -1. Doubles the value -2. Filters out values less than 10 -3. Times out after 1 second -4. Returns 0 if any error occurs -5. Executes the entire chain - -The power of combinators lies in their ability to compose effects in a declarative way, making your code more readable and maintainable. \ No newline at end of file diff --git a/docs/concurrency.md b/docs/concurrency.md index 34e29a3..e8618d3 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -207,130 +207,63 @@ Remember that parallel execution is not a silver bullet. Consider: When in doubt, prefer simpler concurrency models like fibers for I/O operations and only use parallel execution for CPU-intensive tasks that benefit from true parallelism. -## Sequence and Traverse +## Race, Either and Timeout -These combinators help you work with collections of effects: +These run an effect where it can be polled and cancelled: a blocking effect in the context it was given, any other effect in the context passed as the last argument, `ParallelExecutionContext` by default. The fiber context cannot be polled, so it is refused with a `RuntimeException`. -### Sequence ```php -use function Phunkie\Effect\Functions\io\io; - -$ios = [ - io(function() { return 1; }), - io(function() { return 2; }), - io(function() { return 3; }) -]; - -$result = sequence($ios)->unsafeRun(); // [1, 2, 3] -``` +use Phunkie\Effect\Concurrent\ParallelExecutionContext; +use Phunkie\Effect\Concurrent\TimeoutException; -### Traverse -```php -use function Phunkie\Effect\Functions\io\io; +$fast = blocking(fn() => fetchFromCache(), new ParallelExecutionContext()); +$slow = blocking(fn() => fetchFromOrigin(), new ParallelExecutionContext()); -$numbers = [1, 2, 3]; +$result = $fast->race($slow)->unsafeRun(); // whichever finished first; the other is cancelled +$which = $fast->either($slow)->unsafeRun(); // Left(result) when $fast won, Right(result) when $slow did -$result = traverse($numbers, fn($n) => - io(function() use ($n) { - return $n * 2; - }) -)->unsafeRun(); // [2, 4, 6] +try { + $result = $slow->timeout(1000)->unsafeRun(); // the result, or TimeoutException after 1000 ms +} catch (TimeoutException $e) { +} ``` -## Race, Both, Either +## Both -These combinators help you handle concurrent operations in different ways: +`both` runs this effect, then the other, and pairs the results: -### Race ```php -$fast = io(function() { - return blocking(function() { - usleep(100000); // 0.1 seconds - return "fast"; - }); -}); +$result = $operation1->both($operation2)->unsafeRun(); // [result1, result2] +``` -$slow = io(function() { - return blocking(function() { - usleep(200000); // 0.2 seconds - return "slow"; - }); -}); +It is sequential; `parMap2` is the parallel form. -$result = $fast->race($slow)->unsafeRun(); // "fast" -``` +## Delay -### Both -```php -$result = $operation1->both($operation2)->unsafeRun(); -// Returns [result1, result2] -``` +`delay` runs the effect once the given milliseconds have passed: -### Either ```php -$result = $operation1->either($operation2)->unsafeRun(); -// Returns Left(result1) or Right(result2) +$result = $operation->delay(1000)->unsafeRun(); // runs after 1 second ``` ## Cancellation -Cancellation allows you to stop operations that are no longer needed: +A handle from the parallel context is a `CancellableHandle`: it can be asked whether the computation has finished and can be cancelled. `race` and `timeout` use that to stop the effect that lost or ran out of time. ```php -$operation = io(function() { - return blocking(function() { - sleep(10); - return "done"; - }); -}); - -$handle = $operation->unsafeRun(); +$handle = $operation->start(new ParallelExecutionContext())->unsafeRun(); // ... some time later ... -$handle->cancel(); -``` - -## Delay - -Delay allows you to schedule operations to run after a specified time: - -```php -$operation = io(function() { - return blocking(function() { - return "delayed"; - }); -}); - -$result = $operation->delay(1000)->unsafeRun(); // Runs after 1 second +if (! $handle->isDone()) { + $handle->cancel(); +} ``` -## Channels - -Channels provide a way to communicate between concurrent operations: - -```php -$channel = new Channel(); - -$producer = io(function() use ($channel) { - return blocking(function() use ($channel) { - $channel->send("message"); - }); -}); - -$consumer = io(function() use ($channel) { - return blocking(function() use ($channel) { - return $channel->receive(); - }); -}); - -$result = $producer->both($consumer)->unsafeRun(); -``` +A handle from the fiber context runs the computation when it is awaited and cannot be cancelled. ## Best Practices 1. **Choose the Right Execution Context** - Use Fiber context for I/O operations - Use Parallel context for CPU-intensive tasks - - Use Process context for isolated operations 2. **Handle Resources Properly** ```php @@ -350,29 +283,3 @@ $result = $producer->both($consumer)->unsafeRun(); $result = $operation1->flatMap(fn($a) => $operation2->map(fn($b) => $a + $b)); ``` -4. **Handle Cancellation** - ```php - $handle = $operation->unsafeRun(); - try { - $result = $handle->await(); - } catch (CancellationException $e) { - // Handle cancellation - } - ``` - -5. **Use Error Handling** - ```php - $result = $operation - ->handleError(fn($e) => "fallback") - ->timeout(1000) - ->unsafeRun(); - ``` - -Remember that concurrency is a powerful tool, but it comes with complexity. Always consider: -- Resource management -- Error handling -- Cancellation -- Timeouts -- State management - -The key is to use the right tool for the job and handle all edge cases appropriately. \ No newline at end of file diff --git a/docs/index.md b/docs/index.md index 3eaa174..1bf8618 100644 --- a/docs/index.md +++ b/docs/index.md @@ -42,6 +42,8 @@ Master the fundamental building blocks: ### 4. [Combinators](combinators.md) - Core IO Operations + - Sequence and Traverse + - Race and Timeout - Error Handling - Composition Patterns @@ -49,18 +51,14 @@ Master the fundamental building blocks: - Blockers - Execution Context - Parallel Execution - - Sequence and Traverse - - Race, Both, Either + - Race, Either and Timeout + - Both and Delay - Cancellation - - Delay - - Channels ### 6. [Resources](resources.md) - - Brackets - - Resource Combinators - - Files - - URL - - Database + - Brackets, bracketCase and bracketE + - Resource + - Files, database and network connections ## Contributing diff --git a/docs/io.md b/docs/io.md index caaafb5..3bb90db 100644 --- a/docs/io.md +++ b/docs/io.md @@ -131,18 +131,16 @@ $workflow = readConfig('config.json') ->flatMap(fn($users) => sendEmails($users)); ``` -### productR (*>) - Sequence and Discard First +### productR - Sequence and Discard First ```php // Run both effects, keep only the second result $program = writeLog('Starting...') - ->productR(performOperation()) - ->productR(writeLog('Done')); + ->productR(performOperation()); // Equivalent to: $program = writeLog('Starting...') - ->flatMap(fn($_) => performOperation()) - ->flatMap(fn($result) => writeLog('Done')->map(fn($_) => $result)); + ->flatMap(fn($_) => performOperation()); ``` ## Combining IO @@ -318,7 +316,7 @@ plain `for { $x <- io } yield ...` (no guard) works as usual, desugaring to $program = fetchFromApi() ->flatMap(fn($data) => io(fn() => processData($data))) ->handleErrorWith(fn($error) => io(fn() => logError($error)) - ->productR(io(fn() => getDefaultData()))); + ->flatMap(fn() => io(fn() => getDefaultData()))); ``` ## Running IO diff --git a/docs/resources.md b/docs/resources.md index a5429af..e001704 100644 --- a/docs/resources.md +++ b/docs/resources.md @@ -50,39 +50,68 @@ $result = bracket( ### BracketCase -`bracketCase` allows different cleanup strategies based on how the resource was used: +`bracketCase` tells the release step how the use step ended, as an `ExitCase`: `Completed`, or `Error` when it threw. The error is rethrown once the release step has run. ```php +use Phunkie\Effect\IO\ExitCase; + +use function Phunkie\Effect\Functions\io\bracketCase; + $result = bracketCase( blocking(function() { return openFile(); }), function($file) { return readFile($file); }, - function($file, $exitCase) { + function($file, ExitCase $exitCase) { return match($exitCase) { ExitCase::Completed => closeFile($file), ExitCase::Error => closeAndLogError($file), - ExitCase::Cancelled => closeAndNotify($file) + ExitCase::Cancelled => closeAndNotify($file), }; } )->unsafeRun(); ``` +`ExitCase::Cancelled` exists for the day cancellation reaches brackets; nothing produces it yet. + ### BracketE -`bracketE` combines error handling with resource management: +`bracketE` hands the release step the error the use step raised, or `null` when it completed: ```php +use function Phunkie\Effect\Functions\io\bracketE; + $result = bracketE( blocking(function() { return openConnection(); }), function($conn) { return useConnection($conn); }, - function($conn, $error) { - return match($error) { - null => closeConnection($conn), - default => closeAndLogError($conn, $error) - }; + function($conn, ?Throwable $error) { + return null === $error ? closeConnection($conn) : closeAndLogError($conn, $error); } )->unsafeRun(); ``` +`bracket` is `bracketE` with a release step that ignores the error. + +## Resource + +A `Resource` is a bracket as a value: something acquired for the duration of a use step and released after it, however the step ends. Resources compose before anything is acquired, and `use()` turns the whole thing back into one `IO`: + +```php +use Phunkie\Effect\IO\Resource; + +$connection = Resource::make( + io(fn() => connect($dsn)), + fn($conn) => io(fn() => $conn->close()), +); + +$statement = $connection->flatMap(fn($conn) => Resource::make( + io(fn() => $conn->prepare($sql)), + fn($stmt) => io(fn() => $stmt->closeCursor()), +)); + +$rows = $statement->use(fn($stmt) => io(fn() => $stmt->fetchAll()))->unsafeRun(); +``` + +`flatMap` nests: the statement is acquired inside the connection and released before it. `map` transforms what was acquired; `Resource::pure($value)` and `Resource::eval($io)` lift a value or an effect that needs no releasing, so they compose with the ones that do. + ## Common Resource Patterns ### File Operations @@ -159,36 +188,7 @@ $result = bracket( } ``` -2. **Handle Errors in Release** - ```php - $result = bracketE( - blocking(function() { return openResource(); }), - function($r) { return useResource($r); }, - function($r, $error) { - return match($error) { - null => closeResource($r), - default => closeAndLogError($r, $error) - }; - } - ); - ``` - -3. **Use Appropriate Cleanup Strategies** - ```php - $result = bracketCase( - blocking(function() { return openResource(); }), - function($r) { return useResource($r); }, - function($r, $exitCase) { - return match($exitCase) { - ExitCase::Completed => closeResource($r), - ExitCase::Error => closeAndLogError($r), - ExitCase::Cancelled => closeAndNotify($r) - }; - } - ); - ``` - -4. **Compose Resource Operations** +2. **Compose Resource Operations** ```php $result = bracket( blocking(function() { return openResource(); }), @@ -203,13 +203,12 @@ $result = bracket( ); ``` -5. **Handle Resource Limits** +3. **Handle Resource Limits** ```php $result = bracket( blocking(function() { return openResource(); }), function($r) { return $operation - ->timeout(1000) ->handleError(fn($e) => "fallback"); }, function($r) { return closeResource($r); } diff --git a/src/Concurrent/CancellableHandle.php b/src/Concurrent/CancellableHandle.php new file mode 100644 index 0000000..10497b3 --- /dev/null +++ b/src/Concurrent/CancellableHandle.php @@ -0,0 +1,27 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Effect\Concurrent; + +/** + * A handle to a computation that is running on its own, so it can be asked whether it has + * finished without waiting for it, and stopped when its result is no longer wanted. + * race, either and timeout need handles of this kind. + * + * @template A + * @extends AsyncHandle + */ +interface CancellableHandle extends AsyncHandle +{ + public function isDone(): bool; + + public function cancel(): void; +} diff --git a/src/Concurrent/ParallelExecutionContext.php b/src/Concurrent/ParallelExecutionContext.php index dc3bbfc..3d07a59 100644 --- a/src/Concurrent/ParallelExecutionContext.php +++ b/src/Concurrent/ParallelExecutionContext.php @@ -58,7 +58,7 @@ public function executeAsync(callable $thunk): AsyncHandle $runtime = new Runtime(); // one thread per async task $future = $runtime->run($thunk); - return new class ($future) implements AsyncHandle { + return new class ($future) implements CancellableHandle { public function __construct(private readonly Future $future) { } @@ -67,6 +67,16 @@ public function await(): mixed { return $this->future->value(); // blocks when awaited } + + public function isDone(): bool + { + return $this->future->done(); + } + + public function cancel(): void + { + $this->future->cancel(); + } }; } else { throw new \RuntimeException("The 'parallel' extension is required for ParallelExecutionContext."); diff --git a/src/Concurrent/TimeoutException.php b/src/Concurrent/TimeoutException.php new file mode 100644 index 0000000..f2ed073 --- /dev/null +++ b/src/Concurrent/TimeoutException.php @@ -0,0 +1,21 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Effect\Concurrent; + +use RuntimeException; + +/** + * Raised by timeout() when the effect has not finished within its time; the effect was cancelled. + */ +final class TimeoutException extends RuntimeException +{ +} diff --git a/src/Functions/io.php b/src/Functions/io.php index 3bb5c3f..d5f4201 100644 --- a/src/Functions/io.php +++ b/src/Functions/io.php @@ -11,7 +11,9 @@ namespace Phunkie\Effect\Functions\io; +use Phunkie\Effect\IO\ExitCase; use Phunkie\Effect\IO\IO; +use Throwable; /** * Creates an IO effect from a value or a callable. @@ -35,30 +37,86 @@ function io($value): IO } /** - * Creates an IO that will acquire a resource, use it, and then release it. - * Ensures the release is always called, even if it fails. + * Acquire a resource, use it, release it, whatever way the use step ends. * - * @template T - * @template R - * @param IO $acquire The IO that acquires the resource - * @param callable(T): IO $use The function that uses the resource - * @param callable(T): IO $release The function that releases the resource - * @return IO + * @template A + * @template B + * @param IO $acquire + * @param callable(A): IO $use + * @param callable(A): IO $release + * @return IO */ function bracket(IO $acquire, callable $use, callable $release): IO +{ + return bracketE($acquire, $use, fn ($resource, ?Throwable $error) => $release($resource)); +} + +/** + * bracket whose release step is told how the use step ended: ExitCase::Completed, or ExitCase::Error. + * + * @template A + * @template B + * @param IO $acquire + * @param callable(A): IO $use + * @param callable(A, ExitCase): IO $release + * @return IO + */ +function bracketCase(IO $acquire, callable $use, callable $release): IO +{ + return bracketE($acquire, $use, fn ($resource, ?Throwable $error) => $release($resource, null === $error ? ExitCase::Completed : ExitCase::Error)); +} + +/** + * bracket whose release step is handed the error the use step raised, or null when it completed. + * The error is rethrown once the release step has run. + * + * @template A + * @template B + * @param IO $acquire + * @param callable(A): IO $use + * @param callable(A, Throwable|null): IO $release + * @return IO + */ +function bracketE(IO $acquire, callable $use, callable $release): IO { return new IO(function () use ($acquire, $use, $release) { $resource = $acquire->unsafeRun(); try { $result = $use($resource)->unsafeRun(); - $release($resource)->unsafeRun(); - - return $result; - } catch (\Throwable $e) { - $release($resource)->unsafeRun(); + } catch (Throwable $e) { + $release($resource, $e)->unsafeRun(); throw $e; } + $release($resource, null)->unsafeRun(); + + return $result; }); } + +/** + * A list of effects as one effect of their results, run in order, keys kept. + * + * @template A + * @param array> $ios + * @return IO> + */ +function sequence(array $ios): IO +{ + return new IO(fn () => array_map(fn (IO $io) => $io->unsafeRunSync(), $ios)); +} + +/** + * Each value through $f, the resulting effects sequenced. + * + * @template A + * @template B + * @param array $values + * @param callable(A): IO $f + * @return IO> + */ +function traverse(array $values, callable $f): IO +{ + return sequence(array_map($f, $values)); +} diff --git a/src/IO/ExitCase.php b/src/IO/ExitCase.php new file mode 100644 index 0000000..8e0673d --- /dev/null +++ b/src/IO/ExitCase.php @@ -0,0 +1,22 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Effect\IO; + +/** + * How the use step of a bracket ended, handed to the release step of bracketCase. + */ +enum ExitCase +{ + case Completed; + case Error; + case Cancelled; +} diff --git a/src/IO/IO.php b/src/IO/IO.php index a223175..c8ea4fe 100644 --- a/src/IO/IO.php +++ b/src/IO/IO.php @@ -19,6 +19,7 @@ use Phunkie\Effect\Concurrent\ExecutionContext; use Phunkie\Effect\Concurrent\FiberExecutionContext; use Phunkie\Effect\Ops\IO\ApplicativeOps; +use Phunkie\Effect\Ops\IO\ConcurrentOps; use Phunkie\Effect\Ops\IO\FunctorOps; use Phunkie\Effect\Ops\IO\MonadErrorOps; use Phunkie\Effect\Ops\IO\MonadOps; @@ -48,6 +49,7 @@ class IO implements Functor, Applicative, Monad, Parallel, Kind use MonadOps; use MonadErrorOps; use ParallelOps; + use ConcurrentOps; private $unsafeRun; @@ -169,6 +171,20 @@ public function mapN(array $fbs, callable $f): IO }); } + /** + * The same effect, run only once $millis have passed. + * + * @return IO + */ + public function delay(int $millis): IO + { + return new IO(function () use ($millis) { + usleep($millis * 1000); + + return ($this->unsafeRun)(); + }); + } + /** * Attempts to execute the effect, capturing any error in a Validation. * diff --git a/src/IO/Resource.php b/src/IO/Resource.php new file mode 100644 index 0000000..654b05e --- /dev/null +++ b/src/IO/Resource.php @@ -0,0 +1,104 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Effect\IO; + +use Closure; + +use function Phunkie\Effect\Functions\io\bracket; + +/** + * A bracket as a value: something acquired for the duration of a use step and released after it, + * whatever way the step ends. Resources compose: flatMap nests them, so the inner one is released + * first, and use() turns the whole thing back into one IO. + * + * @template A + */ +final class Resource +{ + /** + * @param Closure(callable(A): IO): IO $use + */ + private function __construct(private readonly Closure $use) + { + } + + /** + * A resource acquired by an IO and released by a function of what was acquired. + * + * @template B + * @param IO $acquire + * @param callable(B): IO $release + * @return Resource + */ + public static function make(IO $acquire, callable $release): self + { + return new self(fn (callable $f): IO => bracket($acquire, $f, $release)); + } + + /** + * A value that needs no acquiring or releasing. + * + * @template B + * @param B $value + * @return Resource + */ + public static function pure(mixed $value): self + { + return new self(fn (callable $f): IO => $f($value)); + } + + /** + * An IO's result, as a resource that needs no releasing. + * + * @template B + * @param IO $io + * @return Resource + */ + public static function eval(IO $io): self + { + return new self(fn (callable $f): IO => $io->flatMap($f)); + } + + /** + * Acquire, run $f on what was acquired, release, as one IO. + * + * @template B + * @param callable(A): IO $f + * @return IO + */ + public function use(callable $f): IO + { + return ($this->use)($f); + } + + /** + * @template B + * @param callable(A): B $f + * @return Resource + */ + public function map(callable $f): self + { + return new self(fn (callable $g): IO => $this->use(fn ($a) => $g($f($a)))); + } + + /** + * A resource that depends on this one: acquired inside it and released before it. + * + * @template B + * @param callable(A): Resource $f + * @return Resource + */ + public function flatMap(callable $f): self + { + return new self(fn (callable $g): IO => $this->use(fn ($a) => $f($a)->use($g))); + } +} diff --git a/src/Ops/IO/ConcurrentOps.php b/src/Ops/IO/ConcurrentOps.php new file mode 100644 index 0000000..0473049 --- /dev/null +++ b/src/Ops/IO/ConcurrentOps.php @@ -0,0 +1,128 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Effect\Ops\IO; + +use function Left; + +use Phunkie\Effect\Concurrent\Blocker; +use Phunkie\Effect\Concurrent\CancellableHandle; +use Phunkie\Effect\Concurrent\ExecutionContext; +use Phunkie\Effect\Concurrent\ParallelExecutionContext; +use Phunkie\Effect\Concurrent\TimeoutException; +use Phunkie\Effect\IO\IO; +use Phunkie\Types\Either; + +use function Right; + +use RuntimeException; + +/** + * Running effects against each other or against the clock. race, either and timeout start the + * effects where they can be polled and cancelled: a blocking effect runs in its own context, any + * other in the context given, the parallel one by default. + */ +trait ConcurrentOps +{ + /** + * Both results, this effect run first, as a pair. + * + * @template B + * @param IO $other + * @return IO + */ + public function both(IO $other): IO + { + return new IO(fn () => [$this->unsafeRunSync(), $other->unsafeRunSync()]); + } + + /** + * The result of whichever effect finishes first; the other is cancelled. + * + * @template B + * @param IO $other + * @return IO + */ + public function race(IO $other, ?ExecutionContext $context = null): IO + { + return new IO(fn () => $this->winner($other, $context)[1]); + } + + /** + * Like race, but says which one won: Left for this effect, Right for the other. + * + * @template B + * @param IO $other + * @return IO> + */ + public function either(IO $other, ?ExecutionContext $context = null): IO + { + return new IO(function () use ($other, $context) { + [$index, $value] = $this->winner($other, $context); + + return 0 === $index ? Left($value) : Right($value); + }); + } + + /** + * The result, provided it arrives within $millis; otherwise the effect is cancelled and a + * TimeoutException is raised. + * + * @return IO + */ + public function timeout(int $millis, ?ExecutionContext $context = null): IO + { + return new IO(function () use ($millis, $context) { + $handle = $this->handleIn($context ?? new ParallelExecutionContext()); + $deadline = hrtime(true) + $millis * 1_000_000; + while (! $handle->isDone()) { + if (hrtime(true) >= $deadline) { + $handle->cancel(); + + throw new TimeoutException(sprintf('Timed out after %d ms.', $millis)); + } + usleep(1000); + } + + return $handle->await(); + }); + } + + /** + * @return array{0: int, 1: mixed} which effect finished first, 0 for this one, and its result + */ + private function winner(IO $other, ?ExecutionContext $context): array + { + $context ??= new ParallelExecutionContext(); + $handles = [$this->handleIn($context), $other->handleIn($context)]; + while (true) { + foreach ($handles as $index => $handle) { + if ($handle->isDone()) { + $handles[1 - $index]->cancel(); + + return [$index, $handle->await()]; + } + } + usleep(1000); + } + } + + private function handleIn(ExecutionContext $context): CancellableHandle + { + $run = $this->unsafeRun; + $handle = $run instanceof Blocker ? $run() : $context->executeAsync($run); + if (! $handle instanceof CancellableHandle) { + throw new RuntimeException('race, either and timeout need an execution context whose handles can be polled and cancelled, such as ParallelExecutionContext; a blocking effect brings its own context.'); + } + + return $handle; + } +} diff --git a/src/Ops/IO/MonadOps.php b/src/Ops/IO/MonadOps.php index ad5634a..4089813 100644 --- a/src/Ops/IO/MonadOps.php +++ b/src/Ops/IO/MonadOps.php @@ -34,6 +34,18 @@ public function flatMap(callable $f): IO }); } + /** + * Runs this effect, then $next, keeping only the result of $next. + * + * @template B + * @param IO $next + * @return IO + */ + public function productR(IO $next): IO + { + return $this->flatMap(fn () => $next); + } + /** * Flattens a nested IO structure (IO> -> IO). * diff --git a/tests/Unit/Concurrent/RacingTest.php b/tests/Unit/Concurrent/RacingTest.php new file mode 100644 index 0000000..f2ea9e0 --- /dev/null +++ b/tests/Unit/Concurrent/RacingTest.php @@ -0,0 +1,177 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit\Phunkie\Effect\Concurrent; + +use parallel\Runtime; +use PHPUnit\Framework\Attributes\Test; +use PHPUnit\Framework\TestCase; +use Phunkie\Effect\Concurrent\AsyncHandle; +use Phunkie\Effect\Concurrent\CancellableHandle; +use Phunkie\Effect\Concurrent\ExecutionContext; +use Phunkie\Effect\Concurrent\FiberExecutionContext; +use Phunkie\Effect\Concurrent\ParallelExecutionContext; +use Phunkie\Effect\Concurrent\TimeoutException; + +use function Phunkie\Effect\Functions\blocking\blocking; +use function Phunkie\Effect\Functions\io\io; + +use RuntimeException; + +class RacingTest extends TestCase +{ + #[Test] + public function race_returns_whichever_finishes_first_and_cancels_the_other() + { + $context = new ScriptedContext([3, 1]); + + $winner = io(fn () => 'slow')->race(io(fn () => 'fast'), $context)->unsafeRun(); + + $this->assertSame('fast', $winner); + $this->assertSame([0], $context->cancelled); + } + + #[Test] + public function race_can_be_won_by_the_first_effect() + { + $context = new ScriptedContext([1, 3]); + + $this->assertSame('first', io(fn () => 'first')->race(io(fn () => 'second'), $context)->unsafeRun()); + $this->assertSame([1], $context->cancelled); + } + + #[Test] + public function either_says_which_effect_won() + { + $this->assertEquals(Left('a'), io(fn () => 'a')->either(io(fn () => 'b'), new ScriptedContext([1, 2]))->unsafeRun()); + $this->assertEquals(Right('b'), io(fn () => 'a')->either(io(fn () => 'b'), new ScriptedContext([2, 1]))->unsafeRun()); + } + + #[Test] + public function timeout_returns_the_result_when_it_arrives_in_time() + { + $this->assertSame(42, io(fn () => 42)->timeout(50, new ScriptedContext([2]))->unsafeRun()); + } + + #[Test] + public function timeout_cancels_the_effect_and_fails_when_the_time_is_up() + { + $context = new ScriptedContext([PHP_INT_MAX]); + + try { + io(fn () => 'never')->timeout(10, $context)->unsafeRun(); + $this->fail('Expected a TimeoutException.'); + } catch (TimeoutException $e) { + $this->assertSame('Timed out after 10 ms.', $e->getMessage()); + } + + $this->assertSame([0], $context->cancelled); + } + + #[Test] + public function racing_needs_handles_that_can_be_polled_and_cancelled() + { + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('polled and cancelled'); + + io(fn () => 1)->race(io(fn () => 2), new FiberExecutionContext())->unsafeRun(); + } + + #[Test] + public function nothing_runs_before_the_io_does() + { + $runs = 0; + $counting = function () use (&$runs) { + $runs++; + + return $runs; + }; + + $race = io($counting)->race(io($counting), new ScriptedContext([1, 1])); + + $this->assertSame(0, $runs); + $this->assertSame(1, $race->unsafeRun()); + } + + #[Test] + public function blocking_effects_race_in_the_parallel_context() + { + if (! class_exists(Runtime::class)) { + $this->markTestSkipped("The 'parallel' extension is not available."); + } + + $slow = blocking(function () { + usleep(200000); + + return 'slow'; + }, new ParallelExecutionContext()); + $fast = blocking(function () { + usleep(20000); + + return 'fast'; + }, new ParallelExecutionContext()); + + $this->assertSame('fast', $slow->race($fast)->unsafeRun()); + } +} + +/** + * An execution context for the specs: each computation reports itself done after a scripted + * number of polls, so races and timeouts are decided by the script, not by timing. + */ +final class ScriptedContext implements ExecutionContext +{ + /** @var list */ + public array $cancelled = []; + + private int $started = 0; + + /** + * @param list $pollsUntilDone one entry per computation started, in order + */ + public function __construct(private array $pollsUntilDone) + { + } + + public function execute(callable $thunk): mixed + { + return $thunk(); + } + + public function executeAsync(callable $thunk): AsyncHandle + { + $index = $this->started++; + $polls = $this->pollsUntilDone[$index]; + + return new class ($thunk, $polls, function () use ($index) { + $this->cancelled[] = $index; + }) implements CancellableHandle { + public function __construct(private $thunk, private int $polls, private \Closure $onCancel) + { + } + + public function isDone(): bool + { + return --$this->polls <= 0; + } + + public function cancel(): void + { + ($this->onCancel)(); + } + + public function await(): mixed + { + return ($this->thunk)(); + } + }; + } +} diff --git a/tests/Unit/Functions/sequenceTest.php b/tests/Unit/Functions/sequenceTest.php new file mode 100644 index 0000000..1e12615 --- /dev/null +++ b/tests/Unit/Functions/sequenceTest.php @@ -0,0 +1,60 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit\Phunkie\Effect\Functions; + +use PHPUnit\Framework\Attributes\Test; +use PHPUnit\Framework\TestCase; + +use function Phunkie\Effect\Functions\blocking\blocking; +use function Phunkie\Effect\Functions\io\io; +use function Phunkie\Effect\Functions\io\sequence; +use function Phunkie\Effect\Functions\io\traverse; + +class sequenceTest extends TestCase +{ + #[Test] + public function sequence_turns_a_list_of_effects_into_an_effect_of_a_list() + { + $this->assertSame([1, 2, 3], sequence([io(fn () => 1), io(fn () => 2), io(fn () => 3)])->unsafeRun()); + } + + #[Test] + public function sequence_runs_the_effects_in_order_and_only_when_run() + { + $log = []; + $logging = function (string $entry) use (&$log) { + return io(function () use (&$log, $entry) { + $log[] = $entry; + + return $entry; + }); + }; + + $all = sequence(['first' => $logging('a'), 'second' => $logging('b')]); + $this->assertSame([], $log); + + $this->assertSame(['first' => 'a', 'second' => 'b'], $all->unsafeRun()); + $this->assertSame(['a', 'b'], $log); + } + + #[Test] + public function sequence_awaits_blocking_effects() + { + $this->assertSame([2, 3], sequence([blocking(fn () => 2), blocking(fn () => 3)])->unsafeRun()); + } + + #[Test] + public function traverse_maps_each_value_to_an_effect_and_sequences_them() + { + $this->assertSame([2, 4, 6], traverse([1, 2, 3], fn (int $n) => io(fn () => $n * 2))->unsafeRun()); + } +} diff --git a/tests/Unit/IO/BracketCaseTest.php b/tests/Unit/IO/BracketCaseTest.php new file mode 100644 index 0000000..dde2be1 --- /dev/null +++ b/tests/Unit/IO/BracketCaseTest.php @@ -0,0 +1,68 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit\Phunkie\Effect\IO; + +use PHPUnit\Framework\Attributes\Test; +use PHPUnit\Framework\TestCase; + +use function Phunkie\Effect\Functions\io\bracketCase; +use function Phunkie\Effect\Functions\io\bracketE; +use function Phunkie\Effect\Functions\io\io; + +use Phunkie\Effect\IO\ExitCase; +use RuntimeException; +use Throwable; + +class BracketCaseTest extends TestCase +{ + #[Test] + public function bracket_case_tells_the_release_step_how_the_use_step_ended() + { + $cases = []; + $release = function (string $resource, ExitCase $case) use (&$cases) { + return io(function () use ($resource, $case, &$cases) { + $cases[$resource] = $case; + }); + }; + + $result = bracketCase(io(fn () => 'ok'), fn () => io(fn () => 'done'), $release)->unsafeRun(); + + try { + bracketCase(io(fn () => 'bad'), fn () => io(fn () => throw new RuntimeException('boom')), $release)->unsafeRun(); + $this->fail('Expected the error to propagate.'); + } catch (RuntimeException $e) { + } + + $this->assertSame('done', $result); + $this->assertSame(['ok' => ExitCase::Completed, 'bad' => ExitCase::Error], $cases); + } + + #[Test] + public function bracket_e_hands_the_release_step_the_error_or_null() + { + $errors = []; + $release = function (string $resource, ?Throwable $error) use (&$errors) { + return io(function () use ($resource, $error, &$errors) { + $errors[$resource] = $error?->getMessage(); + }); + }; + + bracketE(io(fn () => 'ok'), fn () => io(fn () => 1), $release)->unsafeRun(); + + try { + bracketE(io(fn () => 'bad'), fn () => io(fn () => throw new RuntimeException('boom')), $release)->unsafeRun(); + } catch (RuntimeException $e) { + } + + $this->assertSame(['ok' => null, 'bad' => 'boom'], $errors); + } +} diff --git a/tests/Unit/IO/CombinatorsTest.php b/tests/Unit/IO/CombinatorsTest.php new file mode 100644 index 0000000..1c5d56c --- /dev/null +++ b/tests/Unit/IO/CombinatorsTest.php @@ -0,0 +1,63 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit\Phunkie\Effect\IO; + +use PHPUnit\Framework\Attributes\Test; +use PHPUnit\Framework\TestCase; + +use function Phunkie\Effect\Functions\blocking\blocking; +use function Phunkie\Effect\Functions\io\io; + +class CombinatorsTest extends TestCase +{ + #[Test] + public function product_r_runs_both_and_keeps_the_second_result() + { + $log = []; + $logging = function (string $entry) use (&$log) { + return io(function () use (&$log, $entry) { + $log[] = $entry; + + return $entry; + }); + }; + + $this->assertSame('second', $logging('first')->productR($logging('second'))->unsafeRun()); + $this->assertSame(['first', 'second'], $log); + } + + #[Test] + public function delay_waits_before_running() + { + $started = hrtime(true); + + $this->assertSame('late', io(fn () => 'late')->delay(20)->unsafeRun()); + $this->assertGreaterThanOrEqual(20, (hrtime(true) - $started) / 1_000_000); + } + + #[Test] + public function both_pairs_the_results_this_effect_first() + { + $log = []; + $logging = function (string $entry) use (&$log) { + return io(function () use (&$log, $entry) { + $log[] = $entry; + + return $entry; + }); + }; + + $this->assertSame(['a', 'b'], $logging('a')->both($logging('b'))->unsafeRun()); + $this->assertSame(['a', 'b'], $log); + $this->assertSame([1, 2], blocking(fn () => 1)->both(blocking(fn () => 2))->unsafeRun()); + } +} diff --git a/tests/Unit/IO/ResourceTest.php b/tests/Unit/IO/ResourceTest.php new file mode 100644 index 0000000..4628266 --- /dev/null +++ b/tests/Unit/IO/ResourceTest.php @@ -0,0 +1,91 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit\Phunkie\Effect\IO; + +use PHPUnit\Framework\Attributes\Test; +use PHPUnit\Framework\TestCase; + +use function Phunkie\Effect\Functions\io\io; + +use Phunkie\Effect\IO\Resource; +use RuntimeException; + +class ResourceTest extends TestCase +{ + private array $log = []; + + #[Test] + public function it_acquires_for_the_use_step_and_releases_after_it() + { + $result = $this->resource('db')->use(fn (string $db) => $this->logged("query $db", 'rows'))->unsafeRun(); + + $this->assertSame('rows', $result); + $this->assertSame(['open db', 'query db', 'close db'], $this->log); + } + + #[Test] + public function it_releases_when_the_use_step_throws_and_rethrows() + { + try { + $this->resource('db')->use(fn () => io(fn () => throw new RuntimeException('boom')))->unsafeRun(); + $this->fail('Expected the error to propagate.'); + } catch (RuntimeException $e) { + $this->assertSame('boom', $e->getMessage()); + } + + $this->assertSame(['open db', 'close db'], $this->log); + } + + #[Test] + public function it_nests_with_flat_map_and_releases_the_inner_one_first() + { + $both = $this->resource('db')->flatMap(fn (string $db) => $this->resource("cache for $db")); + + $both->use(fn (string $cache) => $this->logged("use $cache", null))->unsafeRun(); + + $this->assertSame(['open db', 'open cache for db', 'use cache for db', 'close cache for db', 'close db'], $this->log); + } + + #[Test] + public function it_maps_what_was_acquired() + { + $this->assertSame('DB', $this->resource('db')->map(strtoupper(...))->use(fn (string $name) => io(fn () => $name))->unsafeRun()); + } + + #[Test] + public function it_lifts_a_value_and_an_io_as_resources_that_need_no_release() + { + $this->assertSame(3, Resource::pure(1)->flatMap(fn (int $a) => Resource::eval(io(fn () => $a + 2)))->use(fn (int $b) => io(fn () => $b))->unsafeRun()); + } + + #[Test] + public function nothing_is_acquired_until_the_io_runs() + { + $this->resource('db')->use(fn () => io(fn () => null)); + + $this->assertSame([], $this->log); + } + + private function resource(string $name): Resource + { + return Resource::make($this->logged("open $name", $name), fn (string $acquired) => $this->logged("close $acquired", null)); + } + + private function logged(string $entry, mixed $value) + { + return io(function () use ($entry, $value) { + $this->log[] = $entry; + + return $value; + }); + } +}