From e159341782dbe42d160a03429dfc50915b4141fc Mon Sep 17 00:00:00 2001 From: Kirill Smorodinnikov Date: Wed, 20 Jul 2016 18:44:55 +0300 Subject: [PATCH 1/2] net: fix addicting net thread on sending responses to one state If there are a lot of small responses which should be sent to some state with good connection, e.g. iterator generates such responses, net thread can get into cycle of sending these responses and will not send responses to other states until cycle is done. If responses is rather small, the cycle will be done only when all responses are sent. I've added break to this cycle if a response is fully sent, so net thread will switch to another state for sending response to it. --- library/pool.c | 1 + 1 file changed, 1 insertion(+) diff --git a/library/pool.c b/library/pool.c index d8b2ec7df..ae8f8f296 100644 --- a/library/pool.c +++ b/library/pool.c @@ -689,6 +689,7 @@ static int dnet_process_send_single(struct dnet_net_state *st) dnet_io_req_free(r); st->send_offset = 0; + break; } if (err) From 5314713d536998d9d193b8a9d8991f16ca3d8f25 Mon Sep 17 00:00:00 2001 From: Kirill Smorodinnikov Date: Fri, 5 Aug 2016 11:31:26 +0300 Subject: [PATCH 2/2] net: add send_limit option --- example/config.cpp | 1 + include/elliptics/core.h | 2 ++ include/elliptics/interface.h | 4 +++- library/elliptics.h | 5 +++++ library/node.c | 1 + library/pool.c | 11 ++++++++++- 6 files changed, 22 insertions(+), 2 deletions(-) diff --git a/example/config.cpp b/example/config.cpp index 3dceb3345..dc828faa8 100644 --- a/example/config.cpp +++ b/example/config.cpp @@ -265,6 +265,7 @@ void parse_options(config_data *data, const config &options) data->cfg_state.flags |= (options.at("join", false) ? DNET_CFG_JOIN_NETWORK : 0); data->cfg_state.flags |= (options.at("flags", 0) & ~DNET_CFG_JOIN_NETWORK); data->cfg_state.io_thread_num = options.at("io_thread_num"); + data->cfg_state.send_limit = options.at("send_limit", DNET_DEFAULT_SEND_LIMIT); data->cfg_state.nonblocking_io_thread_num = options.at("nonblocking_io_thread_num"); data->cfg_state.net_thread_num = options.at("net_thread_num"); data->cfg_state.bg_ionice_class = options.at("bg_ionice_class", 0); diff --git a/include/elliptics/core.h b/include/elliptics/core.h index 3da52cc94..609075b81 100644 --- a/include/elliptics/core.h +++ b/include/elliptics/core.h @@ -85,6 +85,8 @@ #define DNET_DEFAULT_CACHE_PAGES_NUMBER 1 +#define DNET_DEFAULT_SEND_LIMIT 1000 + #define ARRAY_SIZE(arr) (sizeof(arr) / sizeof((arr)[0])) #ifndef dnet_offsetof diff --git a/include/elliptics/interface.h b/include/elliptics/interface.h index d20924c28..15259d839 100644 --- a/include/elliptics/interface.h +++ b/include/elliptics/interface.h @@ -379,7 +379,9 @@ struct dnet_config /* Config values for srw backend */ struct srw_init_ctl srw; - int reserved_for_future_use_2[5]; + int send_limit; + + int reserved_for_future_use_2[4]; /* Config file name for handystats library */ const char *handystats_config; diff --git a/library/elliptics.h b/library/elliptics.h index 19480cf10..f267ee7dd 100644 --- a/library/elliptics.h +++ b/library/elliptics.h @@ -661,6 +661,11 @@ struct dnet_node struct dnet_test_settings *test_settings; /* Lock for test_settings */ pthread_rwlock_t test_settings_lock; + + /* Maximum number of packets sent to one state in a row. It is quota for fast connections + * after which net thread will switch to next ready connection. + */ + uint32_t send_limit; }; diff --git a/library/node.c b/library/node.c index 77d8e6250..2306be5fb 100644 --- a/library/node.c +++ b/library/node.c @@ -760,6 +760,7 @@ struct dnet_node *dnet_node_create(struct dnet_config *cfg) n->removal_delay = cfg->removal_delay; n->flags = cfg->flags; n->indexes_shard_count = cfg->indexes_shard_count; + n->send_limit = cfg->send_limit; if (!n->log) dnet_log_init(n, cfg->log); diff --git a/library/pool.c b/library/pool.c index ae8f8f296..fc5a7bdde 100644 --- a/library/pool.c +++ b/library/pool.c @@ -650,6 +650,7 @@ static int dnet_process_send_single(struct dnet_net_state *st) { struct dnet_io_req *r = NULL; int err; + uint32_t counter = 0; while (1) { r = NULL; @@ -689,7 +690,15 @@ static int dnet_process_send_single(struct dnet_net_state *st) dnet_io_req_free(r); st->send_offset = 0; - break; + /* exit the loop, if @send_limit was set and it has been reached, and switch net thread to + * another ready state. + */ + if (st->n->send_limit && ++counter >= st->n->send_limit) { + dnet_log(st->n, DNET_LOG_INFO, "Limit on number of packet sent to one state in a row " + "has been reached: limit: %" PRIu32, + st->n->send_limit); + break; + } } if (err)