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 d8b2ec7df..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,6 +690,15 @@ static int dnet_process_send_single(struct dnet_net_state *st) dnet_io_req_free(r); st->send_offset = 0; + /* 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)