Reuse TCP/IP socket connection and pipeline outgoing queries

clue · clue · commit 44a8c92e9c62 · 2019-08-11T15:51:22.000+02:00
The TcpTransportExecutor will now keep the TCP/IP socket connection to the DNS server open for a short period. Additional queries can reuse the same socket connection and pipeline multiple queries over this single socket. This avoids creating multiple socket resources for concurrent requests and implies the TCP/IP connection overhead only applies once instead of for each query. Among others, this helps avoiding excessive resources and significantly improves query performance when using multiple (concurrent) queries. This implements RFC 7766 as per https://tools.ietf.org/html/rfc7766
diff --git a/README.md b/README.md
@@ -319,12 +319,20 @@ Unlike the `UdpTransportExecutor`, this class uses a reliable TCP/IP
 transport, so you do not necessarily have to implement any retry logic.
 
 Note that this executor is entirely async and as such allows you to execute
-any number of queries concurrently. You should probably limit the number of
-concurrent queries in your application or you're very likely going to face
-rate limitations and bans on the resolver end. For many common applications,
-you may want to avoid sending the same query multiple times when the first
-one is still pending, so you will likely want to use this in combination with
-a `CoopExecutor` like this:
+queries concurrently. The first query will establish a TCP/IP socket
+connection to the DNS server which will be kept open for a short period.
+Additional queries will automatically reuse this existing socket connection
+to the DNS server, will pipeline multiple requests over this single
+connection and will keep an idle connection open for a short period. The
+initial TCP/IP connection overhead may incur a slight delay if you only send
+occasional queries – when sending a larger number of concurrent queries over
+an existing connection, it becomes increasingly more efficient and avoids
+creating many concurrent sockets like the UDP-based executor. You may still
+want to limit the number of (concurrent) queries in your application or you
+may be facing rate limitations and bans on the resolver end. For many common
+applications, you may want to avoid sending the same query multiple times
+when the first one is still pending, so you will likely want to use this in
+combination with a `CoopExecutor` like this:
 
 ```php
 $executor = new CoopExecutor(
diff --git a/composer.json b/composer.json
@@ -6,7 +6,7 @@
     "require": {
         "php": ">=5.3.0",
         "react/cache": "^1.0 || ^0.6 || ^0.5",
-        "react/event-loop": "^1.0 || ^0.5 || ^0.4 || ^0.3.5",
+        "react/event-loop": "^1.0 || ^0.5",
         "react/promise": "^2.1 || ^1.2.1",
         "react/promise-timer": "^1.2"
     },
diff --git a/src/Query/TcpTransportExecutor.php b/src/Query/TcpTransportExecutor.php
@@ -48,12 +48,20 @@
  * transport, so you do not necessarily have to implement any retry logic.
  *
  * Note that this executor is entirely async and as such allows you to execute
- * any number of queries concurrently. You should probably limit the number of
- * concurrent queries in your application or you're very likely going to face
- * rate limitations and bans on the resolver end. For many common applications,
- * you may want to avoid sending the same query multiple times when the first
- * one is still pending, so you will likely want to use this in combination with
- * a `CoopExecutor` like this:
+ * queries concurrently. The first query will establish a TCP/IP socket
+ * connection to the DNS server which will be kept open for a short period.
+ * Additional queries will automatically reuse this existing socket connection
+ * to the DNS server, will pipeline multiple requests over this single
+ * connection and will keep an idle connection open for a short period. The
+ * initial TCP/IP connection overhead may incur a slight delay if you only send
+ * occasional queries – when sending a larger number of concurrent queries over
+ * an existing connection, it becomes increasingly more efficient and avoids
+ * creating many concurrent sockets like the UDP-based executor. You may still
+ * want to limit the number of (concurrent) queries in your application or you
+ * may be facing rate limitations and bans on the resolver end. For many common
+ * applications, you may want to avoid sending the same query multiple times
+ * when the first one is still pending, so you will likely want to use this in
+ * combination with a `CoopExecutor` like this:
  *
  * ```php
  * $executor = new CoopExecutor(
@@ -78,6 +86,51 @@ class TcpTransportExecutor implements ExecutorInterface
     private $parser;
     private $dumper;
 
+    /**
+     * @var ?resource
+     */
+    private $socket;
+
+    /**
+     * @var Deferred[]
+     */
+    private $pending = array();
+
+    /**
+     * @var string[]
+     */
+    private $names = array();
+
+    /**
+     * Maximum idle time when socket is current unused (i.e. no pending queries outstanding)
+     *
+     * If a new query is to be sent during the idle period, we can reuse the
+     * existing socket without having to wait for a new socket connection.
+     * This uses a rather small, hard-coded value to not keep any unneeded
+     * sockets open and to not keep the loop busy longer than needed.
+     *
+     * A future implementation may take advantage of `edns-tcp-keepalive` to keep
+     * the socket open for longer periods. This will likely require explicit
+     * configuration because this may consume additional resources and also keep
+     * the loop busy for longer than expected in some applications.
+     *
+     * @var float
+     * @link https://tools.ietf.org/html/rfc7766#section-6.2.1
+     * @link https://tools.ietf.org/html/rfc7828
+     */
+    private $idlePeriod = 0.001;
+
+    /**
+     * @var ?\React\EventLoop\TimerInterface
+     */
+    private $idleTimer;
+
+    private $writeBuffer = '';
+    private $writePending = false;
+
+    private $readBuffer = '';
+    private $readPending = false;
+
     /**
      * @param string        $nameserver
      * @param LoopInterface $loop
@@ -103,6 +156,12 @@ public function __construct($nameserver, LoopInterface $loop)
     public function query(Query $query)
     {
         $request = Message::createRequestForQuery($query);
+
+        // keep shuffing message ID to avoid using the same message ID for two pending queries at the same time
+        while (isset($this->pending[$request->id])) {
+            $request->id = \mt_rand(0, 0xffff); // @codeCoverageIgnore
+        }
+
         $queryData = $this->dumper->toBinary($request);
         $length = \strlen($queryData);
         if ($length > 0xffff) {
@@ -113,106 +172,175 @@ public function query(Query $query)
 
         $queryData = \pack('n', $length) . $queryData;
 
-        // create async TCP/IP connection (may take a while)
-        $socket = @\stream_socket_client($this->nameserver, $errno, $errstr, 0, \STREAM_CLIENT_CONNECT | \STREAM_CLIENT_ASYNC_CONNECT);
-        if ($socket === false) {
-            return \React\Promise\reject(new \RuntimeException(
-                'DNS query for ' . $query->name . ' failed: Unable to connect to DNS server ('  . $errstr . ')',
-                $errno
-            ));
+        if ($this->socket === null) {
+            // create async TCP/IP connection (may take a while)
+            $socket = @\stream_socket_client($this->nameserver, $errno, $errstr, 0, \STREAM_CLIENT_CONNECT | \STREAM_CLIENT_ASYNC_CONNECT);
+            if ($socket === false) {
+                return \React\Promise\reject(new \RuntimeException(
+                    'DNS query for ' . $query->name . ' failed: Unable to connect to DNS server ('  . $errstr . ')',
+                    $errno
+                ));
+            }
+
+            // set socket to non-blocking and wait for it to become writable (connection success/rejected)
+            \stream_set_blocking($socket, false);
+            $this->socket = $socket;
+        }
+
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
+        }
+
+        // wait for socket to become writable to actually write out data
+        $this->writeBuffer .= $queryData;
+        if (!$this->writePending) {
+            $this->writePending = true;
+            $this->loop->addWriteStream($this->socket, array($this, 'handleWritable'));
         }
 
-        $loop = $this->loop;
-        $deferred = new Deferred(function () use ($loop, $socket, $query) {
-            // cancellation should remove socket from loop and close socket
-            $loop->removeReadStream($socket);
-            $loop->removeWriteStream($socket);
-            \fclose($socket);
+        $names =& $this->names;
+        $that = $this;
+        $deferred = new Deferred(function () use ($that, &$names, $request) {
+            // remove from list of pending names, but remember pending query
+            $name = $names[$request->id];
+            unset($names[$request->id]);
+            $that->checkIdle();
 
-            throw new CancellationException('DNS query for ' . $query->name . ' has been cancelled');
+            throw new CancellationException('DNS query for ' . $name . ' has been cancelled');
         });
 
-        // set socket to non-blocking and wait for it to become writable (connection success/rejected)
-        \stream_set_blocking($socket, false);
-        $loop->addWriteStream($socket, function ($socket) use ($loop, $query, $queryData, $deferred) {
-            $loop->removeWriteStream($socket);
-            $name = @\stream_socket_get_name($socket, true);
+        $this->pending[$request->id] = $deferred;
+        $this->names[$request->id] = $query->name;
+
+        return $deferred->promise();
+    }
+
+    /**
+     * @internal
+     */
+    public function handleWritable()
+    {
+        if ($this->readPending === false) {
+            $name = @\stream_socket_get_name($this->socket, true);
             if ($name === false) {
-                $loop->removeReadStream($socket);
-                @\fclose($socket);
-                $deferred->reject(new \RuntimeException(
-                    'DNS query for ' . $query->name . ' failed: Connection to DNS server rejected'
-                ));
+                $this->closeError('Connection to DNS server rejected');
                 return;
             }
 
-            $written = @\fwrite($socket, $queryData);
-            if ($written !== \strlen($queryData)) {
-                $loop->removeReadStream($socket);
-                \fclose($socket);
-                $deferred->reject(new \RuntimeException(
-                    'DNS query for ' . $query->name . ' failed: Unable to write DNS query message in one chunk'
-                ));
-            }
-        });
+            $this->readPending = true;
+            $this->loop->addReadStream($this->socket, array($this, 'handleRead'));
+        }
 
-        $buffer = '';
-        $parser = $this->parser;
-        $loop->addReadStream($socket, function ($socket) use (&$buffer, $loop, $deferred, $query, $parser, $request) {
-            // read one chunk of data from the DNS server
-            // any error is fatal, this is a stream of TCP/IP data
-            $chunk = @\fread($socket, 65536);
-            if ($chunk === false || $chunk === '') {
-                $loop->removeReadStream($socket);
-                \fclose($socket);
-                $deferred->reject(new \RuntimeException(
-                    'DNS query for ' . $query->name . ' failed: Connection to DNS server lost'
-                ));
-                return;
-            }
+        $written = @\fwrite($this->socket, $this->writeBuffer);
+        if ($written === false || $written === 0) {
+            $this->closeError('Unable to write to closed socket');
+            return;
+        }
 
-            // reassemble complete message by concatenating all chunks.
-            // response message header contains at least 12 bytes
-            $buffer .= $chunk;
-            if (!isset($buffer[11])) {
-                return;
-            }
+        if (isset($this->writeBuffer[$written])) {
+            $this->writeBuffer = \substr($this->writeBuffer, $written);
+        } else {
+            $this->loop->removeWriteStream($this->socket);
+            $this->writePending = false;
+            $this->writeBuffer = '';
+        }
+    }
+
+    /**
+     * @internal
+     */
+    public function handleRead()
+    {
+        // read one chunk of data from the DNS server
+        // any error is fatal, this is a stream of TCP/IP data
+        $chunk = @\fread($this->socket, 65536);
+        if ($chunk === false || $chunk === '') {
+            $this->closeError('Connection to DNS server lost');
+            return;
+        }
 
+        // reassemble complete message by concatenating all chunks.
+        $this->readBuffer .= $chunk;
+
+        // response message header contains at least 12 bytes
+        while (isset($this->readBuffer[11])) {
             // read response message length from first 2 bytes and ensure we have length + data in buffer
-            list(, $length) = \unpack('n', $buffer);
-            if (!isset($buffer[$length + 1])) {
+            list(, $length) = \unpack('n', $this->readBuffer);
+            if (!isset($this->readBuffer[$length + 1])) {
                 return;
             }
 
-            // we only react to the first complete message, so remove socket from loop and close
-            $loop->removeReadStream($socket);
-            \fclose($socket);
-            $data = \substr($buffer, 2, $length);
-            $buffer = '';
+            $data = \substr($this->readBuffer, 2, $length);
+            $this->readBuffer = (string)substr($this->readBuffer, $length + 2);
 
             try {
-                $response = $parser->parseMessage($data);
+                $response = $this->parser->parseMessage($data);
             } catch (\Exception $e) {
-                // reject if we received an invalid message from remote server
-                $deferred->reject(new \RuntimeException(
-                    'DNS query for ' . $query->name . ' failed: Invalid message received from DNS server',
-                    0,
-                    $e
-                ));
+                // reject all pending queries if we received an invalid message from remote server
+                $this->closeError('Invalid message received from DNS server');
                 return;
             }
 
-            // reject if we received an unexpected response ID or truncated response
-            if ($response->id !== $request->id || $response->tc) {
-                $deferred->reject(new \RuntimeException(
-                    'DNS query for ' . $query->name . ' failed: Invalid response message received from DNS server'
-                ));
+            // reject all pending queries if we received an unexpected response ID or truncated response
+            if (!isset($this->pending[$response->id]) || $response->tc) {
+                $this->closeError('Invalid response message received from DNS server');
                 return;
             }
 
+            $deferred = $this->pending[$response->id];
+            unset($this->pending[$response->id], $this->names[$response->id]);
+
             $deferred->resolve($response);
-        });
 
-        return $deferred->promise();
+            $this->checkIdle();
+        }
+    }
+
+    /**
+     * @internal
+     * @param string $reason
+     */
+    public function closeError($reason)
+    {
+        $this->readBuffer = '';
+        if ($this->readPending) {
+            $this->loop->removeReadStream($this->socket);
+            $this->readPending = false;
+        }
+
+        $this->writeBuffer = '';
+        if ($this->writePending) {
+            $this->loop->removeWriteStream($this->socket);
+            $this->writePending = false;
+        }
+
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
+        }
+
+        @\fclose($this->socket);
+        $this->socket = null;
+
+        foreach ($this->names as $id => $name) {
+            $this->pending[$id]->reject(new \RuntimeException(
+                'DNS query for ' . $name . ' failed: ' . $reason
+            ));
+        }
+        $this->pending = $this->names = array();
+    }
+
+    /**
+     * @internal
+     */
+    public function checkIdle()
+    {
+        if ($this->idleTimer === null && !$this->names) {
+            $that = $this;
+            $this->idleTimer = $this->loop->addTimer($this->idlePeriod, function () use ($that) {
+                $that->closeError('Idle timeout');
+            });
+        }
     }
 }
diff --git a/tests/Query/TcpTransportExecutorTest.php b/tests/Query/TcpTransportExecutorTest.php
