[tarantool-patches] Re: [PATCH] iproto: update readahead in existing connections

[Georgy Kirichenko <georgy@tarantool.org>]

[-- Attachment #1: Type: text/plain, Size: 5222 bytes --]

Looks good for me, thank you

On Monday, February 18, 2019 8:25:50 PM MSK Serge Petrenko wrote:
> Iproto connections keep old readahead values for input buffers even
> after box.cfg.readahead reconfiguration. This means that for the changes
> to take place for the old clients, they have to reconnect. Otherwise
> tarantool log will be spammed with 'readahead limit is reached' errors.
> 
> To fix this, start updating input buffer size for iproto connections
> if needed every time the buffer is empty.
> 
> Closes #3958
> ---
> https://github.com/tarantool/tarantool/issues/3958
> https://github.com/tarantool/tarantool/tree/sp/gh-3958-iproto-update-readahe
> ad
> 
>  src/box/iproto.cc               |  8 +++-
>  test/box/iproto_stress.result   | 66 +++++++++++++++++++++++++++++++--
>  test/box/iproto_stress.test.lua | 40 +++++++++++++++++++-
>  3 files changed, 108 insertions(+), 6 deletions(-)
> 
> diff --git a/src/box/iproto.cc b/src/box/iproto.cc
> index a08c8c5cb..745221841 100644
> --- a/src/box/iproto.cc
> +++ b/src/box/iproto.cc
> @@ -145,7 +145,8 @@ iproto_reset_input(struct ibuf *ibuf)
>  	 * move the pos to the start of the input buffer.
>  	 */
>  	assert(ibuf_used(ibuf) == 0);
> -	if (ibuf_capacity(ibuf) < iproto_max_input_size()) {
> +	if (ibuf_capacity(ibuf) < iproto_max_input_size() &&
> +	    ibuf->start_capacity == iproto_readahead) {
>  		ibuf_reset(ibuf);
>  	} else {
>  		struct slab_cache *slabc = ibuf->slabc;
> @@ -670,6 +671,11 @@ iproto_connection_input_buffer(struct iproto_connection
> *con) */
>  		return NULL;
>  	}
> +	/* Update buffer size if readahead has changed. */
> +	if (new_ibuf->start_capacity != iproto_readahead) {
> +		ibuf_destroy(new_ibuf);
> +		ibuf_create(new_ibuf, cord_slab_cache(), 
iproto_readahead);
> +	}
> 
>  	ibuf_reserve_xc(new_ibuf, to_read + con->parse_size);
>  	/*
> diff --git a/test/box/iproto_stress.result b/test/box/iproto_stress.result
> index 4239b49b8..6ced92d70 100644
> --- a/test/box/iproto_stress.result
> +++ b/test/box/iproto_stress.result
> @@ -86,12 +86,72 @@ n_errors -- 0
>  ---
>  - 0
>  ...
> -box.schema.user.revoke('guest', 'read,write,execute', 'universe')
> +s:drop()
>  ---
>  ...
> -s:drop()
> +-- gh-3958 updating box.cfg.readahead doesn't affect existing connections.
> +test_run:cmd('restart server default')
> +fiber = require('fiber')
> +---
> +...
> +netbox = require('net.box')
> +---
> +...
> +readahead = box.cfg.readahead
> +---
> +...
> +net_msg_max = box.cfg.net_msg_max
> +---
> +...
> +box.cfg{net_msg_max=1000, readahead = 128}
> +---
> +...
> +test_run:cmd('setopt delimiter ";"')
> +---
> +- true
> +...
> +function do_call(arg)
> +    fiber.sleep(0.1)
> +    return arg
> +end;
> +---
> +...
> +test_run:cmd('setopt delimiter ""');
> +---
> +- true
> +...
> +-- connection is created with small readahead value,
> +-- make sure it is updated if box.cfg.readahead is changed.
> +c = netbox.connect(box.cfg.listen)
> +---
> +...
> +box.cfg{readahead = 100 * 1024}
> +---
> +...
> +test_run:cmd('setopt delimiter ";"')
> +---
> +- true
> +...
> +for i = 1,500 do
> +    local pad = string.rep('x', 100)
> +    fiber.create(function()
> +        for i = 1,10 do
> +            c:call('do_call', {pad})
> +        end
> +    end)
> +end;
> +---
> +...
> +test_run:cmd('setopt delimiter ""');
> +---
> +- true
> +...
> +test_run:wait_log('default', 'readahead limit is reached', nil, 1.0)
> +---
> +...
> +box.schema.user.revoke('guest', 'read,write,execute', 'universe')
>  ---
>  ...
> -box.cfg{net_msg_max = net_msg_max}
> +box.cfg{net_msg_max = net_msg_max, readahead = readahead}
>  ---
>  ...
> diff --git a/test/box/iproto_stress.test.lua
> b/test/box/iproto_stress.test.lua index 2f3071450..931e66d59 100644
> --- a/test/box/iproto_stress.test.lua
> +++ b/test/box/iproto_stress.test.lua
> @@ -46,7 +46,43 @@ while n_workers > 0 and attempt < 100 do fiber.sleep(0.1)
> attempt = attempt + 1 n_workers -- 0
>  n_errors -- 0
> 
> -box.schema.user.revoke('guest', 'read,write,execute', 'universe')
>  s:drop()
> 
> -box.cfg{net_msg_max = net_msg_max}
> +-- gh-3958 updating box.cfg.readahead doesn't affect existing connections.
> +test_run:cmd('restart server default')
> +fiber = require('fiber')
> +netbox = require('net.box')
> +
> +readahead = box.cfg.readahead
> +net_msg_max = box.cfg.net_msg_max
> +
> +box.cfg{net_msg_max=1000, readahead = 128}
> +
> +test_run:cmd('setopt delimiter ";"')
> +function do_call(arg)
> +    fiber.sleep(0.1)
> +    return arg
> +end;
> +test_run:cmd('setopt delimiter ""');
> +
> +-- connection is created with small readahead value,
> +-- make sure it is updated if box.cfg.readahead is changed.
> +c = netbox.connect(box.cfg.listen)
> +
> +box.cfg{readahead = 100 * 1024}
> +
> +test_run:cmd('setopt delimiter ";"')
> +for i = 1,500 do
> +    local pad = string.rep('x', 100)
> +    fiber.create(function()
> +        for i = 1,10 do
> +            c:call('do_call', {pad})
> +        end
> +    end)
> +end;
> +test_run:cmd('setopt delimiter ""');
> +
> +test_run:wait_log('default', 'readahead limit is reached', nil, 1.0)
> +
> +box.schema.user.revoke('guest', 'read,write,execute', 'universe')
> +box.cfg{net_msg_max = net_msg_max, readahead = readahead}


[-- Attachment #2: This is a digitally signed message part. --]
[-- Type: application/pgp-signature, Size: 488 bytes --]