r20647: add cluster code

(This used to be commit 5870830b99a8d76bda1ff5af3fcf8dda9aba50ec)

2 files changed, 47 insertions, 6 deletions

diff --git a/source4/torture/nbench/nbench.c b/source4/torture/nbench/nbench.c
index f671a5b0de..eb67e15fbb 100644
--- a/source4/torture/nbench/nbench.c
+++ b/source4/torture/nbench/nbench.c
@@ -24,6 +24,7 @@
 #include "torture/util.h"
 #include "torture/torture.h"
 #include "system/filesys.h"
+#include "system/locale.h"
 #include "pstring.h"
 
 #include "torture/nbench/proto.h"
@@ -44,7 +45,6 @@ static BOOL run_netbench(struct torture_context *tctx, struct smbcli_state *cli,
 	pstring line;
 	char *cname;
 	FILE *f;
-	const char **params;
 	BOOL correct = True;
 
 	if (torture_nprocs == 1) {
@@ -64,9 +64,13 @@ static BOOL run_netbench(struct torture_context *tctx, struct smbcli_state *cli,
 		return False;
 	}
 
+
 again:
+	nbio_time_reset();
+
 	while (fgets(line, sizeof(line)-1, f)) {
 		NTSTATUS status;
+		const char **params0, **params;
 
 		nbench_line_count++;
 
@@ -74,9 +78,16 @@ again:
 
 		all_string_sub(line,"client1", cname, sizeof(line));
 		
-		params = str_list_make_shell(NULL, line, " ");
+		params = params0 = str_list_make_shell(NULL, line, " ");
 		i = str_list_length(params);
 
+		if (i > 0 && isdigit(params[0][0])) {
+			double targett = strtod(params[0], NULL);
+			nbio_time_delay(targett);
+			params++;
+			i--;
+		}
+
 		if (i < 2 || params[0][0] == '#') continue;
 
 		if (!strncmp(params[0],"SMB", 3)) {
@@ -146,7 +157,7 @@ again:
 			printf("[%d] Unknown operation %s\n", nbench_line_count, params[0]);
 		}
 
-		talloc_free(params);
+		talloc_free(params0);
 		
 		if (nb_tick()) goto done;
 	}
diff --git a/source4/torture/nbench/nbio.c b/source4/torture/nbench/nbio.c
index cd68085169..6335d53c92 100644
--- a/source4/torture/nbench/nbio.c
+++ b/source4/torture/nbench/nbio.c
@@ -53,8 +53,25 @@ static struct {
 	double bytes, warmup_bytes;
 	int line;
 	int done;
+	double max_latency;
+	struct timeval starttime;
 } *children;
 
+void nbio_time_reset(void)
+{
+	children[nbio_id].starttime = timeval_current();	
+}
+
+void nbio_time_delay(double targett)
+{
+	double elapsed = timeval_elapsed(&children[nbio_id].starttime);
+	if (targett > elapsed) {
+		msleep(1000*(targett - elapsed));
+	} else if (elapsed - targett > children[nbio_id].max_latency) {
+		children[nbio_id].max_latency = elapsed - targett;
+	}
+}
+
 double nbio_result(void)
 {
 	int i;
@@ -65,6 +82,19 @@ double nbio_result(void)
 	return 1.0e-6 * total / timeval_elapsed2(&tv_start, &tv_end);
 }
 
+double nbio_latency(void)
+{
+	int i;
+	double max_latency = 0;
+	for (i=0;i<nprocs;i++) {
+		if (children[i].max_latency > max_latency) {
+			max_latency = children[i].max_latency;
+			children[i].max_latency = 0;
+		}
+	}
+	return max_latency;
+}
+
 BOOL nb_tick(void)
 {
 	return children[nbio_id].done;
@@ -122,9 +152,9 @@ void nb_alarm(int sig)
 		       nprocs, lines/nprocs, 
 		       nbio_result(), t);
 	} else {
-		printf("%4d  %8d  %.2f MB/sec  execute %.0f sec   \n", 
+		printf("%4d  %8d  %.2f MB/sec  execute %.0f sec  latency %.2f msec \n", 
 		       nprocs, lines/nprocs, 
-		       nbio_result(), t);
+		       nbio_result(), t, nbio_latency() * 1.0e3);
 	}
 
 	fflush(stdout);
@@ -443,7 +473,7 @@ void nb_readx(int handle, off_t offset, int size, int ret_size, NTSTATUS status)
 	io.readx.in.remaining = 0;
 	io.readx.in.read_for_execute = False;
 	io.readx.out.data     = buf;
-		
+	
 	ret = smb_raw_read(c->tree, &io);
 
 	free(buf);