light: add limiters for io and network bandwidth

lib_limiter.h

@@ -0,0 +1,64 @@
+// (c) 2012 Thomas Schoebel-Theuer / 1&1 Internet AG
+#ifndef MARS_LIB_LIMITER_H
+#define MARS_LIB_LIMITER_H
+
+#include <linux/utsname.h>
+
+#define LIMITER_TIME_RESOLUTION NSEC_PER_SEC
+
+struct mars_limiter {
+	/* tunables */
+	int lim_max_rate;
+	/* internal */
+	int lim_accu;
+	unsigned long long lim_stamp;
+};
+
+extern inline
+int mars_limit(struct mars_limiter *lim, int amount)
+{
+	int res = 0;
+	unsigned long long now;
+
+	now = cpu_clock(raw_smp_processor_id());
+
+	if (lim->lim_max_rate > 0 && likely(lim->lim_stamp)) {
+		long long elapsed = now - lim->lim_stamp;
+		long long rate;
+
+		/* Races are possible, but taken into account.
+		 * There is no real harm from rarely lost updates.
+		 */
+		lim->lim_accu += amount;
+
+		rate = (long long)lim->lim_accu * LIMITER_TIME_RESOLUTION / elapsed;
+
+		if (rate > lim->lim_max_rate) {
+			res = 1001 - lim->lim_max_rate * 1000 / rate;
+		}
+
+		elapsed -= LIMITER_TIME_RESOLUTION * 2;
+		if (elapsed > LIMITER_TIME_RESOLUTION) {
+			lim->lim_stamp += elapsed;
+			if (lim->lim_accu > 0)
+				lim->lim_accu -= (long long)lim->lim_max_rate * elapsed / LIMITER_TIME_RESOLUTION;
+		}
+	} else {
+		lim->lim_accu = amount;
+		lim->lim_stamp = now;
+	}
+	return res;
+}
+
+extern inline
+void mars_limit_sleep(struct mars_limiter *lim, int amount)
+{
+	int sleep = mars_limit(lim, amount);
+	if (sleep > 0) {
+		if (sleep > 1000)
+			sleep = 1000;
+		msleep(sleep);
+	}
+}
+
+#endif

mars_client.c

@@ -475,6 +475,13 @@ static int sender_thread(void *data)
 		mref_a = container_of(tmp, struct client_mref_aspect, io_head);
 		mref = mref_a->object;
 
+		if (brick->limit_mode) {
+			int amount = 0;
+			if (mref->ref_cs_mode < 2)
+				amount = (mref->ref_len - 1) / 1024 + 1;
+			mars_limit_sleep(&client_limiter, amount);
+		}
+
 		MARS_IO("sending mref, id = %d pos = %lld len = %d rw = %d\n", mref->ref_id, mref->ref_pos, mref->ref_len, mref->ref_rw);
 
 		status = mars_send_mref(&output->socket, mref);
@@ -693,6 +700,11 @@ EXPORT_SYMBOL_GPL(client_brick_type);
 
 ////////////////// module init stuff /////////////////////////
 
+struct mars_limiter client_limiter = {
+	.lim_max_rate = 0,
+};
+EXPORT_SYMBOL_GPL(client_limiter);
+
 int __init init_mars_client(void)
 {
 	MARS_INF("init_client()\n");

mars_client.h

@@ -3,6 +3,10 @@
 #define MARS_CLIENT_H
 
 #include "mars_net.h"
+#include "lib_limiter.h"
+
+extern struct mars_limiter client_limiter;
+
 
 #define CLIENT_HASH_MAX 256
 
@@ -20,6 +24,7 @@ struct client_brick {
 	// tunables
 	int max_flying; // limit on parallelism
 	int io_timeout;    // > 0: report IO errors after timeout (in seconds)
+	bool limit_mode;
 };
 
 struct client_input {

mars_server.c

@@ -143,6 +143,7 @@ int server_io(struct server_brick *brick, struct mars_socket *sock, struct mars_
 {
 	struct mref_object *mref;
 	struct server_mref_aspect *mref_a;
+	int amount;
 	int status = -ENOTRECOVERABLE;
 
 	if (!brick->cb_running || !mars_socket_is_alive(sock))
@@ -167,6 +168,11 @@ int server_io(struct server_brick *brick, struct mars_socket *sock, struct mars_
 	
 	mref_a->brick = brick;
 	SETUP_CALLBACK(mref, server_endio, mref_a);
+
+	amount = 0;
+	if (!mref->ref_cs_mode < 2)
+		amount = (mref->ref_len - 1) / 1024 + 1;
+	mars_limit_sleep(&server_limiter, amount);
 	
 	status = GENERIC_INPUT_CALL(brick->inputs[0], mref_get, mref);
 	if (unlikely(status < 0)) {
@@ -254,7 +260,7 @@ int _set_server_bio_params(struct mars_brick *_brick, void *private)
 		return -EINVAL;
 	}
 	bio_brick = (void*)_brick;
-	bio_brick->ra_pages = 1;
+	bio_brick->ra_pages = 0;
 	bio_brick->do_noidle = true;
 	bio_brick->do_sync = true;
 	bio_brick->do_unplug = true;
@@ -768,6 +774,11 @@ static int _server_thread(void *data)
 
 ////////////////// module init stuff /////////////////////////
 
+struct mars_limiter server_limiter = {
+	.lim_max_rate = 0,
+};
+EXPORT_SYMBOL_GPL(server_limiter);
+
 int __init init_mars_server(void)
 {
 	struct sockaddr_storage sockaddr = {};

mars_server.h

@@ -5,6 +5,9 @@
 #include <linux/wait.h>
 
 #include "mars_net.h"
+#include "lib_limiter.h"
+
+extern struct mars_limiter server_limiter;
 
 struct server_mref_aspect {
 	GENERIC_ASPECT(mref);

sy_old/mars_light.c

@@ -160,11 +160,17 @@ int _set_trans_params(struct mars_brick *_brick, void *private)
 	return 1;
 }
 
+struct client_cookie {
+	bool limit_mode;
+};
+
 static
 int _set_client_params(struct mars_brick *_brick, void *private)
 {
 	struct client_brick *client_brick = (void*)_brick;
+	struct client_cookie *clc = private;
 	client_brick->io_timeout = CONFIG_MARS_NETIO_TIMEOUT;
+	client_brick->limit_mode = clc ? clc->limit_mode : false;
 	MARS_INF("name = '%s' path = '%s'\n", _brick->brick_name, _brick->brick_path);
 	return 1;
 }
@@ -609,11 +615,15 @@ int __make_copy(
 		const char *argv[],
 		loff_t start_pos, // -1 means at EOF
 		bool verify_mode,
+		bool limit_mode,
 		struct copy_brick **__copy)
 {
 	struct mars_brick *copy;
 	struct copy_brick *_copy;
 	struct copy_cookie cc = {};
+	struct client_cookie clc = {
+		.limit_mode = limit_mode,
+	};
 	int i;
 	int status = -EINVAL;
 
@@ -646,7 +656,7 @@ int __make_copy(
 				       NULL,
 				       false,
 				       _set_bio_params_nocache,
-				       NULL,
+				       &clc,
 				       10 * HZ,
 				       NULL,
 				       (const struct generic_brick_type*)&bio_brick_type,
@@ -770,7 +780,7 @@ int _update_file(struct mars_rotate *rot, const char *switch_path, const char *c
 		goto done;
 
 	MARS_DBG("src = '%s' dst = '%s'\n", tmp, file);
-	status = __make_copy(global, NULL, switch_path, copy_path, NULL, argv, -1, false, &copy);
+	status = __make_copy(global, NULL, switch_path, copy_path, NULL, argv, -1, false, false, &copy);
 	if (status >= 0 && copy && (!copy->append_mode || copy->power.led_off)) {
 		if (end_pos > copy->copy_end) {
 			MARS_DBG("appending to '%s' %lld => %lld\n", copy_path, copy->copy_end, end_pos);
@@ -2727,7 +2737,7 @@ static int _make_copy(void *buf, struct mars_dent *dent)
 	// check whether connection is allowed
 	switch_path = path_make("%s/todo-%s/connect", dent->d_parent->d_path, my_id());
 
-	status = __make_copy(global, dent, switch_path, copy_path, dent->d_parent->d_path, (const char**)dent->d_argv, -1, false, NULL);
+	status = __make_copy(global, dent, switch_path, copy_path, dent->d_parent->d_path, (const char**)dent->d_argv, -1, false, true, NULL);
 
 done:
 	MARS_DBG("status = %d\n", status);
@@ -2857,7 +2867,7 @@ static int make_sync(void *buf, struct mars_dent *dent)
 #else
 # define VERIFY_MODE false
 #endif
-		status = __make_copy(global, dent, do_start ? switch_path : "", copy_path, dent->d_parent->d_path, argv, start_pos, VERIFY_MODE, &copy);
+		status = __make_copy(global, dent, do_start ? switch_path : "", copy_path, dent->d_parent->d_path, argv, start_pos, VERIFY_MODE, true, &copy);
 		rot->sync_brick = copy;
 		rot->allow_replay = (!copy || copy->power.led_off);
 	}

sy_old/mars_proc.c

@@ -13,6 +13,8 @@
 
 #include "strategy.h"
 #include "mars_proc.h"
+#include "../mars_client.h"
+#include "../mars_server.h"
 
 mars_info_fn mars_info = NULL;
 
@@ -181,6 +183,24 @@ ctl_table mars_table[] = {
 		.strategy       = &sysctl_intvec,
 	},
 #endif
+	{
+		.ctl_name       = CTL_UNNUMBERED,
+		.procname	= "network_traffic_limit",
+		.data           = &client_limiter.lim_max_rate,
+		.maxlen         = sizeof(int),
+		.mode		= 0600,
+		.proc_handler	= &proc_dointvec,
+		.strategy       = &sysctl_intvec,
+	},
+	{
+		.ctl_name       = CTL_UNNUMBERED,
+		.procname	= "server_io_limit",
+		.data           = &server_limiter.lim_max_rate,
+		.maxlen         = sizeof(int),
+		.mode		= 0600,
+		.proc_handler	= &proc_dointvec,
+		.strategy       = &sysctl_intvec,
+	},
 	{}
 };