ntttcp-for-linux 1.3.3: add real-time throughput display

1) on both sender and receiver endpoints, added new thread to manage the throughput (collect "number of bytes transferred" periodically and calculate the real-time throughput)
2) updated tcpstream and udpstream streams to make them report "number of bytes transferred" to its data structure so that the throughtput_management thread cal collect that number from those test streams.

Author: simonxiaoss

src/Makefile

@@ -2,14 +2,16 @@ CC=gcc
 CFLAGS=-Wall -W -O2 -fno-strict-aliasing
 CLIBS=-pthread
 OUT=ntttcp
-OBJS=ntttcp.o util.o tcpstream.o udpstream.o endpointsync.o multithreading.o main.o
+OBJS=ntttcp.o util.o throughputmanagement.o tcpstream.o udpstream.o endpointsync.o multithreading.o main.o
 
-$(OUT): ntttcp.o util.o tcpstream.o udpstream.o endpointsync.o multithreading.o main.o
+$(OUT): ntttcp.o util.o throughputmanagement.o tcpstream.o udpstream.o endpointsync.o multithreading.o main.o
 	$(CC) $(CFLAGS) $(CLIBS) -o $(OUT) $(OBJS)
 ntttcp.o: ntttcp.c
 	$(CC) $(CFLAGS) -c ntttcp.c -o ntttcp.o
 util.o: util.c
 	$(CC) $(CFLAGS) -c util.c -o util.o
+throughputmanagement.o: throughputmanagement.c
+	$(CC) $(CFLAGS) -c throughputmanagement.c -o throughputmanagement.o
 tcpstream.o: tcpstream.c
 	$(CC) $(CFLAGS) -c tcpstream.c -o tcpstream.o
 udpstream.o:udpstream.c

src/const.h

@@ -5,7 +5,7 @@
 // ----------------------------------------------------------------------------------
 
 #define TOOL_NAME "NTTTCP for Linux"
-#define TOOL_VERSION "1.3.2"
+#define TOOL_VERSION "1.3.3"
 #define AUTHOR_NAME "Shihua (Simon) Xiao, [email protected]"
 
 #define TCP 				SOCK_STREAM
@@ -19,7 +19,6 @@
 #define TEST_RUNNING			12
 #define TEST_FINISHED			13
 
-
 /*
  * NUM_SERVER_PORTS:		how many PORTs opened by receiver;
  * CONNS_PER_SERVER_PORT: 	how many THREADs on sender side will be created for each receiver PORT;

src/main.c

@@ -80,10 +80,23 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep)
 		PRINT_INFO("Starting sender activity (no sync) ...");
 	}
 
-	/* create threads */
+	/* prepare to create threads */
 	pthread_attr_init(&pth_attrs);
 	pthread_attr_setstacksize(&pth_attrs, THREAD_STACK_SIZE);
 
+	/* create throughput management thread */
+	rc = pthread_create(tep->throughput_mgmt_thread,
+			    &pth_attrs,
+			    run_ntttcp_throughput_management,
+			    (void*)tep);
+	if (rc) {
+		ASPRINTF(&log, "pthread_create(): failed to create throughput management thread. errno = %d", errno);
+		PRINT_ERR_FREE(log);
+		pthread_attr_destroy(&pth_attrs);
+		return ERROR_PTHREAD_CREATE;
+	}
+
+	/* create test threads */
 	for (t = 0; t < test->server_ports; t++) {
 		for (n = 0; n < test->threads_per_server_port; n++ ) {
 			cs = tep->client_streams[t * test->threads_per_server_port + n];
@@ -183,6 +196,8 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep)
 		tep->results->threads[n]->total_bytes = nbytes;
 		tep->results->threads[n]->actual_test_time = actual_test_time;
 	}
+
+	pthread_join(*(tep->throughput_mgmt_thread), &p_retval);
 	ASPRINTF(&log, "%d connections tested", total_conns_created);
 	PRINT_INFO_FREE(log);
 
@@ -211,7 +226,18 @@ int run_ntttcp_receiver(struct ntttcp_test_endpoint *tep)
 	struct timeval now;
 	double actual_test_time = 0;
 
-	/* create threads */
+	/* create throughput management thread */
+	rc = pthread_create(tep->throughput_mgmt_thread,
+			    NULL,
+			    run_ntttcp_throughput_management,
+			    (void*)tep);
+	if (rc) {
+		ASPRINTF(&log, "pthread_create(): failed to create throughput management thread. errno = %d", errno);
+		PRINT_ERR_FREE(log);
+		return ERROR_PTHREAD_CREATE;
+	}
+
+	/* create test threads */
 	for (t = 0; t < test->server_ports; t++) {
 		ss = tep->server_streams[t];
 		ss->server_port = test->server_base_port + t;
@@ -358,6 +384,7 @@ int run_ntttcp_receiver(struct ntttcp_test_endpoint *tep)
 	for (t=0; t < threads_created; t++) {
 		pthread_join(tep->threads[t], NULL);
 	}
+	pthread_join(*(tep->throughput_mgmt_thread), NULL);
 
 	return err_code;
 }

src/main.h

@@ -8,4 +8,5 @@
 #include <stdio.h>
 #include <limits.h>
 #include "endpointsync.h"
+#include "throughputmanagement.h"
 #include "udpstream.h"

src/ntttcp.c

@@ -127,6 +127,8 @@ struct ntttcp_test_endpoint *new_ntttcp_test_endpoint(struct ntttcp_test *test,
 			return NULL;
 		}
 	}
+	/* for throughput management thread */
+	e->throughput_mgmt_thread = malloc( sizeof(pthread_t) );
 
 	/* for test results */
 	e->results = (struct ntttcp_test_endpoint_results *) malloc(sizeof(struct ntttcp_test_endpoint_results));
@@ -216,6 +218,7 @@ void free_ntttcp_test_endpoint_and_test(struct ntttcp_test_endpoint* e)
 	free( e->results->final_tcp_retrans);
 	free( e->results );
 	free( e->threads );
+	free( e->throughput_mgmt_thread );
 	free( e->test );
 	free( e );
 }

src/ntttcp.h

@@ -66,9 +66,11 @@ struct ntttcp_test_endpoint{
 	struct	ntttcp_stream_client **client_streams; /* alloc memory for this if client/sender role */
 	struct	ntttcp_stream_server **server_streams; /* alloc memory for this if server/receiver role */
 
-	pthread_t	*threads;
+	pthread_t	*threads;      /* linux threads created to transfer test data */
 	struct	ntttcp_test_endpoint_results	*results;	/* test results */
 
+	pthread_t	*throughput_mgmt_thread;	/* linux thread created to manage the throughput on endpoint */
+
 	/* to support testing with multiple senders */
 	int	num_remote_endpoints; /* number to test client/sender endpoints */
 	int	remote_endpoints[MAX_REMOTE_ENDPOINTS]; /* list of the TCP listeners of those endpoints */

src/tcpstream.c

@@ -69,7 +69,6 @@ void *run_ntttcp_sender_tcp_stream( void *ptr )
 	uint i           = 0; //for loop iterator
 	char *buffer;         //send buffer
 	int n            = 0; //write n bytes to socket
-	uint64_t nbytes  = 0; //total bytes sent
 	int ret          = 0; //hold function return value
 	uint total_sub_conn_created = 0; //track how many sub connections created in this thread
 	struct ntttcp_stream_client *sc;
@@ -235,11 +234,10 @@ void *run_ntttcp_sender_tcp_stream( void *ptr )
 //				PRINT_ERR("cannot write data to a socket");
 				continue;
 			}
-			nbytes += n;
+			sc->total_bytes_transferred += n;
 		}
 	}
 	sc->num_conns_created = total_sub_conn_created;
-	sc->total_bytes_transferred = nbytes;
 	free(buffer);
 
 CLEANUP:

src/throughputmanagement.c

@@ -0,0 +1,43 @@
+// ----------------------------------------------------------------------------------
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+// Author: Shihua (Simon) Xiao, [email protected]
+// ----------------------------------------------------------------------------------
+
+#include "throughputmanagement.h"
+
+void *run_ntttcp_throughput_management(void *ptr)
+{
+	struct ntttcp_test_endpoint *tep = (struct ntttcp_test_endpoint *) ptr;
+	uint n;
+	uint64_t last_total_bytes, this_total_bytes, total_bytes;
+	uint total_test_threads = 0;
+	last_total_bytes = this_total_bytes = total_bytes = 0;
+
+	if (tep->test->client_role == true )
+		total_test_threads = tep->test->server_ports * tep->test->threads_per_server_port;
+	else
+		total_test_threads = tep->test->server_ports;
+
+	wait_light_on();
+
+	while( is_light_turned_on(tep->test->duration == 0) ) {
+		for (n = 0; n < total_test_threads; n++) {
+			if (tep->test->client_role == true )
+				this_total_bytes += tep->client_streams[n]->total_bytes_transferred;
+			else
+				this_total_bytes += tep->server_streams[n]->total_bytes_transferred;
+		}
+
+		total_bytes = this_total_bytes - last_total_bytes;
+		last_total_bytes = this_total_bytes;
+		this_total_bytes = 0;
+
+		/* sleep 0.5 second, then calculate the throughput for this period */
+		printf("%s: %s\r", "Real-time throughput", format_throughput(total_bytes, 0.5));
+		fflush(stdout);
+		usleep(500000);
+	}
+
+	return NULL;
+}

src/throughputmanagement.h

@@ -0,0 +1,11 @@
+// ----------------------------------------------------------------------------------
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+// Author: Shihua (Simon) Xiao, [email protected]
+// ----------------------------------------------------------------------------------
+
+#define _GNU_SOURCE
+#include "multithreading.h"
+#include "util.h"
+
+void *run_ntttcp_throughput_management(void *ptr);

src/udpstream.c

@@ -32,7 +32,6 @@ void *run_ntttcp_sender_udp4_stream( struct ntttcp_stream_client * sc )
 	char *buffer;         //send buffer
 
 	int n            = 0; //write n bytes to socket
-	uint64_t nbytes  = 0; //total bytes sent
 	int ret          = 0; //hold function return value
 	uint i           = 0; //for loop iterator
 	uint total_sub_conn_created = 0; //track how many sub connections created in this thread
@@ -115,11 +114,10 @@ void *run_ntttcp_sender_udp4_stream( struct ntttcp_stream_client * sc )
 	//			printf("error: %d \n", errno);
 				continue;
 			}
-			nbytes += n;
+			sc->total_bytes_transferred += n;
 		}
 	}
 	sc->num_conns_created = total_sub_conn_created;
-	sc->total_bytes_transferred = nbytes;
 	free(buffer);
 
 CLEANUP: