Skip to content

Fix separately removing readable and writable side of stream when closing #139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Dec 22, 2017
Merged

Fix separately removing readable and writable side of stream when closing #139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5e95b14
Test removing both readable and writable side of stream when closing
clue Dec 19, 2017
88b5c2a
Use separate event listeners for readable and writable side
clue Dec 19, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 40 additions & 88 deletions src/ExtEventLoop.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,12 @@ final class ExtEventLoop implements LoopInterface
private $timerCallback;
private $timerEvents;
private $streamCallback;
private $streamEvents = [];
private $streamFlags = [];
private $streamRefs = [];
private $readEvents = [];
private $writeEvents = [];
private $readListeners = [];
private $writeListeners = [];
private $readRefs = [];
private $writeRefs = [];
private $running;
private $signals;
private $signalEvents = [];
Expand Down Expand Up @@ -70,56 +71,65 @@ function ($signal) {
public function addReadStream($stream, callable $listener)
{
$key = (int) $stream;
if (isset($this->readListeners[$key])) {
return;
}

if (!isset($this->readListeners[$key])) {
$this->readListeners[$key] = $listener;
$this->subscribeStreamEvent($stream, Event::READ);
$event = new Event($this->eventBase, $stream, Event::PERSIST | Event::READ, $this->streamCallback);
$event->add();
$this->readEvents[$key] = $event;
$this->readListeners[$key] = $listener;

// ext-event does not increase refcount on stream resources for PHP 7+
// manually keep track of stream resource to prevent premature garbage collection
if (PHP_VERSION_ID >= 70000) {
$this->readRefs[$key] = $stream;
}
}

public function addWriteStream($stream, callable $listener)
{
$key = (int) $stream;

if (!isset($this->writeListeners[$key])) {
$this->writeListeners[$key] = $listener;
$this->subscribeStreamEvent($stream, Event::WRITE);
if (isset($this->writeListeners[$key])) {
return;
}
}

public function removeReadStream($stream)
{
$key = (int) $stream;
$event = new Event($this->eventBase, $stream, Event::PERSIST | Event::WRITE, $this->streamCallback);
$event->add();
$this->writeEvents[$key] = $event;
$this->writeListeners[$key] = $listener;

if (isset($this->readListeners[$key])) {
unset($this->readListeners[$key]);
$this->unsubscribeStreamEvent($stream, Event::READ);
// ext-event does not increase refcount on stream resources for PHP 7+
// manually keep track of stream resource to prevent premature garbage collection
if (PHP_VERSION_ID >= 70000) {
$this->writeRefs[$key] = $stream;
}
}

public function removeWriteStream($stream)
public function removeReadStream($stream)
{
$key = (int) $stream;

if (isset($this->writeListeners[$key])) {
unset($this->writeListeners[$key]);
$this->unsubscribeStreamEvent($stream, Event::WRITE);
if (isset($this->readEvents[$key])) {
$this->readEvents[$key]->free();
unset(
$this->readEvents[$key],
$this->readListeners[$key],
$this->readRefs[$key]
);
}
}

private function removeStream($stream)
public function removeWriteStream($stream)
{
$key = (int) $stream;

if (isset($this->streamEvents[$key])) {
$this->streamEvents[$key]->free();

if (isset($this->writeEvents[$key])) {
$this->writeEvents[$key]->free();
unset(
$this->streamFlags[$key],
$this->streamEvents[$key],
$this->readListeners[$key],
$this->writeEvents[$key],
$this->writeListeners[$key],
$this->streamRefs[$key]
$this->writeRefs[$key]
);
}
}
Expand Down Expand Up @@ -175,7 +185,7 @@ public function run()
$flags = EventBase::LOOP_ONCE;
if (!$this->running || !$this->futureTickQueue->isEmpty()) {
$flags |= EventBase::LOOP_NONBLOCK;
} elseif (!$this->streamEvents && !$this->timerEvents->count()) {
} elseif (!$this->readEvents && !$this->writeEvents && !$this->timerEvents->count()) {
break;
}

Expand Down Expand Up @@ -207,64 +217,6 @@ private function scheduleTimer(TimerInterface $timer)
$event->add($timer->getInterval());
}

/**
* Create a new ext-event Event object, or update the existing one.
*
* @param resource $stream
* @param integer $flag Event::READ or Event::WRITE
*/
private function subscribeStreamEvent($stream, $flag)
{
$key = (int) $stream;

if (isset($this->streamEvents[$key])) {
$event = $this->streamEvents[$key];
$flags = ($this->streamFlags[$key] |= $flag);

$event->del();
$event->set($this->eventBase, $stream, Event::PERSIST | $flags, $this->streamCallback);
} else {
$event = new Event($this->eventBase, $stream, Event::PERSIST | $flag, $this->streamCallback);

$this->streamEvents[$key] = $event;
$this->streamFlags[$key] = $flag;

// ext-event does not increase refcount on stream resources for PHP 7+
// manually keep track of stream resource to prevent premature garbage collection
if (PHP_VERSION_ID >= 70000) {
$this->streamRefs[$key] = $stream;
}
}

$event->add();
}

/**
* Update the ext-event Event object for this stream to stop listening to
* the given event type, or remove it entirely if it's no longer needed.
*
* @param resource $stream
* @param integer $flag Event::READ or Event::WRITE
*/
private function unsubscribeStreamEvent($stream, $flag)
{
$key = (int) $stream;

$flags = $this->streamFlags[$key] &= ~$flag;

if (0 === $flags) {
$this->removeStream($stream);

return;
}

$event = $this->streamEvents[$key];

$event->del();
$event->set($this->eventBase, $stream, Event::PERSIST | $flags, $this->streamCallback);
$event->add();
}

/**
* Create a callback used as the target of timer events.
*
Expand Down
117 changes: 33 additions & 84 deletions src/ExtLibeventLoop.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,8 @@ final class ExtLibeventLoop implements LoopInterface
private $timerCallback;
private $timerEvents;
private $streamCallback;
private $streamEvents = [];
private $streamFlags = [];
private $readEvents = [];
private $writeEvents = [];
private $readListeners = [];
private $writeListeners = [];
private $running;
Expand Down Expand Up @@ -88,30 +88,48 @@ function ($signal) {
public function addReadStream($stream, callable $listener)
{
$key = (int) $stream;

if (!isset($this->readListeners[$key])) {
$this->readListeners[$key] = $listener;
$this->subscribeStreamEvent($stream, EV_READ);
if (isset($this->readListeners[$key])) {
return;
}

$event = event_new();
event_set($event, $stream, EV_PERSIST | EV_READ, $this->streamCallback);
event_base_set($event, $this->eventBase);
event_add($event);

$this->readEvents[$key] = $event;
$this->readListeners[$key] = $listener;
}

public function addWriteStream($stream, callable $listener)
{
$key = (int) $stream;

if (!isset($this->writeListeners[$key])) {
$this->writeListeners[$key] = $listener;
$this->subscribeStreamEvent($stream, EV_WRITE);
if (isset($this->writeListeners[$key])) {
return;
}

$event = event_new();
event_set($event, $stream, EV_PERSIST | EV_WRITE, $this->streamCallback);
event_base_set($event, $this->eventBase);
event_add($event);

$this->writeEvents[$key] = $event;
$this->writeListeners[$key] = $listener;
}

public function removeReadStream($stream)
{
$key = (int) $stream;

if (isset($this->readListeners[$key])) {
unset($this->readListeners[$key]);
$this->unsubscribeStreamEvent($stream, EV_READ);
$event = $this->readEvents[$key];
event_del($event);
event_free($event);

unset(
$this->readEvents[$key],
$this->readListeners[$key]
);
}
}

Expand All @@ -120,25 +138,12 @@ public function removeWriteStream($stream)
$key = (int) $stream;

if (isset($this->writeListeners[$key])) {
unset($this->writeListeners[$key]);
$this->unsubscribeStreamEvent($stream, EV_WRITE);
}
}

private function removeStream($stream)
{
$key = (int) $stream;

if (isset($this->streamEvents[$key])) {
$event = $this->streamEvents[$key];

$event = $this->writeEvents[$key];
event_del($event);
event_free($event);

unset(
$this->streamFlags[$key],
$this->streamEvents[$key],
$this->readListeners[$key],
$this->writeEvents[$key],
$this->writeListeners[$key]
);
}
Expand Down Expand Up @@ -166,7 +171,6 @@ public function cancelTimer(TimerInterface $timer)
{
if ($this->timerEvents->contains($timer)) {
$event = $this->timerEvents[$timer];

event_del($event);
event_free($event);

Expand Down Expand Up @@ -199,7 +203,7 @@ public function run()
$flags = EVLOOP_ONCE;
if (!$this->running || !$this->futureTickQueue->isEmpty()) {
$flags |= EVLOOP_NONBLOCK;
} elseif (!$this->streamEvents && !$this->timerEvents->count()) {
} elseif (!$this->readEvents && !$this->writeEvents && !$this->timerEvents->count()) {
break;
}

Expand All @@ -226,61 +230,6 @@ private function scheduleTimer(TimerInterface $timer)
event_add($event, $timer->getInterval() * self::MICROSECONDS_PER_SECOND);
}

/**
* Create a new ext-libevent event resource, or update the existing one.
*
* @param resource $stream
* @param integer $flag EV_READ or EV_WRITE
*/
private function subscribeStreamEvent($stream, $flag)
{
$key = (int) $stream;

if (isset($this->streamEvents[$key])) {
$event = $this->streamEvents[$key];
$flags = $this->streamFlags[$key] |= $flag;

event_del($event);
event_set($event, $stream, EV_PERSIST | $flags, $this->streamCallback);
} else {
$event = event_new();

event_set($event, $stream, EV_PERSIST | $flag, $this->streamCallback);
event_base_set($event, $this->eventBase);

$this->streamEvents[$key] = $event;
$this->streamFlags[$key] = $flag;
}

event_add($event);
}

/**
* Update the ext-libevent event resource for this stream to stop listening to
* the given event type, or remove it entirely if it's no longer needed.
*
* @param resource $stream
* @param integer $flag EV_READ or EV_WRITE
*/
private function unsubscribeStreamEvent($stream, $flag)
{
$key = (int) $stream;

$flags = $this->streamFlags[$key] &= ~$flag;

if (0 === $flags) {
$this->removeStream($stream);

return;
}

$event = $this->streamEvents[$key];

event_del($event);
event_set($event, $stream, EV_PERSIST | $flags, $this->streamCallback);
event_add($event);
}

/**
* Create a callback used as the target of timer events.
*
Expand Down
Loading