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

@@ -4,6 +4,7 @@
 # pylint: disable=too-many-lines,duplicate-code
 
 import os
+import select
 import time
 import errno
 from enum import Enum
@@ -623,7 +624,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 +898,25 @@ 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.
+
+        :meta private:
+        """
+        self.connected = True
+        if self._socket:
+            self._socket.close()
+
+        socket_fd = self.port
+        self._socket = socket.socket(fileno=socket_fd)
 
     def connect_tcp(self):
         """
@@ -1124,6 +1140,9 @@ def _recv(self, to_read):
         while to_read > 0:
             try:
                 tmp = self._socket.recv(to_read)
+            except BlockingIOError:
+                select.select([self._socket.fileno()], [], [])
+                continue
             except OverflowError as exc:
                 self._socket.close()
                 err = socket.error(
@@ -1163,6 +1182,39 @@ 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
+                select.select([], [self._socket.fileno()], [])
+                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 +1243,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
         ]

test/suites/test_socket_fd.py

@@ -0,0 +1,175 @@
+"""
+This module tests work with connection over socket fd.
+"""
+import os.path
+# pylint: disable=missing-class-docstring,missing-function-docstring
+
+import socket
+import sys
+import unittest
+
+import tarantool
+from .lib.skip import skip_or_run_box_session_new_tests
+from .lib.tarantool_server import TarantoolServer, find_port
+from .utils import assert_admin_success
+
+
+def find_python():
+    for _dir in os.environ["PATH"].split(os.pathsep):
+        exe = os.path.join(_dir, "python")
+        if os.access(exe, os.X_OK):
+            return os.path.abspath(exe)
+    raise RuntimeError("Can't find python executable in " + os.environ["PATH"])
+
+
+class TestSuiteSocketFD(unittest.TestCase):
+    EVAL_USER = "return box.session.user()"
+
+    @classmethod
+    def setUpClass(cls):
+        print(' SOCKET FD '.center(70, '='), file=sys.stderr)
+        print('-' * 70, file=sys.stderr)
+
+        cls.srv = TarantoolServer()
+        cls.srv.script = 'test/suites/box.lua'
+        cls.srv.start()
+        cls.tcp_port = find_port()
+
+        # Start tcp server to test work with blocking sockets.
+        # pylint:disable=C0209
+        resp = cls.srv.admin("""
+            local socket = require('socket')
+
+            box.cfg{}
+            box.schema.user.create('test', {password = 'test', if_not_exists = true})
+            box.schema.user.grant('test', 'read,write,execute,create', 'universe',
+                                   nil, {if_not_exists = true})
+            box.schema.user.grant('guest', 'execute', 'universe',
+                                   nil, {if_not_exists = true})
+
+            socket.tcp_server('0.0.0.0', %d, function(s)
+                if not s:nonblock(true) then
+                    s:close()
+                    return
+                end
+                box.session.new({
+                    type = 'binary',
+                    fd = s:fd(),
+                    user = 'test',
+                })
+                s:detach()
+            end)
+
+            box.schema.create_space('test', {
+                format = {{type='unsigned', name='id'}},
+                if_not_exists = true,
+            })
+            box.space.test:create_index('primary')
+
+            return true
+        """ % cls.tcp_port)
+        assert_admin_success(resp)
+
+    @skip_or_run_box_session_new_tests
+    def setUp(self):
+        # Prevent a remote tarantool from clean our session.
+        if self.srv.is_started():
+            self.srv.touch_lock()
+
+    def _get_tt_sock(self):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.connect((self.srv.host, self.tcp_port))
+        return sock
+
+    def test_01_socket_fd_connect(self):
+        sock = self._get_tt_sock()
+        conn = tarantool.connect(host=None, port=sock.fileno())
+        sock.detach()
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["test"])
+        finally:
+            conn.close()
+
+    def test_02_socket_fd_re_auth(self):
+        sock = self._get_tt_sock()
+        conn = tarantool.connect(host=None, port=sock.fileno(), user="guest")
+        sock.detach()
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["guest"])
+        finally:
+            conn.close()
+
+    def test_03_socket_fd_mesh(self):
+        sock = self._get_tt_sock()
+        pool = tarantool.ConnectionPool(
+            addrs=[{'host': None, 'port': sock.fileno()}]
+        )
+        sock.detach()
+        try:
+            self.assertSequenceEqual(pool.eval(self.EVAL_USER, mode=tarantool.Mode.ANY), ["test"])
+        finally:
+            pool.close()
+
+    def test_04_socket_fd_pool(self):
+        sock = self._get_tt_sock()
+        mesh = tarantool.MeshConnection(
+            host=None, port=sock.fileno()
+        )
+        sock.detach()
+        try:
+            self.assertSequenceEqual(mesh.eval(self.EVAL_USER), ["test"])
+        finally:
+            mesh.close()
+
+    @unittest.skipIf(sys.platform.startswith("win"),
+                     "Skip on Windows since it uses remote server")
+    def test_05_tarantool_made_socket(self):
+        python_exe = find_python()
+        cwd = os.getcwd()
+        side_script_path = os.path.join(cwd, "test", "suites", "sidecar.py")
+
+        # pylint:disable=C0209
+        ret_code, err = self.srv.admin("""
+            local socket = require('socket')
+            local popen = require('popen')
+            local os = require('os')
+            local s1, s2 = socket.socketpair('AF_UNIX', 'SOCK_STREAM', 0)
+
+            --[[ Tell sidecar which fd use to connect. --]]
+            os.setenv('SOCKET_FD', tostring(s2:fd()))
+
+            --[[ Tell sidecar where find `tarantool` module. --]]
+            os.setenv('PYTHONPATH', (os.getenv('PYTHONPATH') or '') .. ':' .. '%s')
+
+            box.session.new({
+                type = 'binary',
+                fd = s1:fd(),
+                user = 'test',
+            })
+            s1:detach()
+
+            local ph, err = popen.new({'%s', '%s'}, {
+                stdout = popen.opts.PIPE,
+                stderr = popen.opts.PIPE,
+                inherit_fds = {s2:fd()},
+            })
+
+            if err ~= nil then
+                return 1, err
+            end
+
+            ph:wait()
+
+            local status_code = ph:info().status.exit_code
+            local stderr = ph:read({stderr=true}):rstrip()
+            return status_code, stderr
+        """ % (cwd, python_exe, side_script_path))
+        if err is not None:
+            print(err)
+        self.assertEqual(ret_code, 0)
+        self.assertIsNone(err)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.srv.stop()
+        cls.srv.clean()