connpool: speed up by using recent conn checks

This patch speeds up `connpool.call()` and `connpool.filter()` methods
by making them not wait for instances that have been unavailable
recently. Thanks to the reconnecting logic introduced previously we can
be sure that if connpool was not able to reconnect to an instance
recently we might continue and assume it is dead instead of waiting. A
detailed description of this idea is presented within the mentioned
patch.

Closes tarantool#10330

@TarantoolBot document
Title: connpool: add information on advanced connection management
Product: Tarantool Platform

`experimental.connpool` module also offers advanced automatic connection
management. It automatically closes unused connections and tries to
reconnect to recently used in case of an error. See also
a `connpool.idle_timeout` configuration option.

Author: georgiy-belyanin

changelogs/unreleased/gh-10330-connpool-works-faster.md

@@ -0,0 +1,5 @@
+## feature/connpool
+
+* `experimental.connpool` methods `call()` and `filter()` became faster. They do
+  not wait for unavailable instances if it is known they have been inaccessible
+  recently.

src/box/lua/connpool.lua

@@ -239,7 +239,8 @@ function pool_methods.connect(self, instance_name, opts)
 end
 
 --- This method connects to the specified instances and returns
---- the set of successfully connected ones.
+--- the set of successfully connected ones. Instances that are
+--- known to have been recently unavailable are skipped.
 ---
 --- If a callback accepting an instance name and returning a
 --- boolean value is provided in `opts.any` the method connects
@@ -279,23 +280,54 @@ function pool_methods.connect_to_multiple(self, instances, opts)
     local timeout = opts.timeout or WATCHER_TIMEOUT
     local connect_deadline = clock.monotonic() + timeout
 
+    -- We divide instances into three categories.
+    -- * Available ones: these instances are already connected.
+    --   Nothing to do with them apart from returning them.
+    -- * Failed: the pool assumes an instance failed if it has
+    --   been recently accessed but the connection has failed.
+    --   Such instances are not returned and not waited. The pool
+    --   tries to automatically reconnect to them in the
+    --   background guaranteeing the failed status is relatively
+    --   actual.
+    -- * Unknown: these are either or the ones that have been
+    --   flushed because they have not been needed for a while
+    --   (see idle timeout and related logic) or the ones not
+    --   accessed since startup. The pool connects to them and
+    --   waits for success/fail.
+    local candidate_instances = {}
     for _, instance_name in pairs(instances) do
-        self:connect(instance_name, {wait_connected = false})
+        local conn = self._connections[instance_name]
+
+        -- If instance has not been accessed recently (it has
+        -- unknown state) start connecting to it. The call assume
+        -- it is a candidate and will wait for the connection to
+        -- fail/succeed.
+        if conn == nil then
+            self:connect(instance_name, {wait_connected = false})
+            table.insert(candidate_instances, instance_name)
+        -- If the connection is already ok it is likely it should
+        -- be returned as is.
+        elseif is_connection_valid(conn,
+            {fetch_schema = (conn.opts or {}).fetch_schema}) then
+            table.insert(candidate_instances, instance_name)
+        -- The remaining connections are the failed ones that has
+        -- been checked rather recently. Skip them.
+        end
     end
 
     local connected_instances = {}
     while clock.monotonic() < connect_deadline do
-        connected_instances = fun.iter(instances)
+        connected_instances = fun.iter(candidate_instances)
             :filter(is_instance_connected)
             :totable()
 
         if opts.any then
-            if fun.iter(instances):any(opts.any) then
+            if fun.iter(candidate_instances):any(opts.any) then
                 break
             end
         end
 
-        if fun.iter(instances):all(is_instance_checked) then
+        if fun.iter(candidate_instances):all(is_instance_checked) then
             break
         end
 

test/config-luatest/rpc_test.lua

@@ -1305,3 +1305,80 @@ g.test_tries_to_reconnect = function()
         t.assert_equals(_G.i2_conn.state, 'error')
     end)
 end
+
+g.test_does_not_wait_if_recently_checked = function()
+    local config = cbuilder:new()
+        :set_global_option('credentials.users.myuser',
+            {password =  'secret',
+             roles = { 'replication' },
+             privileges = {{permissions = {'execute'}, universe = true}}})
+        :set_global_option('iproto.advertise.peer.login', 'myuser')
+        :add_instance('i-001', { database = { mode = 'rw' } })
+        :add_instance('i-002', {})
+        :config()
+
+    local cluster = cluster:new(config)
+
+    -- The reconnect after interval for connpool is hardcoded in
+    -- connpool and equals to 3 seconds.
+    local reconnect_after = 3
+
+    -- Add a counter to count netbox.connect() calls.
+    treegen.write_file(cluster._dir, 'override/net/box.lua',
+        string.dump(function()
+            local loaders = require('internal.loaders')
+
+            rawset(_G, 'connects', 0)
+
+            local builtin_netbox = loaders.builtin['net.box']
+            local builtin_connect = builtin_netbox.connect
+            builtin_netbox.connect = function(...)
+                _G.connects = _G.connects + 1
+                return builtin_connect(...)
+            end
+
+            return builtin_netbox
+        end))
+
+    cluster:start()
+
+    cluster['i-001']:exec(function()
+        local connpool = require('experimental.connpool')
+
+        connpool.call('box.info', nil, {mode = 'ro'})
+        t.assert_equals(_G.connects, 2)
+
+        -- Save a connection to i-002.
+        rawset(_G, 'i2_conn', connpool.connect('i-002'))
+    end)
+
+    cluster['i-002']:stop()
+
+    -- The connpool might try to reconnect. Let's wait
+    -- for one attempt.
+    fiber.sleep(reconnect_after + 1)
+
+    cluster['i-001']:exec(function()
+        local connpool = require('experimental.connpool')
+        local clock = require('clock')
+
+        -- The connection timeout for filter() is hardcoded in
+        -- connpool and equals to 10 seconds.
+        local CONNECT_TIMEOUT = 10
+
+        -- Check no new connection attempts has been done
+        -- and the call has quickly failed due to no candidates.
+        local timestamp_before_call = clock.monotonic()
+        local cur_connects = _G.connects
+
+        local exp_err = 'no candidates are available with these conditions'
+        local opts = {mode = 'ro'}
+        t.assert_error_msg_contains(exp_err, connpool.call, 'box.info', nil,
+                                    opts)
+
+        local elapsed_time = clock.monotonic() - timestamp_before_call
+
+        t.assert_equals(_G.connects, cur_connects)
+        t.assert_lt(elapsed_time, CONNECT_TIMEOUT)
+    end)
+end