Fix possible accept() race condition

Multiple socket servers can listen on the same socket address (using
`SO_REUSEPORT` or shared file descriptors). Accordingly, when a new
connection is incoming, multiple event-loops could report the socket to
be readable and try to a `accept()` an incoming connection from the same
socket.

This wouldn't be an issue with the underlying `accept()` system call.
The first call would succeed and subsequent calls would report `EAGAIN`
or `EWOULDBLOCK` for the other servers given the socket resource is in
non-blocking mode.

However, PHP's `stream_socket_accept()` implementation first runs a
`poll()` on the socket resource before performing an `accept()`. This
means multiple instances listening on the same address could end up in a
race condition where some may be "stuck" in the pending `poll()`.

We work around this by specifiying a `0` timeout to ensure the `poll()`
doesn't block in this case. This allows all servers to either complete
successfully or report an error and continue processing right away.

Author: clue

src/TcpServer.php

@@ -204,7 +204,7 @@ public function resume()
 
         $that = $this;
         $this->loop->addReadStream($this->master, function ($master) use ($that) {
-            $newSocket = @\stream_socket_accept($master);
+            $newSocket = @\stream_socket_accept($master, 0);
             if (false === $newSocket) {
                 $that->emit('error', array(new \RuntimeException('Error accepting new connection')));
 

src/UnixServer.php

@@ -106,7 +106,7 @@ public function resume()
 
         $that = $this;
         $this->loop->addReadStream($this->master, function ($master) use ($that) {
-            $newSocket = @\stream_socket_accept($master);
+            $newSocket = @\stream_socket_accept($master, 0);
             if (false === $newSocket) {
                 $that->emit('error', array(new \RuntimeException('Error accepting new connection')));
 

tests/TcpServerTest.php

@@ -280,8 +280,12 @@ public function testEmitsErrorWhenAcceptListenerFails()
 
         $this->assertNotNull($listener);
         $socket = stream_socket_server('tcp://127.0.0.1:0');
-        fclose($socket);
+
+        $time = microtime(true);
         $listener($socket);
+        $time = microtime(true) - $time;
+
+        $this->assertLessThan(1, $time);
     }
 
     public function testListenOnBusyPortThrows()

tests/UnixServerTest.php

@@ -285,8 +285,12 @@ public function testEmitsErrorWhenAcceptListenerFails()
 
         $this->assertNotNull($listener);
         $socket = stream_socket_server('tcp://127.0.0.1:0');
-        fclose($socket);
+
+        $time = microtime(true);
         $listener($socket);
+        $time = microtime(true) - $time;
+
+        $this->assertLessThan(1, $time);
     }
 
     public function testListenOnBusyPortThrows()