Wait for all buffered data to be sent before closing TCP Streams

[davidBar-On]

Context

• Version of iperf3:
  iperf 3.7+ (cJSON 1.5.2)

• Hardware:
  Dell PC DESKTOP-L81E90U

• Operating system (and distribution, if any):
  Ubunto Linux 18.04 LTS VM under Windows 10

• Other relevant information (for example, non-default compilers,
  libraries, cross-compiling, etc.):

Enhancement Request

• Current behavior
  When using -k/-t/-n parameters to limit amount of data sent, once condition is met, all streams are closed. However, in this case data that is buffered at the sender end is not sent and the total data received is not all the data sent.
  The issue is raised in Debobroto Das email to ipef_dev that asks how to ensure all data sent by the client is received by the server before the connection is closed: https://groups.google.com/d/topic/iperf-dev/okHU7g0C1i8/discussion.

• Desired behavior
  Close the streams only when all data that was sent is received by the other end.

• Implementation notes
  Searching the internet, it seems that using socket shutdown and SO_LINGER option can improve iperf3 related behavior. See for example: https://blog.netherlabs.nl/articles/2009/01/18/the-ultimate-so_linger-page-or-why-is-my-tcp-not-reliable.

The suggested code changes are below. Unfortunately, I don't have the infrastructure that can allow to test the effectiveness of these changes. I did check with basic scenarios that the changes do not cause problems (at least in Ubunto VM under Windows 10). This is also why I am not sending a pull request.

The approach is to set SO_LINGER timeout for each stream (I set 10 seconds) and to call shotdown() for each stream before closing it. This should be enough to make sure that the sender sent all the data. However, it may not be enough to ensure that the receiver received all the data. That may also depend on the shutdown() implementation in each operating system.

Following are the enhancements I did. The code I added is marked between the /* DBO >>> …*/ and the /* <<< DBO */ comments.

iperf_client_api.c

iperf_create_streams(…) {
    ….
   for (i = 0; i < test->num_streams; ++i) {

        test->bind_port = orig_bind_port;
	if (orig_bind_port)
	    test->bind_port += i;
        if ((s = test->protocol->connect(test)) < 0)
            return -1;

    /* DBO >>> Set socket Linger value */
        int saved_errno;
        struct linger ling;
	ling.l_onoff = 1; 
        ling.l_linger = 10;
	if (setsockopt(s, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling))) {
	    saved_errno = errno;
	    close(s);
	    errno = saved_errno;
            i_errno = IESETLINGER;
            return -1;
	}
    /* <<< DBO */
    ….
}

……..

iperf_run_client(…) {
    ……

		/* Yes, done!  Send TEST_END. */
		test->done = 1;
		cpu_util(test->cpu_util);
		test->stats_callback(test);
		if (iperf_set_send_state(test, TEST_END) != 0)
		    return -1;

		/* >>> DBO - When Sender - Make sure all sockets are flushed and received by other end (that sends FIN) */
		SLIST_FOREACH(sp, &test->streams, streams) {
			if (sp->sender) {
				printf("*** CLNT BEFORE SHUTDOWN\n");
				shutdown(sp->socket, SHUT_WR);
				printf("*** CNLT AFTER SHUTDOWN\n");
			}
		}

		/* When Receiver - Make sure all sockets are received */
		int all_blocks_received = 0;
		int r;
		while (all_blocks_received == 0) {
			all_blocks_received = 1;
			SLIST_FOREACH(sp, &test->streams, streams) {
				if (!sp->sender) {
					r = sp->rcv(sp);
					printf("*** RECEIVED LAST BLOCK SIZE %i\n", r);
					if (r > 0)
					all_blocks_received = 0;
				}
			}
		}
		/* <<<< DBO */
    ……..

}

iperf_server_api.c

iperf_handle_message_server(struct iperf_test *test)
{
    ….
  switch(test->state) {
        case TEST_START:
            break;
        case TEST_END:
	    /* >>> DBO - When Sender - Make sure all sockets are flushed and received by other end (that sends FIN) */
		SLIST_FOREACH(sp, &test->streams, streams) {
		    if (sp->sender) {
			printf("*** SRV BEFORE SHUTDOWN\n");
			setnonblocking(sp->socket, 0);
			shutdown(sp->socket, SHUT_WR);
			setnonblocking(sp->socket, 1);
			printf("*** SRV AFTER SHUTDOWN\n");
		    }
		}

		/* When Receiver - Make sure all sockets are received */
		int all_blocks_received = 0;
		int r;
		while (all_blocks_received == 0) {
		    all_blocks_received = 1;
		    SLIST_FOREACH(sp, &test->streams, streams) {
			if (!sp->sender) {
			    r = sp->rcv(sp);
			    printf("*** RECEIVED LAST BLOCK SIZE %i\n", r);
			    if (r > 0)
				all_blocks_received = 0;
			}
		    }
		}
	    /* <<< DBO */

            test->done = 1;
            ………..

……..

iperf_run_server(struct iperf_test *test)
{
      …….
            if (test->state == CREATE_STREAMS) {
                if (FD_ISSET(test->prot_listener, &read_set)) {
    
                    if ((s = test->protocol->accept(test)) < 0) {
			cleanup_server(test);
                        return -1;
		    }

		    /* DBO >>> Set socket Linger value */
		    int saved_errno;
		    struct linger ling;
		    ling.l_onoff = 1; 
		    ling.l_linger = 10;
		    if (setsockopt(s, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling))) {
			saved_errno = errno;
			close(s);
			errno = saved_errno;
			i_errno = IESETLINGER;
			return -1;
		    }
		    /* <<< DBO */
    …….

iperf_ api.c

iperf_send(struct iperf_test *test, fd_set *write_setP)
{
    register int multisend, r, streams_active;
    register struct iperf_stream *sp;
    struct iperf_time now;

    /* >>> DBO - Don't send if test ended */
    if (test->state == TEST_END)
        return 0;
    /* >>> DBO */

    ……….