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

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

[Serge Petrenko]

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-readahead

 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}
-- 
2.17.2 (Apple Git-113)

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

[Georgy Kirichenko]

[-- 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 --]

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

[Vladimir Davydov]

On Mon, Feb 18, 2019 at 08:25:50PM +0300, 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

The ticket also mentions that a connection may be established before the
iproto_readahead option is applied on initial box.cfg{}. Please fix this
race in the scope of this ticket as well (probably in a separate patch).

> ---
> https://github.com/tarantool/tarantool/issues/3958
> https://github.com/tarantool/tarantool/tree/sp/gh-3958-iproto-update-readahead
> 
>  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) {

A comment's missing.

>  		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);
> +	}

Do we really need to have this code in two places? Can we leave this
check only in iproto_reset_input() or iproto_connection_input_buffer()?

>  
>  	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 ""');

Can we somehow speed up this test? E.g. do one big INSERT, which is
stalled on WAL and blocks the iproto buffer?

Also, I don't think it should be a part of iproto_stress, which is
marked long BTW.

> +
> +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}