<repo_name>b1rdex/sync<file_sep>/test/FileMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\FileMutex;
use Amp\Sync\Mutex;

class FileMutexTest extends AbstractMutexTest
{
    public function createMutex(): Mutex
    {
        return new FileMutex(\tempnam(\sys_get_temp_dir(), 'mutex-') . '.lock');
    }
}
<file_sep>/test/AbstractKeyedSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Loop;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\KeyedSemaphore;

abstract class AbstractKeyedSemaphoreTest extends AsyncTestCase
{
    abstract public function createSemaphore(int $size): KeyedSemaphore;

    public function testAcquire(): \Generator
    {
        $mutex = $this->createSemaphore(1);
        $lock = yield $mutex->acquire('test');
        $lock->release();
        $this->assertTrue($lock->isReleased());
    }

    public function testAcquireMultiple(): \Generator
    {
        $this->setMinimumRuntime(300);
        $this->setTimeout(500);

        $mutex = $this->createSemaphore(5);

        for ($i = 0; $i < 15; $i++) {
            $lock = yield $mutex->acquire('test');
            Loop::delay(100, function () use ($lock) {
                $lock->release();
            });
        }
    }
}
<file_sep>/test/AbstractSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Loop;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\Semaphore;

abstract class AbstractSemaphoreTest extends AsyncTestCase
{
    /**
     * @var \Amp\Sync\Semaphore
     */
    protected $semaphore;

    /**
     * @param int $locks Number of locks in the semaphore.
     *
     * @return \Amp\Sync\Semaphore
     */
    abstract public function createSemaphore(int $locks): Semaphore;

    public function tearDown(): void
    {
        parent::tearDown();
        $this->semaphore = null; // Force Semaphore::__destruct() to be invoked.
    }

    public function testConstructorOnInvalidMaxLocks(): void
    {
        $this->expectException(\Error::class);

        $this->semaphore = $this->createSemaphore(-1);
    }

    public function testAcquire(): \Generator
    {
        $this->semaphore = $this->createSemaphore(1);

        $lock = yield $this->semaphore->acquire();

        $this->assertFalse($lock->isReleased());

        $lock->release();

        $this->assertTrue($lock->isReleased());
    }

    public function testAcquireMultipleFromSingleLockSemaphore(): \Generator
    {
        $this->setMinimumRuntime(300);

        $this->semaphore = $this->createSemaphore(1);

        $lock1 = yield $this->semaphore->acquire();
        $this->assertSame(0, $lock1->getId());
        Loop::delay(100, function () use ($lock1) {
            $lock1->release();
        });

        $lock2 = yield $this->semaphore->acquire();
        $this->assertSame(0, $lock2->getId());
        Loop::delay(100, function () use ($lock2) {
            $lock2->release();
        });

        $lock3 = yield $this->semaphore->acquire();
        $this->assertSame(0, $lock3->getId());
        Loop::delay(100, function () use ($lock3) {
            $lock3->release();
        });
    }

    public function testAcquireMultipleFromMultipleLockSemaphore(): \Generator
    {
        $this->setMinimumRuntime(300);

        $this->semaphore = $this->createSemaphore(3);

        $lock1 = yield $this->semaphore->acquire();
        Loop::delay(100, function () use ($lock1) {
            $lock1->release();
        });

        $lock2 = yield $this->semaphore->acquire();
        $this->assertNotSame($lock1->getId(), $lock2->getId());
        Loop::delay(200, function () use ($lock2) {
            $lock2->release();
        });

        $lock3 = yield $this->semaphore->acquire();
        $this->assertNotSame($lock1->getId(), $lock3->getId());
        $this->assertNotSame($lock2->getId(), $lock3->getId());
        Loop::delay(200, function () use ($lock3) {
            $lock3->release();
        });

        $lock4 = yield $this->semaphore->acquire();
        $this->assertSame($lock1->getId(), $lock4->getId());
        Loop::delay(200, function () use ($lock4) {
            $lock4->release();
        });
    }

    public function getSemaphoreSizes(): array
    {
        return [
            [5],
            [10],
            [20],
            [30],
        ];
    }

    /**
     * @dataProvider getSemaphoreSizes
     *
     * @param int $count Number of locks to test.
     */
    public function testAcquireFromMultipleSizeSemaphores(int $count): \Generator
    {
        $this->setMinimumRuntime(200);

        $this->semaphore = $this->createSemaphore($count);

        foreach (\range(0, $count - 1) as $value) {
            $this->semaphore->acquire()->onResolve(function ($exception, $lock) {
                if ($exception) {
                    throw $exception;
                }

                Loop::delay(100, [$lock, "release"]);
            });
        }

        $lock = yield $this->semaphore->acquire();
        Loop::delay(100, [$lock, "release"]);
    }

    public function testSimultaneousAcquire(): \Generator
    {
        $this->setMinimumRuntime(100);

        $this->semaphore = $this->createSemaphore(1);

        $promise1 = $this->semaphore->acquire();
        $promise2 = $this->semaphore->acquire();

        Loop::delay(100, function () use ($promise1) {
            (yield $promise1)->release();
        });

        (yield $promise2)->release();
    }
}
<file_sep>/test/ThreadedSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Delayed;
use Amp\Loop;
use Amp\Sync\Semaphore;
use Amp\Sync\ThreadedSemaphore;

/**
 * @requires extension pthreads
 */
class ThreadedSemaphoreTest extends AbstractSemaphoreTest
{
    public function createSemaphore(int $locks): Semaphore
    {
        return new ThreadedSemaphore($locks);
    }

    public function testWithinThread(): \Generator
    {
        $semaphore = $this->createSemaphore(1);

        $thread = new class($semaphore) extends \Thread {
            private $semaphore;

            public function __construct(Semaphore $semaphore)
            {
                $this->semaphore = $semaphore;
            }

            public function run()
            {
                // Protect scope by using an unbound closure (protects static access as well).
                (static function () {
                    $paths = [
                        \dirname(__DIR__) . \DIRECTORY_SEPARATOR . "vendor" . \DIRECTORY_SEPARATOR . "autoload.php",
                        \dirname(__DIR__, 3) . \DIRECTORY_SEPARATOR . "autoload.php",
                    ];

                    foreach ($paths as $path) {
                        if (\file_exists($path)) {
                            $autoloadPath = $path;
                            break;
                        }
                    }

                    if (!isset($autoloadPath)) {
                        throw new \Error("Could not locate autoload.php");
                    }

                    require $autoloadPath;
                })->bindTo(null, null)();

                Loop::run(function () {
                    $lock = yield $this->semaphore->acquire();
                    Loop::delay(1000, [$lock, "release"]);
                });
            }
        };

        $this->setMinimumRuntime(1100);

        $thread->start(\PTHREADS_INHERIT_INI);

        yield new Delayed(500); // Wait for thread to start and obtain lock.
        $lock = yield $semaphore->acquire();
        Loop::delay(100, [$lock, "release"]);
    }
}
<file_sep>/test/LocalKeyedSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\KeyedSemaphore;
use Amp\Sync\LocalKeyedSemaphore;

class LocalKeyedSemaphoreTest extends AbstractKeyedSemaphoreTest
{
    public function createSemaphore(int $size): KeyedSemaphore
    {
        return new LocalKeyedSemaphore($size);
    }
}
<file_sep>/test/ThreadedMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Delayed;
use Amp\Loop;
use Amp\Sync\Mutex;
use Amp\Sync\ThreadedMutex;

/**
 * @requires extension pthreads
 */
class ThreadedMutexTest extends AbstractMutexTest
{
    public function createMutex(): Mutex
    {
        return new ThreadedMutex;
    }

    public function testWithinThread(): \Generator
    {
        $mutex = $this->createMutex();

        $thread = new class($mutex) extends \Thread {
            private $mutex;

            public function __construct(Mutex $mutex)
            {
                $this->mutex = $mutex;
            }

            public function run()
            {
                // Protect scope by using an unbound closure (protects static access as well).
                (static function () {
                    $paths = [
                        \dirname(__DIR__) . \DIRECTORY_SEPARATOR . "vendor" . \DIRECTORY_SEPARATOR . "autoload.php",
                        \dirname(__DIR__, 3) . \DIRECTORY_SEPARATOR . "autoload.php",
                    ];

                    foreach ($paths as $path) {
                        if (\file_exists($path)) {
                            $autoloadPath = $path;
                            break;
                        }
                    }

                    if (!isset($autoloadPath)) {
                        throw new \Error("Could not locate autoload.php");
                    }

                    require $autoloadPath;
                })->bindTo(null, null)();

                Loop::run(function () {
                    $lock = yield $this->mutex->acquire();
                    Loop::delay(1000, [$lock, "release"]);
                });
            }
        };

        $this->setMinimumRuntime(1100);

        $thread->start(\PTHREADS_INHERIT_INI);

        yield new Delayed(500); // Wait for thread to start and obtain lock.
        $lock = yield $mutex->acquire();
        Loop::delay(100, [$lock, "release"]);
    }
}
<file_sep>/test/ConcurrentEachTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Iterator;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\LocalSemaphore;
use function Amp\delay;
use function Amp\Sync\ConcurrentIterator\each;

class ConcurrentEachTest extends AsyncTestCase
{
    public function test(): \Generator
    {
        $this->expectOutputString('123');

        $processor = static function ($job) {
            print $job;
        };

        $this->assertSame(
            3,
            yield each(Iterator\fromIterable([1, 2, 3]), new LocalSemaphore(3), $processor)
        );
    }

    public function testOutputOrder(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);
        };

        $this->assertSame(
            3,
            yield each(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(3), $processor)
        );
    }

    public function testOutputOrderWithoutConcurrency(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);
        };

        $this->assertSame(
            3,
            yield each(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(1), $processor)
        );
    }

    public function testErrorHandling(): \Generator
    {
        $processor = static function ($job) {
            print $job;

            yield delay(0);

            if ($job === 2) {
                throw new \Exception('Failure');
            }

            return $job;
        };

        // Job 2 errors, so only job 3 and 4 should be executed
        $this->expectOutputString('1234');
        $this->expectException(\Exception::class);
        $this->expectExceptionMessage('Failure');

        yield each(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);
    }

    protected function tearDownAsync()
    {
        // Required to make testBackpressure fail instead of the following test
        \gc_collect_cycles();
    }
}
<file_sep>/test/LocalMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\LocalMutex;
use Amp\Sync\Mutex;

class LocalMutexTest extends AbstractMutexTest
{
    public function createMutex(): Mutex
    {
        return new LocalMutex;
    }
}
<file_sep>/docs/mutex.md
---
title: Mutex
permalink: /mutex
---
[Mutual exclusion](https://en.wikipedia.org/wiki/Mutual_exclusion) can be achieved using `Amp\Sync\synchronized()` and any `Mutex` implementation, or by manually using the `Mutex` instance to acquire a lock. 

Locks are acquired using `Mutex::acquire()`, which returns a `Promise` that resolves to an instance of `Lock` once the lock as been successfully acquired.

As long as the resulting `Lock` object isn't released using `Lock::release()` or by being garbage collected, the holder of the lock can exclusively run some code as long as all other parties running the same code also acquire a lock before doing so.

### Examples

```php
function writeExclusively(Amp\Sync\Mutex $mutex, string $filePath, string $data) {
    return Amp\call(function () use ($mutex, $filePath, $data) {
        /** @var Amp\Sync\Lock $lock */
        $lock = yield $mutex->acquire();
        
        $this->fileHandle = yield Amp\File\open($filePath, 'w');
        yield $this->fileHandle->write($data);
        
        $lock->release();
    });
}
```

```php
function writeExclusively(Amp\Sync\Mutex $mutex, string $filePath, string $data) {
    return Amp\Sync\synchronized($mutex, function () use ($filePath, $data) {
        $this->fileHandle = yield Amp\File\open($filePath, 'w');
        yield $this->fileHandle->write($data);
    });
}
```<file_sep>/test/PrefixedKeyedSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\KeyedSemaphore;
use Amp\Sync\LocalKeyedSemaphore;
use Amp\Sync\PrefixedKeyedSemaphore;

class PrefixedKeyedSemaphoreTest extends AbstractKeyedSemaphoreTest
{
    public function createSemaphore(int $size): KeyedSemaphore
    {
        return new PrefixedKeyedSemaphore(new LocalKeyedSemaphore($size), 'prefix.');
    }
}
<file_sep>/examples/mutex.php
<?php

use Amp\Sync\LocalMutex;
use Amp\Sync\Lock;
use function Amp\call;
use function Amp\delay;
use function Amp\Promise\all;
use function Amp\Promise\wait;

require __DIR__ . '/../vendor/autoload.php';

$mutex = new LocalMutex;

$task = function (string $identifier) use ($mutex) {
    print "[$identifier] Starting" . \PHP_EOL;

    for ($i = 0; $i < 3; $i++) {
        print "[$identifier][$i] Acquiring lock" . \PHP_EOL;

        /** @var Lock $lock */
        $lock = yield $mutex->acquire();

        try {
            print "[$identifier][$i] Acquired lock" . \PHP_EOL;

            // do anything exclusively
            yield delay(\random_int(0, 1000));
        } finally {
            print "[$identifier][$i] Releasing lock" . \PHP_EOL;

            $lock->release();
        }
    }

    print "[$identifier] Finished" . \PHP_EOL;
};

$promiseA = call($task, 'A');
$promiseB = call($task, 'B');
$promiseC = call($task, 'C');

wait(all([$promiseA, $promiseB, $promiseC]));<file_sep>/test/LocalKeyedMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\KeyedMutex;
use Amp\Sync\LocalKeyedMutex;

class LocalKeyedMutexTest extends AbstractKeyedMutexTest
{
    public function createMutex(): KeyedMutex
    {
        return new LocalKeyedMutex();
    }
}
<file_sep>/test/BarrierTest.php
<?php

namespace Amp\Sync\Test;

use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\Barrier;

class BarrierTest extends AsyncTestCase
{
    /** @var Barrier */
    private $barrier;

    public function testArriveUntilResolved(): void
    {
        $resolved = false;

        $this->barrier->await()->onResolve(static function () use (&$resolved) {
            $resolved = true;
        });

        $this->assertFalse($resolved);

        $this->barrier->arrive();
        $this->assertSame(1, $this->barrier->getCount());

        $this->assertFalse($resolved);

        $this->barrier->arrive();

        $this->assertTrue($resolved);
        $this->assertSame(0, $this->barrier->getCount());
    }

    public function testArriveAfterResolved(): void
    {
        $this->barrier->arrive();
        $this->barrier->arrive();

        $this->expectException(\Error::class);
        $this->barrier->arrive();
    }

    public function testArriveWithCount(): void
    {
        $resolved = false;

        $this->barrier->await()->onResolve(static function () use (&$resolved) {
            $resolved = true;
        });

        $this->assertFalse($resolved);

        $this->barrier->arrive(2);

        $this->assertTrue($resolved);
    }

    public function testArriveWithInvalidCount(): void
    {
        $this->expectException(\Error::class);

        $this->barrier->arrive(0);
    }

    public function testArriveTooHighCount(): void
    {
        $this->expectException(\Error::class);

        $this->barrier->arrive(3);
    }

    public function testGetCurrentCount(): void
    {
        $this->barrier->arrive();
        $this->assertEquals(1, $this->barrier->getCount());
    }

    public function testInvalidSignalCountInConstructor(): void
    {
        $this->expectException(\Error::class);
        new Barrier(0);
    }

    public function testRegisterCount(): void
    {
        $resolved = false;

        $this->barrier->await()->onResolve(static function () use (&$resolved) {
            $resolved = true;
        });

        $this->barrier->arrive();
        $this->barrier->register();
        $this->barrier->arrive();

        $this->assertFalse($resolved);

        $this->barrier->arrive();

        $this->assertTrue($resolved);
    }

    public function testRegisterCountWithCustomCount(): void
    {
        $resolved = false;

        $this->barrier->await()->onResolve(static function () use (&$resolved) {
            $resolved = true;
        });

        $this->barrier->arrive();
        $this->barrier->register(2);
        $this->barrier->arrive();

        $this->assertFalse($resolved);

        $this->barrier->arrive();

        $this->assertFalse($resolved);

        $this->barrier->arrive();

        $this->assertTrue($resolved);
    }

    public function testRegisterCountWithInvalidCount(): void
    {
        $this->expectException(\Error::class);
        $this->expectExceptionMessage('Count must be at least 1, got 0');

        $this->barrier->register(0);
    }

    public function testRegisterCountWithResolvedBarrier(): void
    {
        $this->barrier->arrive();
        $this->barrier->arrive();

        $this->expectException(\Error::class);
        $this->expectExceptionMessage('Can\'t increase count, because the barrier already broke');

        $this->barrier->register(1);
    }

    protected function setUp(): void
    {
        parent::setUp();

        $this->barrier = new Barrier(2);
    }
}
<file_sep>/docs/concurrent-iterator.md
---
title: Concurrent Iterators
permalink: /concurrent-iterator
---
As already stated in the [preamble of our documentation](https://amphp.org/amp/), the weak link when managing concurrency is humans; so `amphp/sync` provides abstractions to hide the complexity of concurrent iteration.

## Concurrency Approaches

Given you have a list of URLs you want to crawl, let's discuss a few possible approaches. For simplicity, we will assume a `fetch` function already exists, which takes a URL and returns the HTTP status code (which is everything we want to know for these examples).

### Approach 1: Sequential

Simple loop using non-blocking I/O, but no concurrency while fetching the individual URLs; starts the second request as soon as the first completed.

```php
$urls = [...];

Amp\call(function () use ($urls) {
    $results = [];

    foreach ($urls as $url) {
        // always wait for the promise to resolve before fetching the next one
        $statusCode = yield fetch($url);
        $results[$url] = $statusCode;
    }
    
    return $results;
});
```

### Approach 2: Everything Concurrently

Almost the same loop, but awaiting all promises at once; starts all requests immediately. Might not be feasible with too many URLs.

```php
$urls = [...];

Amp\call(function () use ($urls) {
    $results = [];

    foreach ($urls as $url) {
        // note the missing yield, we're adding the promises to the array
        $statusCodePromise = fetch($url);
        $results[$url] = $statusCodePromise;
    }

    // yielding an array of promises awaits them all at once
    $results = yield $results;
    
    return $results;
});
```

### Approach 3: Concurrent Chunks

Splitting the jobs into chunks of ten; all requests within a chunk are made concurrently, but each chunk sequentially, so the timing for each chunk depends on the slowest response; starts the eleventh request as soon as the first ten requests completed.

```php
$urls = [...];

Amp\call(function () use ($urls) {
    $results = [];

    foreach (\array_chunk($urls, 10) as $chunk) {
        $promises = [];

        foreach ($chunk as $url) {
            // note the missing yield, we're adding the promises to the array
            $statusCodePromise = fetch($url);
            $promises[$url] = $statusCodePromise;
        }

        // yielding an array of promises awaits them all at once
        $results = \array_merge($results, yield $promises);
    }
    
    return $results;
});
```

### Approach 4: Concurrent Iterator

Concurrent iteration, keeping the concurrency at a maximum of ten; starts the eleventh request as soon as any of the first ten requests completes.

```php
$urls = [...];

Amp\call(function () use ($urls) {
    $results = [];

    yield Amp\Sync\ConcurrentIterator\each(
        Amp\Iterator\fromIterable($urls),
        new Amp\Sync\LocalSemaphore(10),
        function (string $url) use (&$results) {
            $statusCode = yield fetch($url);
            $results[$url] = $statusCode;
        }
    );

    return $results;
});
```

## Provided APIs

### `Amp\Sync\ConcurrentIterator\each`

Calls `$processor` for each item in the iterator while acquiring a lock from `$semaphore` during each operation.
The returned `Promise` resolves as soon as the iterator is empty and all operations are completed.

Use `LocalSemaphore` if you don't need to synchronize beyond a single process.

```php
function each(Iterator $iterator, Semaphore $semaphore, callable $processor): Promise
{
    // ...
}
```

### `Amp\Sync\ConcurrentIterator\map`

Calls `$processor` for each item in the iterator while acquiring a lock from `$semaphore` during each operation.
Returns a new `Iterator` instance with the return values of `$processor`.
 
Use `LocalSemaphore` if you don't need to synchronize beyond a single process.

```php
function map(Iterator $iterator, Semaphore $semaphore, callable $processor): Iterator
{
    // ...
}
```

### `Amp\Sync\ConcurrentIterator\filter`

Calls `$filter` for each item in the iterator while acquiring a lock from `$semaphore` during each operation.
Returns a new `Iterator` instance with the original values where `$filter` resolves to `true`.
 
Use `LocalSemaphore` if you don't need to synchronize beyond a single process.

```php
function filter(Iterator $iterator, Semaphore $semaphore, callable $filter): Iterator
{
    // ...
}
```

### `Amp\Sync\ConcurrentIterator\transform`

Calls `$processor` for each item in the iterator while acquiring a lock from `$semaphore` during each operation.
`$processor` receives the current element and an `$emit` callable as arguments.

This function can be used to implement additional concurrent iteration functions and is the base for `map`, `filter`, and `each`. 

Use `LocalSemaphore` if you don't need to synchronize beyond a single process.

```php
function transform(Iterator $iterator, Semaphore $semaphore, callable $processor): Iterator
{
    // ...
}
```<file_sep>/docs/index.md
---
title: Synchronization Tools
permalink: /
---
This package defines synchronization primitives for PHP applications and libraries using Amp, such as locks, semaphores and concurrency limited iterator operations.

## Installation

This package can be installed as a [Composer](https://getcomposer.org/) dependency.

```bash
composer require amphp/sync
```

## Usage

See [`Mutex`](./mutex.md), [`Semaphore`](./semaphore.md), and [concurrent iteration](./concurrent-iterator.md).<file_sep>/test/PosixSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Loop;
use Amp\Sync\PosixSemaphore;
use Amp\Sync\Semaphore;
use Amp\Sync\SyncException;

/**
 * @group posix
 * @requires extension sysvmsg
 */
class PosixSemaphoreTest extends AbstractSemaphoreTest
{
    const ID = __CLASS__ . '/4';

    public function makeId(): string
    {
        return \spl_object_hash($this);
    }

    /**
     * @param $locks
     *
     * @return \Amp\Sync\PosixSemaphore
     */
    public function createSemaphore(int $locks): Semaphore
    {
        return PosixSemaphore::create(self::ID, $locks);
    }

    public function testConstructorOnInvalidMaxLocks(): void
    {
        $this->expectException(\Error::class);
        $this->expectExceptionMessage("Number of locks must be greater than 0");

        $this->semaphore = $this->createSemaphore(-1);
    }

    public function testCreateOnInvalidMaxLocks(): void
    {
        $this->expectException(\Error::class);

        PosixSemaphore::create(self::ID, -1);
    }

    public function testGetPermissions(): void
    {
        $this->semaphore = PosixSemaphore::create(self::ID, 1);
        $used = PosixSemaphore::use(self::ID);
        $used->setPermissions(0644);

        $this->assertSame(0644, $this->semaphore->getPermissions());
    }

    public function testGetId(): void
    {
        $this->semaphore = $this->createSemaphore(1);

        $this->assertSame(self::ID, $this->semaphore->getId());
    }

    public function testUseOnInvalidSemaphoreId(): void
    {
        $this->expectException(SyncException::class);
        $this->expectExceptionMessage("No semaphore with that ID found");

        PosixSemaphore::use(1);
    }

    public function testCreateOnDuplicatedSemaphoreId(): void
    {
        $this->expectException(SyncException::class);
        $this->expectExceptionMessage("A semaphore with that ID already exists");

        $semaphore = PosixSemaphore::create(self::ID, 1);
        $semaphore::create(self::ID, 1);
    }

    public function testUse()
    {
        $this->setMinimumRuntime(500);

        $this->semaphore = $this->createSemaphore(1);

        $used = PosixSemaphore::use(self::ID);

        $promise1 = $used->acquire();
        $promise2 = $this->semaphore->acquire();

        Loop::delay(500, function () use ($promise1) {
            (yield $promise1)->release();
        });

        (yield $promise2)->release();
    }
}
<file_sep>/test/ConcurrentMapTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Iterator;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\LocalSemaphore;
use function Amp\delay;
use function Amp\Iterator\toArray;
use function Amp\Sync\ConcurrentIterator\map;

class ConcurrentMapTest extends AsyncTestCase
{
    public function test(): \Generator
    {
        $this->expectOutputString('123');

        $processor = static function ($job) {
            print $job;
        };

        $this->assertSame(
            [null, null, null],
            yield toArray(map(Iterator\fromIterable([1, 2, 3]), new LocalSemaphore(3), $processor))
        );
    }

    public function testOutputOrder(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);

            return $job;
        };

        $this->assertSame(
            [1, 2, 3],
            yield toArray(map(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(3), $processor))
        );
    }

    public function testOutputOrderWithoutConcurrency(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);

            return $job;
        };

        $this->assertSame(
            [3, 2, 1],
            yield toArray(map(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(1), $processor))
        );
    }

    public function testBackpressure(): void
    {
        $this->expectOutputString('12');

        $processor = static function ($job) {
            print $job;

            return $job;
        };

        map(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);
    }

    public function testBackpressurePartialConsume1(): \Generator
    {
        $this->expectOutputString('123');

        $processor = static function ($job) {
            print $job;

            return $job;
        };

        $iterator = map(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        yield $iterator->advance();
    }

    public function testBackpressurePartialConsume2(): \Generator
    {
        $this->expectOutputString('1234');

        $processor = static function ($job) {
            print $job;

            return $job;
        };

        $iterator = map(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        yield $iterator->advance();
        yield $iterator->advance();
    }

    public function testErrorHandling(): \Generator
    {
        $processor = static function ($job) {
            print $job;

            yield delay(0);

            if ($job === 2) {
                throw new \Exception('Failure');
            }

            return $job;
        };

        $iterator = map(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        // Job 2 errors, so only job 3 and 4 should be executed
        $this->expectOutputString('1234');
        $this->expectException(\Exception::class);
        $this->expectExceptionMessage('Failure');

        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
    }

    public function testErrorHandlingCompletesPending(): \Generator
    {
        $processor = static function ($job) {
            print $job;

            if ($job === 2) {
                throw new \Exception('Failure');
            }

            yield delay(0);

            print $job;

            return $job;
        };

        $iterator = map(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        // Job 2 errors, so only job 3 and 4 should be executed
        $this->expectOutputString('121');
        $this->expectException(\Exception::class);
        $this->expectExceptionMessage('Failure');

        yield $iterator->advance();
        yield $iterator->advance();
    }

    protected function tearDownAsync()
    {
        // Required to make testBackpressure fail instead of the following test
        \gc_collect_cycles();
    }
}
<file_sep>/test/LocalSemaphoreTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\LocalSemaphore;
use Amp\Sync\Semaphore;

class LocalSemaphoreTest extends AbstractSemaphoreTest
{
    public function createSemaphore(int $locks): Semaphore
    {
        return new LocalSemaphore($locks);
    }
}
<file_sep>/test/ConcurrentFilterTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Iterator;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\LocalSemaphore;
use function Amp\delay;
use function Amp\Iterator\toArray;
use function Amp\Sync\ConcurrentIterator\filter;

class ConcurrentFilterTest extends AsyncTestCase
{
    public function test(): \Generator
    {
        $this->expectOutputString('123');

        $processor = static function ($job) {
            print $job;

            return $job === 2;
        };

        $this->assertSame(
            [2],
            yield toArray(filter(Iterator\fromIterable([1, 2, 3]), new LocalSemaphore(3), $processor))
        );
    }

    public function testOutputOrder(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);

            return true;
        };

        $this->assertSame(
            [1, 2, 3],
            yield toArray(filter(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(3), $processor))
        );
    }

    public function testOutputOrderWithoutConcurrency(): \Generator
    {
        $processor = static function ($job) {
            yield delay($job * 100);

            return true;
        };

        $this->assertSame(
            [3, 2, 1],
            yield toArray(filter(Iterator\fromIterable([3, 2, 1]), new LocalSemaphore(1), $processor))
        );
    }

    public function testBackpressure(): void
    {
        $this->expectOutputString('12');

        $processor = static function ($job) {
            print $job;

            return true;
        };

        filter(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);
    }

    public function testBackpressurePartialConsume1(): \Generator
    {
        $this->expectOutputString('123');

        $processor = static function ($job) {
            print $job;

            return true;
        };

        $iterator = filter(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        yield $iterator->advance();
    }

    public function testBackpressurePartialConsume2(): \Generator
    {
        $this->expectOutputString('1234');

        $processor = static function ($job) {
            print $job;

            return true;
        };

        $iterator = filter(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        yield $iterator->advance();
        yield $iterator->advance();
    }

    public function testErrorHandling(): \Generator
    {
        $processor = static function ($job) {
            print $job;

            yield delay(0);

            if ($job === 2) {
                throw new \Exception('Failure');
            }

            return true;
        };

        $iterator = filter(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        // Job 2 errors, so only job 3 and 4 should be executed
        $this->expectOutputString('1234');
        $this->expectException(\Exception::class);
        $this->expectExceptionMessage('Failure');

        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
    }

    public function testInvalidReturn(): \Generator
    {
        $processor = static function ($job) {
            print $job;

            yield delay(0);

            if ($job === 2) {
                return 0;
            }

            return true;
        };

        $iterator = filter(Iterator\fromIterable([1, 2, 3, 4, 5]), new LocalSemaphore(2), $processor);

        // Job 2 errors, so only job 3 and 4 should be executed
        $this->expectOutputString('1234');
        $this->expectException(\TypeError::class);
        $this->expectExceptionMessage('Amp\Sync\ConcurrentIterator\filter\'s callable must resolve to a boolean value, got integer');

        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
        yield $iterator->advance();
    }

    protected function tearDownAsync()
    {
        // Required to make testBackpressure fail instead of the following test
        \gc_collect_cycles();
    }
}
<file_sep>/examples/queue.php
<?php

use Amp\Emitter;
use Amp\Sync\LocalSemaphore;
use function Amp\delay;
use function Amp\Promise\wait;
use function Amp\Sync\ConcurrentIterator\each;

require __DIR__ . '/../vendor/autoload.php';

$emitter = new Emitter;

$jobId = 0;

for ($i = 0; $i < 10; $i++) {
    print 'enqueued ' . $jobId . \PHP_EOL;
    $emitter->emit($jobId++);
}

wait(each(
    $emitter->iterate(),
    new LocalSemaphore(3),
    static function ($job) use ($emitter, &$jobId) {
        print 'starting ' . $job . \PHP_EOL;

        yield delay(1000);

        if ($job < 10) {
            if (\random_int(0, 1)) {
                print 'enqueued ' . $jobId . \PHP_EOL;
                $emitter->emit($jobId++);
            }
        } elseif ($job === 10) {
            $emitter->complete();
        }

        print 'finished ' . $job . \PHP_EOL;
    }
));<file_sep>/test/PrefixedKeyedMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\KeyedMutex;
use Amp\Sync\LocalKeyedMutex;
use Amp\Sync\PrefixedKeyedMutex;

class PrefixedKeyedMutexTest extends AbstractKeyedMutexTest
{
    public function createMutex(): KeyedMutex
    {
        return new PrefixedKeyedMutex(new LocalKeyedMutex, 'prefix.');
    }
}
<file_sep>/test/AbstractMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Loop;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\Mutex;

abstract class AbstractMutexTest extends AsyncTestCase
{
    abstract public function createMutex(): Mutex;

    public function testAcquire(): \Generator
    {
        $mutex = $this->createMutex();
        $lock = yield $mutex->acquire();
        $lock->release();
        $this->assertTrue($lock->isReleased());
    }

    public function testAcquireMultiple(): \Generator
    {
        $this->setMinimumRuntime(300);

        $mutex = $this->createMutex();

        $lock1 = yield $mutex->acquire();
        Loop::delay(100, function () use ($lock1) {
            $lock1->release();
        });

        $lock2 = yield $mutex->acquire();
        Loop::delay(100, function () use ($lock2) {
            $lock2->release();
        });

        $lock3 = yield $mutex->acquire();
        Loop::delay(100, function () use ($lock3) {
            $lock3->release();
        });
    }
}
<file_sep>/test/SemaphoreMutexTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Sync\LocalSemaphore;
use Amp\Sync\Mutex;
use Amp\Sync\SemaphoreMutex;

class SemaphoreMutexTest extends AbstractMutexTest
{
    const ID = __CLASS__;

    public function createMutex(): Mutex
    {
        return new SemaphoreMutex(new LocalSemaphore(1));
    }

    public function testSemaphoreWithMultipleLocks(): \Generator
    {
        $this->expectException(\Error::class);
        $this->expectExceptionMessage('Cannot use a semaphore with more than a single lock');

        $mutex = new SemaphoreMutex(new LocalSemaphore(2));
        while (yield $mutex->acquire());
    }
}
<file_sep>/test/SynchronizedTest.php
<?php

namespace Amp\Sync\Test;

use Amp\Delayed;
use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\LocalMutex;
use function Amp\Sync\synchronized;

class SynchronizedTest extends AsyncTestCase
{
    public function testSynchronized(): \Generator
    {
        $this->setMinimumRuntime(300);

        $mutex = new LocalMutex;
        $callback = function (int $value) {
            return yield new Delayed(100, $value);
        };

        $promises = [];
        foreach (\range(0, 2) as $value) {
            $promises[] = synchronized($mutex, $callback, $value);
        }
        $result = yield $promises;
        $this->assertSame(\range(0, 2), $result);
    }
}
<file_sep>/docs/semaphore.md
---
title: Semaphore
permalink: /semaphore
---
[Semaphores](https://en.wikipedia.org/wiki/Semaphore_%28programming%29) are another synchronization primitive in addition to [mutual exclusion](./mutex.md).

Instead of providing exclusive access to a single party, they provide access to a limited set of N parties at the same time.
This makes them great to control concurrency, e.g. limiting an HTTP client to X concurrent requests, so the HTTP server doesn't get overwhelmed.

Similar to [`Mutex`](./mutex.md), `Lock` instances can be acquired using `Semaphore::acquire()`.
Please refer to the `Mutex` documentation for additional usage documentation, as they're basically equivalent except for the fact that `Mutex` is always a `Semaphore` with a count of exactly one party.

In many cases you can use [concurrent iterators](./concurrent-iterator.md) instead of directly using a `Semaphore`.<file_sep>/test/LockTest.php
<?php

namespace Amp\Sync\Test;

use Amp\PHPUnit\AsyncTestCase;
use Amp\Sync\Lock;

class LockTest extends AsyncTestCase
{
    public function testIsReleased()
    {
        $lock = new Lock(0, $this->createCallback(1));
        $this->assertFalse($lock->isReleased());
        $lock->release();
        $this->assertTrue($lock->isReleased());
    }

    public function testIsReleasedOnDestruct()
    {
        $lock = new Lock(0, $this->createCallback(1));
        unset($lock);
    }

    public function testThrowsOnMultiRelease()
    {
        $lock = new Lock(0, $this->createCallback(1));
        $lock->release();
        $this->assertTrue($lock->isReleased());
        $lock->release();
        $this->assertTrue($lock->isReleased());
    }

    public function testGetId()
    {
        $id = 42;
        $lock = new Lock($id, $this->createCallback(1));
        $this->assertSame($id, $lock->getId());
        $lock->release();
    }
}
