From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Subject: Re: [tarantool-patches] Re: [PATCH 6/8] netbox: introduce fiber-async API References: <49a50d32a154959aa786ec2a85a4f74792d7ae09.1523903144.git.v.shpilevoy@tarantool.org> <20180423164450.yd7o7deawhjxbt7f@esperanza> From: Vladislav Shpilevoy Message-ID: Date: Mon, 23 Apr 2018 21:59:29 +0300 MIME-Version: 1.0 In-Reply-To: <20180423164450.yd7o7deawhjxbt7f@esperanza> Content-Type: text/plain; charset=utf-8; format=flowed Content-Language: en-US Content-Transfer-Encoding: 7bit To: Vladimir Davydov Cc: tarantool-patches@freelists.org List-ID: Hello. Thanks for review! On 23/04/2018 19:44, Vladimir Davydov wrote: > On Mon, Apr 16, 2018 at 09:39:16PM +0300, Vladislav Shpilevoy wrote: >> Now any netbox call blocks a caller-fiber until a result is read >> from a socket, or time is out. To use it asynchronously it is >> necessary to create a fiber per request. Sometimes it is >> unwanted - for example if RPS is very high (for example, about >> 100k), and latency is about 1 second. Or when it is neccessary >> to send multiple requests in paralles and then collect responses >> (map-reduce). >> >> The patch introduces a new option for all netbox requests: >> is_async. With this option any called netbox method returns >> immediately (but still yields for a moment) a 'future' object. >> >> By a future object a user can check if the request is finalized, >> get a result or error, wait for a timeout, discard a response. >> >> Example of is_async usage: >> future = conn:call(func, {params}, {..., is_async = true}) >> -- Do some work ... >> if not future.is_ready() then >> result, err = future:wait_result(timeout) >> end >> -- Or: >> result, error = future:result() >> >> A future:result() and :wait_result() returns either an error or >> a response in the same format, as the sync versions of the called >> methods. >> >> Part of #3107 >> + -- It is possible that multiple fibers are waiting for >> + -- a result. In such a case a first, who got it, must >> + -- wakeup the previous waiting client. This one wakes >> + -- up another. Another wakes up third one, etc. >> + wakeup_client(old_client) > > This is rather difficult for understanding IMO. Can we use a fiber.cond > instead? Sure, we can. Done on the branch. diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua index 1f4828a7e..9bbc047d5 100644 --- a/src/box/lua/net_box.lua +++ b/src/box/lua/net_box.lua @@ -233,12 +233,6 @@ local function create_transport(host, port, user, password, callback, local send_buf = buffer.ibuf(buffer.READAHEAD) local recv_buf = buffer.ibuf(buffer.READAHEAD) - local function wakeup_client(client) - if client and client:status() ~= 'dead' then - client:wakeup() - end - end - -- -- Async request metamethods. -- @@ -287,22 +281,14 @@ local function create_transport(host, port, user, password, callback, if not self:is_ready() then -- When a response is ready before timeout, the -- waiting client is waked up prematurely. - local old_client = self.client - self.client = fiber.self() while timeout > 0 and not self:is_ready() do local ts = fiber.clock() - state_cond:wait(timeout) + self.cond:wait(timeout) timeout = timeout - (fiber.clock() - ts) end - self.client = old_client if not self:is_ready() then return nil, box.error.new(E_TIMEOUT) end - -- It is possible that multiple fibers are waiting for - -- a result. In such a case a first, who got it, must - -- wakeup the previous waiting client. This one wakes - -- up another. Another wakes up third one, etc. - wakeup_client(old_client) end return self:result() end @@ -333,6 +319,7 @@ local function create_transport(host, port, user, password, callback, request.id = nil request.errno = new_errno request.response = new_error + request.cond:broadcast() end requests = {} end @@ -428,11 +415,15 @@ local function create_transport(host, port, user, password, callback, local id = next_request_id method_encoder[method](send_buf, id, schema_version, ...) next_request_id = next_id(id) + -- Request has maximum 7 members: + -- method, schema_version, buffer, id, cond, errno, + -- response. local request = setmetatable(table_new(0, 7), request_mt) request.method = method request.schema_version = schema_version request.buffer = buffer request.id = id + request.cond = fiber.cond() requests[id] = request return request end @@ -468,7 +459,7 @@ local function create_transport(host, port, user, password, callback, assert(body_end == body_end_check, "invalid xrow length") request.errno = band(status, IPROTO_ERRNO_MASK) request.response = body[IPROTO_ERROR_KEY] - wakeup_client(request.client) + request.cond:broadcast() return end @@ -479,7 +470,7 @@ local function create_transport(host, port, user, password, callback, local wpos = buffer:alloc(body_len) ffi.copy(wpos, body_rpos, body_len) request.response = tonumber(body_len) - wakeup_client(request.client) + request.cond:broadcast() return end @@ -490,7 +481,7 @@ local function create_transport(host, port, user, password, callback, request.response, request.errno = method_decoder[request.method](body[IPROTO_DATA_KEY]) end - wakeup_client(request.client) + request.cond:broadcast() end local function new_request_id() @@ -601,7 +592,7 @@ local function create_transport(host, port, user, password, callback, request.id = nil requests[rid] = nil request.response = response - wakeup_client(request.client) + request.cond:broadcast() return console_sm(next_id(rid)) end end