net: fix addicting net thread on sending responses to one state

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<unsigned>("io_thread_num");
+	data->cfg_state.send_limit = options.at<unsigned>("send_limit", DNET_DEFAULT_SEND_LIMIT);
 	data->cfg_state.nonblocking_io_thread_num = options.at<unsigned>("nonblocking_io_thread_num");
 	data->cfg_state.net_thread_num = options.at<unsigned>("net_thread_num");
 	data->cfg_state.bg_ionice_class = options.at("bg_ionice_class", 0);

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

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;

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

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

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)