diff options
Diffstat (limited to 'src/iperf_client_api.c')
| -rw-r--r-- | src/iperf_client_api.c | 819 |
1 files changed, 819 insertions, 0 deletions
diff --git a/src/iperf_client_api.c b/src/iperf_client_api.c new file mode 100644 index 0000000..5b33369 --- /dev/null +++ b/src/iperf_client_api.c @@ -0,0 +1,819 @@ +/* + * iperf, Copyright (c) 2014-2023, The Regents of the University of + * California, through Lawrence Berkeley National Laboratory (subject + * to receipt of any required approvals from the U.S. Dept. of + * Energy). All rights reserved. + * + * If you have questions about your rights to use or distribute this + * software, please contact Berkeley Lab's Technology Transfer + * Department at TTD@lbl.gov. + * + * NOTICE. This software is owned by the U.S. Department of Energy. + * As such, the U.S. Government has been granted for itself and others + * acting on its behalf a paid-up, nonexclusive, irrevocable, + * worldwide license in the Software to reproduce, prepare derivative + * works, and perform publicly and display publicly. Beginning five + * (5) years after the date permission to assert copyright is obtained + * from the U.S. Department of Energy, and subject to any subsequent + * five (5) year renewals, the U.S. Government is granted for itself + * and others acting on its behalf a paid-up, nonexclusive, + * irrevocable, worldwide license in the Software to reproduce, + * prepare derivative works, distribute copies to the public, perform + * publicly and display publicly, and to permit others to do so. + * + * This code is distributed under a BSD style license, see the LICENSE + * file for complete information. + */ +#include <errno.h> +#include <setjmp.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> +#include <signal.h> +#include <sys/types.h> +#include <netinet/in.h> +#include <sys/select.h> +#include <sys/uio.h> +#include <arpa/inet.h> + +#include "iperf.h" +#include "iperf_api.h" +#include "iperf_util.h" +#include "iperf_locale.h" +#include "iperf_time.h" +#include "net.h" +#include "timer.h" + +#if defined(HAVE_TCP_CONGESTION) +#if !defined(TCP_CA_NAME_MAX) +#define TCP_CA_NAME_MAX 16 +#endif /* TCP_CA_NAME_MAX */ +#endif /* HAVE_TCP_CONGESTION */ + +void * +iperf_client_worker_run(void *s) { + struct iperf_stream *sp = (struct iperf_stream *) s; + struct iperf_test *test = sp->test; + + /* Allow this thread to be cancelled even if it's in a syscall */ + pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL); + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); + + while (! (test->done) && ! (sp->done)) { + if (sp->sender) { + if (iperf_send_mt(sp) < 0) { + goto cleanup_and_fail; + } + } + else { + if (iperf_recv_mt(sp) < 0) { + goto cleanup_and_fail; + } + } + } + return NULL; + + cleanup_and_fail: + return NULL; +} + +int +iperf_create_streams(struct iperf_test *test, int sender) +{ + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + return -1; + } + int i, s; +#if defined(HAVE_TCP_CONGESTION) + int saved_errno; +#endif /* HAVE_TCP_CONGESTION */ + struct iperf_stream *sp; + + int orig_bind_port = test->bind_port; + for (i = 0; i < test->num_streams; ++i) { + + test->bind_port = orig_bind_port; + if (orig_bind_port) { + test->bind_port += i; + // If Bidir make sure send and receive ports are different + if (!sender && test->mode == BIDIRECTIONAL) + test->bind_port += test->num_streams; + } + s = test->protocol->connect(test); + test->bind_port = orig_bind_port; + if (s < 0) + return -1; + +#if defined(HAVE_TCP_CONGESTION) + if (test->protocol->id == Ptcp) { + if (test->congestion) { + if (setsockopt(s, IPPROTO_TCP, TCP_CONGESTION, test->congestion, strlen(test->congestion)) < 0) { + saved_errno = errno; + close(s); + errno = saved_errno; + i_errno = IESETCONGESTION; + return -1; + } + } + { + socklen_t len = TCP_CA_NAME_MAX; + char ca[TCP_CA_NAME_MAX + 1]; + int rc; + rc = getsockopt(s, IPPROTO_TCP, TCP_CONGESTION, ca, &len); + if (rc < 0 && test->congestion) { + saved_errno = errno; + close(s); + errno = saved_errno; + i_errno = IESETCONGESTION; + return -1; + } + // Set actual used congestion alg, or set to unknown if could not get it + if (rc < 0) + test->congestion_used = strdup("unknown"); + else + test->congestion_used = strdup(ca); + if (test->debug) { + printf("Congestion algorithm is %s\n", test->congestion_used); + } + } + } +#endif /* HAVE_TCP_CONGESTION */ + + sp = iperf_new_stream(test, s, sender); + if (!sp) + return -1; + + /* Perform the new stream callback */ + if (test->on_new_stream) + test->on_new_stream(sp); + } + + return 0; +} + +static void +test_timer_proc(TimerClientData client_data, struct iperf_time *nowP) +{ + struct iperf_test *test = client_data.p; + + test->timer = NULL; + test->done = 1; +} + +static void +client_stats_timer_proc(TimerClientData client_data, struct iperf_time *nowP) +{ + struct iperf_test *test = client_data.p; + + if (test->done) + return; + if (test->stats_callback) + test->stats_callback(test); +} + +static void +client_reporter_timer_proc(TimerClientData client_data, struct iperf_time *nowP) +{ + struct iperf_test *test = client_data.p; + + if (test->done) + return; + if (test->reporter_callback) + test->reporter_callback(test); +} + +static int +create_client_timers(struct iperf_test * test) +{ + struct iperf_time now; + TimerClientData cd; + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + i_errno = IEINITTEST; + return -1; + } + + if (iperf_time_now(&now) < 0) { + i_errno = IEINITTEST; + return -1; + } + cd.p = test; + test->timer = test->stats_timer = test->reporter_timer = NULL; + if (test->duration != 0) { + test->done = 0; + test->timer = tmr_create(&now, test_timer_proc, cd, ( test->duration + test->omit ) * SEC_TO_US, 0); + if (test->timer == NULL) { + i_errno = IEINITTEST; + return -1; + } + } + if (test->stats_interval != 0) { + test->stats_timer = tmr_create(&now, client_stats_timer_proc, cd, test->stats_interval * SEC_TO_US, 1); + if (test->stats_timer == NULL) { + i_errno = IEINITTEST; + return -1; + } + } + if (test->reporter_interval != 0) { + test->reporter_timer = tmr_create(&now, client_reporter_timer_proc, cd, test->reporter_interval * SEC_TO_US, 1); + if (test->reporter_timer == NULL) { + i_errno = IEINITTEST; + return -1; + } + } + return 0; +} + +static void +client_omit_timer_proc(TimerClientData client_data, struct iperf_time *nowP) +{ + struct iperf_test *test = client_data.p; + + test->omit_timer = NULL; + test->omitting = 0; + iperf_reset_stats(test); + if (test->verbose && !test->json_output && test->reporter_interval == 0) + iperf_printf(test, "%s", report_omit_done); + + /* Reset the timers. */ + if (test->stats_timer != NULL) + tmr_reset(nowP, test->stats_timer); + if (test->reporter_timer != NULL) + tmr_reset(nowP, test->reporter_timer); +} + +static int +create_client_omit_timer(struct iperf_test * test) +{ + struct iperf_time now; + TimerClientData cd; + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + return -1; + } + + if (test->omit == 0) { + test->omit_timer = NULL; + test->omitting = 0; + } else { + if (iperf_time_now(&now) < 0) { + i_errno = IEINITTEST; + return -1; + } + test->omitting = 1; + cd.p = test; + test->omit_timer = tmr_create(&now, client_omit_timer_proc, cd, test->omit * SEC_TO_US, 0); + if (test->omit_timer == NULL) { + i_errno = IEINITTEST; + return -1; + } + } + return 0; +} + +int +iperf_handle_message_client(struct iperf_test *test) +{ + int rval; + int32_t err; + + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + i_errno = IEINITTEST; + return -1; + } + /*!!! Why is this read() and not Nread()? */ + if ((rval = read(test->ctrl_sck, (char*) &test->state, sizeof(signed char))) <= 0) { + if (rval == 0) { + i_errno = IECTRLCLOSE; + return -1; + } else { + i_errno = IERECVMESSAGE; + return -1; + } + } + + switch (test->state) { + case PARAM_EXCHANGE: + if (iperf_exchange_parameters(test) < 0) + return -1; + if (test->on_connect) + test->on_connect(test); + break; + case CREATE_STREAMS: + if (test->mode == BIDIRECTIONAL) + { + if (iperf_create_streams(test, 1) < 0) + return -1; + if (iperf_create_streams(test, 0) < 0) + return -1; + } + else if (iperf_create_streams(test, test->mode) < 0) + return -1; + break; + case TEST_START: + if (iperf_init_test(test) < 0) + return -1; + if (create_client_timers(test) < 0) + return -1; + if (create_client_omit_timer(test) < 0) + return -1; + if (test->mode) + if (iperf_create_send_timers(test) < 0) + return -1; + break; + case TEST_RUNNING: + break; + case EXCHANGE_RESULTS: + if (iperf_exchange_results(test) < 0) + return -1; + break; + case DISPLAY_RESULTS: + if (test->on_test_finish) + test->on_test_finish(test); + iperf_client_end(test); + break; + case IPERF_DONE: + break; + case SERVER_TERMINATE: + i_errno = IESERVERTERM; + + /* + * Temporarily be in DISPLAY_RESULTS phase so we can get + * ending summary statistics. + */ + signed char oldstate = test->state; + cpu_util(test->cpu_util); + test->state = DISPLAY_RESULTS; + test->reporter_callback(test); + test->state = oldstate; + return -1; + case ACCESS_DENIED: + i_errno = IEACCESSDENIED; + return -1; + case SERVER_ERROR: + if (Nread(test->ctrl_sck, (char*) &err, sizeof(err), Ptcp) < 0) { + i_errno = IECTRLREAD; + return -1; + } + i_errno = ntohl(err); + if (Nread(test->ctrl_sck, (char*) &err, sizeof(err), Ptcp) < 0) { + i_errno = IECTRLREAD; + return -1; + } + errno = ntohl(err); + return -1; + default: + i_errno = IEMESSAGE; + return -1; + } + + return 0; +} + + + +/* iperf_connect -- client to server connection function */ +int +iperf_connect(struct iperf_test *test) +{ + int opt; + socklen_t len; + + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + return -1; + } + FD_ZERO(&test->read_set); + FD_ZERO(&test->write_set); + + make_cookie(test->cookie); + + /* Create and connect the control channel */ + if (test->ctrl_sck < 0) + // Create the control channel using an ephemeral port + test->ctrl_sck = netdial(test->settings->domain, Ptcp, test->bind_address, test->bind_dev, 0, test->server_hostname, test->server_port, test->settings->connect_timeout); + if (test->ctrl_sck < 0) { + i_errno = IECONNECT; + return -1; + } + + // set TCP_NODELAY for lower latency on control messages + int flag = 1; + if (setsockopt(test->ctrl_sck, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int))) { + i_errno = IESETNODELAY; + return -1; + } + +#if defined(HAVE_TCP_USER_TIMEOUT) + if ((opt = test->settings->snd_timeout)) { + if (setsockopt(test->ctrl_sck, IPPROTO_TCP, TCP_USER_TIMEOUT, &opt, sizeof(opt)) < 0) { + i_errno = IESETUSERTIMEOUT; + return -1; + } + } +#endif /* HAVE_TCP_USER_TIMEOUT */ + + if (Nwrite(test->ctrl_sck, test->cookie, COOKIE_SIZE, Ptcp) < 0) { + i_errno = IESENDCOOKIE; + return -1; + } + + FD_SET(test->ctrl_sck, &test->read_set); + if (test->ctrl_sck > test->max_fd) test->max_fd = test->ctrl_sck; + + len = sizeof(opt); + if (getsockopt(test->ctrl_sck, IPPROTO_TCP, TCP_MAXSEG, &opt, &len) < 0) { + test->ctrl_sck_mss = 0; + } + else { + if (opt > 0 && opt <= MAX_UDP_BLOCKSIZE) { + test->ctrl_sck_mss = opt; + } + else { + char str[WARN_STR_LEN]; + snprintf(str, sizeof(str), + "Ignoring nonsense TCP MSS %d", opt); + warning(str); + + test->ctrl_sck_mss = 0; + } + } + + if (test->verbose) { + printf("Control connection MSS %d\n", test->ctrl_sck_mss); + } + + /* + * If we're doing a UDP test and the block size wasn't explicitly + * set, then use the known MSS of the control connection to pick + * an appropriate default. If we weren't able to get the + * MSS for some reason, then default to something that should + * work on non-jumbo-frame Ethernet networks. The goal is to + * pick a reasonable default that is large but should get from + * sender to receiver without any IP fragmentation. + * + * We assume that the control connection is routed the same as the + * data packets (thus has the same PMTU). Also in the case of + * --reverse tests, we assume that the MTU is the same in both + * directions. Note that even if the algorithm guesses wrong, + * the user always has the option to override. + */ + if (test->protocol->id == Pudp) { + if (test->settings->blksize == 0) { + if (test->ctrl_sck_mss) { + test->settings->blksize = test->ctrl_sck_mss; + } + else { + test->settings->blksize = DEFAULT_UDP_BLKSIZE; + } + if (test->verbose) { + printf("Setting UDP block size to %d\n", test->settings->blksize); + } + } + + /* + * Regardless of whether explicitly or implicitly set, if the + * block size is larger than the MSS, print a warning. + */ + if (test->ctrl_sck_mss > 0 && + test->settings->blksize > test->ctrl_sck_mss) { + char str[WARN_STR_LEN]; + snprintf(str, sizeof(str), + "UDP block size %d exceeds TCP MSS %d, may result in fragmentation / drops", test->settings->blksize, test->ctrl_sck_mss); + warning(str); + } + } + + return 0; +} + + +int +iperf_client_end(struct iperf_test *test) +{ + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + return -1; + } + struct iperf_stream *sp; + + /* Close all stream sockets */ + SLIST_FOREACH(sp, &test->streams, streams) { + close(sp->socket); + } + + /* show final summary */ + test->reporter_callback(test); + + /* Send response only if no error in server */ + if (test->state > 0) { + if (iperf_set_send_state(test, IPERF_DONE) != 0) + return -1; + } + + /* Close control socket */ + if (test->ctrl_sck >= 0) + close(test->ctrl_sck); + + return 0; +} + + +int +iperf_run_client(struct iperf_test * test) +{ + int startup; + int result = 0; + fd_set read_set, write_set; + struct iperf_time now; + struct timeval* timeout = NULL; + struct iperf_stream *sp; + struct iperf_time last_receive_time; + struct iperf_time diff_time; + struct timeval used_timeout; + iperf_size_t last_receive_blocks; + int64_t t_usecs; + int64_t timeout_us; + int64_t rcv_timeout_us; + int i_errno_save; + + if (NULL == test) + { + iperf_err(NULL, "No test\n"); + return -1; + } + + if (test->logfile) + if (iperf_open_logfile(test) < 0) + return -1; + + if (test->affinity != -1) + if (iperf_setaffinity(test, test->affinity) != 0) + return -1; + + if (test->json_output) + if (iperf_json_start(test) < 0) + return -1; + + if (test->json_output) { + cJSON_AddItemToObject(test->json_start, "version", cJSON_CreateString(version)); + cJSON_AddItemToObject(test->json_start, "system_info", cJSON_CreateString(get_system_info())); + } else if (test->verbose) { + iperf_printf(test, "%s\n", version); + iperf_printf(test, "%s", ""); + iperf_printf(test, "%s\n", get_system_info()); + iflush(test); + } + + /* Start the client and connect to the server */ + if (iperf_connect(test) < 0) + goto cleanup_and_fail; + + /* Begin calculating CPU utilization */ + cpu_util(NULL); + if (test->mode != SENDER) + rcv_timeout_us = (test->settings->rcv_timeout.secs * SEC_TO_US) + test->settings->rcv_timeout.usecs; + else + rcv_timeout_us = 0; + + iperf_time_now(&last_receive_time); // Initialize last time something was received + last_receive_blocks = 0; + + startup = 1; + while (test->state != IPERF_DONE) { + memcpy(&read_set, &test->read_set, sizeof(fd_set)); + memcpy(&write_set, &test->write_set, sizeof(fd_set)); + iperf_time_now(&now); + timeout = tmr_timeout(&now); + + // In reverse active mode client ensures data is received + if (test->state == TEST_RUNNING && rcv_timeout_us > 0) { + timeout_us = -1; + if (timeout != NULL) { + used_timeout.tv_sec = timeout->tv_sec; + used_timeout.tv_usec = timeout->tv_usec; + timeout_us = (timeout->tv_sec * SEC_TO_US) + timeout->tv_usec; + } + /* Cap the maximum select timeout at 1 second */ + if (timeout_us > SEC_TO_US) { + timeout_us = SEC_TO_US; + } + if (timeout_us < 0 || timeout_us > rcv_timeout_us) { + used_timeout.tv_sec = test->settings->rcv_timeout.secs; + used_timeout.tv_usec = test->settings->rcv_timeout.usecs; + } + timeout = &used_timeout; + } + + result = select(test->max_fd + 1, &read_set, &write_set, NULL, timeout); + if (result < 0 && errno != EINTR) { + i_errno = IESELECT; + goto cleanup_and_fail; + } else if (result == 0 && test->state == TEST_RUNNING && rcv_timeout_us > 0) { + /* + * If nothing was received in non-reverse running state + * then probably something got stuck - either client, + * server or network, and test should be terminated./ + */ + iperf_time_now(&now); + if (iperf_time_diff(&now, &last_receive_time, &diff_time) == 0) { + t_usecs = iperf_time_in_usecs(&diff_time); + if (t_usecs > rcv_timeout_us) { + /* Idle timeout if no new blocks received */ + if (test->blocks_received == last_receive_blocks) { + i_errno = IENOMSG; + goto cleanup_and_fail; + } + } + + } + } + + /* See if the test is making progress */ + if (test->blocks_received > last_receive_blocks) { + last_receive_blocks = test->blocks_received; + last_receive_time = now; + } + + if (result > 0) { + if (FD_ISSET(test->ctrl_sck, &read_set)) { + if (iperf_handle_message_client(test) < 0) { + goto cleanup_and_fail; + } + FD_CLR(test->ctrl_sck, &read_set); + } + } + + if (test->state == TEST_RUNNING) { + + /* Is this our first time really running? */ + if (startup) { + startup = 0; + + /* Create and spin up threads */ + pthread_attr_t attr; + if (pthread_attr_init(&attr) != 0) { + i_errno = IEPTHREADATTRINIT; + goto cleanup_and_fail; + } + + SLIST_FOREACH(sp, &test->streams, streams) { + if (pthread_create(&(sp->thr), &attr, &iperf_client_worker_run, sp) != 0) { + i_errno = IEPTHREADCREATE; + goto cleanup_and_fail; + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Thread FD %d created\n", sp->socket); + } + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "All threads created\n"); + } + if (pthread_attr_destroy(&attr) != 0) { + i_errno = IEPTHREADATTRDESTROY; + goto cleanup_and_fail; + } + + } + + /* Run the timers. */ + iperf_time_now(&now); + tmr_run(&now); + + /* + * Is the test done yet? We have to be out of omitting + * mode, and then we have to have fulfilled one of the + * ending criteria, either by times, bytes, or blocks. + * The bytes and blocks tests needs to handle both the + * cases of the client being the sender and the client + * being the receiver. + */ + if ((!test->omitting) && + (test->done || + (test->settings->bytes != 0 && (test->bytes_sent >= test->settings->bytes || + test->bytes_received >= test->settings->bytes)) || + (test->settings->blocks != 0 && (test->blocks_sent >= test->settings->blocks || + test->blocks_received >= test->settings->blocks)))) { + + /* Cancel outstanding sender threads */ + SLIST_FOREACH(sp, &test->streams, streams) { + if (sp->sender) { + int rc; + sp->done = 1; + rc = pthread_cancel(sp->thr); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADCANCEL; + errno = rc; + iperf_err(test, "sender cancel in pthread_cancel - %s", iperf_strerror(i_errno)); + goto cleanup_and_fail; + } + rc = pthread_join(sp->thr, NULL); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADJOIN; + errno = rc; + iperf_err(test, "sender cancel in pthread_join - %s", iperf_strerror(i_errno)); + goto cleanup_and_fail; + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Thread FD %d stopped\n", sp->socket); + } + } + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Sender threads stopped\n"); + } + + /* 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) + goto cleanup_and_fail; + } + } + } + + /* Cancel outstanding receiver threads */ + SLIST_FOREACH(sp, &test->streams, streams) { + if (!sp->sender) { + int rc; + sp->done = 1; + rc = pthread_cancel(sp->thr); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADCANCEL; + errno = rc; + iperf_err(test, "receiver cancel in pthread_cancel - %s", iperf_strerror(i_errno)); + goto cleanup_and_fail; + } + rc = pthread_join(sp->thr, NULL); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADJOIN; + errno = rc; + iperf_err(test, "receiver cancel in pthread_join - %s", iperf_strerror(i_errno)); + goto cleanup_and_fail; + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Thread FD %d stopped\n", sp->socket); + } + } + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Receiver threads stopped\n"); + } + + if (test->json_output) { + if (iperf_json_finish(test) < 0) + return -1; + } else { + iperf_printf(test, "\n"); + iperf_printf(test, "%s", report_done); + } + + iflush(test); + + return 0; + + cleanup_and_fail: + /* Cancel all outstanding threads */ + i_errno_save = i_errno; + SLIST_FOREACH(sp, &test->streams, streams) { + sp->done = 1; + int rc; + rc = pthread_cancel(sp->thr); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADCANCEL; + errno = rc; + iperf_err(test, "cleanup_and_fail in pthread_cancel - %s", iperf_strerror(i_errno)); + } + rc = pthread_join(sp->thr, NULL); + if (rc != 0 && rc != ESRCH) { + i_errno = IEPTHREADJOIN; + errno = rc; + iperf_err(test, "cleanup_and_fail in pthread_join - %s", iperf_strerror(i_errno)); + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "Thread FD %d stopped\n", sp->socket); + } + } + if (test->debug_level >= DEBUG_LEVEL_INFO) { + iperf_printf(test, "All threads stopped\n"); + } + i_errno = i_errno_save; + + iperf_client_end(test); + if (test->json_output) { + cJSON_AddStringToObject(test->json_top, "error", iperf_strerror(i_errno)); + iperf_json_finish(test); + } + iflush(test); + return -1; +} |