danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2024-11-30 04:39:01 +01:00
Code Issues Projects Releases Wiki Activity

Improve semaphore tests

Browse Source
This commit is contained in:
Aaron Piotrowski 2015-09-03 23:23:12 -05:00
parent 348f7c7461
commit 3071dee6c8
4 changed files with 179 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/Sync/PosixSemaphore.php
View File

@ -134,7 +134,7 @@ class PosixSemaphore implements SemaphoreInterface, \Serializable
*/
public function acquire()
{
while (true) {
do {
// Attempt to acquire a lock from the semaphore.
if (@msg_receive($this->queue, 0, $type, 1, $chr, false, MSG_IPC_NOWAIT, $errno)) {
// A free lock was found, so resolve with a lock object that can
@ -149,12 +149,7 @@ class PosixSemaphore implements SemaphoreInterface, \Serializable
if ($errno !== MSG_ENOMSG) {
throw new SemaphoreException('Failed to acquire a lock.');
}
// Sleep for a while, giving a chance for other threads to release
// their locks. After we finish sleeping, we can check again to see
// if it is our turn to acquire a lock.
yield Coroutine\sleep(self::LATENCY_TIMEOUT);
}
} while (yield Coroutine\sleep(self::LATENCY_TIMEOUT));
}
/**

125
tests/Sync/AbstractSemaphoreTest.php Normal file
View File

@ -0,0 +1,125 @@
<?php
namespace Icicle\Tests\Concurrent\Sync;
use Icicle\Concurrent\Sync\Lock;
use Icicle\Concurrent\Sync\PosixSemaphore;
use Icicle\Coroutine;
use Icicle\Loop;
use Icicle\Tests\Concurrent\TestCase;
abstract class AbstractSemaphoreTest extends TestCase
{
/**
* @var \Icicle\Concurrent\Sync\SemaphoreInterface
*/
protected $semaphore;
/**
* @return \Icicle\Concurrent\Sync\SemaphoreInterface
*/
abstract public function createSemaphore($locks);
public function testCount()
{
$this->semaphore = new PosixSemaphore(4);
$this->assertCount(4, $this->semaphore);
$this->semaphore->free();
}
public function testAcquire()
{
Coroutine\create(function () {
$this->semaphore = $this->createSemaphore(1);
$lock = (yield $this->semaphore->acquire());
$this->assertFalse($lock->isReleased());
$lock->release();
$this->assertTrue($lock->isReleased());
})->done();
Loop\run();
}
public function testAcquireMultiple()
{
$this->assertRunTimeGreaterThan(function () {
$this->semaphore = $this->createSemaphore(1);
Coroutine\create(function () {
$lock1 = (yield $this->semaphore->acquire());
Loop\timer(0.5, function () use ($lock1) {
$lock1->release();
});
$lock2 = (yield $this->semaphore->acquire());
Loop\timer(0.5, function () use ($lock2) {
$lock2->release();
});
$lock3 = (yield $this->semaphore->acquire());
Loop\timer(0.5, function () use ($lock3) {
$lock3->release();
});
})->done();
Loop\run();
}, 1.5);
}
public function tesCloneIsSameSemaphore()
{
Coroutine\create(function () {
$this->semaphore = $this->createSemaphore(1);
$clone = clone $this->semaphore;
$lock = (yield $clone->acquire());
$this->assertCount(1, $this->semaphore);
$this->assertCount(0, $clone);
$lock->release();
})->done();
Loop\run();
}
public function testSerializedIsSameSemaphore()
{
Coroutine\create(function () {
$this->semaphore = $this->createSemaphore(1);
$unserialized = unserialize(serialize($this->semaphore));
$lock = (yield $unserialized->acquire());
$this->assertCount(0, $this->semaphore);
$this->assertCount(0, $unserialized);
$lock->release();
})->done();
Loop\run();
}
public function testSimultaneousAcquire()
{
$this->semaphore = $this->createSemaphore(1);
$coroutine1 = new Coroutine\Coroutine($this->semaphore->acquire());
$coroutine2 = new Coroutine\Coroutine($this->semaphore->acquire());
$coroutine1->delay(0.5)->then(function (Lock $lock) {
$lock->release();
});
$coroutine2->delay(0.5)->then(function (Lock $lock) {
$lock->release();
});
$this->assertRunTimeGreaterThan('Icicle\Loop\run', 1);
}
}

117
tests/Sync/PosixSemaphoreTest.php
View File

@ -1,111 +1,78 @@
<?php
namespace Icicle\Tests\Concurrent\Sync;
use Icicle\Concurrent\Forking\Fork;
use Icicle\Concurrent\Sync\PosixSemaphore;
use Icicle\Concurrent\Sync\SemaphoreInterface;
use Icicle\Coroutine;
use Icicle\Loop;
use Icicle\Tests\Concurrent\TestCase;
/**
* @group posix
* @requires extension sysvmsg
*/
class PosixSemaphoreTest extends TestCase
class PosixSemaphoreTest extends AbstractSemaphoreTest
{
public function createSemaphore($locks)
{
return new PosixSemaphore($locks);
}
public function tearDown()
{
if (!$this->semaphore->isFreed()) {
$this->semaphore->free();
}
}
public function testFree()
{
$semaphore = new PosixSemaphore(1);
$this->semaphore = $this->createSemaphore(1);
$this->assertFalse($semaphore->isFreed());
$this->assertFalse($this->semaphore->isFreed());
$semaphore->free();
$this->semaphore->free();
$this->assertTrue($semaphore->isFreed());
$this->assertTrue($this->semaphore->isFreed());
}
public function testCount()
{
$semaphore = new PosixSemaphore(4);
$this->assertCount(4, $semaphore);
$semaphore->free();
}
public function testAcquire()
/**
* @requires extension pcntl
*/
public function testAcquireInMultipleForks()
{
Coroutine\create(function () {
$semaphore = new PosixSemaphore(1);
$this->semaphore = $this->createSemaphore(1);
$lock = (yield $semaphore->acquire());
$lock->release();
$fork1 = new Fork(function (SemaphoreInterface $semaphore) {
$lock = (yield $semaphore->acquire());
$this->assertTrue($lock->isReleased());
usleep(1e5);
$semaphore->free();
});
$lock->release();
Loop\run();
}
yield 0;
}, $this->semaphore);
public function testAcquireMultiple()
{
$this->assertRunTimeGreaterThan(function () {
$semaphore = new PosixSemaphore(1);
$fork2 = new Fork(function (SemaphoreInterface $semaphore) {
$lock = (yield $semaphore->acquire());
Coroutine\create(function () use ($semaphore) {
$lock1 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock1) {
$lock1->release();
});
usleep(1e5);
$lock2 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock2) {
$lock2->release();
});
$lock->release();
$lock3 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock3) {
$lock3->release();
});
});
yield 1;
}, $this->semaphore);
Loop\run();
$semaphore->free();
}, 1.5);
}
$start = microtime(true);
public function tesCloneIsSameSemaphore()
{
Coroutine\create(function () {
$semaphore = new PosixSemaphore(1);
$clone = clone $semaphore;
$fork1->start();
$fork2->start();
$lock = (yield $clone->acquire());
yield $fork1->join();
yield $fork2->join();
$this->assertCount(0, $semaphore);
$this->assertCount(0, $clone);
$lock->release();
$semaphore->free();
});
Loop\run();
}
public function testSerializedIsSameSemaphore()
{
Coroutine\create(function () {
$semaphore = new PosixSemaphore(1);
$unserialized = unserialize(serialize($semaphore));
$lock = (yield $unserialized->acquire());
$this->assertCount(0, $semaphore);
$this->assertCount(0, $unserialized);
$lock->release();
$semaphore->free();
$this->assertGreaterThan(1, microtime(true) - $start);
});
Loop\run();

89
tests/Threading/SemaphoreTest.php
View File

@ -1,99 +1,30 @@
<?php
namespace Icicle\Tests\Concurrent\Threading;
use Icicle\Concurrent\Sync\Lock;
use Icicle\Concurrent\Sync\SemaphoreInterface;
use Icicle\Concurrent\Threading\Semaphore;
use Icicle\Concurrent\Threading\Thread;
use Icicle\Coroutine;
use Icicle\Loop;
use Icicle\Tests\Concurrent\TestCase;
use Icicle\Tests\Concurrent\Sync\AbstractSemaphoreTest;
/**
* @group threading
* @requires extension pthreads
*/
class SemaphoreTest extends TestCase
class SemaphoreTest extends AbstractSemaphoreTest
{
public function testCount()
public function createSemaphore($locks)
{
$semaphore = new Semaphore(1);
$this->assertEquals(1, $semaphore->count());
return new Semaphore($locks);
}
/**
* @depends testCount
*/
public function testInvalidLockCount()
{
$semaphore = new Semaphore(0);
$this->assertEquals(1, $semaphore->count());
}
public function testAcquire()
{
Coroutine\create(function () {
$semaphore = new Semaphore(1);
$lock = (yield $semaphore->acquire());
$lock->release();
$this->assertTrue($lock->isReleased());
});
Loop\run();
}
public function testAcquireMultiple()
{
$this->assertRunTimeGreaterThan(function () {
Coroutine\create(function () {
$semaphore = new Semaphore(1);
$lock1 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock1) {
$lock1->release();
});
$lock2 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock2) {
$lock2->release();
});
$lock3 = (yield $semaphore->acquire());
Loop\timer(0.5, function () use ($lock3) {
$lock3->release();
});
});
Loop\run();
}, 1.5);
}
public function testSimultaneousAcquire()
{
$semaphore = new Semaphore(1);
$coroutine1 = new Coroutine\Coroutine($semaphore->acquire());
$coroutine2 = new Coroutine\Coroutine($semaphore->acquire());
$coroutine1->delay(0.5)->then(function (Lock $lock) {
$lock->release();
});
$coroutine2->delay(0.5)->then(function (Lock $lock) {
$lock->release();
});
$this->assertRunTimeGreaterThan('Icicle\Loop\run', 1);
}
/**
* @depends testAcquireMultiple
*/
public function testAcquireInMultipleThreads()
{
Coroutine\create(function () {
$semaphore = new Semaphore(1);
$this->semaphore = $this->createSemaphore(1);
$thread1 = new Thread(function (Semaphore $semaphore) {
$thread1 = new Thread(function (SemaphoreInterface $semaphore) {
$lock = (yield $semaphore->acquire());
usleep(1e5);
@ -101,9 +32,9 @@ class SemaphoreTest extends TestCase
$lock->release();
yield 0;
}, $semaphore);
}, $this->semaphore);
$thread2 = new Thread(function (Semaphore $semaphore) {
$thread2 = new Thread(function (SemaphoreInterface $semaphore) {
$lock = (yield $semaphore->acquire());
usleep(1e5);
@ -111,7 +42,7 @@ class SemaphoreTest extends TestCase
$lock->release();
yield 1;
}, $semaphore);
}, $this->semaphore);
$start = microtime(true);