[tarantool-patches] [PATCH v2 0/3] console: do not use netbox for console text connections

[tarantool-patches] [PATCH v2 0/3] console: do not use netbox for console text connections

[Vladislav Shpilevoy]

Branch: http://github.com/tarantool/tarantool/tree/gh-2677-prepare-console-for-push
Issue: https://github.com/tarantool/tarantool/issues/2677

Netbox console support complicates both netbox and console. Lets
use sockets directly for text protocol. But the problem is that
before a connection is established, the connection protocol is
unknown (binary or text). It can be determined only when a
greeting is read and decoded. After that the socket must be either
wrapped by netbox API for binary connections, or by text console
API for text connections. To be able to do that the new netbox
functions are introduced: establish_connection and wrap.

Netbox.establish_connection allows to connect to a server, read
and decode greeting.

Netbox.wrap creates a netbox state machine around a given socket.

Vladislav Shpilevoy (3):
  netbox: allow to create a netbox connection from existing socket
  console: do not use netbox for console text connections
  netbox: deprecate console support

 src/box/lua/console.lua   | 162 +++++++++++++++++++++++++++++-------
 src/box/lua/net_box.lua   | 204 ++++++++++++++++++++++++++++++++--------------
 test/box/net.box.result   |  91 ++++++++++++++++++++-
 test/box/net.box.test.lua |  41 +++++++++-
 4 files changed, 400 insertions(+), 98 deletions(-)

-- 
2.14.3 (Apple Git-98)

[tarantool-patches] [PATCH v2 1/3] netbox: allow to create a netbox connection from existing socket

[Vladislav Shpilevoy]

It is needed to create a binary console connection, when a
socket is already created and a greeting is read and decoded.
---
 src/box/lua/net_box.lua   | 199 ++++++++++++++++++++++++++++++++--------------
 test/box/net.box.result   |  91 ++++++++++++++++++++-
 test/box/net.box.test.lua |  41 +++++++++-
 3 files changed, 264 insertions(+), 67 deletions(-)

diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
index 87c8c548b..5691dcb9e 100644
--- a/src/box/lua/net_box.lua
+++ b/src/box/lua/net_box.lua
@@ -31,6 +31,7 @@ local sequence_mt      = { __serialize = 'sequence' }
 local TIMEOUT_INFINITY = 500 * 365 * 86400
 local VSPACE_ID        = 281
 local VINDEX_ID        = 289
+local DEFAULT_CONNECT_TIMEOUT = 10
 
 local IPROTO_STATUS_KEY    = 0x00
 local IPROTO_ERRNO_MASK    = 0x7FFF
@@ -70,9 +71,37 @@ local method_codec           = {
 
 local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
 
--- function create_transport(host, port, user, password, callback)
 --
--- Transport methods: connect(), close(), perfrom_request(), wait_state()
+-- Connect to a remote server, do handshake.
+-- @param host Hostname.
+-- @param port TCP port.
+-- @param timeout Timeout to connect and receive greeting.
+--
+-- @retval nil, err Error occured. The reason is returned.
+-- @retval two non-nils A connected socket and a decoded greeting.
+--
+local function establish_connection(host, port, timeout)
+    local timeout = timeout or DEFAULT_CONNECT_TIMEOUT
+    local begin = fiber.clock()
+    local s = socket.tcp_connect(host, port, timeout)
+    if not s then
+        return nil, errno.strerror(errno())
+    end
+    local msg = s:read({chunk = IPROTO_GREETING_SIZE},
+                        timeout - (fiber.clock() - begin))
+    if not msg then
+        local err = s:error()
+        s:close()
+        return nil, err
+    end
+    local greeting, err = decode_greeting(msg)
+    if not greeting then
+        s:close()
+        return nil, err
+    end
+    return s, greeting
+end
+
 --
 -- Basically, *transport* is a TCP connection speaking one of
 -- Tarantool network protocols. This is a low-level interface.
@@ -85,16 +114,17 @@ local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
 --
 -- Transport state machine:
 --
--- State machine starts in 'initial' state. Connect method changes
--- the state to 'connecting' and spawns a worker fiber. Close
--- method sets the state to 'closed' and kills the worker.
--- If the transport is already in 'error' state close() does nothing.
+-- State machine starts in 'initial' state. New_sm method
+-- accepts an established connection and spawns a worker fiber.
+-- Stop method sets the state to 'closed' and kills the worker.
+-- If the transport is already in 'error' state stop() does
+-- nothing.
 --
 -- State chart:
 --
---  initial -> connecting -> active
+-- connecting -> initial +-> active
 --                        \
---                          -> auth -> fetch_schema <-> active
+--                         +-> auth -> fetch_schema <-> active
 --
 --  (any state, on error) -> error_reconnect -> connecting -> ...
 --                                           \
@@ -121,15 +151,15 @@ local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
 -- Suggestion for callback writers: sleep a few secs before approving
 -- reconnect.
 --
-local function create_transport(host, port, user, password, callback)
+local function create_transport(host, port, user, password, callback,
+                                connection, greeting)
     -- check / normalize credentials
     if user == nil and password ~= nil then
         box.error(E_PROC_LUA, 'net.box: user is not defined')
     end
     if user ~= nil and password == nil then password = '' end
 
-    -- state: current state, only the worker fiber and connect method
-    -- change state
+    -- Current state machine's state.
     local state            = 'initial'
     local last_errno
     local last_error
@@ -145,7 +175,6 @@ local function create_transport(host, port, user, password, callback)
     local next_request_id  = 1
 
     local worker_fiber
-    local connection
     local send_buf         = buffer.ibuf(buffer.READAHEAD)
     local recv_buf         = buffer.ibuf(buffer.READAHEAD)
 
@@ -185,16 +214,19 @@ local function create_transport(host, port, user, password, callback)
         return state == target_state or target_state[state] or false
     end
 
-    -- CONNECT/CLOSE --
+    -- START/STOP --
     local protocol_sm
 
-    local function connect()
+    local function start()
         if state ~= 'initial' then return not is_final_state[state] end
-        set_state('connecting')
+        if not connection then
+            set_state('error', E_NO_CONNECTION)
+            return
+        end
         fiber.create(function()
             worker_fiber = fiber_self()
             fiber.name(string.format('%s:%s (net.box)', host, port), {truncate=true})
-    ::reconnect::
+    ::handle_connection::
             local ok, err = pcall(protocol_sm)
             if not (ok or is_final_state[state]) then
                 set_state('error', E_UNKNOWN, err)
@@ -204,21 +236,34 @@ local function create_transport(host, port, user, password, callback)
                 connection = nil
             end
             local timeout = callback('reconnect_timeout')
-            if timeout and state == 'error_reconnect' then
+            -- Non-nil reconnect_timeout is an indicator of set
+            -- reconnect_after option.
+            while timeout and state == 'error_reconnect' do
                 fiber.sleep(timeout)
+                -- During sleep, a connection could be garbage
+                -- collected, or explicitly closed by a user, or
+                -- the reconnect_after could be reset. Check the
+                -- timeout again.
                 timeout = callback('reconnect_timeout')
-                if timeout and state == 'error_reconnect' then
-                    goto reconnect
+                if not timeout or state ~= 'error_reconnect' then
+                    break
+                end
+                connection, greeting =
+                    establish_connection(host, port,
+                                         callback('fetch_connect_timeout'))
+                if connection then
+                    goto handle_connection
                 end
+                set_state('error_reconnect', E_NO_CONNECTION, greeting)
+                timeout = callback('reconnect_timeout')
             end
             send_buf:recycle()
             recv_buf:recycle()
             worker_fiber = nil
         end)
-        return true
     end
 
-    local function close()
+    local function stop()
         if not is_final_state[state] then
             set_state('closed', E_NO_CONNECTION, 'Connection closed')
         end
@@ -367,27 +412,14 @@ local function create_transport(host, port, user, password, callback)
     -- collected. See gh-3164, where because of reconnect sleeps
     -- in this function, a connection could not be deleted.
     --
-    protocol_sm = function ()
-        local tm_begin, tm = fiber.clock(), callback('fetch_connect_timeout')
-        connection = socket.tcp_connect(host, port, tm)
-        if connection == nil then
-            return error_sm(E_NO_CONNECTION, errno.strerror(errno()))
-        end
-        local size = IPROTO_GREETING_SIZE
-        local err, msg = send_and_recv(size, tm - (fiber.clock() - tm_begin))
-        if err then
-            return error_sm(err, msg)
-        end
-        local g = decode_greeting(ffi.string(recv_buf.rpos, size))
-        recv_buf.rpos = recv_buf.rpos + size
-        if not g then
-            return error_sm(E_NO_CONNECTION, 'Can\'t decode handshake')
-        end
-        err, msg = callback('handshake', g)
+    protocol_sm = function()
+        assert(connection)
+        assert(greeting)
+        local err, msg = callback('handshake', greeting)
         if err then
             return error_sm(err, msg)
         end
-        if g.protocol == 'Lua console' then
+        if greeting.protocol == 'Lua console' then
             local setup_delimiter = 'require("console").delimiter("$EOF$")\n'
             method_codec.inject(send_buf, nil, nil, setup_delimiter)
             local err, response = send_and_recv_console()
@@ -399,10 +431,11 @@ local function create_transport(host, port, user, password, callback)
             local rid = next_request_id
             set_state('active')
             return console_sm(rid)
-        elseif g.protocol == 'Binary' then
-            return iproto_auth_sm(g.salt)
+        elseif greeting.protocol == 'Binary' then
+            return iproto_auth_sm(greeting.salt)
         else
-            return error_sm(E_NO_CONNECTION, 'Unknown protocol: ' .. g.protocol)
+            return error_sm(E_NO_CONNECTION,
+                            'Unknown protocol: '..greeting.protocol)
         end
     end
 
@@ -524,14 +557,14 @@ local function create_transport(host, port, user, password, callback)
     end
 
     return {
-        close           = close,
-        connect         = connect,
+        stop            = stop,
+        start           = start,
         wait_state      = wait_state,
         perform_request = perform_request
     }
 end
 
--- Wrap create_transport, adding auto-close-on-GC feature.
+-- Wrap create_transport, adding auto-stop-on-GC feature.
 -- All the GC magic is neatly encapsulated!
 -- The tricky part is the callback:
 --  * callback (typically) references the transport (indirectly);
@@ -539,23 +572,24 @@ end
 --  * fibers are GC roots - i.e. transport is never GC-ed!
 -- We solve the issue by making the worker->callback ref weak.
 -- Now it is necessary to have a strong ref to callback somewhere or
--- it is GC-ed prematurely. We wrap close() method, stashing the
--- ref in an upvalue (close() performance doesn't matter much.)
-local create_transport = function(host, port, user, password, callback)
+-- it is GC-ed prematurely. We wrap stop() method, stashing the
+-- ref in an upvalue (stop() performance doesn't matter much.)
+local create_transport = function(host, port, user, password, callback,
+                                  connection, greeting)
     local weak_refs = setmetatable({callback = callback}, {__mode = 'v'})
     local function weak_callback(...)
         local callback = weak_refs.callback
         if callback then return callback(...) end
     end
-    local transport = create_transport(host, port, user,
-                                       password, weak_callback)
-    local transport_close = transport.close
+    local transport = create_transport(host, port, user, password,
+                                       weak_callback, connection, greeting)
+    local transport_stop = transport.stop
     local gc_hook = ffi.gc(ffi.new('char[1]'), function()
-        pcall(transport_close)
+        pcall(transport_stop)
     end)
-    transport.close = function()
+    transport.stop = function()
         -- dummy gc_hook, callback refs prevent premature GC
-        return transport_close(gc_hook, callback)
+        return transport_stop(gc_hook, callback)
     end
     return transport
 end
@@ -612,8 +646,7 @@ local console_mt = {
 
 local space_metatable, index_metatable
 
-local function connect(...)
-    local host, port, opts = parse_connect_params(...)
+local function new_sm(host, port, opts, connection, greeting)
     local user, password = opts.user, opts.password; opts.password = nil
     local last_reconnect_error
     local remote = {host = host, port = port, opts = opts, state = 'initial'}
@@ -652,7 +685,7 @@ local function connect(...)
         elseif what == 'will_fetch_schema' then
             return not opts.console
         elseif what == 'fetch_connect_timeout' then
-            return opts.connect_timeout or 10
+            return opts.connect_timeout or DEFAULT_CONNECT_TIMEOUT
         elseif what == 'did_fetch_schema' then
             remote:_install_schema(...)
         elseif what == 'reconnect_timeout' then
@@ -679,14 +712,56 @@ local function connect(...)
     remote._on_schema_reload = trigger.new("on_schema_reload")
     remote._on_disconnect = trigger.new("on_disconnect")
     remote._on_connect = trigger.new("on_connect")
-    remote._transport = create_transport(host, port, user, password, callback)
-    remote._transport.connect()
+    remote._transport = create_transport(host, port, user, password, callback,
+                                         connection, greeting)
+    remote._transport.start()
     if opts.wait_connected ~= false then
         remote._transport.wait_state('active', tonumber(opts.wait_connected))
     end
     return remote
 end
 
+--
+-- Wrap an existing connection into net.box API.
+-- @param connection Connected socket.
+-- @param greeting Decoded greeting, received from a server.
+-- @param host Hostname to which @a connection is established.
+-- @param port TCP port to which @a connection is established.
+-- @param opts Options like reconnect_after, connect_timeout,
+--        wait_connected, login, password, ...
+--
+-- @retval Net.box object.
+--
+local function wrap(connection, greeting, host, port, opts)
+    if connection == nil or type(greeting) ~= 'table' then
+        error('Usage: netbox.wrap(socket, greeting, [opts])')
+    end
+    opts = opts or {}
+    return new_sm(host, port, opts, connection, greeting)
+end
+
+--
+-- Connect to a remote server.
+-- @param uri OR host and port. URI is a string like
+--        hostname:port@login:password. Host and port can be
+--        passed separately with login and password in the next
+--        parameter.
+-- @param opts @Sa wrap().
+--
+-- @retval Net.box object.
+--
+local function connect(...)
+    local host, port, opts = parse_connect_params(...)
+    local connection, greeting =
+        establish_connection(host, port, opts.connect_timeout)
+    if not connection then
+        local dummy_conn = new_sm(host, port, opts)
+        dummy_conn.error = greeting
+        return dummy_conn
+    end
+    return new_sm(host, port, opts, connection, greeting)
+end
+
 local function check_remote_arg(remote, method)
     if type(remote) ~= 'table' then
         local fmt = 'Use remote:%s(...) instead of remote.%s(...):'
@@ -710,7 +785,7 @@ end
 
 function remote_methods:close()
     check_remote_arg(self, 'close')
-    self._transport.close()
+    self._transport.stop()
 end
 
 function remote_methods:on_schema_reload(...)
@@ -1112,7 +1187,9 @@ end
 local this_module = {
     create_transport = create_transport,
     connect = connect,
-    new = connect -- Tarantool < 1.7.1 compatibility
+    new = connect, -- Tarantool < 1.7.1 compatibility,
+    wrap = wrap,
+    establish_connection = establish_connection,
 }
 
 function this_module.timeout(timeout, ...)
diff --git a/test/box/net.box.result b/test/box/net.box.result
index 46d85b327..c541cad60 100644
--- a/test/box/net.box.result
+++ b/test/box/net.box.result
@@ -1754,7 +1754,7 @@ nb = net.new('localhost:3392', {
 });
 ---
 ...
-nb.error == "Timeout exceeded" or nb.error == "Connection timed out";
+assert(nb.error == 'Operation timed out');
 ---
 - true
 ...
@@ -2292,12 +2292,97 @@ weak.c
 ---
 - null
 ...
-box.schema.user.revoke('guest', 'execute', 'universe')
+--
+-- gh-2677: netbox supports console connections, that complicates
+-- both console and netbox. It was necessary because before a
+-- connection is established, a console does not known is it
+-- binary or text protocol, and netbox could not be created from
+-- existing socket.
+--
+box.schema.user.grant('guest','read,write,execute','universe')
+---
+...
+urilib = require('uri')
+---
+...
+uri = urilib.parse(tostring(box.cfg.listen))
+---
+...
+s, greeting = net.establish_connection(uri.host, uri.service)
+---
+...
+c = net.wrap(s, greeting, uri.host, uri.service, {reconnect_after = 0.01})
+---
+...
+c.state
+---
+- active
+...
+a = 100
+---
+...
+function kek(args) return {1, 2, 3, args} end
+---
+...
+c:eval('a = 200')
+---
+...
+a
+---
+- 200
+...
+c:call('kek', {300})
+---
+- [1, 2, 3, 300]
+...
+s = box.schema.create_space('test')
+---
+...
+pk = s:create_index('pk')
+---
+...
+c:reload_schema()
+---
+...
+c.space.test:replace{1}
+---
+- [1]
+...
+c.space.test:get{1}
+---
+- [1]
+...
+c.space.test:delete{1}
+---
+- [1]
+...
+--
+-- Break a connection to test reconnect_after.
+--
+_ = c._transport.perform_request(nil, nil, 'inject', nil, '\x80')
+---
+...
+c.state
+---
+- error_reconnect
+...
+while not c:is_connected() do fiber.sleep(0.01) end
+---
+...
+c:ping()
+---
+- true
+...
+s:drop()
 ---
 ...
 c:close()
 ---
 ...
-c = nil
+c.state
+---
+- closed
+...
+box.schema.user.revoke('guest', 'read,write,execute', 'universe')
 ---
 ...
diff --git a/test/box/net.box.test.lua b/test/box/net.box.test.lua
index 87e26f84c..d828ac529 100644
--- a/test/box/net.box.test.lua
+++ b/test/box/net.box.test.lua
@@ -701,7 +701,7 @@ nb = net.new('localhost:3392', {
     wait_connected = true, console = true,
     connect_timeout = 0.01
 });
-nb.error == "Timeout exceeded" or nb.error == "Connection timed out";
+assert(nb.error == 'Operation timed out');
 nb:close();
 -- we must get peer closed
 nb = net.new('localhost:3392', {
@@ -937,6 +937,41 @@ collectgarbage('collect')
 -- connection is deleted by 'collect'.
 weak.c
 
-box.schema.user.revoke('guest', 'execute', 'universe')
+--
+-- gh-2677: netbox supports console connections, that complicates
+-- both console and netbox. It was necessary because before a
+-- connection is established, a console does not known is it
+-- binary or text protocol, and netbox could not be created from
+-- existing socket.
+--
+box.schema.user.grant('guest','read,write,execute','universe')
+urilib = require('uri')
+uri = urilib.parse(tostring(box.cfg.listen))
+s, greeting = net.establish_connection(uri.host, uri.service)
+c = net.wrap(s, greeting, uri.host, uri.service, {reconnect_after = 0.01})
+c.state
+
+a = 100
+function kek(args) return {1, 2, 3, args} end
+c:eval('a = 200')
+a
+c:call('kek', {300})
+s = box.schema.create_space('test')
+pk = s:create_index('pk')
+c:reload_schema()
+c.space.test:replace{1}
+c.space.test:get{1}
+c.space.test:delete{1}
+--
+-- Break a connection to test reconnect_after.
+--
+_ = c._transport.perform_request(nil, nil, 'inject', nil, '\x80')
+c.state
+while not c:is_connected() do fiber.sleep(0.01) end
+c:ping()
+
+s:drop()
 c:close()
-c = nil
+c.state
+
+box.schema.user.revoke('guest', 'read,write,execute', 'universe')
-- 
2.14.3 (Apple Git-98)

[tarantool-patches] [PATCH v2 2/3] console: do not use netbox for console text connections

[Vladislav Shpilevoy]

Netbox console support complicates both netbox and console. Lets
use sockets directly for text protocol.

Part of #2677
---
 src/box/lua/console.lua | 162 +++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 131 insertions(+), 31 deletions(-)

diff --git a/src/box/lua/console.lua b/src/box/lua/console.lua
index d49cf42be..b853d523c 100644
--- a/src/box/lua/console.lua
+++ b/src/box/lua/console.lua
@@ -8,6 +8,9 @@ local log = require('log')
 local errno = require('errno')
 local urilib = require('uri')
 local yaml = require('yaml')
+local net_box = require('net.box')
+
+local YAML_TERM = '\n...\n'
 
 -- admin formatter must be able to encode any Lua variable
 local formatter = yaml.new()
@@ -76,26 +79,117 @@ local function eval(line)
     return local_eval(nil, line)
 end
 
+local text_connection_mt = {
+    __index = {
+        --
+        -- Close underlying socket.
+        --
+        close = function(self)
+            self._socket:close()
+        end,
+        --
+        -- Write a text into a socket.
+        -- @param test Text to send.
+        -- @retval not nil Bytes sent.
+        -- @retval     nil Error.
+        --
+        write = function(self, text)
+            -- It is the hack to protect from SIGPIPE, which is
+            -- not ignored under debugger (gdb, lldb) on send in
+            -- a socket, that is actually closed. If a socket is
+            -- readable and read() returns nothing then the socket
+            -- is closed, and writing into it will raise SIGPIPE.
+            if self._socket:readable(0) then
+                local rc = self._socket:read({chunk = 1})
+                if not rc or rc == '' then
+                    return nil
+                else
+                    assert(#rc == 1)
+                    -- Make the char be unread.
+                    self._socket.rbuf.wpos = self._socket.rbuf.wpos - 1
+                end
+            end
+            return self._socket:write(text)
+        end,
+        --
+        -- Read a text from a socket until YAML terminator.
+        -- @retval not nil Well formatted YAML.
+        -- @retval     nil Error.
+        --
+        read = function(self)
+            local ret = self._socket:read(YAML_TERM)
+            if ret and ret ~= '' then
+                return ret
+            end
+        end,
+        --
+        -- Write + Read.
+        --
+        eval = function(self, text)
+            text = text..'$EOF$\n'
+            if self:write(text) then
+                local rc = self:read()
+                if rc then
+                    return rc
+                end
+            end
+            error(self:set_error())
+        end,
+        --
+        -- Make the connection be in error state, set error
+        -- message.
+        -- @retval Error message.
+        --
+        set_error = function(self)
+            self.state = 'error'
+            self.error = self._socket:error()
+            if not self.error then
+                self.error = 'Peer closed'
+            end
+            return self.error
+        end,
+    }
+}
+
+--
+-- Wrap an existing socket inside a netbox-like object.
+-- @param connection Socket to wrap.
+-- @param url Parsed destination URL.
+-- @retval nil, err Error, and err contains an error message.
+-- @retval not nil Netbox-like object.
+--
+local function wrap_text_socket(connection, url)
+    local conn = setmetatable({
+        _socket = connection,
+        state = 'active',
+        host = url.host or 'localhost',
+        port = url.service,
+    }, text_connection_mt)
+    if not conn:write('require("console").delimiter("$EOF$")\n') or
+       not conn:read() then
+        conn:set_error()
+    end
+    return conn
+end
+
 --
 -- Evaluate command on remote instance
 --
 local function remote_eval(self, line)
-    if not line or self.remote.state ~= 'active' then
-        local err = self.remote.error
-        self.remote:close()
-        self.remote = nil
-        -- restore local REPL mode
-        self.eval = nil
-        self.prompt = nil
-        self.completion = nil
-        pcall(self.on_client_disconnect, self)
-        return (err and format(false, err)) or ''
+    if line and self.remote.state == 'active' then
+        local ok, res = pcall(self.remote.eval, self.remote, line)
+        if self.remote.state == 'active' then
+            return ok and res or format(false, res)
+        end
     end
-    --
-    -- execute line
-    --
-    local ok, res = pcall(self.remote.eval, self.remote, line)
-    return ok and res or format(false, res)
+    local err = self.remote.error
+    self.remote:close()
+    self.remote = nil
+    self.eval = nil
+    self.prompt = nil
+    self.completion = nil
+    pcall(self.on_client_disconnect, self)
+    return (err and format(false, err)) or ''
 end
 
 --
@@ -285,12 +379,7 @@ end
 --
 -- Connect to remove instance
 --
-local netbox_connect
 local function connect(uri, opts)
-    if not netbox_connect then -- workaround the broken loader
-        netbox_connect = require('net.box').connect
-    end
-
     opts = opts or {}
 
     local self = fiber.self().storage.console
@@ -306,18 +395,29 @@ local function connect(uri, opts)
         error('Usage: console.connect("[login:password@][host:]port")')
     end
 
-    -- connect to remote host
+    local connection, greeting =
+        net_box.establish_connection(u.host, u.service, opts.timeout)
+    if not connection then
+        log.verbose(greeting)
+        box.error(box.error.NO_CONNECTION)
+    end
     local remote
-    remote = netbox_connect(u.host, u.service, {
-        user = u.login, password = u.password,
-        console = true, connect_timeout = opts.timeout
-    })
-    remote.host, remote.port = u.host or 'localhost', u.service
-
-    -- run disconnect trigger if connection failed
-    if not remote:is_connected() then
-        pcall(self.on_client_disconnect, self)
-        error('Connection is not established: '..remote.error)
+    if greeting.protocol == 'Lua console' then
+        remote = wrap_text_socket(connection, u)
+    else
+        opts = {
+            connect_timeout = opts.timeout,
+            user = u.login,
+            password = u.password,
+        }
+        remote = net_box.wrap(connection, greeting, u.host, u.service, opts)
+        if not remote.host then
+            remote.host = 'localhost'
+        end
+        local old_eval = remote.eval
+        remote.eval = function(con, line)
+            return old_eval(con, 'return require("console").eval(...)', {line})
+        end
     end
 
     -- check connection && permissions
-- 
2.14.3 (Apple Git-98)

[tarantool-patches] [PATCH v2 3/3] netbox: deprecate console support

[Vladislav Shpilevoy]

Print warning about that. After a while the cosole support will
be deleted from netbox.
---
 src/box/lua/net_box.lua | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
index 5691dcb9e..0309eef17 100644
--- a/src/box/lua/net_box.lua
+++ b/src/box/lua/net_box.lua
@@ -650,6 +650,11 @@ local function new_sm(host, port, opts, connection, greeting)
     local user, password = opts.user, opts.password; opts.password = nil
     local last_reconnect_error
     local remote = {host = host, port = port, opts = opts, state = 'initial'}
+    -- @deprecated since 1.10
+    if greeting and greeting.protocol == 'Lua console' or opts.console then
+        log.warn("Netbox text protocol support is deprecated since 1.10, "..
+                 "please use require('console').connect() instead")
+    end
     local function callback(what, ...)
         if what == 'state_changed' then
             local state, errno, err = ...
-- 
2.14.3 (Apple Git-98)

[tarantool-patches] Re: [PATCH v2 1/3] netbox: allow to create a netbox connection from existing socket

[Konstantin Osipov]

* Vladislav Shpilevoy <v.shpilevoy@tarantool.org> [18/03/28 16:50]:
> ---

This patch looks good to me, I will push it now.

>  src/box/lua/net_box.lua   | 199 ++++++++++++++++++++++++++++++++--------------
>  test/box/net.box.result   |  91 ++++++++++++++++++++-
>  test/box/net.box.test.lua |  41 +++++++++-
>  3 files changed, 264 insertions(+), 67 deletions(-)
> 
> diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
> index 87c8c548b..5691dcb9e 100644
> --- a/src/box/lua/net_box.lua
> +++ b/src/box/lua/net_box.lua
> @@ -31,6 +31,7 @@ local sequence_mt      = { __serialize = 'sequence' }
>  local TIMEOUT_INFINITY = 500 * 365 * 86400
>  local VSPACE_ID        = 281
>  local VINDEX_ID        = 289
> +local DEFAULT_CONNECT_TIMEOUT = 10
>  
>  local IPROTO_STATUS_KEY    = 0x00
>  local IPROTO_ERRNO_MASK    = 0x7FFF
> @@ -70,9 +71,37 @@ local method_codec           = {
>  
>  local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
>  
> --- function create_transport(host, port, user, password, callback)
>  --
> --- Transport methods: connect(), close(), perfrom_request(), wait_state()
> +-- Connect to a remote server, do handshake.
> +-- @param host Hostname.
> +-- @param port TCP port.
> +-- @param timeout Timeout to connect and receive greeting.
> +--
> +-- @retval nil, err Error occured. The reason is returned.
> +-- @retval two non-nils A connected socket and a decoded greeting.
> +--
> +local function establish_connection(host, port, timeout)
> +    local timeout = timeout or DEFAULT_CONNECT_TIMEOUT
> +    local begin = fiber.clock()
> +    local s = socket.tcp_connect(host, port, timeout)
> +    if not s then
> +        return nil, errno.strerror(errno())
> +    end
> +    local msg = s:read({chunk = IPROTO_GREETING_SIZE},
> +                        timeout - (fiber.clock() - begin))
> +    if not msg then
> +        local err = s:error()
> +        s:close()
> +        return nil, err
> +    end
> +    local greeting, err = decode_greeting(msg)
> +    if not greeting then
> +        s:close()
> +        return nil, err
> +    end
> +    return s, greeting
> +end
> +
>  --
>  -- Basically, *transport* is a TCP connection speaking one of
>  -- Tarantool network protocols. This is a low-level interface.
> @@ -85,16 +114,17 @@ local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
>  --
>  -- Transport state machine:
>  --
> --- State machine starts in 'initial' state. Connect method changes
> --- the state to 'connecting' and spawns a worker fiber. Close
> --- method sets the state to 'closed' and kills the worker.
> --- If the transport is already in 'error' state close() does nothing.
> +-- State machine starts in 'initial' state. New_sm method
> +-- accepts an established connection and spawns a worker fiber.
> +-- Stop method sets the state to 'closed' and kills the worker.
> +-- If the transport is already in 'error' state stop() does
> +-- nothing.
>  --
>  -- State chart:
>  --
> ---  initial -> connecting -> active
> +-- connecting -> initial +-> active
>  --                        \
> ---                          -> auth -> fetch_schema <-> active
> +--                         +-> auth -> fetch_schema <-> active
>  --
>  --  (any state, on error) -> error_reconnect -> connecting -> ...
>  --                                           \
> @@ -121,15 +151,15 @@ local function next_id(id) return band(id + 1, 0x7FFFFFFF) end
>  -- Suggestion for callback writers: sleep a few secs before approving
>  -- reconnect.
>  --
> -local function create_transport(host, port, user, password, callback)
> +local function create_transport(host, port, user, password, callback,
> +                                connection, greeting)
>      -- check / normalize credentials
>      if user == nil and password ~= nil then
>          box.error(E_PROC_LUA, 'net.box: user is not defined')
>      end
>      if user ~= nil and password == nil then password = '' end
>  
> -    -- state: current state, only the worker fiber and connect method
> -    -- change state
> +    -- Current state machine's state.
>      local state            = 'initial'
>      local last_errno
>      local last_error
> @@ -145,7 +175,6 @@ local function create_transport(host, port, user, password, callback)
>      local next_request_id  = 1
>  
>      local worker_fiber
> -    local connection
>      local send_buf         = buffer.ibuf(buffer.READAHEAD)
>      local recv_buf         = buffer.ibuf(buffer.READAHEAD)
>  
> @@ -185,16 +214,19 @@ local function create_transport(host, port, user, password, callback)
>          return state == target_state or target_state[state] or false
>      end
>  
> -    -- CONNECT/CLOSE --
> +    -- START/STOP --
>      local protocol_sm
>  
> -    local function connect()
> +    local function start()
>          if state ~= 'initial' then return not is_final_state[state] end
> -        set_state('connecting')
> +        if not connection then
> +            set_state('error', E_NO_CONNECTION)
> +            return
> +        end
>          fiber.create(function()
>              worker_fiber = fiber_self()
>              fiber.name(string.format('%s:%s (net.box)', host, port), {truncate=true})
> -    ::reconnect::
> +    ::handle_connection::
>              local ok, err = pcall(protocol_sm)
>              if not (ok or is_final_state[state]) then
>                  set_state('error', E_UNKNOWN, err)
> @@ -204,21 +236,34 @@ local function create_transport(host, port, user, password, callback)
>                  connection = nil
>              end
>              local timeout = callback('reconnect_timeout')
> -            if timeout and state == 'error_reconnect' then
> +            -- Non-nil reconnect_timeout is an indicator of set
> +            -- reconnect_after option.
> +            while timeout and state == 'error_reconnect' do
>                  fiber.sleep(timeout)
> +                -- During sleep, a connection could be garbage
> +                -- collected, or explicitly closed by a user, or
> +                -- the reconnect_after could be reset. Check the
> +                -- timeout again.
>                  timeout = callback('reconnect_timeout')
> -                if timeout and state == 'error_reconnect' then
> -                    goto reconnect
> +                if not timeout or state ~= 'error_reconnect' then
> +                    break
> +                end
> +                connection, greeting =
> +                    establish_connection(host, port,
> +                                         callback('fetch_connect_timeout'))
> +                if connection then
> +                    goto handle_connection
>                  end
> +                set_state('error_reconnect', E_NO_CONNECTION, greeting)
> +                timeout = callback('reconnect_timeout')
>              end
>              send_buf:recycle()
>              recv_buf:recycle()
>              worker_fiber = nil
>          end)
> -        return true
>      end
>  
> -    local function close()
> +    local function stop()
>          if not is_final_state[state] then
>              set_state('closed', E_NO_CONNECTION, 'Connection closed')
>          end
> @@ -367,27 +412,14 @@ local function create_transport(host, port, user, password, callback)
>      -- collected. See gh-3164, where because of reconnect sleeps
>      -- in this function, a connection could not be deleted.
>      --
> -    protocol_sm = function ()
> -        local tm_begin, tm = fiber.clock(), callback('fetch_connect_timeout')
> -        connection = socket.tcp_connect(host, port, tm)
> -        if connection == nil then
> -            return error_sm(E_NO_CONNECTION, errno.strerror(errno()))
> -        end
> -        local size = IPROTO_GREETING_SIZE
> -        local err, msg = send_and_recv(size, tm - (fiber.clock() - tm_begin))
> -        if err then
> -            return error_sm(err, msg)
> -        end
> -        local g = decode_greeting(ffi.string(recv_buf.rpos, size))
> -        recv_buf.rpos = recv_buf.rpos + size
> -        if not g then
> -            return error_sm(E_NO_CONNECTION, 'Can\'t decode handshake')
> -        end
> -        err, msg = callback('handshake', g)
> +    protocol_sm = function()
> +        assert(connection)
> +        assert(greeting)
> +        local err, msg = callback('handshake', greeting)
>          if err then
>              return error_sm(err, msg)
>          end
> -        if g.protocol == 'Lua console' then
> +        if greeting.protocol == 'Lua console' then
>              local setup_delimiter = 'require("console").delimiter("$EOF$")\n'
>              method_codec.inject(send_buf, nil, nil, setup_delimiter)
>              local err, response = send_and_recv_console()
> @@ -399,10 +431,11 @@ local function create_transport(host, port, user, password, callback)
>              local rid = next_request_id
>              set_state('active')
>              return console_sm(rid)
> -        elseif g.protocol == 'Binary' then
> -            return iproto_auth_sm(g.salt)
> +        elseif greeting.protocol == 'Binary' then
> +            return iproto_auth_sm(greeting.salt)
>          else
> -            return error_sm(E_NO_CONNECTION, 'Unknown protocol: ' .. g.protocol)
> +            return error_sm(E_NO_CONNECTION,
> +                            'Unknown protocol: '..greeting.protocol)
>          end
>      end
>  
> @@ -524,14 +557,14 @@ local function create_transport(host, port, user, password, callback)
>      end
>  
>      return {
> -        close           = close,
> -        connect         = connect,
> +        stop            = stop,
> +        start           = start,
>          wait_state      = wait_state,
>          perform_request = perform_request
>      }
>  end
>  
> --- Wrap create_transport, adding auto-close-on-GC feature.
> +-- Wrap create_transport, adding auto-stop-on-GC feature.
>  -- All the GC magic is neatly encapsulated!
>  -- The tricky part is the callback:
>  --  * callback (typically) references the transport (indirectly);
> @@ -539,23 +572,24 @@ end
>  --  * fibers are GC roots - i.e. transport is never GC-ed!
>  -- We solve the issue by making the worker->callback ref weak.
>  -- Now it is necessary to have a strong ref to callback somewhere or
> --- it is GC-ed prematurely. We wrap close() method, stashing the
> --- ref in an upvalue (close() performance doesn't matter much.)
> -local create_transport = function(host, port, user, password, callback)
> +-- it is GC-ed prematurely. We wrap stop() method, stashing the
> +-- ref in an upvalue (stop() performance doesn't matter much.)
> +local create_transport = function(host, port, user, password, callback,
> +                                  connection, greeting)
>      local weak_refs = setmetatable({callback = callback}, {__mode = 'v'})
>      local function weak_callback(...)
>          local callback = weak_refs.callback
>          if callback then return callback(...) end
>      end
> -    local transport = create_transport(host, port, user,
> -                                       password, weak_callback)
> -    local transport_close = transport.close
> +    local transport = create_transport(host, port, user, password,
> +                                       weak_callback, connection, greeting)
> +    local transport_stop = transport.stop
>      local gc_hook = ffi.gc(ffi.new('char[1]'), function()
> -        pcall(transport_close)
> +        pcall(transport_stop)
>      end)
> -    transport.close = function()
> +    transport.stop = function()
>          -- dummy gc_hook, callback refs prevent premature GC
> -        return transport_close(gc_hook, callback)
> +        return transport_stop(gc_hook, callback)
>      end
>      return transport
>  end
> @@ -612,8 +646,7 @@ local console_mt = {
>  
>  local space_metatable, index_metatable
>  
> -local function connect(...)
> -    local host, port, opts = parse_connect_params(...)
> +local function new_sm(host, port, opts, connection, greeting)
>      local user, password = opts.user, opts.password; opts.password = nil
>      local last_reconnect_error
>      local remote = {host = host, port = port, opts = opts, state = 'initial'}
> @@ -652,7 +685,7 @@ local function connect(...)
>          elseif what == 'will_fetch_schema' then
>              return not opts.console
>          elseif what == 'fetch_connect_timeout' then
> -            return opts.connect_timeout or 10
> +            return opts.connect_timeout or DEFAULT_CONNECT_TIMEOUT
>          elseif what == 'did_fetch_schema' then
>              remote:_install_schema(...)
>          elseif what == 'reconnect_timeout' then
> @@ -679,14 +712,56 @@ local function connect(...)
>      remote._on_schema_reload = trigger.new("on_schema_reload")
>      remote._on_disconnect = trigger.new("on_disconnect")
>      remote._on_connect = trigger.new("on_connect")
> -    remote._transport = create_transport(host, port, user, password, callback)
> -    remote._transport.connect()
> +    remote._transport = create_transport(host, port, user, password, callback,
> +                                         connection, greeting)
> +    remote._transport.start()
>      if opts.wait_connected ~= false then
>          remote._transport.wait_state('active', tonumber(opts.wait_connected))
>      end
>      return remote
>  end
>  
> +--
> +-- Wrap an existing connection into net.box API.
> +-- @param connection Connected socket.
> +-- @param greeting Decoded greeting, received from a server.
> +-- @param host Hostname to which @a connection is established.
> +-- @param port TCP port to which @a connection is established.
> +-- @param opts Options like reconnect_after, connect_timeout,
> +--        wait_connected, login, password, ...
> +--
> +-- @retval Net.box object.
> +--
> +local function wrap(connection, greeting, host, port, opts)
> +    if connection == nil or type(greeting) ~= 'table' then
> +        error('Usage: netbox.wrap(socket, greeting, [opts])')
> +    end
> +    opts = opts or {}
> +    return new_sm(host, port, opts, connection, greeting)
> +end
> +
> +--
> +-- Connect to a remote server.
> +-- @param uri OR host and port. URI is a string like
> +--        hostname:port@login:password. Host and port can be
> +--        passed separately with login and password in the next
> +--        parameter.
> +-- @param opts @Sa wrap().
> +--
> +-- @retval Net.box object.
> +--
> +local function connect(...)
> +    local host, port, opts = parse_connect_params(...)
> +    local connection, greeting =
> +        establish_connection(host, port, opts.connect_timeout)
> +    if not connection then
> +        local dummy_conn = new_sm(host, port, opts)
> +        dummy_conn.error = greeting
> +        return dummy_conn
> +    end
> +    return new_sm(host, port, opts, connection, greeting)
> +end
> +
>  local function check_remote_arg(remote, method)
>      if type(remote) ~= 'table' then
>          local fmt = 'Use remote:%s(...) instead of remote.%s(...):'
> @@ -710,7 +785,7 @@ end
>  
>  function remote_methods:close()
>      check_remote_arg(self, 'close')
> -    self._transport.close()
> +    self._transport.stop()
>  end
>  
>  function remote_methods:on_schema_reload(...)
> @@ -1112,7 +1187,9 @@ end
>  local this_module = {
>      create_transport = create_transport,
>      connect = connect,
> -    new = connect -- Tarantool < 1.7.1 compatibility
> +    new = connect, -- Tarantool < 1.7.1 compatibility,
> +    wrap = wrap,
> +    establish_connection = establish_connection,
>  }
>  
>  function this_module.timeout(timeout, ...)
> diff --git a/test/box/net.box.result b/test/box/net.box.result
> index 46d85b327..c541cad60 100644
> --- a/test/box/net.box.result
> +++ b/test/box/net.box.result
> @@ -1754,7 +1754,7 @@ nb = net.new('localhost:3392', {
>  });
>  ---
>  ...
> -nb.error == "Timeout exceeded" or nb.error == "Connection timed out";
> +assert(nb.error == 'Operation timed out');
>  ---
>  - true
>  ...
> @@ -2292,12 +2292,97 @@ weak.c
>  ---
>  - null
>  ...
> -box.schema.user.revoke('guest', 'execute', 'universe')
> +--
> +-- gh-2677: netbox supports console connections, that complicates
> +-- both console and netbox. It was necessary because before a
> +-- connection is established, a console does not known is it
> +-- binary or text protocol, and netbox could not be created from
> +-- existing socket.
> +--
> +box.schema.user.grant('guest','read,write,execute','universe')
> +---
> +...
> +urilib = require('uri')
> +---
> +...
> +uri = urilib.parse(tostring(box.cfg.listen))
> +---
> +...
> +s, greeting = net.establish_connection(uri.host, uri.service)
> +---
> +...
> +c = net.wrap(s, greeting, uri.host, uri.service, {reconnect_after = 0.01})
> +---
> +...
> +c.state
> +---
> +- active
> +...
> +a = 100
> +---
> +...
> +function kek(args) return {1, 2, 3, args} end
> +---
> +...
> +c:eval('a = 200')
> +---
> +...
> +a
> +---
> +- 200
> +...
> +c:call('kek', {300})
> +---
> +- [1, 2, 3, 300]
> +...
> +s = box.schema.create_space('test')
> +---
> +...
> +pk = s:create_index('pk')
> +---
> +...
> +c:reload_schema()
> +---
> +...
> +c.space.test:replace{1}
> +---
> +- [1]
> +...
> +c.space.test:get{1}
> +---
> +- [1]
> +...
> +c.space.test:delete{1}
> +---
> +- [1]
> +...
> +--
> +-- Break a connection to test reconnect_after.
> +--
> +_ = c._transport.perform_request(nil, nil, 'inject', nil, '\x80')
> +---
> +...
> +c.state
> +---
> +- error_reconnect
> +...
> +while not c:is_connected() do fiber.sleep(0.01) end
> +---
> +...
> +c:ping()
> +---
> +- true
> +...
> +s:drop()
>  ---
>  ...
>  c:close()
>  ---
>  ...
> -c = nil
> +c.state
> +---
> +- closed
> +...
> +box.schema.user.revoke('guest', 'read,write,execute', 'universe')
>  ---
>  ...
> diff --git a/test/box/net.box.test.lua b/test/box/net.box.test.lua
> index 87e26f84c..d828ac529 100644
> --- a/test/box/net.box.test.lua
> +++ b/test/box/net.box.test.lua
> @@ -701,7 +701,7 @@ nb = net.new('localhost:3392', {
>      wait_connected = true, console = true,
>      connect_timeout = 0.01
>  });
> -nb.error == "Timeout exceeded" or nb.error == "Connection timed out";
> +assert(nb.error == 'Operation timed out');
>  nb:close();
>  -- we must get peer closed
>  nb = net.new('localhost:3392', {
> @@ -937,6 +937,41 @@ collectgarbage('collect')
>  -- connection is deleted by 'collect'.
>  weak.c
>  
> -box.schema.user.revoke('guest', 'execute', 'universe')
> +--
> +-- gh-2677: netbox supports console connections, that complicates
> +-- both console and netbox. It was necessary because before a
> +-- connection is established, a console does not known is it
> +-- binary or text protocol, and netbox could not be created from
> +-- existing socket.
> +--
> +box.schema.user.grant('guest','read,write,execute','universe')
> +urilib = require('uri')
> +uri = urilib.parse(tostring(box.cfg.listen))
> +s, greeting = net.establish_connection(uri.host, uri.service)
> +c = net.wrap(s, greeting, uri.host, uri.service, {reconnect_after = 0.01})
> +c.state
> +
> +a = 100
> +function kek(args) return {1, 2, 3, args} end
> +c:eval('a = 200')
> +a
> +c:call('kek', {300})
> +s = box.schema.create_space('test')
> +pk = s:create_index('pk')
> +c:reload_schema()
> +c.space.test:replace{1}
> +c.space.test:get{1}
> +c.space.test:delete{1}
> +--
> +-- Break a connection to test reconnect_after.
> +--
> +_ = c._transport.perform_request(nil, nil, 'inject', nil, '\x80')
> +c.state
> +while not c:is_connected() do fiber.sleep(0.01) end
> +c:ping()
> +
> +s:drop()
>  c:close()
> -c = nil
> +c.state
> +
> +box.schema.user.revoke('guest', 'read,write,execute', 'universe')
> -- 
> 2.14.3 (Apple Git-98)
> 
> 

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

[tarantool-patches] Re: [PATCH v2 2/3] console: do not use netbox for console text connections

[Konstantin Osipov]

* Vladislav Shpilevoy <v.shpilevoy@tarantool.org> [18/03/28 16:50]:

You mentioned you were going to get rid of the read() to detect
sigpipe, and investigate sigpipe problem more thoroughly.

Please do.

> +        write = function(self, text)
> +            -- It is the hack to protect from SIGPIPE, which is
> +            -- not ignored under debugger (gdb, lldb) on send in
> +            -- a socket, that is actually closed. If a socket is
> +            -- readable and read() returns nothing then the socket
> +            -- is closed, and writing into it will raise SIGPIPE.
> +            if self._socket:readable(0) then
> +                local rc = self._socket:read({chunk = 1})
> +                if not rc or rc == '' then
> +                    return nil
> +                else
> +                    assert(#rc == 1)
> +                    -- Make the char be unread.
> +                    self._socket.rbuf.wpos = self._socket.rbuf.wpos - 1
> +                end
> +            end
> +            return self._socket:write(text)

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov