Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
61 / 61
100.00%
12 / 12
 CRAP
100.00%
1 / 1
FileCalls
100.00%
61 / 61
100.00%
12 / 12
 21
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 current
100.00%
7 / 7
100.00%
1 / 1
 2
 use
100.00%
1 / 1
100.00%
1 / 1
 2
 open
100.00%
11 / 11
100.00%
1 / 1
 4
 run
100.00%
1 / 1
100.00%
1 / 1
 1
 close
100.00%
1 / 1
100.00%
1 / 1
 1
 detach
100.00%
1 / 1
100.00%
1 / 1
 1
 begin
100.00%
7 / 7
100.00%
1 / 1
 1
 closing
100.00%
6 / 6
100.00%
1 / 1
 1
 await
100.00%
4 / 4
100.00%
1 / 1
 2
 take
100.00%
14 / 14
100.00%
1 / 1
 3
 give
100.00%
7 / 7
100.00%
1 / 1
 2
1<?php
2
3/**
4 * SPDX-License-Identifier: EUPL-1.2
5 *
6 * This file is part of icap-flow.
7 *
8 * Licensed under the EUPL, Version 1.2 only (the "Licence");
9 * you may not use this work except in compliance with the Licence.
10 * You may obtain a copy of the Licence at:
11 *
12 *     https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the Licence is distributed on an "AS IS" basis,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 */
18
19declare(strict_types=1);
20
21namespace Ndrstmr\Icap;
22
23use Amp\ByteStream\StreamException;
24use Amp\Cancellation;
25use Amp\CancelledException;
26use Amp\File\Driver\ParallelFilesystemDriver;
27use Amp\File\File;
28use Amp\File\FilesystemDriver;
29use Amp\File\FilesystemException;
30use Amp\Future;
31use Amp\NullCancellation;
32use Amp\Parallel\Context\ContextException;
33use Amp\Parallel\Worker\ContextWorkerPool;
34use Amp\Parallel\Worker\WorkerException;
35use Ndrstmr\Icap\Exception\IcapStorageException;
36use Revolt\EventLoop;
37use Revolt\EventLoop\Driver;
38use Revolt\EventLoop\Suspension;
39
40use function Amp\async;
41
42/**
43 * The calls of the client to its file driver: worker processes of amphp/parallel, one set per event loop, whatever
44 * driver the application configured for its own files. Only these are known to give a regular file a size and an end;
45 * a blocking driver would hold the event loop with the first open.
46 *
47 * A worker that hangs in the operating system does not answer, and no cancellation reaches it. Each call therefore
48 * runs in a fiber of its own and the operation only stops waiting for it; the call stays in the worker. It holds one
49 * of {@see MAX_PENDING} places until the driver has returned, not until the wait is over, and so does the close of a
50 * file. A caller that finds no place waits for one as long as its operation lives, and sends nothing to the driver,
51 * and a call whose operation ended while it waited or was scheduled does not start.
52 *
53 * The type of a path is looked up by a worker of its own. The workers of the files are leased to them for as long as
54 * they are open, so a lookup that waited for one would wait for the end of scans that wait for the places it holds.
55 *
56 * @internal
57 */
58final class FileCalls
59{
60    /** Driver calls that may be on their way at the same time, those nobody waits for any more included. */
61    public const int MAX_PENDING = 64;
62
63    /** The bits of the mode of a stat result that give the type of the file, and the type of a regular file. */
64    private const int TYPE_MASK = 0o170000;
65
66    private const int REGULAR = 0o100000;
67
68    /** @var \WeakMap<Driver, self>|null */
69    private static ?\WeakMap $perLoop = null;
70
71    private static ?self $replacement = null;
72
73    private int $free = self::MAX_PENDING;
74
75    private int $turns = 0;
76
77    /** @var array<int, Suspension<null>> callers waiting for a place, in the order they came */
78    private array $waiting = [];
79
80    private function __construct(
81        private readonly FilesystemDriver $files,
82        private readonly FilesystemDriver $paths,
83    ) {
84    }
85
86    public static function current(): self
87    {
88        if (self::$replacement !== null) {
89            return self::$replacement;
90        }
91        self::$perLoop ??= new \WeakMap();
92
93        return self::$perLoop[EventLoop::getDriver()] ??= new self(
94            new ParallelFilesystemDriver(),
95            new ParallelFilesystemDriver(new ContextWorkerPool(1)),
96        );
97    }
98
99    /**
100     * For tests: serves the calls from another driver with places of its own; null puts the worker processes back.
101     */
102    public static function use(?FilesystemDriver $driver): void
103    {
104        self::$replacement = $driver === null ? null : new self($driver, $driver);
105    }
106
107    /**
108     * Opens a regular file. Its type is looked up first, so that a FIFO is not opened and waited for, and a device or a
109     * directory is not read as if it were a file; a path that cannot be looked up is none (a stream wrapper, a path
110     * that does not exist).
111     */
112    public function open(string $path, Cancellation $cancellation): File
113    {
114        $opening = $this->begin(function () use ($path, $cancellation): File {
115            $mode = $this->paths->getStatus($path)['mode'] ?? null;
116            if (!is_int($mode) || ($mode & self::TYPE_MASK) !== self::REGULAR) {
117                throw new FilesystemException('Not a regular file: ' . $path);
118            }
119            // The lookup may have outlived the operation.
120            $cancellation->throwIfRequested();
121
122            return $this->files->openFile($path, 'r');
123        }, $cancellation);
124
125        try {
126            return $this->await($opening, $cancellation, 'Unable to open file: ' . $path);
127        } catch (CancelledException $e) {
128            // The worker may still deliver the file: it is closed, not used.
129            $opening->map($this->detach(...))->ignore();
130            throw $e;
131        }
132    }
133
134    /**
135     * @template T
136     *
137     * @param \Closure(): T $call
138     *
139     * @return T
140     */
141    public function run(Cancellation $cancellation, \Closure $call, string $failure): mixed
142    {
143        return $this->await($this->begin($call, $cancellation), $cancellation, $failure);
144    }
145
146    /**
147     * Closes a file whose scan is done. The close is begun in any case, waited for as long as the operation lives, and
148     * a close that fails is reported.
149     */
150    public function close(File $file, Cancellation $cancellation): void
151    {
152        $this->await($this->closing($file), $cancellation, 'Unable to close the file.');
153    }
154
155    /**
156     * Closes a file nobody uses any more because the scan failed or was given up. Nobody waits for the close, and a
157     * failure of it has nobody to tell.
158     */
159    public function detach(File $file): void
160    {
161        $this->closing($file)->ignore();
162    }
163
164    /**
165     * @template T
166     *
167     * @param \Closure(): T $call
168     *
169     * @return Future<T>
170     */
171    private function begin(\Closure $call, Cancellation $cancellation): Future
172    {
173        $this->take($cancellation);
174
175        /** @var Future<T> $future */
176        $future = async(function () use ($call, $cancellation): mixed {
177            try {
178                // The call starts a loop turn after the place came, and the operation may have ended by then.
179                $cancellation->throwIfRequested();
180
181                return $call();
182            } finally {
183                $this->give();
184            }
185        });
186
187        return $future;
188    }
189
190    /**
191     * @return Future<null>
192     */
193    private function closing(File $file): Future
194    {
195        /** @var Future<null> $future */
196        $future = async(function () use ($file): void {
197            $this->take(new NullCancellation());
198            try {
199                $file->close();
200            } finally {
201                $this->give();
202            }
203        });
204
205        return $future;
206    }
207
208    /**
209     * Waits for the call as long as the operation lives. What the file driver reports as a failure of the file is an
210     * {@see IcapStorageException}, unless the operation is over by then; anything else passes as it is.
211     *
212     * @template T
213     *
214     * @param Future<T> $call
215     *
216     * @return T
217     */
218    private function await(Future $call, Cancellation $cancellation, string $failure): mixed
219    {
220        try {
221            return $call->await($cancellation);
222        } catch (FilesystemException|StreamException|WorkerException|ContextException $e) {
223            // The end of the operation outranks the failure that came in with it.
224            $cancellation->throwIfRequested();
225
226            throw new IcapStorageException($failure, 0, $e);
227        }
228    }
229
230    private function take(Cancellation $cancellation): void
231    {
232        $cancellation->throwIfRequested();
233        if ($this->free > 0) {
234            --$this->free;
235
236            return;
237        }
238
239        $suspension = EventLoop::getSuspension();
240        $turn = $this->turns++;
241        $this->waiting[$turn] = $suspension;
242        $watch = $cancellation->subscribe(function (CancelledException $e) use ($turn, $suspension): void {
243            // Either the place or the end of the operation reaches the caller first, never both.
244            if (isset($this->waiting[$turn])) {
245                unset($this->waiting[$turn]);
246                $suspension->throw($e);
247            }
248        });
249
250        try {
251            $suspension->suspend();
252        } finally {
253            $cancellation->unsubscribe($watch);
254        }
255    }
256
257    private function give(): void
258    {
259        $turn = array_key_first($this->waiting);
260        if ($turn === null) {
261            ++$this->free;
262
263            return;
264        }
265
266        $suspension = $this->waiting[$turn];
267        unset($this->waiting[$turn]);
268        $suspension->resume();
269    }
270}