iperf3代码分析
iperf3是一个网络性能测试工具，主要用来测试两个主机之间的TCP/UDP带宽，在很多网络链路测试中会使用它的输出结果作为参考。
iperf3的源码地址在：https://github.com/esnet/iperf
目录src是iperf3的具体实现，入口文件是main.c。主函数main中会首先使用iperf_new_test创建一个iperf_test 结构体 test, 再使用iperf_defaults设置一些默认参数，然后使用iperf_parse_arguments解析命令行输入参数并赋值给test, 最后调用run运行这个test。函数run中会根据test->role是server还是client分别执行不同的流程，对应的函数分别是iperf_run_server和iperf_run_client，其中server端支持在后台运行。主要实现代码如下：
int
main(int argc, char **argv)
{
struct iperf_test *test;
/*这里检查对于 64 位数据类型的操作是否是原子性的，如果不支持也打印错误提示。
在 32 位的cpu 上或者 32 位的编译器上是不是就不是原子性的呢？*/
/*
* Atomics check. We prefer to have atomic types (which is
* basically on any compiler supporting C11 or better). If we
* don't have them, we try to approximate the type we need with a
* regular integer, but complain if they're not lock-free. We only
* know how to check this on GCC. GCC on CentOS 7 / RHEL 7 is the
* targeted use case for these check.
*/
#ifndef HAVE_STDATOMIC_H
#ifdef __GNUC__
if (! __atomic_always_lock_free (sizeof (u_int64_t), 0)) {
#endif // __GNUC__
fprintf(stderr, "Warning: Cannot guarantee lock-free operation with 64-bit data types\n");
#ifdef __GNUC__
}
#endif // __GNUC__
#endif // HAVE_STDATOMIC_H
// 设置进程的 cpu 亲核性
// XXX: Setting the process affinity requires root on most systems.
// Is this a feature we really need?
#ifdef TEST_PROC_AFFINITY
/* didn't seem to work.... */
/*
* increasing the priority of the process to minimise packet generation
* delay
*/
int rc = setpriority(PRIO_PROCESS, 0, -15);
if (rc < 0) {
perror("setpriority:");
fprintf(stderr, "setting priority to valid level\n");
rc = setpriority(PRIO_PROCESS, 0, 0);
}
/* setting the affinity of the process */
cpu_set_t cpu_set;
int affinity = -1;
int ncores = 1;
sched_getaffinity(0, sizeof(cpu_set_t), &cpu_set);
if (errno)
perror("couldn't get affinity:");
if ((ncores = sysconf(_SC_NPROCESSORS_CONF)) <= 0)
err("sysconf: couldn't get _SC_NPROCESSORS_CONF");
CPU_ZERO(&cpu_set);
CPU_SET(affinity, &cpu_set);
if (sched_setaffinity(0, sizeof(cpu_set_t), &cpu_set) != 0)
err("couldn't change CPU affinity");
#endif
// 创建test并设置默认值
test = iperf_new_test();
if (!test)
iperf_errexit(NULL, "create new test error - %s", iperf_strerror(i_errno));
iperf_defaults(test); /* sets defaults */
// 解析命令行报数
if (iperf_parse_arguments(test, argc, argv) < 0) {
iperf_err(test, "parameter error - %s", iperf_strerror(i_errno));
fprintf(stderr, "\n");
usage();
exit(1);
}
// 运行测试，直到到达指定的退出条件或者被用户中断
if (run(test) < 0)
iperf_errexit(test, "error - %s", iperf_strerror(i_errno));
// 释放测试资源 test
iperf_free_test(test);
return 0;
}
static int
run(struct iperf_test *test)
{
// 设置信号处理函数，注意实现方式
/* Termination signals. */
iperf_catch_sigend(sigend_handler);
if (setjmp(sigend_jmp_buf))
iperf_got_sigend(test, signed_sig);
// 忽略SIGPIPE 信号
/* Ignore SIGPIPE to simplify error handling */
signal(SIGPIPE, SIG_IGN);
switch (test->role) {
case 's':
if (test->daemon) {
// 通过 daemon 设置服务器运行模式为后台驻留方式
int rc;
rc = daemon(1, 0);
if (rc < 0) {
i_errno = IEDAEMON;
iperf_errexit(test, "error - %s", iperf_strerror(i_errno));
}
}
if (iperf_create_pidfile(test) < 0) {
i_errno = IEPIDFILE;
iperf_errexit(test, "error - %s", iperf_strerror(i_errno));
}
for (;;) {
// 服务器事常驻模式，正常用户不通过 ctrl+c 是不会退出的，因此采用循环方式执行每个测试
int rc;
// 运行服务器模式
rc = iperf_run_server(test);
test->server_last_run_rc = rc;
if (rc < 0) {
iperf_err(test, "error - %s", iperf_strerror(i_errno));
if (test->json_output) {
if (iperf_json_finish(test) < 0)
return -1;
}
iflush(test);
if (rc < -1) {
iperf_errexit(test, "exiting");
}
}
iperf_reset_test(test);
if (iperf_get_test_one_off(test) && rc != 2) {
/* Authentication failure doesn't count for 1-off test */
if (rc < 0 && i_errno == IEAUTHTEST) {
continue;
}
break;
}
}
iperf_delete_pidfile(test);
break;
case 'c':
if (iperf_create_pidfile(test) < 0) {
i_errno = IEPIDFILE;
iperf_errexit(test, "error - %s", iperf_strerror(i_errno));
}
// 运行客户端测试
if (iperf_run_client(test) < 0)
iperf_errexit(test, "error - %s", iperf_strerror(i_errno));
iperf_delete_pidfile(test);
break;
default:
usage();
break;
}
// 服务器和客户端都支持创建pidfile，用来做重复性检查
iperf_catch_sigend(SIG_DFL);
signal(SIGPIPE, SIG_DFL);
return 0;
}
iperf_run_server在iperf_server_api.c中实现。
我们先看iperf_run_server的实现代码：
int
iperf_run_server(struct iperf_test *test)
{
int result, s;
int send_streams_accepted, rec_streams_accepted;
int streams_to_send = 0, streams_to_rec = 0;
#if defined(HAVE_TCP_CONGESTION)
int saved_errno;
#endif /* HAVE_TCP_CONGESTION */
fd_set read_set, write_set;
struct iperf_stream *sp;
struct iperf_time now;
struct iperf_time last_receive_time;
struct iperf_time diff_time;
struct timeval* timeout;
struct timeval used_timeout;
iperf_size_t last_receive_blocks;
int flag;
int64_t t_usecs;
int64_t timeout_us;
int64_t rcv_timeout_us;
int32_t err;
if (test->logfile) {
if (iperf_open_logfile(test) < 0)
return -2;
}
if (test->affinity != -1) {
if (iperf_setaffinity(test, test->affinity) != 0) {
cleanup_server(test);
return -2;
}
}
if (test->json_output) {
if (iperf_json_start(test) < 0) {
cleanup_server(test);
return -2;
}
}
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);
}
// 打开socket并侦听，这里是指控制通道
// Open socket and listen
if (iperf_server_listen(test) < 0) {
cleanup_server(test);
return -2;
}
// 更新最后接收时间和统计信息 iperf_time_now(&last_receive_time); // Initialize last time something was received
last_receive_blocks = 0;
// 开始启动状态机，设置为IPERF_START状态
iperf_set_test_state(test, IPERF_START);
// 发送流接收统计，接收流接收统计清零
send_streams_accepted = 0;
rec_streams_accepted = 0;
// 计算接收超时时间，单位为微秒
rcv_timeout_us = (test->settings->rcv_timeout.secs * SEC_TO_US) + test->settings->rcv_timeout.usecs;
// 如果测试的状态不是IPERF_DONE则一直循环执行
while (test->state != IPERF_DONE) {
// Check if average transfer rate was exceeded (condition set in the callback routines)
// 这里查看是否查过了显示速率，如果超出则发送错误信息并退出
if (test->bitrate_limit_exceeded) {
i_errno = IETOTALRATE;
if (iperf_set_send_state(test, SERVER_ERROR) != 0) {
cleanup_server(test);
return -1;
}
// 发送错误码i_errno给对端
err = htonl(i_errno);
if (Nwrite(test->ctrl_sck, (char*) &err, sizeof(err), Ptcp) < 0) {
cleanup_server(test);
i_errno = IECTRLWRITE;
return -1;
}
// 发送错误码errno给对端，注意和上面的不一样，是两种不同的错误码
err = htonl(errno);
if (Nwrite(test->ctrl_sck, (char*) &err, sizeof(err), Ptcp) < 0) {
cleanup_server(test);
i_errno = IECTRLWRITE;
return -1;
}
cleanup_server(test);
return -1;
} // 这里对应的是if (test->bitrate_limit_exceeded)
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);
// Ensure select() will timeout to allow handling error cases that require server restart
// 根据状态的不同计算select的超时时间
if (test->state == IPERF_START) { // In idle mode server may need to restart
if (timeout == NULL && test->settings->idle_timeout > 0) {
used_timeout.tv_sec = test->settings->idle_timeout;
used_timeout.tv_usec = 0;
timeout = &used_timeout;
}
} else if (test->mode != SENDER) { // In non-reverse active mode server ensures data is received
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;
}
// select等待事件的发生
result = select(test->max_fd + 1, &read_set, &write_set, NULL, timeout);
if (result < 0 && errno != EINTR) {
cleanup_server(test);
i_errno = IESELECT;
return -1;
} else if (result == 0) {
/*
* If nothing was received during the specified time (per
* state) then probably something got stuck either at the
* client, server or network, and test should be forced to
* end.
*/
// 这里需要结束测试
iperf_time_now(&now);
t_usecs = 0;
if (iperf_time_diff(&now, &last_receive_time, &diff_time) == 0) {
t_usecs = iperf_time_in_usecs(&diff_time);
/* We're in the state where we're still accepting connections */
if (test->state == IPERF_START) {
if (test->settings->idle_timeout > 0 && t_usecs >= test->settings->idle_timeout * SEC_TO_US) {
test->server_forced_idle_restarts_count += 1;
if (test->debug)
printf("Server restart (#%d) in idle state as no connection request was received for %d sec\n",
test->server_forced_idle_restarts_count, test->settings->idle_timeout);
cleanup_server(test);
if ( iperf_get_test_one_off(test) ) {
if (test->debug)
printf("No connection request was received for %d sec in one-off mode; exiting.\n",
test->settings->idle_timeout);
exit(0);
}
return 2;
}
}
/*
* Running a test. If we're receiving, be sure we're making
* progress (sender hasn't died/crashed).
*/
else if (test->mode != SENDER && t_usecs > rcv_timeout_us) {
/* Idle timeout if no new blocks received */
if (test->blocks_received == last_receive_blocks) {
test->server_forced_no_msg_restarts_count += 1;
i_errno = IENOMSG;
if (iperf_get_verbose(test))
iperf_err(test, "Server restart (#%d) during active test due to idle timeout for receiving data",
test->server_forced_no_msg_restarts_count);
cleanup_server(test);
return -1;
}
}
}
}
/* See if the test is making progress */
// 判断测试是否改继续进行，if条件成立说明接收的个数已经超过了上次接收的个数，更新上次接收个数和接收时间
if (test->blocks_received > last_receive_blocks) {
last_receive_blocks = test->blocks_received;
last_receive_time = now;
}
if (result > 0) {
if (FD_ISSET(test->listener, &read_set)) {
// 这里处理数据通道，通过accept接收服务器请求，格局测试模式设置发送和接收的stream数量
if (test->state != CREATE_STREAMS) {
if (iperf_accept(test) < 0) {
cleanup_server(test);
return -1;
}
FD_CLR(test->listener, &read_set);
// Set streams number
if (test->mode == BIDIRECTIONAL) {
streams_to_send = test->num_streams;
streams_to_rec = test->num_streams;
} else if (test->mode == RECEIVER) {
streams_to_rec = test->num_streams;
streams_to_send = 0;
} else {
streams_to_send = test->num_streams;
streams_to_rec = 0;
}
}
}
if (FD_ISSET(test->ctrl_sck, &read_set)) {
// 这里处理控制通道，通过iperf_handle_message_server处理消息，驱动测试状态机
if (iperf_handle_message_server(test) < 0) {
cleanup_server(test);
return -1;
}
FD_CLR(test->ctrl_sck, &read_set);
}
if (test->state == CREATE_STREAMS) {
// 当前是CREATE_STREAMS状态，创建测试stream
if (FD_ISSET(test->prot_listener, &read_set)) {
// 调用协议对应的access回调
if ((s = test->protocol->accept(test)) < 0) {
cleanup_server(test);
return -1;
}
/* apply other common socket options */
if (iperf_common_sockopts(test, s) < 0)
{
cleanup_server(test);
return -1;
}
if (!is_closed(s)) {
// 下面是根据编译选项，分别设置tcp的超时时间，拥塞控制
#if defined(HAVE_TCP_USER_TIMEOUT)
if (test->protocol->id == Ptcp) {
int opt;
if ((opt = test->settings->snd_timeout)) {
if (setsockopt(s, IPPROTO_TCP, TCP_USER_TIMEOUT, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
cleanup_server(test);
errno = saved_errno;
i_errno = IESETUSERTIMEOUT;
return -1;
}
}
}
#endif /* HAVE_TCP_USER_TIMEOUT */
#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) {
/*
* ENOENT means we tried to set the
* congestion algorithm but the algorithm
* specified doesn't exist. This can happen
* if the client and server have different
* congestion algorithms available. In this
* case, print a warning, but otherwise
* continue.
*/
if (errno == ENOENT) {
warning("TCP congestion control algorithm not supported");
}
else {
saved_errno = errno;
close(s);
cleanup_server(test);
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);
cleanup_server(test);
errno = saved_errno;
i_errno = IESETCONGESTION;
return -1;
}
/*
* If not the first connection, discard prior
* congestion algorithm name so we don't leak
* duplicated strings. We probably don't need
* the old string anyway.
*/
if (test->congestion_used != NULL) {
free(test->congestion_used);
}
// 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 */
// 这个根据发送和接收的预期个数来更新flag,以及对应的变量
if (rec_streams_accepted != streams_to_rec) {
flag = 0;
++rec_streams_accepted;
} else if (send_streams_accepted != streams_to_send) {
flag = 1;
++send_streams_accepted;
}
if (flag != -1) {
// flag不等于说明需要创建stream
sp = iperf_new_stream(test, s, flag);
if (!sp) {
cleanup_server(test);
return -1;
}
if (s > test->max_fd) test->max_fd = s;
if (test->on_new_stream)
test->on_new_stream(sp);
// flag置-1，以便后续不会重复创建stream
flag = -1;
}
}
FD_CLR(test->prot_listener, &read_set);
}
// 这里判断已经接收的数量和配置数量是否相同，同时已经发送的数量和配置的数量是否相同
if (rec_streams_accepted == streams_to_rec && send_streams_accepted == streams_to_send) {
if (test->protocol->id != Ptcp) {
// 非tcp协议，直接close对应的侦听socket
FD_CLR(test->prot_listener, &test->read_set);
close(test->prot_listener);
test->prot_listener = -1;
} else {
if (test->no_delay || test->settings->mss || test->settings->socket_bufsize) {
// 如果是tcp，且设置了no_delay,mss或者socket_bufsize选项，除了close侦听的socket之外，还要创建一个tcp链接？
FD_CLR(test->listener, &test->read_set);
close(test->listener);
test->listener = -1;
if ((s = netannounce(test->settings->domain, Ptcp, test->bind_address, test->bind_dev, test->server_port)) < 0) {
cleanup_server(test);
i_errno = IELISTEN;
return -1;
}
test->listener = s;
FD_SET(test->listener, &test->read_set);
if (test->listener > test->max_fd) test->max_fd = test->listener;
}
}
test->prot_listener = -1;
/* Ensure that total requested data rate is not above limit */
// total_requested_rate 是stream个数，设置的速率，以及模式的乘积(双向乘以2，单向乘以1)
iperf_size_t total_requested_rate = test->num_streams * test->settings->rate * (test->mode == BIDIRECTIONAL? 2 : 1);
if (test->settings->bitrate_limit > 0 && total_requested_rate > test->settings->bitrate_limit) {
// 执行到这里，说明配置了bitrate_limit，且根据上述stream计算出的速率大于这个bitrate_limit，配置不合理，返回错误
if (iperf_get_verbose(test))
iperf_err(test, "Client total requested throughput rate of %" PRIu64 " bps exceeded %" PRIu64 " bps limit",
total_requested_rate, test->settings->bitrate_limit);
cleanup_server(test);
i_errno = IETOTALRATE;
return -1;
}
// Begin calculating CPU utilization
// 计算CPU利用率，后面会用到
cpu_util(NULL);
// 通知对端启动测试
if (iperf_set_send_state(test, TEST_START) != 0) {
cleanup_server(test);
return -1;
}
// 初始化测试
if (iperf_init_test(test) < 0) {
cleanup_server(test);
return -1;
}
// 创建测试定时器
if (create_server_timers(test) < 0) {
cleanup_server(test);
return -1;
}
// 创建测试omit定时器
if (create_server_omit_timer(test) < 0) {
cleanup_server(test);
return -1;
}
// 如果测试模式不等于只接收，说明有发送的角色，创建发送定时器
if (test->mode != RECEIVER)
if (iperf_create_send_timers(test) < 0) {
cleanup_server(test);
return -1;
}
// 设置对端状态为运行
if (iperf_set_send_state(test, TEST_RUNNING) != 0) {
cleanup_server(test);
return -1;
}
// 开始创建测试线程，执行具体的接收和发送测试
/* Create and spin up threads */
pthread_attr_t attr;
if (pthread_attr_init(&attr) != 0) {
i_errno = IEPTHREADATTRINIT;
cleanup_server(test);
};
// 遍历每个stream并创建对应的pthread线程，执行函数iperf_server_worker_run
SLIST_FOREACH(sp, &test->streams, streams) {
if (pthread_create(&(sp->thr), &attr, &iperf_server_worker_run, sp) != 0) {
i_errno = IEPTHREADCREATE;
cleanup_server(test);
return -1;
}
sp->thread_created = 1;
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;
cleanup_server(test);
};
/* Reset receive-progress baseline after workers are ready */
last_receive_blocks = test->blocks_received;
iperf_time_now(&last_receive_time);
}
}
}
// result=0或者超时时间为0， 运行定时器tmr_run
if (result == 0 ||
(timeout != NULL && timeout->tv_sec == 0 && timeout->tv_usec == 0)) {
/* Run the timers. */
iperf_time_now(&now);
tmr_run(&now);
}
}
// 测试完成后，支持json输出
if (test->json_output) {
if (iperf_json_finish(test) < 0)
return -1;
}
// 测试完成后，清理资源
iflush(test);
cleanup_server(test);
if (test->server_affinity != -1)
if (iperf_clearaffinity(test) != 0)
return -1;
return 0;
}
在iperf_run_server中会单独启动任务iperf_server_worker_run，实现代码如下：
void *
iperf_server_worker_run(void *s) {
struct iperf_stream *sp = (struct iperf_stream *) s;
struct iperf_test *test = sp->test;
/* Blocking signal to make sure that signal will be handled by main thread */
sigset_t set;
sigemptyset(&set);
#ifdef SIGTERM
sigaddset(&set, SIGTERM);
#endif
#ifdef SIGHUP
sigaddset(&set, SIGHUP);
#endif
#ifdef SIGINT
sigaddset(&set, SIGINT);
#endif
if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0) {
i_errno = IEPTHREADSIGMASK;
goto cleanup_and_fail;
}
/* Allow this thread to be cancelled even if it's in a syscall */
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
// 当test->done 和 sp->done都不是完成状态的时候，当sender存在执行iperf_send_mt,否则执行iperf_recv_mt
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;
}
iperf_run_client在iperf_client_api.c中实现。
我们看看iperf_run_client的实现代码：
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 */
// 启动和server之间的连接控制通道
if (iperf_connect(test) < 0)
goto cleanup_and_fail;
/* Begin calculating CPU utilization */
cpu_util(NULL);
// 计算接收超时时间，转为为微秒，接收端才需要，发送端为0
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);
// 获取tmr定时器的超时时间
timeout = tmr_timeout(&now);
// In reverse active mode client ensures data is received
// 计算selet的超时时间，条件是当前是测试运行状态且设置的超时时间大于0(如果不修改，默认超时时间是120000毫秒：DEFAULT_NO_MSG_RCVD_TIMEOUT)
if (test->state == TEST_RUNNING && rcv_timeout_us > 0) {
timeout_us = -1; //默认值，表示无tmr定时器超时
if (timeout != NULL) {
// 如果定时器有超时时间，首先使用定时器的时间，一种格式是sec+usec(select使用)，一种格式是usec(判断是否超过1秒)
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) {
// 无tmr定时器超时，或者定时器超时时间大于接收超时时间(定时器超时更长，使用短的时间作为超时时间),使用配置的超时时间
used_timeout.tv_sec = test->settings->rcv_timeout.secs;
used_timeout.tv_usec = test->settings->rcv_timeout.usecs;
}
// TODO 如果上面的if流程都没走到，used_timeout是多少，没有看到初始化的值。
timeout = &used_timeout;
}
// vxworks和linux对于select的最后一个参数timeout的配置略有不同，这里分别处理。
// 比如超时时间为0的时候，vxworks会先延时一个tick,然后再select返回，而linux下直接返回
// 另外，linux超时时间内返回后会修改timeout的超时时间而vxworks不会
#if (defined(__vxworks)) || (defined(__VXWORKS__))
if (timeout != NULL && timeout->tv_sec == 0 && timeout->tv_usec == 0) {
taskDelay (1);
}
result = select(test->max_fd + 1,
&read_set,
(test->state == TEST_RUNNING && !test->reverse) ? &write_set : NULL,
NULL,
timeout);
#else
result = select(test->max_fd + 1, &read_set, &write_set, NULL, timeout);
#endif // __vxworks or __VXWORKS__
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./
*/
// result返回0表示超时时间到，但是没有fd可以读写，表示无网络数据
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) {
// 处理控制通道接收报文，客户端只有这个控制通道的报文要处理，没有listener等socekt。
if (FD_ISSET(test->ctrl_sck, &read_set)) {
// 客户端消息处理回调iperf_handle_message_client，主要处理控制信息状态机
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;
}
sp->thread_created = 1;
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;
}
} // 完成首次启动处理
// 运行定时器，注意iperf的定时器是自定义实现的，且在任务创建完成后再运行
/* Run the timers. */
iperf_time_now(&now);
tmr_run(&now);
// 测试完成判断，包括done标记，以及发送和接收字节个数是否达到预期，bloc发送或者接收的数量是否达到设置的值
/*
* 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 the periodic timers. As the timers are not re-set in this stage their expiration time
* is in the past, so without the cancellation the select() timeout is set to 0,
* and it enters a loop while waiting the exchanged results, etc.
*/
if (test->stats_timer != NULL) {
tmr_cancel(test->stats_timer);
test->stats_timer = NULL;
}
if (test->reporter_timer != NULL) {
tmr_cancel(test->reporter_timer);
test->reporter_timer = NULL;
}
/* Cancel outstanding sender threads */
SLIST_FOREACH(sp, &test->streams, streams) {
if (sp->sender) {
// 取消发送线程
int rc;
sp->done = 1;
if (sp->thread_created == 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);
}
sp->thread_created = 0;
}
}
}
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;
}
}
}
// 完成上述发送流程后，再取消接收线程,注意sender和receiver角色是互斥的，只能是其中一个
// 如果是单向测试，server和client之间只有1条stream，如果是双向测试，需要创建两条stream，每个角色都有sender和receiver的stream。
/* Cancel outstanding receiver threads */
SLIST_FOREACH(sp, &test->streams, streams) {
if (!sp->sender) {
int rc;
sp->done = 1;
if (sp->thread_created == 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);
}
sp->thread_created = 0;
}
}
}
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) {
if (sp->done) {
continue;
}
sp->done = 1;
int rc;
if (sp->thread_created == 1) {
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);
}
sp->thread_created = 0;
}
}
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;
}
、
iperf_run_client会单独创建任务iperf_client_work_run，代码实现如下：
void *
iperf_client_worker_run(void *s) {
struct iperf_stream *sp = (struct iperf_stream *) s;
struct iperf_test *test = sp->test;
/* Blocking signal to make sure that signal will be handled by main thread */
sigset_t set;
sigemptyset(&set);
#ifdef SIGTERM
sigaddset(&set, SIGTERM);
#endif
#ifdef SIGHUP
sigaddset(&set, SIGHUP);
#endif
#ifdef SIGINT
sigaddset(&set, SIGINT);
#endif
if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0) {
i_errno = IEPTHREADSIGMASK;
goto cleanup_and_fail;
}
/* Allow this thread to be cancelled even if it's in a syscall */
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, 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;
}
如果你仔细看一下，iperf_server_work_run和iperf_client_work_run的代码，发现他们其实是一模一样的，就是函数名字不一样，连代码中的注释都是一样的，其实这两个函数可以合并成一个，叫做iperf_work_run，不区分server和client。之所以可以实现这种效果，因为上面这两个函数中分别调用iperf_send_mt和iperf_recv_mt实现报文发送和接收，让我们看看这两个函数的实现代码。
int
iperf_send_mt(struct iperf_stream *sp)
{
register int multisend, r, message_sent;
register struct iperf_test *test = sp->test;
struct iperf_time now;
int throttle_check_per_message;
#if defined(HAVE_CLOCK_NANOSLEEP) || defined(HAVE_NANOSLEEP)
int throttle_check;
#endif /* HAVE_CLOCK_NANOSLEEP, HAVE_NANOSLEEP */
/* Can we do multisend mode? */
// multisend是什么功能
// burst默认为0，除非用户通过-b进行了设置
// rate等于0表示用户未设置速率，使用multisend默认值，默认为10
// 如果rate不是0，说明用户配置了速率，这时候multisend固定为0
if (test->settings->burst != 0)
multisend = test->settings->burst;
else if (test->settings->rate == 0)
multisend = test->multisend;
else
multisend = 1; /* nope */
// rate是速率，如果非0表示用户设置了预期测试速率，需要进行周期性检查
/* Should bitrate throttle be checked for every send */
#if defined(HAVE_CLOCK_NANOSLEEP) || defined(HAVE_NANOSLEEP)
if (test->settings->rate != 0) {
throttle_check = 1;
if (test->settings->burst == 0)
throttle_check_per_message = 1;
else
throttle_check_per_message = 0;
} else {
throttle_check = 0;
throttle_check_per_message = 0;
}
#else /* !HAVE_CLOCK_NANOSLEEP && !HAVE_NANOSLEEP */
throttle_check_per_message = test->settings->rate != 0 && test->settings->burst == 0;
#endif /* HAVE_CLOCK_NANOSLEEP, HAVE_NANOSLEEP */
// 看这里，如果multisend不是0，是进不了下面的snd回调的
for (message_sent = 0; sp->green_light && multisend > 0; --multisend) {
// XXX If we hit one of these ending conditions maybe
// want to stop even trying to send something?
if (multisend > 1 && test->settings->bytes != 0 && test->bytes_sent >= test->settings->bytes)
break;
if (multisend > 1 && test->settings->blocks != 0 && test->blocks_sent >= test->settings->blocks)
break;
if ((r = sp->snd(sp)) < 0) {
if (r == NET_SOFTERROR)
break;
i_errno = IESTREAMWRITE;
return r;
}
test->bytes_sent += r;
if (!sp->pending_size)
++test->blocks_sent;
if (throttle_check_per_message) {
if (message_sent == 0)
iperf_time_now(&now);
iperf_check_throttle(sp, &now);
}
message_sent = 1;
}
#if defined(HAVE_CLOCK_NANOSLEEP) || defined(HAVE_NANOSLEEP)
/* Should check if green light can be set, as pacing timer is not supported in this case */
if (throttle_check && (!throttle_check_per_message || message_sent == 0)) {
#else /* !HAVE_CLOCK_NANOSLEEP && !HAVE_NANOSLEEP */
if (!throttle_check_per_message || message_sent == 0) { /* Throttle check if was not checked for each send */
#endif /* HAVE_CLOCK_NANOSLEEP, HAVE_NANOSLEEP */
iperf_time_now(&now);
iperf_check_throttle(sp, &now);
}
return 0;
}
int
iperf_recv_mt(struct iperf_stream *sp)
{
int r;
struct iperf_test *test = sp->test;
// 执行rcv回调，没有条件判断
if ((r = sp->rcv(sp)) < 0) {
i_errno = IESTREAMREAD;
return r;
}
/* Collect statistics only if receive did not timeout (e.g. `Nread()` may timeout).
* This is also important for `--rcv-timeout` to work properly.
*/
if (r > 0) {
test->bytes_received += r;
++test->blocks_received;
}
return 0;
}
可以看到iperf_send_mt和iperf_recv_mt中分别调用了iperf_stream的成员函数snd和rcv实现发包和收包。那么snd和rcv回调又指向谁呢？在iperf_new_stream函数中，会分别给snd和rcv赋值，而来源是test->protocol->send和test->protocol->recv。
struct iperf_stream *
iperf_new_stream(struct iperf_test *test, int s, int sender)
{
struct iperf_stream *sp;
int ret = 0;
int size;
char template[1024];
if (test->tmp_template) {
snprintf(template, sizeof(template) / sizeof(char), "%s", test->tmp_template);
} else {
//find the system temporary dir *unix, windows, cygwin support
char* tempdir = getenv("TMPDIR");
if (tempdir == 0){
tempdir = getenv("TEMP");
}
if (tempdir == 0){
tempdir = getenv("TMP");
}
if (tempdir == 0){
#if defined(__ANDROID__)
tempdir = "/data/local/tmp";
#else
tempdir = "/tmp";
#endif
}
snprintf(template, sizeof(template) / sizeof(char), "%s/iperf3.XXXXXX", tempdir);
}
sp = (struct iperf_stream *) malloc(sizeof(struct iperf_stream));
if (!sp) {
i_errno = IECREATESTREAM;
goto err_exit;
}
memset(sp, 0, sizeof(struct iperf_stream));
// sender代表发送或者接收的角色，等于1代表发送端(或者非0)，等于0为接收端。
sp->sender = sender;
sp->test = test;
sp->settings = test->settings;
sp->result = (struct iperf_stream_result *) malloc(sizeof(struct iperf_stream_result));
if (!sp->result) {
i_errno = IECREATESTREAM;
goto err_exit_free_sp;
}
memset(sp->result, 0, sizeof(struct iperf_stream_result));
TAILQ_INIT(&sp->result->interval_results);
/* Create and randomize the buffer */
// mkstemp创建一个临时文件，用于后续报文发送或者接收(接收端应该不需要吧)
sp->buffer_fd = mkstemp(template);
if (sp->buffer_fd == -1) {
i_errno = IECREATESTREAM;
goto err_exit_free_result;
}
// unlink没有删除文件，只是删除了文件名，只有当改文件没有被应用的时候才被删除，也就是close的时候才删除
if (unlink(template) < 0) {
i_errno = IECREATESTREAM;
goto err_exit_close_buffer;
}
size = test->settings->blksize;
if (test->protocol->id == Pudp && test->settings->gso && (size < test->settings->gso_bf_size))
size = test->settings->gso_bf_size;
if (test->protocol->id == Pudp && test->settings->gro && (size < test->settings->gro_bf_size))
size = test->settings->gro_bf_size;
if (sp->test->debug)
printf("Buffer %d bytes\n", size);
if (ftruncate(sp->buffer_fd, size) < 0) {
i_errno = IECREATESTREAM;
goto err_exit_close_buffer;
}
// mmap内存映射，便于使用标准socket接口发送和接收报文
sp->buffer = (char *) mmap(NULL, size, PROT_READ|PROT_WRITE, MAP_SHARED, sp->buffer_fd, 0);
if (sp->buffer == MAP_FAILED) {
i_errno = IECREATESTREAM;
goto err_exit_close_buffer;
}
sp->pending_size = 0;
/* Set socket */
sp->socket = s;
sp->snd = test->protocol->send;
sp->rcv = test->protocol->recv;
if (test->diskfile_name != (char*) 0) {
sp->diskfile_fd = open(test->diskfile_name, sender ? O_RDONLY : (O_WRONLY|O_CREAT|O_TRUNC), S_IRUSR|S_IWUSR);
if (sp->diskfile_fd == -1) {
i_errno = IEFILE;
goto err_exit_munmap_buffer;
}
sp->snd2 = sp->snd;
sp->snd = diskfile_send;
sp->rcv2 = sp->rcv;
sp->rcv = diskfile_recv;
} else
sp->diskfile_fd = -1;
/* Initialize stream */
if (test->repeating_payload)
fill_with_repeating_pattern(sp->buffer, test->settings->blksize);
else
ret = readentropy(sp->buffer, test->settings->blksize);
if ((ret < 0) || (iperf_init_stream(sp, test) < 0)) {
goto err_exit_close_diskfile;
}
// 添加test对应的strems的记录中
iperf_add_stream(test, sp);
return sp;
err_exit_close_diskfile:
/* The file may not be open because it depends on user given -F option. */
if (sp->diskfile_fd >= 0) {
close(sp->diskfile_fd);
}
err_exit_munmap_buffer:
munmap(sp->buffer, sp->test->settings->blksize);
err_exit_close_buffer:
close(sp->buffer_fd);
err_exit_free_result:
free(sp->result);
err_exit_free_sp:
free(sp);
err_exit:
return NULL;
}
在函数iperf_defaults中创建了3个协议tcp, udp和sctp, 在iperf_parse_arguments中会根据用户输入解析协议类型，然后调用set_protocol设置test中的protocol成员。