conn: create from socket fd

This patch adds the ability to create Tarantool connection
using an existing socket fd.

To achieve this, several changes have been made to work with
non-blocking sockets, as `socket.socketpair` creates such [1].

The authentication [2] might have already occured when we establish
such a connection. If that's the case, there is no need to pass
'user' argument.

On success, connect takes ownership of the `fd`.

1. tarantool/tarantool#8944
2. https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/authentication/

Closes #304

Author: askalt

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Added
+- The ability to connect to the Tarantool using an existing socket fd (#304).
+
 ## 1.1.2 - 2023-09-20
 
 ### Fixed

tarantool/connection.py

@@ -1,9 +1,11 @@
 """
 This module provides API for interaction with a Tarantool server.
 """
+import fcntl
 # pylint: disable=too-many-lines,duplicate-code
 
 import os
+import select
 import time
 import errno
 from enum import Enum
@@ -623,7 +625,7 @@ def __init__(self, host, port,
             Unix sockets.
         :type host: :obj:`str` or :obj:`None`
 
-        :param port: Server port or Unix socket path.
+        :param port: Server port, socket fd or Unix socket path.
         :type port: :obj:`int` or :obj:`str`
 
         :param user: User name for authentication on the Tarantool
@@ -897,10 +899,34 @@ def connect_basic(self):
         :meta private:
         """
 
-        if self.host is None:
-            self.connect_unix()
-        else:
+        if self.host is not None:
             self.connect_tcp()
+        elif isinstance(self.port, int):
+            self.connect_socket_fd()
+        else:
+            self.connect_unix()
+
+    def connect_socket_fd(self):
+        """
+        Establish a connection using an existing socket fd.
+
+        :raise: :exc:`OSError`
+
+        :meta private:
+        """
+
+        # If old socket already exists - close it and re-create.
+        self.connected = True
+        if self._socket:
+            self._socket.close()
+
+        socket_fd = self.port
+        self._socket = socket.socket(fileno=socket_fd)
+
+        is_non_blocking = fcntl.fcntl(socket_fd, fcntl.F_GETFL) & os.O_NONBLOCK != 0
+        if is_non_blocking:
+            # Explicitly specify, because otherwise it will not be initialized correctly.
+            self._socket.settimeout(0)
 
     def connect_tcp(self):
         """
@@ -1120,9 +1146,12 @@ def _recv(self, to_read):
         :meta private:
         """
 
+        is_non_blocking = self._socket.gettimeout() == 0
         buf = b""
         while to_read > 0:
             try:
+                if is_non_blocking:
+                    select.select([self._socket.fileno()], [], [])
                 tmp = self._socket.recv(to_read)
             except OverflowError as exc:
                 self._socket.close()
@@ -1163,6 +1192,41 @@ def _read_response(self):
         # Read the packet
         return self._recv(length)
 
+    def _sendall(self, bytes):
+        """
+        Sends bytes to the transport (socket).
+
+        :param bytes: message to send.
+        :type bytes: :obj:`bytes`
+
+        :raise: :exc:`~tarantool.error.NetworkError`
+
+        :meta: private:
+        """
+        is_non_blocking = self._socket.gettimeout() == 0
+        total_sent = 0
+        while total_sent < len(bytes):
+            try:
+                if is_non_blocking:
+                    select.select([], [self._socket.fileno()], [])
+                sent = self._socket.send(bytes[total_sent:])
+                if sent == 0:
+                    err = socket.error(
+                        errno.ECONNRESET,
+                        "Lost connection to server during query"
+                    )
+                    raise NetworkError(err)
+                total_sent += sent
+            except BlockingIOError as exc:
+                total_sent += exc.characters_written
+                continue
+            except socket.error as exc:
+                err = socket.error(
+                    errno.ECONNRESET,
+                    "Lost connection to server during query"
+                )
+                raise NetworkError(err) from exc
+
     def _send_request_wo_reconnect(self, request, on_push=None, on_push_ctx=None):
         """
         Send request without trying to reconnect.
@@ -1191,7 +1255,7 @@ def _send_request_wo_reconnect(self, request, on_push=None, on_push_ctx=None):
         response = None
         while True:
             try:
-                self._socket.sendall(bytes(request))
+                self._sendall(bytes(request))
                 response = request.response_class(self, self._read_response())
                 break
             except SchemaReloadException as exc:

tarantool/mesh_connection.py

@@ -139,23 +139,23 @@ def format_error(address, err):
             result[key] = val
 
     if isinstance(result['port'], int):
-        # Looks like an inet address.
+        # Looks like an inet address or socket fd.
 
         # Validate host.
-        if 'host' not in result or result['host'] is None:
-            return format_error(result,
-                                'host is mandatory for an inet result')
-        if not isinstance(result['host'], str):
-            return format_error(result,
-                                'host must be a string for an inet result')
+        if ('host' in result and result['host'] is not None
+                and not isinstance(result['host'], str)):
+            return format_error(result, 'host must be a string for an inet result, '
+                                        'or None for a socket fd result')
 
         # Validate port.
         if not isinstance(result['port'], int):
             return format_error(result,
-                                'port must be an int for an inet result')
-        if result['port'] < 1 or result['port'] > 65535:
-            return format_error(result, 'port must be in range [1, 65535] '
-                                        'for an inet result')
+                                'port must be an int for an inet/socket fd result')
+        if 'host' in result and isinstance(result['host'], str):
+            # Check that the port is correct.
+            if result['port'] < 1 or result['port'] > 65535:
+                return format_error(result, 'port must be in range [1, 65535] '
+                                            'for an inet result')
 
         # Looks okay.
         return result, None
@@ -447,7 +447,8 @@ def __init__(self, host=None, port=None,
             # Don't change user provided arguments.
             addrs = addrs[:]
 
-        if host and port:
+        if port:
+            # host can be None in the case of socket fd or Unix socket.
             addrs.insert(0, {'host': host,
                              'port': port,
                              'transport': transport,

test/suites/__init__.py

@@ -15,6 +15,7 @@
 from .test_execute import TestSuiteExecute
 from .test_dbapi import TestSuiteDBAPI
 from .test_encoding import TestSuiteEncoding
+from .test_socket_fd import TestSuiteSocketFD
 from .test_ssl import TestSuiteSsl
 from .test_decimal import TestSuiteDecimal
 from .test_uuid import TestSuiteUUID
@@ -33,7 +34,7 @@
               TestSuiteEncoding, TestSuitePool, TestSuiteSsl,
               TestSuiteDecimal, TestSuiteUUID, TestSuiteDatetime,
               TestSuiteInterval, TestSuitePackage, TestSuiteErrorExt,
-              TestSuitePush, TestSuiteConnection, TestSuiteCrud,)
+              TestSuitePush, TestSuiteConnection, TestSuiteCrud, TestSuiteSocketFD)
 
 
 def load_tests(loader, tests, pattern):

test/suites/lib/skip.py

@@ -306,3 +306,14 @@ def skip_or_run_iproto_basic_features_test(func):
 
     return skip_or_run_test_tarantool(func, '2.10.0',
                                       'does not support iproto ID and iproto basic features')
+
+
+def skip_or_run_box_session_new_tests(func):
+    """
+    Decorator to skip or run tests that use box.session.new.
+
+    Tarantool supports box.session.new only in current master since
+    commit 324872a.
+    See https://github.com/tarantool/tarantool/issues/8801.
+    """
+    return skip_or_run_test_tarantool(func, '3.0.0', 'does not support box.session.new')

test/suites/sidecar.py

@@ -0,0 +1,16 @@
+# pylint: disable=missing-module-docstring
+import os
+
+import tarantool
+
+socket_fd = int(os.environ["SOCKET_FD"])
+
+conn = tarantool.connect(host=None, port=socket_fd)
+
+# Check user.
+assert conn.eval("return box.session.user()").data[0] == "test"
+
+# Check db operations.
+conn.insert("test", [1])
+conn.insert("test", [2])
+assert conn.select("test").data == [[1], [2]]

test/suites/test_mesh.py

@@ -135,7 +135,6 @@ def test_01_contructor(self):
         # Verify that a bad address given at initialization leads
         # to an error.
         bad_addrs = [
-            {"port": 1234},                         # no host
             {"host": "localhost"},                  # no port
             {"host": "localhost", "port": "1234"},  # port is str
         ]