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 `socket_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/__init__.py

@@ -51,7 +51,7 @@
     __version__ = '0.0.0-dev'
 
 
-def connect(host="localhost", port=33013, user=None, password=None,
+def connect(host="localhost", port=33013, socket_fd=None, user=None, password=None,
             encoding=ENCODING_DEFAULT, transport=DEFAULT_TRANSPORT,
             ssl_key_file=DEFAULT_SSL_KEY_FILE,
             ssl_cert_file=DEFAULT_SSL_CERT_FILE,
@@ -64,6 +64,8 @@ def connect(host="localhost", port=33013, user=None, password=None,
 
     :param port: Refer to :paramref:`~tarantool.Connection.params.port`.
 
+    :param socket_fd: Refer to :paramref:`~tarantool.Connection.params.socket_fd`.
+
     :param user: Refer to :paramref:`~tarantool.Connection.params.user`.
 
     :param password: Refer to
@@ -93,6 +95,7 @@ def connect(host="localhost", port=33013, user=None, password=None,
     """
 
     return Connection(host, port,
+                      socket_fd=socket_fd,
                       user=user,
                       password=password,
                       socket_timeout=SOCKET_TIMEOUT,

tarantool/connection.py

@@ -4,6 +4,7 @@
 # pylint: disable=too-many-lines,duplicate-code
 
 import os
+import select
 import time
 import errno
 from enum import Enum
@@ -594,7 +595,10 @@ class Connection(ConnectionInterface):
     :value: :exc:`~tarantool.error.CrudModuleError`
     """
 
-    def __init__(self, host, port,
+    def __init__(self,
+                 host=None,
+                 port=None,
+                 socket_fd=None,
                  user=None,
                  password=None,
                  socket_timeout=SOCKET_TIMEOUT,
@@ -623,8 +627,11 @@ def __init__(self, host, port,
             Unix sockets.
         :type host: :obj:`str` or :obj:`None`
 
-        :param port: Server port or Unix socket path.
-        :type port: :obj:`int` or :obj:`str`
+        :param port: Server port, or Unix socket path.
+        :type port: :obj:`int` or :obj:`str` or :obj:`None`
+
+        :param socket_fd: socket fd number.
+        :type socket_fd: :obj:`int` or :obj:`None`
 
         :param user: User name for authentication on the Tarantool
             server.
@@ -803,6 +810,8 @@ def __init__(self, host, port,
         .. _mp_array: https://github.com/msgpack/msgpack/blob/master/spec.md#array-format-family
         """
         # pylint: disable=too-many-arguments,too-many-locals,too-many-statements
+        if socket_fd is not None and (host is not None or port is not None):
+            raise ConfigurationError("specifying both socket_fd and host/port is not allowed")
 
         if msgpack.version >= (1, 0, 0) and encoding not in (None, 'utf-8'):
             raise ConfigurationError("msgpack>=1.0.0 only supports None and "
@@ -820,6 +829,7 @@ def __init__(self, host, port,
         recv.restype = ctypes.c_int
         self.host = host
         self.port = port
+        self.socket_fd = socket_fd
         self.user = user
         self.password = password
         self.socket_timeout = socket_timeout
@@ -897,10 +907,35 @@ def connect_basic(self):
         :meta private:
         """
 
-        if self.host is None:
-            self.connect_unix()
-        else:
+        if self.socket_fd is not None:
+            self.connect_socket_fd()
+        elif self.host is not None:
             self.connect_tcp()
+        else:
+            self.connect_unix()
+
+    def connect_socket_fd(self):
+        """
+        Establish a connection using an existing socket fd.
+
+        socket_fd / timeout|         >= 0            |         None
+        -------------------|-------------------------|-------------------------
+              blocking     | set non-blocking        |  don't change
+                           | socket lib call `select`| `select` isn't needed
+        -------------------|-------------------------|-------------------------
+            non-blocking   | don't change            |  don't change
+                           | socket lib call `select`|  call `select` ourselves
+        -------------------|-------------------------|-------------------------
+
+        :meta private:
+        """
+        self.connected = True
+        if self._socket:
+            self._socket.close()
+
+        self._socket = socket.socket(fileno=self.socket_fd)
+        if self.socket_timeout is not None:
+            self._socket.settimeout(self.socket_timeout)
 
     def connect_tcp(self):
         """
@@ -1124,6 +1159,11 @@ def _recv(self, to_read):
         while to_read > 0:
             try:
                 tmp = self._socket.recv(to_read)
+            except BlockingIOError:
+                ready, _, _ = select.select([self._socket.fileno()], [], [], self.socket_timeout)
+                if not ready:
+                    raise NetworkError(TimeoutError())  # pylint: disable=raise-missing-from
+                continue
             except OverflowError as exc:
                 self._socket.close()
                 err = socket.error(
@@ -1163,6 +1203,40 @@ def _read_response(self):
         # Read the packet
         return self._recv(length)
 
+    def _sendall(self, bytes_to_send):
+        """
+        Sends bytes to the transport (socket).
+
+        :param bytes_to_send: message to send.
+        :type bytes_to_send: :obj:`bytes`
+
+        :raise: :exc:`~tarantool.error.NetworkError`
+
+        :meta: private:
+        """
+        total_sent = 0
+        while total_sent < len(bytes_to_send):
+            try:
+                sent = self._socket.send(bytes_to_send[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
+                _, ready, _ = select.select([], [self._socket.fileno()], [], self.socket_timeout)
+                if not ready:
+                    raise NetworkError(TimeoutError())  # pylint: disable=raise-missing-from
+            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 +1265,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/connection_pool.py

@@ -115,7 +115,7 @@ class PoolUnit():
 
     addr: dict
     """
-    ``{"host": host, "port": port}`` info.
+    ``{"host": host, "port": port, "socket_fd": socket_fd}`` info.
 
     :type: :obj:`dict`
     """
@@ -161,6 +161,14 @@ class PoolUnit():
     :type: :obj:`bool`
     """
 
+    def get_address(self):
+        """
+        Get an address string representation.
+        """
+        if self.addr['socket_fd'] is not None:
+            return f'fd://{self.addr["socket_fd"]}'
+        return f'{self.addr["host"]}:{self.addr["port"]}'
+
 
 # Based on https://realpython.com/python-interface/
 class StrategyInterface(metaclass=abc.ABCMeta):
@@ -398,6 +406,7 @@ def __init__(self,
                 {
                     "host': "str" or None,     # mandatory
                     "port": int or "str",      # mandatory
+                    "socket_fd": int,          # optional
                     "transport": "str",        # optional
                     "ssl_key_file": "str",     # optional
                     "ssl_cert_file": "str",    # optional
@@ -499,6 +508,7 @@ def __init__(self,
                 conn=Connection(
                     host=addr['host'],
                     port=addr['port'],
+                    socket_fd=addr['socket_fd'],
                     user=user,
                     password=password,
                     socket_timeout=socket_timeout,
@@ -529,15 +539,16 @@ def _make_key(self, addr):
         """
         Make a unique key for a server based on its address.
 
-        :param addr: `{"host": host, "port": port}` dictionary.
+        :param addr: `{"host": host, "port": port, "socket_fd": socket_fd}` dictionary.
         :type addr: :obj:`dict`
 
         :rtype: :obj:`str`
 
         :meta private:
         """
-
-        return f"{addr['host']}:{addr['port']}"
+        if addr['socket_fd'] is None:
+            return f"{addr['host']}:{addr['port']}"
+        return addr['socket_fd']
 
     def _get_new_state(self, unit):
         """
@@ -557,23 +568,23 @@ def _get_new_state(self, unit):
             try:
                 conn.connect()
             except NetworkError as exc:
-                msg = (f"Failed to connect to {unit.addr['host']}:{unit.addr['port']}, "
+                msg = (f"Failed to connect to {unit.get_address()}, "
                        f"reason: {repr(exc)}")
                 warn(msg, ClusterConnectWarning)
                 return InstanceState(Status.UNHEALTHY)
 
         try:
             resp = conn.call('box.info')
         except NetworkError as exc:
-            msg = (f"Failed to get box.info for {unit.addr['host']}:{unit.addr['port']}, "
+            msg = (f"Failed to get box.info for {unit.get_address()}, "
                    f"reason: {repr(exc)}")
             warn(msg, PoolTolopogyWarning)
             return InstanceState(Status.UNHEALTHY)
 
         try:
             read_only = resp.data[0]['ro']
         except (IndexError, KeyError) as exc:
-            msg = (f"Incorrect box.info response from {unit.addr['host']}:{unit.addr['port']}"
+            msg = (f"Incorrect box.info response from {unit.get_address()}"
                    f"reason: {repr(exc)}")
             warn(msg, PoolTolopogyWarning)
             return InstanceState(Status.UNHEALTHY)
@@ -582,11 +593,11 @@ def _get_new_state(self, unit):
             status = resp.data[0]['status']
 
             if status != 'running':
-                msg = f"{unit.addr['host']}:{unit.addr['port']} instance status is not 'running'"
+                msg = f"{unit.get_address()} instance status is not 'running'"
                 warn(msg, PoolTolopogyWarning)
                 return InstanceState(Status.UNHEALTHY)
         except (IndexError, KeyError) as exc:
-            msg = (f"Incorrect box.info response from {unit.addr['host']}:{unit.addr['port']}"
+            msg = (f"Incorrect box.info response from {unit.get_address()}"
                    f"reason: {repr(exc)}")
             warn(msg, PoolTolopogyWarning)
             return InstanceState(Status.UNHEALTHY)

tarantool/const.py

@@ -103,6 +103,8 @@
 IPROTO_FEATURE_SPACE_AND_INDEX_NAMES = 5
 IPROTO_FEATURE_WATCH_ONCE = 6
 
+# Default value for socket_fd param.
+DEFAULT_SOCKET_FD = None
 # Default value for connection timeout (seconds)
 CONNECTION_TIMEOUT = None
 # Default value for socket timeout (seconds)

tarantool/mesh_connection.py

@@ -28,13 +28,15 @@
     DEFAULT_SSL_PASSWORD,
     DEFAULT_SSL_PASSWORD_FILE,
     CLUSTER_DISCOVERY_DELAY,
+    DEFAULT_SOCKET_FD,
 )
 
 from tarantool.request import (
     RequestCall
 )
 
 default_addr_opts = {
+    'socket_fd': DEFAULT_SOCKET_FD,
     'transport': DEFAULT_TRANSPORT,
     'ssl_key_file': DEFAULT_SSL_KEY_FILE,
     'ssl_cert_file': DEFAULT_SSL_CERT_FILE,
@@ -127,9 +129,6 @@ def format_error(address, err):
     if not isinstance(address, dict):
         return format_error(address, 'address must be a dict')
 
-    if 'port' not in address or address['port'] is None:
-        return format_error(address, 'port is not set or None')
-
     result = {}
     for key, val in address.items():
         result[key] = val
@@ -138,6 +137,13 @@ def format_error(address, err):
         if key not in result:
             result[key] = val
 
+    if result['socket_fd'] is not None:
+        # Looks like socket fd.
+        if not isinstance(result['socket_fd'], int):
+            return format_error(result,
+                                'socket_fd must be an int')
+        return result, None
+
     if isinstance(result['port'], int):
         # Looks like an inet address.
 
@@ -192,6 +198,7 @@ def update_connection(conn, address):
 
     conn.host = address["host"]
     conn.port = address["port"]
+    conn.socket_fd = address["socket_fd"]
     conn.transport = address['transport']
     conn.ssl_key_file = address['ssl_key_file']
     conn.ssl_cert_file = address['ssl_cert_file']
@@ -271,6 +278,7 @@ class MeshConnection(Connection):
     def __init__(self, host=None, port=None,
                  user=None,
                  password=None,
+                 socket_fd=None,
                  socket_timeout=SOCKET_TIMEOUT,
                  reconnect_max_attempts=RECONNECT_MAX_ATTEMPTS,
                  reconnect_delay=RECONNECT_DELAY,
@@ -302,6 +310,11 @@ def __init__(self, host=None, port=None,
             Value would be used to add one more server in
             :paramref:`~tarantool.MeshConnection.params.addrs` list.
 
+        :param socket_fd: Refer to
+            :paramref:`~tarantol.Connection.params.socket_fd`.
+            Value would be used to add one more server in
+            :paramref:`~tarantool.MeshConnection.params.addrs` list.
+
         :param user: Refer to
             :paramref:`~tarantool.Connection.params.user`.
             Value would be used to add one more server in
@@ -447,9 +460,10 @@ def __init__(self, host=None, port=None,
             # Don't change user provided arguments.
             addrs = addrs[:]
 
-        if host and port:
+        if (host and port) or socket_fd:
             addrs.insert(0, {'host': host,
                              'port': port,
+                             'socket_fd': socket_fd,
                              'transport': transport,
                              'ssl_key_file': ssl_key_file,
                              'ssl_cert_file': ssl_cert_file,
@@ -484,6 +498,7 @@ def __init__(self, host=None, port=None,
         super().__init__(
             host=addr['host'],
             port=addr['port'],
+            socket_fd=addr['socket_fd'],
             user=user,
             password=password,
             socket_timeout=socket_timeout,
@@ -604,6 +619,7 @@ def _opt_refresh_instances(self):
         # an instance list and connect to one of new instances.
         current_addr = {'host': self.host,
                         'port': self.port,
+                        'socket_fd': self.socket_fd,
                         'transport': self.transport,
                         'ssl_key_file': self.ssl_key_file,
                         'ssl_cert_file': self.ssl_cert_file,