iperf3代码分析(2)
上一篇文章分析了iperf的源码，把整个iperf的代码都贴了上来，并做了代码分析的注释。在走读代码的过程中，发现iperf看似很简单的网络测速工具，代码中却有很多实现细节值得分析，所以延续上一篇文章，让我们再具体分析一下iperf的实现细节。
上一篇文章主要关注iperf的框架实现，包括server端和client端。程序启动后根据命令行参数分析执行server或者client的初始化流程，然后运行主处理循环，根据strem个数创建对应个数的数据收发线程，在数据收发线程中执行实际的数据发送和接收，主处理循环只处理控制通道的报文收发，以及数据平面的建立和销毁。
本篇文章让我们关注一下server和client的状态机是如何运转的，也就是如何从初始化到数据收发，再到测试结束，输出结果的。
对于server端来说，在iperf_run_server中的开始处会调用iperf_server_listen创建控制通道侦听socket。然后调用iperf_set_test_state(test, IPERF_START)设置server为启动状态IPERF_START。因为server必须在client之前启动，所以此时client还没有启动，只需要设置本端状态就可以了，因此用的接口是iperf_set_test_state。当进入while (test->state != IPERF_DONE)循环体之后，select函数返回有数据到来之后，条件(FD_ISSET(test->listener, &read_set))成立，且(test->state != CREATE_STREAMS)之后，则调用iperf_accept创建控制通道连接，让我们看看具体实现。
int
iperf_accept(struct iperf_test *test)
{
int s;
int ret = -1;
signed char rbuf = ACCESS_DENIED;
socklen_t len;
struct sockaddr_storage addr;
len = sizeof(addr);
// 通过accept创建实际控制通道socket
if ((s = accept(test->listener, (struct sockaddr *) &addr, &len)) < 0) {
i_errno = IEACCEPT;
return ret;
}
// 如果为-1表示未创建过，执行初始化流程，ctrl_sck即为控制通道实际socket
if (test->ctrl_sck == -1) {
/* Server free, accept new client */
test->ctrl_sck = s;
// set TCP_NODELAY for lower latency on control messages
int flag = 1;
// 控制通道使能TCP_NODELAY功能，即禁用 Nagle 算法，降低报文延迟
if (setsockopt(test->ctrl_sck, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int))) {
i_errno = IESETNODELAY;
goto error_handling;
}
#if defined(HAVE_TCP_USER_TIMEOUT)
int opt;
// TCP_USER_TIMEOUT含义：已发送但未收到 ACK 的数据包允许等待的最大时长，超时之后recv/send返回ETIMEOUT
if ((opt = test->settings->snd_timeout)) {
if (setsockopt(s, IPPROTO_TCP, TCP_USER_TIMEOUT, &opt, sizeof(opt)) < 0) {
i_errno = IESETUSERTIMEOUT;
goto error_handling;
}
}
#endif /* HAVE_TCP_USER_TIMEOUT */
#if defined (HAVE_TCP_KEEPALIVE)
// Set Control Connection TCP Keepalive (especially useful for long UDP test sessions)
// 设置tcp保活，对于长时间的udp测试这个比较有用，因为如果链路断了控制通道快速感知到，避免无意义的测试。
if (iperf_set_control_keepalive(test) < 0)
return -1;
#endif //HAVE_TCP_KEEPALIVE
// 从通道读取cookie信息，也就是client发给server的标识，让我们分析client的时候看看是如何发送的
if (Nread(test->ctrl_sck, test->cookie, COOKIE_SIZE, Ptcp) != COOKIE_SIZE) {
/*
* Note this error covers both the case of a system error
* or the inability to read the correct amount of data
* (i.e. timed out).
*/
i_errno = IERECVCOOKIE;
goto error_handling;
}
// 把ctrl_sck加入主处理循环的侦听fd_set和max_fd中。
FD_SET(test->ctrl_sck, &test->read_set);
if (test->ctrl_sck > test->max_fd) test->max_fd = test->ctrl_sck;
// 设置server状态为PARAM_EXCHANGE并把这个状态发送给client。
if (iperf_set_send_state(test, PARAM_EXCHANGE) != 0)
goto error_handling;
// 交换配置参数，其实是把客户端配置的参数发送给服务器端
if (iperf_exchange_parameters(test) < 0)
goto error_handling;
// 设置cpu亲和性
if (test->server_affinity != -1) {
if (iperf_setaffinity(test, test->server_affinity) != 0)
goto error_handling;
}
// 执行测试on_connect的回调函数
if (test->on_connect)
test->on_connect(test);
} else {
// 如果已经创立了连接，说明当前有client在和server进行测试，不再接收其他client的请求，发送ACCESS_DENIED错误消息，客户端显示对应错误提示。
/*
* Don't try to read from the socket. It could block an ongoing test.
* Just send ACCESS_DENIED.
* Also, if sending failed, don't return an error, as the request is not related
* to the ongoing test, and returning an error will terminate the test.
*/
if (Nwrite(s, (char*) &rbuf, sizeof(rbuf), Ptcp) < 0) {
if (test->debug)
printf("failed to send ACCESS_DENIED to an unsolicited connection request during active test\n");
} else {
if (test->debug)
printf("successfully sent ACCESS_DENIED to an unsolicited connection request during active test\n");
}
close(s);
}
return 0;
error_handling:
return ret;
}
上面的流程中设置完本地和对端的PARAM_EXCHANGE状态后，最重要的实现就是iperf_exchange_parameters，让我们看看它的实现。
/**
* iperf_exchange_parameters - handles the param_Exchange part for client
*
*/
int
iperf_exchange_parameters(struct iperf_test *test)
{
int s;
if (test->role == 'c') {
// 对于客户端，发送本地参数给服务器
if (send_parameters(test) < 0)
return -1;
} else {
// 对于服务器端，获取客户端发送过来的参数
if (get_parameters(test) < 0) {
return -1;
}
// 是否支持安全授权检查
#if defined(HAVE_SSL)
if (test_is_authorized(test) < 0){
return -1;
}
#endif //HAVE_SSL
// 启动对应协议的listen回调，开始侦听数据面的连接请求
if ((s = test->protocol->listen(test)) < 0) {
return -1;
}
// 把上面侦听返回的socket加入到fd_set中，以及max_fd中，从这里可以看到控制通道和数据通道的侦听结构是一致的。
FD_SET(s, &test->read_set);
test->max_fd = (s > test->max_fd) ? s : test->max_fd;
test->prot_listener = s;
// 这里发送CREATE_STREAMS状态给客户端，通知客户端可以创建测试流了。
// Send the control message to create streams and start the test
if (iperf_set_send_state(test, CREATE_STREAMS) != 0)
return -1;
}
return 0;
}
上面我们分析的是server服务器端，在此时客户端也有状态在运行，让我们具体分析一下。在iperf_run_client的开始处，iperf_connect负责向服务器发起控制通道连接，具体实现如下。
/* iperf_connect -- client to server connection function */
int
iperf_connect(struct iperf_test *test)
{
int opt, n;
socklen_t len;
if (NULL == test)
{
iperf_err(NULL, "No test\n");
return -1;
}
FD_ZERO(&test->read_set);
FD_ZERO(&test->write_set);
// 生成cookie,一般采用类似于随机数生成机制，
make_cookie(test->cookie);
/* Create and connect the control channel */
// 如果是未创建则创建控制通道socket
if (test->ctrl_sck < 0)
// Create the control channel using an ephemeral port
// netdial内部创建控制通道socket，并通过connect机制连接服务器
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;
}
// 和服务器端类似，设置TCP_NODELAY, keepalive和TCP_USER_TIMEOUT
// 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_KEEPALIVE)
// Set Control Connection TCP Keepalive (especially useful for long UDP test sessions)
if (iperf_set_control_keepalive(test) < 0)
return -1;
#endif //HAVE_TCP_KEEPALIVE
#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 */
// 发送cookie给服务器端
if (Nwrite(test->ctrl_sck, test->cookie, COOKIE_SIZE, Ptcp) < 0) {
i_errno = IESENDCOOKIE;
return -1;
}
// 把控制通道socket加入fd_set和max_fd中
FD_SET(test->ctrl_sck, &test->read_set);
if (test->ctrl_sck > test->max_fd) test->max_fd = test->ctrl_sck;
// tcp和mss设置，从socket获取，然后设置到ctrl_sck_mss中。
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);
}
// 对于udp测试来说，如果命令行未设置块大小，通过tcp mss来确定udp报文的大小，避免ip层分片，并且假设双向是一致的。
/*
* 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);
}
}
/* Initialize GSO parameters when --gsro is used */
if (test->settings->gso) {
test->settings->gso_dg_size = test->settings->blksize;
/* use the multiple of datagram size for the best efficiency. */
if (test->settings->gso_dg_size > 0) {
n = test->settings->gso_bf_size / test->settings->gso_dg_size;
if (n > GSO_MAX_DG_IN_BF) {
n = GSO_MAX_DG_IN_BF;
}
test->settings->gso_bf_size = n * test->settings->gso_dg_size;
} else {
/* If gso_dg_size is 0 (unlimited bandwidth), use default UDP datagram size */
test->settings->gso_dg_size = DEFAULT_UDP_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;
}
完成上述流程后，客户端就通过select监听控制通道，直到FD_ISSET(test->ctrl_sck, &read_set)，此时会执行iperf_handle_message_client处理服务器发过来的状态，驱动客户端状态机运行，其具体实现如下。
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;
}
if (test->debug_level >= DEBUG_LEVEL_INFO) {
iperf_printf(test, "Reading new State from the Server - current state is %d-%s\n", test->state, state_to_text(test->state));
}
/*!!! 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;
}
}
if (test->debug_level >= DEBUG_LEVEL_INFO) {
iperf_printf(test, "State change: client received and changed State to %d-%s\n", test->state, state_to_text(test->state));
}
switch (test->state) {
case PARAM_EXCHANGE:
// 参数交换状态，执行iperf_exchange_parameters将参数发送给服务器端
if (iperf_exchange_parameters(test) < 0)
return -1;
if (test->on_connect)
test->on_connect(test);
break;
case CREATE_STREAMS:
// 服务器获取参数后，发送过来的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);
if (errno > 0) {
iperf_err(test, "SERVER ERROR - %s, errno: %s", iperf_strerror(i_errno), strerror(errno));
} else {
iperf_err(test, "SERVER ERROR - %s", iperf_strerror(i_errno));
}
return -1;
default:
i_errno = IEMESSAGE;
return -1;
}
return 0;
}
在客户端的iperf_create_streams实现中，和分别创建服务器的流和客户端的流，让我们看看具体实现。
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;
}
// 调用具体协议层的connect连接服务器
s = test->protocol->connect(test);
test->bind_port = orig_bind_port;
if (s < 0)
return -1;
/配置tcp的拥塞控制/
#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;
}
if (test->congestion_used) {
if (test->debug)
printf("Overriding existing congestion algorithm: %s\n", test->congestion_used);
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 */
// iperf_new_stream是实际的创建流实现
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;
}
client端完成上述处理后，server端的主循环判断if (test->state == CREATE_STREAMS)的条件就会成立，因为上面iperf_exchange_parameters在获取完参数后已经设置过了，同时当客户端执行connect后，服务器的条件if (FD_ISSET(test->prot_listener, &read_set))也会成立，因此就会执行服务器的s = test->protocol->accept(test)建立实际的数据通道。然后服务器通过iperf_common_sockopts设置socket选项，根据发送和接收需要的stream个数，调用iperf_new_stream创建对应的流结构。完成上述创建后，服务器端通过iperf_set_send_state(test, TEST_START)设置状态为TEST_START并发送给客户端。客户端在iperf_handle_message_client中完成测试的初始化，而服务器端在iperf_run_server的循环中完成初始化，之后通过iperf_set_send_state(test, TEST_RUNNING)设置状态为TEST_RUNNING并发送给客户端。之后服务器端通过strem的个数创建对应个数的数据处理线程，开始发送和接收数据，而客户端在iperf_handle_message_client处理该状态，什么操作都不做，而是在iperf_run_client中处理TEST_RUNNING状态，也是通过strem的个数创建对应个数的数据处理线程，开始发送和接收数据。因为本地处理速度一般比网络快，所以应该是服务器的数据处理线程先创建，客户端的后创建。这样客户端作为发送端，服务器作为接收端不会有什么问题，因为接收端的数据处理线程早于发送端。这里有个疑惑，如果配置了-R参数，那么服务器线程如果先创建，如何保证服务器发送报文的时候客户端线程肯定已经创建了呢？
下面我们看看服务器和客户端的数据处理线程实现。上一篇文章其实已经说了，服务器数据线程的实现iperf_server_worker_run和客户端数据线程的实现iperf_client_worker_run其实都是一样的，关键的代码部分如下。
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;
}
}
}
iperf_send_mt和iperf_recv_mt内部分别会调用strem中的snd和rcv回调来发送和接收数据，需要注意的是iperf_send_mt中可以通过multisend支持报文突发，具体是在用-b配置速率的时候通过在速率后面增加/x设置突发，x是突发的个数。snd和rcv和具体的协议有关，目前tcp, udp和sctp都实现了各自的send和recv功能，在test的stream初始化的时候赋值给snd和rcv(iperf_new_stream)。
经过上面的分析，我们涉及到了待测量的传输协议需要实现的回调函数，分别是listen, accept,connect, send和recv。让我们看看tcp和udp的具体实现，sctp的实现可以留给读者自己分析。其实上面这些回调，从名字看和tcp的网络模型就是对应的，listen和accept用于服务器端侦听和接收连接的api，connect是客户端发起三次握手的api, send和recv是报文发送和接收的流程。对于udp来说，创建socket之后，因为没有三次握手的流程，所以只需要用sendto, sendmsg发送报文，用recvfrom, recvmsg接收报文，没有listen和accept这两个流程， 而connect是支持的。在udp的socket上调用connect只是将路由绑定到socket上，下次就可以用send/recv来发送和接收报文了。
接下来我们分别分析tcp和udp的协议实现的上述回调接口，首先看tcp的实现。
/* iperf_tcp_recv
*
* receives the data for TCP
*/
int
iperf_tcp_recv(struct iperf_stream *sp)
{
int r;
int sock_opt;
// 支持配置为MSG_TRUNC模式，返回实际接收报文的大小
#if defined(HAVE_MSG_TRUNC)
sock_opt = sp->test->settings->skip_rx_copy ? MSG_TRUNC : 0;
#else
sock_opt = 0;
#endif /* HAVE_MSG_TRUNC */
// 使用no select形式的 recv api接口接收数据
r = Nrecv_no_select(sp->socket, sp->buffer, sp->settings->blksize, Ptcp, sock_opt);
if (r < 0)
return r;
/* Only count bytes received while we're in the correct state. */
/只有测试运行中才更新接收统计/
if (sp->test->state == TEST_RUNNING) {
sp->result->bytes_received += r;
sp->result->bytes_received_this_interval += r;
}
else {
if (sp->test->debug)
printf("Late receive, state = %d-%s\n", sp->test->state, state_to_text(sp->test->state));
}
return r;
}
/* iperf_tcp_send
*
* sends the data for TCP
*/
int
iperf_tcp_send(struct iperf_stream *sp)
{
int r;
if (!sp->pending_size)
sp->pending_size = sp->settings->blksize;
// 支持零拷贝方式发送报文，也就是不经过用户态的内容拷贝，内核直接将buffer数据发送出去，Nwrite是普通发送方式
if (sp->test->zerocopy)
r = Nsendfile(sp->buffer_fd, sp->socket, sp->buffer, sp->pending_size);
else
r = Nwrite(sp->socket, sp->buffer, sp->pending_size, Ptcp);
if (r < 0)
return r;
// 这里不需要判断当前运行状态么？
sp->pending_size -= r;
sp->result->bytes_sent += r;
sp->result->bytes_sent_this_interval += r;
if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG)
printf("sent %d bytes of %d, pending %d, total %" PRIu64 "\n",
r, sp->settings->blksize, sp->pending_size, sp->result->bytes_sent);
return r;
}
/* iperf_tcp_accept
*
* accept a new TCP stream connection
*/
int
iperf_tcp_accept(struct iperf_test * test)
{
int s;
signed char rbuf = ACCESS_DENIED;
char cookie[COOKIE_SIZE] = {0};
socklen_t len;
struct sockaddr_storage addr;
len = sizeof(addr);
if ((s = accept(test->listener, (struct sockaddr *) &addr, &len)) < 0) {
i_errno = IESTREAMCONNECT;
return -1;
}
#if defined(HAVE_SO_MAX_PACING_RATE)
/* If fq socket pacing is specified, enable it. */
// 是否配置使用TCP的SO_MAX_PACING_RATE选项，也就是通过FQ限制发送速率
if (test->settings->fqrate) {
/* Convert bits per second to bytes per second */
uint64_t fqrate = test->settings->fqrate / 8;
if (fqrate > 0) {
if (test->debug) {
printf("Setting fair-queue socket pacing to %"PRIu64"\n", fqrate);
}
if (setsockopt(s, SOL_SOCKET, SO_MAX_PACING_RATE, &fqrate, sizeof(fqrate)) < 0) {
warning("Unable to set socket pacing");
}
}
}
#endif /* HAVE_SO_MAX_PACING_RATE */
// 读取cookie消息
if (Nread(s, cookie, COOKIE_SIZE, Ptcp) < 0) {
i_errno = IERECVCOOKIE;
close(s);
return -1;
}
// 比较cookie是否匹配，如果不匹配则返回错误ACCESS_DENIED
if (strncmp(test->cookie, cookie, COOKIE_SIZE) != 0) {
if (Nwrite(s, (char*) &rbuf, sizeof(rbuf), Ptcp) < 0) {
iperf_err(test, "failed to send access denied from busy server to new connecting client, errno = %d\n", errno);
}
close(s);
}
return s;
}
/* iperf_tcp_listen
*
* start up a listener for TCP stream connections
*/
int
iperf_tcp_listen(struct iperf_test *test)
{
int s, opt;
socklen_t optlen;
int saved_errno;
int rcvbuf_actual, sndbuf_actual;
s = test->listener;
// 如果支持配置tcp的一些选项，则需要重新创建socket
/*
* If certain parameters are specified (such as socket buffer
* size), then throw away the listening socket (the one for which
* we just accepted the control connection) and recreate it with
* those parameters. That way, when new data connections are
* set, they'll have all the correct parameters in place.
*
* It's not clear whether this is a requirement or a convenience.
*/
if (test->no_delay || test->mptcp || test->settings->mss || test->settings->socket_bufsize) {
struct addrinfo hints, *res;
char portstr[6];
int proto = 0;
// 关闭之前已经listen的socket，并清除fd_set
FD_CLR(s, &test->read_set);
close(s);
snprintf(portstr, 6, "%d", test->server_port);
memset(&hints, 0, sizeof(hints));
// 如果地址族未指定，且未指定绑定地址，则使用AF_INET6地址族
/*
* If binding to the wildcard address with no explicit address
* family specified, then force us to get an AF_INET6 socket.
* More details in the comments in netannounce().
*/
if (test->settings->domain == AF_UNSPEC && !test->bind_address) {
hints.ai_family = AF_INET6;
}
else {
hints.ai_family = test->settings->domain;
}
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_PASSIVE;
if ((gerror = getaddrinfo(test->bind_address, portstr, &hints, &res)) != 0) {
i_errno = IESTREAMLISTEN;
return -1;
}
// 支持mptcp协议
#if defined(HAVE_IPPROTO_MPTCP)
if (test->mptcp)
proto = IPPROTO_MPTCP;
#endif
// 重新创建socket
if ((s = socket(res->ai_family, SOCK_STREAM, proto)) < 0) {
freeaddrinfo(res);
i_errno = IESTREAMLISTEN;
return -1;
}
// 下面就是配置具体的socket选项
if (test->no_delay) {
opt = 1;
if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IESETNODELAY;
return -1;
}
}
// XXX: Setting MSS is very buggy!
if ((opt = test->settings->mss)) {
if (setsockopt(s, IPPROTO_TCP, TCP_MAXSEG, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IESETMSS;
return -1;
}
}
if ((opt = test->settings->socket_bufsize)) {
if (setsockopt(s, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (setsockopt(s, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
}
// 这个缩进看起来很别扭，可能之前有if条件判断后来删除了。
{
unsigned int rate = test->settings->rate / 8;
if (rate > 0) {
if (test->debug) {
printf("Setting application pacing to %u\n", rate);
}
}
}
opt = 1;
if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IEREUSEADDR;
return -1;
}
/*
* If we got an IPv6 socket, figure out if it should accept IPv4
* connections as well. See documentation in netannounce() for
* more details.
*/
// 是否只支持IPv6，还是创建IPv6支持连接ipv4 socket
#if defined(IPV6_V6ONLY) && !defined(__OpenBSD__)
if (res->ai_family == AF_INET6 && (test->settings->domain == AF_UNSPEC || test->settings->domain == AF_INET)) {
if (test->settings->domain == AF_UNSPEC)
opt = 0;
else
opt = 1;
if (setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY,
(char *) &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IEV6ONLY;
return -1;
}
}
#endif /* IPV6_V6ONLY */
// 绑定地址，地址从刚才的getaddrinfo获取到的
if (bind(s, (struct sockaddr *) res->ai_addr, res->ai_addrlen) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(res);
errno = saved_errno;
i_errno = IESTREAMLISTEN;
return -1;
}
freeaddrinfo(res);
// 开始侦听
if (listen(s, INT_MAX) < 0) {
i_errno = IESTREAMLISTEN;
close(s);
return -1;
}
// 更新listener
test->listener = s;
}
// 再次校验socket send buffer和recv buffuer,如果实际的值小于配置的值返回错误
/* Read back and verify the sender socket buffer size */
optlen = sizeof(sndbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_SNDBUF, &sndbuf_actual, &optlen) < 0) {
saved_errno = errno;
close(s);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("SNDBUF is %u, expecting %u\n", sndbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > sndbuf_actual) {
i_errno = IESETBUF2;
close(s);
return -1;
}
/* Read back and verify the receiver socket buffer size */
optlen = sizeof(rcvbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_RCVBUF, &rcvbuf_actual, &optlen) < 0) {
saved_errno = errno;
close(s);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("RCVBUF is %u, expecting %u\n", rcvbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > rcvbuf_actual) {
i_errno = IESETBUF2;
close(s);
return -1;
}
/* Set common socket options */
if (iperf_common_sockopts(test, s) < 0) {
close(s);
return -1;
}
if (test->json_output) {
cJSON_AddNumberToObject(test->json_start, "sock_bufsize", test->settings->socket_bufsize);
cJSON_AddNumberToObject(test->json_start, "sndbuf_actual", sndbuf_actual);
cJSON_AddNumberToObject(test->json_start, "rcvbuf_actual", rcvbuf_actual);
}
return s;
}
/* iperf_tcp_connect
*
* connect to a TCP stream listener
* This function is roughly similar to netdial(), and may indeed have
* been derived from it at some point, but it sets many TCP-specific
* options between socket creation and connection.
*/
int
iperf_tcp_connect(struct iperf_test *test)
{
struct addrinfo *server_res;
int s, opt;
socklen_t optlen;
int saved_errno;
int rcvbuf_actual, sndbuf_actual;
int proto = 0;
#if defined(HAVE_IPPROTO_MPTCP)
if (test->mptcp)
proto = IPPROTO_MPTCP;
#endif
// 客户端socket创建
s = create_socket(test->settings->domain, SOCK_STREAM, proto, test->bind_address, test->bind_dev, test->bind_port, test->server_hostname, test->server_port, &server_res);
if (s < 0) {
i_errno = IESTREAMCONNECT;
return -1;
}
// 设置tcp选项
/* Set socket options */
if (test->no_delay) {
opt = 1;
if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETNODELAY;
return -1;
}
}
if ((opt = test->settings->mss)) {
if (setsockopt(s, IPPROTO_TCP, TCP_MAXSEG, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETMSS;
return -1;
}
}
if ((opt = test->settings->socket_bufsize)) {
if (setsockopt(s, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (setsockopt(s, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
}
#if defined(HAVE_TCP_USER_TIMEOUT)
if ((opt = test->settings->snd_timeout)) {
if (setsockopt(s, IPPROTO_TCP, TCP_USER_TIMEOUT, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETUSERTIMEOUT;
return -1;
}
}
#endif /* HAVE_TCP_USER_TIMEOUT */
// 再次校验socket send buffer和recv buffuer,如果实际的值小于配置的值返回错误
/* Read back and verify the sender socket buffer size */
optlen = sizeof(sndbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_SNDBUF, &sndbuf_actual, &optlen) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("SNDBUF is %u, expecting %u\n", sndbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > sndbuf_actual) {
close(s);
freeaddrinfo(server_res);
i_errno = IESETBUF2;
return -1;
}
/* Read back and verify the receiver socket buffer size */
optlen = sizeof(rcvbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_RCVBUF, &rcvbuf_actual, &optlen) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("RCVBUF is %u, expecting %u\n", rcvbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > rcvbuf_actual) {
close(s);
freeaddrinfo(server_res);
i_errno = IESETBUF2;
return -1;
}
if (test->json_output) {
cJSON *sock_bufsize_item = cJSON_GetObjectItem(test->json_start, "sock_bufsize");
if (sock_bufsize_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "sock_bufsize", test->settings->socket_bufsize);
}
cJSON *sndbuf_actual_item = cJSON_GetObjectItem(test->json_start, "sndbuf_actual");
if (sndbuf_actual_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "sndbuf_actual", sndbuf_actual);
}
cJSON *rcvbuf_actual_item = cJSON_GetObjectItem(test->json_start, "rcvbuf_actual");
if (rcvbuf_actual_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "rcvbuf_actual", rcvbuf_actual);
}
}
// 支持配置ipv6的flow label
#if defined(HAVE_FLOWLABEL)
if (test->settings->flowlabel) {
if (server_res->ai_addr->sa_family != AF_INET6) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETFLOW;
return -1;
} else {
struct sockaddr_in6* sa6P = (struct sockaddr_in6*) server_res->ai_addr;
char freq_buf[sizeof(struct in6_flowlabel_req)];
struct in6_flowlabel_req *freq = (struct in6_flowlabel_req *)freq_buf;
int freq_len = sizeof(*freq);
memset(freq, 0, sizeof(*freq));
freq->flr_label = htonl(test->settings->flowlabel & IPV6_FLOWINFO_FLOWLABEL);
freq->flr_action = IPV6_FL_A_GET;
freq->flr_flags = IPV6_FL_F_CREATE;
freq->flr_share = IPV6_FL_S_ANY;
memcpy(&freq->flr_dst, &sa6P->sin6_addr, 16);
if (setsockopt(s, IPPROTO_IPV6, IPV6_FLOWLABEL_MGR, freq, freq_len) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETFLOW;
return -1;
}
sa6P->sin6_flowinfo = freq->flr_label;
opt = 1;
if (setsockopt(s, IPPROTO_IPV6, IPV6_FLOWINFO_SEND, &opt, sizeof(opt)) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESETFLOW;
return -1;
}
}
}
#endif /* HAVE_FLOWLABEL */
// 支持设置TCP的SO_MAX_PACING_RATE，也就是发送速率
#if defined(HAVE_SO_MAX_PACING_RATE)
/* If socket pacing is specified try to enable it. */
if (test->settings->fqrate) {
/* Convert bits per second to bytes per second */
uint64_t fqrate = test->settings->fqrate / 8;
if (fqrate > 0) {
if (test->debug) {
printf("Setting fair-queue socket pacing to %"PRIu64"\n", fqrate);
}
if (setsockopt(s, SOL_SOCKET, SO_MAX_PACING_RATE, &fqrate, sizeof(fqrate)) < 0) {
warning("Unable to set socket pacing");
}
}
}
#endif /* HAVE_SO_MAX_PACING_RATE */
{
unsigned int rate = test->settings->rate / 8;
if (rate > 0) {
if (test->debug) {
printf("Setting application pacing to %u\n", rate);
}
}
}
/* Set common socket options */
if (iperf_common_sockopts(test, s) < 0) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
return -1;
}
// 使用connect连接服务器端
if (timeout_connect(s, (struct sockaddr *) server_res->ai_addr, server_res->ai_addrlen,
DEFAULT_NO_MSG_RCVD_TIMEOUT) < 0 && errno != EINPROGRESS) {
saved_errno = errno;
close(s);
freeaddrinfo(server_res);
errno = saved_errno;
i_errno = IESTREAMCONNECT;
return -1;
}
freeaddrinfo(server_res);
// 发送cookie给服务器端进行校验
/* Send cookie for verification */
if (Nwrite(s, test->cookie, COOKIE_SIZE, Ptcp) < 0) {
saved_errno = errno;
close(s);
errno = saved_errno;
i_errno = IESENDCOOKIE;
return -1;
}
return s;
}
接下来我们分析udp的实现，就像之前分析的那样，udp是没有listen, accept的实现的，那么是如何实现类似效果的回调，如何适配到iperf的整体框架中的，也就是使用同一套的connect, send和recv实现流程的呢，让我们看具体的代码吧。
/* iperf_udp_recv
*
* receives the data for UDP
*/
int
iperf_udp_recv(struct iperf_stream *sp)
{
uint32_t sec, usec;
uint64_t pcount;
int r;
int size = sp->settings->blksize;
int first_packet = 0;
double transit = 0, d = 0;
struct iperf_time sent_time, arrival_time, temp_time;
struct iperf_test *test = sp->test;
int sock_opt = 0;
int dgram_sz;
int buf_sz;
char *dgram_buf;
char *dgram_buf_end;
const int min_pkt_size = sizeof(uint32_t) * 3; /* sec + usec + pcount (32-bit) */
#if defined(HAVE_MSG_TRUNC)
// UDP recv() with MSG_TRUNC reads only the size bytes, but return the length of the full packet
if (sp->test->settings->skip_rx_copy) {
sock_opt = MSG_TRUNC;
size = sizeof(sec) + sizeof(usec) + sizeof(pcount);
}
#endif /* HAVE_MSG_TRUNC */
// 支持gro和非gro两种方式，dgram_sz是配置的块大小，buf_sz是实际返回的大小
/* Configure loop parameters based on GRO availability */
if (sp->test->settings->gro) {
size = sp->test->settings->gro_bf_size;
r = Nread_gro(sp->socket, sp->buffer, size, Pudp, &dgram_sz);
/* Use negotiated block size for GRO segment stride to ensure correct parsing. */
dgram_sz = sp->settings->blksize;
buf_sz = r;
} else {
/* GRO disabled or unavailable - use normal UDP receive and single packet size */
r = Nrecv_no_select(sp->socket, sp->buffer, size, Pudp, sock_opt);
dgram_sz = sp->settings->blksize;
buf_sz = r;
}
/*
* If we got an error in the read, or if we didn't read anything
* because the underlying read(2) got a EAGAIN, then skip packet
* processing.
*/
if (r <= 0)
return r;
// Check dgram_sz is greater than 0
if (dgram_sz <= 0){
return -1;
}
// 只有测试运行中才需要更新统计
/* Only count bytes received while we're in the correct state. */
if (test->state == TEST_RUNNING) {
/*
* For jitter computation below, it's important to know if this
* packet is the first packet received.
*/
// jitter计算需要直到收到的是否是首包
if (sp->result->bytes_received == 0) {
first_packet = 1;
}
// 更新接收统计
sp->result->bytes_received += r;
sp->result->bytes_received_this_interval += r;
if (sp->test->debug)
printf("received %d bytes of %d, total %" PRIu64 "\n", r, size, sp->result->bytes_received);
// 下面的流程计算udp报文是否乱序，计算jitter(不需要两端时钟同步)，具体流程待分析
/* Unified loop: processes single packet when GRO off, multiple when GRO on */
dgram_buf = sp->buffer;
dgram_buf_end = sp->buffer + buf_sz;
while (buf_sz >= dgram_sz && dgram_buf + dgram_sz <= dgram_buf_end) {
/* Ensure we have enough bytes for the packet header */
if (buf_sz < min_pkt_size)
break;
/* Extract packet headers */
if (sp->test->udp_counters_64bit) {
/* Verify we have enough space for 64-bit counter */
if (buf_sz < sizeof(uint32_t) * 2 + sizeof(uint64_t))
break;
memcpy(&sec, dgram_buf, sizeof(sec));
memcpy(&usec, dgram_buf+4, sizeof(usec));
memcpy(&pcount, dgram_buf+8, sizeof(pcount));
sec = ntohl(sec);
usec = ntohl(usec);
pcount = be64toh(pcount);
sent_time.secs = sec;
sent_time.usecs = usec;
} else {
uint32_t pc;
memcpy(&sec, dgram_buf, sizeof(sec));
memcpy(&usec, dgram_buf+4, sizeof(usec));
memcpy(&pc, dgram_buf+8, sizeof(pc));
sec = ntohl(sec);
usec = ntohl(usec);
pcount = ntohl(pc);
sent_time.secs = sec;
sent_time.usecs = usec;
}
/*
* Try to handle out of order packets. The way we do this
* uses a constant amount of storage but might not be
* correct in all cases. In particular we seem to have the
* assumption that packets can't be duplicated in the network,
* because duplicate packets will possibly cause some problems here.
*
* First figure out if the sequence numbers are going forward.
* Note that pcount is the sequence number read from the packet,
* and sp->packet_count is the highest sequence number seen so
* far (so we're expecting to see the packet with sequence number
* sp->packet_count + 1 arrive next).
*/
if (pcount >= sp->packet_count + 1) {
/* Forward, but is there a gap in sequence numbers? */
if (pcount > sp->packet_count + 1) {
/* There's a gap so count that as a loss. */
sp->cnt_error += (pcount - 1) - sp->packet_count;
if (test->debug_level >= DEBUG_LEVEL_INFO)
fprintf(stderr, "LOST %" PRIu64 " PACKETS - received packet %" PRIu64 " but expected sequence %" PRIu64 " on stream %d\n", (pcount - sp->packet_count + 1), pcount, sp->packet_count + 1, sp->socket);
}
/* Update the highest sequence number seen so far. */
sp->packet_count = pcount;
} else {
/*
* Sequence number went backward (or was stationary?!?).
* This counts as an out-of-order packet.
*/
sp->outoforder_packets++;
/*
* If we have lost packets, then the fact that we are now
* seeing an out-of-order packet offsets a prior sequence
* number gap that was counted as a loss. So we can take
* away a loss.
*/
if (sp->cnt_error > 0)
sp->cnt_error--;
/* Log the out-of-order packet */
if (test->debug_level >= DEBUG_LEVEL_INFO)
fprintf(stderr, "OUT OF ORDER - received packet %" PRIu64 " but expected sequence %" PRIu64 " on stream %d\n", pcount, sp->packet_count + 1, sp->socket);
}
/*
* jitter measurement
*
* This computation is based on RFC 1889 (specifically
* sections 6.3.1 and A.8).
*
* Note that synchronized clocks are not required since
* the source packet delta times are known. Also this
* computation does not require knowing the round-trip
* time.
*/
iperf_time_now(&arrival_time);
iperf_time_diff(&arrival_time, &sent_time, &temp_time);
transit = iperf_time_in_secs(&temp_time);
/* Hack to handle the first packet by initializing prev_transit. */
if (first_packet)
sp->prev_transit = transit;
d = transit - sp->prev_transit;
if (d < 0)
d = -d;
sp->prev_transit = transit;
sp->jitter += (d - sp->jitter) / 16.0;
first_packet = 0;
dgram_buf += dgram_sz;
buf_sz -= dgram_sz;
}
}
else {
if (test->debug_level >= DEBUG_LEVEL_INFO)
printf("Late receive, state = %d\n", test->state);
}
return r;
}
/* iperf_udp_send
*
* sends the data for UDP
*/
int
iperf_udp_send(struct iperf_stream *sp)
{
int r;
int size = sp->settings->blksize;
struct iperf_time before;
int dgram_sz;
int buf_sz;
int cnt = 0;
char *dgram_buf;
char *dgram_buf_end;
const int min_pkt_size = sizeof(uint32_t) * 3; /* sec + usec + pcount (32-bit) */
// 支持gro和非gro场景下的dgram_sz 计算
/* Configure loop parameters based on GSO availability */
if (sp->test->settings->gso) {
dgram_sz = sp->test->settings->gso_dg_size;
buf_sz = sp->test->settings->gso_bf_size;
/* Validate GSO parameters */
if (dgram_sz <= 0 || dgram_sz < min_pkt_size || dgram_sz > buf_sz) {
if (sp->test->debug_level >= DEBUG_LEVEL_INFO)
printf("Invalid GSO dgram_sz %d for buf_sz %d, disabling GSO\n", dgram_sz, buf_sz);
dgram_sz = buf_sz = size;
sp->test->settings->gso = 0; /* Disable GSO for safety */
}
} else {
/* GSO disabled or unavailable - single packet */
dgram_sz = buf_sz = size;
}
// 发送流程处理，注意udp的净荷中包含了时间戳和序列号，以便接收端计算是否乱序，计算jitter抖动。
dgram_buf = sp->buffer;
dgram_buf_end = sp->buffer + buf_sz;
/* Unified loop: processes single packet when GSO off, multiple when GSO on */
while (buf_sz > 0 && dgram_buf + dgram_sz <= dgram_buf_end) {
cnt++;
if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG)
printf("%d (%d) remaining %d\n", cnt, dgram_sz, buf_sz);
/* Prevent buffer underflow */
if (buf_sz < dgram_sz) {
if (sp->test->debug_level >= DEBUG_LEVEL_INFO)
printf("Buffer underflow protection: buf_sz %d < dgram_sz %d\n", buf_sz, dgram_sz);
break;
}
iperf_time_now(&before);
++sp->packet_count;
if (sp->test->udp_counters_64bit) {
uint32_t sec, usec;
uint64_t pcount;
sec = htonl(before.secs);
usec = htonl(before.usecs);
pcount = htobe64(sp->packet_count);
memcpy(dgram_buf, &sec, sizeof(sec));
memcpy(dgram_buf+4, &usec, sizeof(usec));
memcpy(dgram_buf+8, &pcount, sizeof(pcount));
} else {
uint32_t sec, usec, pcount;
sec = htonl(before.secs);
usec = htonl(before.usecs);
pcount = htonl(sp->packet_count);
memcpy(dgram_buf, &sec, sizeof(sec));
memcpy(dgram_buf+4, &usec, sizeof(usec));
memcpy(dgram_buf+8, &pcount, sizeof(pcount));
}
dgram_buf += dgram_sz;
buf_sz -= dgram_sz;
}
/* Warn if we didn't process all the buffer due to size mismatch */
if (buf_sz > 0 && sp->test->debug_level >= DEBUG_LEVEL_INFO) {
printf("GSO: %d bytes remaining unprocessed\n", buf_sz);
}
// 支持gso和非gso两种发送方式
if (sp->test->settings->gso) {
size = sp->test->settings->gso_bf_size;
r = Nwrite_gso(sp->socket, sp->buffer, size, Pudp, sp->test->settings->gso_dg_size);
} else {
r = Nwrite(sp->socket, sp->buffer, size, Pudp);
}
// 发送失败处理，如果返回0则会重新发送，小于0则返回错误
if (r <= 0) {
--sp->packet_count; /* Don't count messages that no data was sent from them.
* Allows "resending" a massage with the same numbering */
if (r < 0) {
if (r == NET_SOFTERROR && sp->test->debug_level >= DEBUG_LEVEL_INFO)
printf("UDP send failed on NET_SOFTERROR. errno=%s\n", strerror(errno));
return r;
}
}
// 更新发送统计信息
sp->result->bytes_sent += r;
sp->result->bytes_sent_this_interval += r;
if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG)
printf("sent %d bytes of %d, total %" PRIu64 "\n", r, size, sp->result->bytes_sent);
return r;
}
// 这种下面的注释所说，因为udp是看向无连接的，喂了模拟tcp行为需要做一些特殊处理
/**************************************************************************/
/*
* The following functions all have to do with managing UDP data sockets.
* UDP of course is connectionless, so there isn't really a concept of
* setting up a connection, although connect(2) can (and is) used to
* bind the remote end of sockets. We need to simulate some of the
* connection management that is built-in to TCP so that each side of the
* connection knows about each other before the real data transfers begin.
*/
// 配置并校验socket 的 send buffer和 recv buffer。
/*
* Set and verify socket buffer sizes.
* Return 0 if no error, -1 if an error, +1 if socket buffers are
* potentially too small to hold a message.
*/
int
iperf_udp_buffercheck(struct iperf_test *test, int s)
{
int rc = 0;
int sndbuf_actual, rcvbuf_actual;
/*
* Set socket buffer size if requested. Do this for both sending and
* receiving so that we can cover both normal and --reverse operation.
*/
int opt;
socklen_t optlen;
if ((opt = test->settings->socket_bufsize)) {
if (setsockopt(s, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(opt)) < 0) {
i_errno = IESETBUF;
return -1;
}
if (setsockopt(s, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(opt)) < 0) {
i_errno = IESETBUF;
return -1;
}
}
/* Read back and verify the sender socket buffer size */
optlen = sizeof(sndbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_SNDBUF, &sndbuf_actual, &optlen) < 0) {
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("SNDBUF is %u, expecting %u\n", sndbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > sndbuf_actual) {
i_errno = IESETBUF2;
return -1;
}
if (test->settings->blksize > sndbuf_actual) {
char str[WARN_STR_LEN];
snprintf(str, sizeof(str),
"Block size %d > sending socket buffer size %d",
test->settings->blksize, sndbuf_actual);
warning(str);
rc = 1;
}
/* Read back and verify the receiver socket buffer size */
optlen = sizeof(rcvbuf_actual);
if (getsockopt(s, SOL_SOCKET, SO_RCVBUF, &rcvbuf_actual, &optlen) < 0) {
i_errno = IESETBUF;
return -1;
}
if (test->debug) {
printf("RCVBUF is %u, expecting %u\n", rcvbuf_actual, test->settings->socket_bufsize);
}
if (test->settings->socket_bufsize && test->settings->socket_bufsize > rcvbuf_actual) {
i_errno = IESETBUF2;
return -1;
}
if (test->settings->blksize > rcvbuf_actual) {
char str[WARN_STR_LEN];
snprintf(str, sizeof(str),
"Block size %d > receiving socket buffer size %d",
test->settings->blksize, rcvbuf_actual);
warning(str);
rc = 1;
}
if (test->json_output) {
cJSON *sock_bufsize_item = cJSON_GetObjectItem(test->json_start, "sock_bufsize");
if (sock_bufsize_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "sock_bufsize", test->settings->socket_bufsize);
}
cJSON *sndbuf_actual_item = cJSON_GetObjectItem(test->json_start, "sndbuf_actual");
if (sndbuf_actual_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "sndbuf_actual", sndbuf_actual);
}
cJSON *rcvbuf_actual_item = cJSON_GetObjectItem(test->json_start, "rcvbuf_actual");
if (rcvbuf_actual_item == NULL) {
cJSON_AddNumberToObject(test->json_start, "rcvbuf_actual", rcvbuf_actual);
}
}
return rc;
}
// 设置udp gso支持情况，即Udp分片卸载功能
#ifdef HAVE_UDP_SEGMENT
int
iperf_udp_gso(struct iperf_test *test, int s)
{
int rc;
int gso = test->settings->gso_dg_size;
rc = setsockopt(s, IPPROTO_UDP, UDP_SEGMENT, (char*) &gso, sizeof(gso));
if (rc) {
if (test->debug)
iperf_printf(test, "No GSO (%d)\n", rc);
test->settings->gso = 0;
} else {
if (test->debug)
iperf_printf(test, "GSO (%d)\n", gso);
}
return rc;
}
#else
int
iperf_udp_gso(struct iperf_test *test, int s)
{
/* GSO not supported on this platform */
test->settings->gso = 0;
return -1;
}
#endif
// 设置udp gro支持情况，即udp分片接收卸载
#ifdef HAVE_UDP_GRO
int
iperf_udp_gro(struct iperf_test *test, int s)
{
int rc;
int gro = 1;
rc = setsockopt(s, IPPROTO_UDP, UDP_GRO, (char*) &gro, sizeof(gro));
if (rc) {
if (test->debug)
iperf_printf(test, "No GRO (%d)\n", rc);
test->settings->gro = 0;
} else {
if (test->debug)
iperf_printf(test, "GRO\n");
}
return rc;
}
#else
int
iperf_udp_gro(struct iperf_test *test, int s)
{
/* GRO not supported on this platform */
test->settings->gro = 0;
return -1;
}
#endif
/*
* iperf_udp_accept
*
* Accepts a new UDP "connection"
*/
int
iperf_udp_accept(struct iperf_test *test)
{
struct sockaddr_storage sa_peer;
unsigned int buf;
socklen_t len;
int sz, s;
int rc;
// 用当前的listener socket作为接收的socket, 然后创建一个新的socket负责listener。
/*
* Get the current outstanding socket. This socket will be used to handle
* data transfers and a new "listening" socket will be created.
*/
s = test->prot_listener;
// 通过recvfrom获取对端的数据，并获取对端信息，以便通过connect绑定连接，后续则可以支持支持send/recv。
/*
* Grab the UDP packet sent by the client. From that we can extract the
* client's address, and then use that information to bind the remote side
* of the socket to the client.
*/
len = sizeof(sa_peer);
if ((sz = recvfrom(test->prot_listener, &buf, sizeof(buf), 0, (struct sockaddr *) &sa_peer, &len)) < 0) {
i_errno = IESTREAMACCEPT;
return -1;
}
if (connect(s, (struct sockaddr *) &sa_peer, len) < 0) {
i_errno = IESTREAMACCEPT;
return -1;
}
/* Check and set socket buffer sizes */
rc = iperf_udp_buffercheck(test, s);
if (rc < 0)
/* error */
return rc;
/*
* If the socket buffer was too small, but it was the default
* size, then try explicitly setting it to something larger.
*/
if (rc > 0) {
if (test->settings->socket_bufsize == 0) {
char str[WARN_STR_LEN];
int bufsize = test->settings->blksize + UDP_BUFFER_EXTRA;
snprintf(str, sizeof(str), "Increasing socket buffer size to %d",
bufsize);
warning(str);
test->settings->socket_bufsize = bufsize;
rc = iperf_udp_buffercheck(test, s);
if (rc < 0)
return rc;
}
}
if (test->settings->gso)
iperf_udp_gso(test, s);
if (test->settings->gro)
iperf_udp_gro(test, s);
// SO_MAX_PACING_RATE设置，udp socket也支持？
#if defined(HAVE_SO_MAX_PACING_RATE)
/* If socket pacing is specified, try it. */
if (test->settings->fqrate) {
/* Convert bits per second to bytes per second */
uint64_t fqrate = test->settings->fqrate / 8;
if (fqrate > 0) {
if (test->debug) {
printf("Setting fair-queue socket pacing to %"PRIu64"\n", fqrate);
}
if (setsockopt(s, SOL_SOCKET, SO_MAX_PACING_RATE, &fqrate, sizeof(fqrate)) < 0) {
warning("Unable to set socket pacing");
}
}
}
#endif /* HAVE_SO_MAX_PACING_RATE */
{
unsigned int rate = test->settings->rate / 8;
if (rate > 0) {
if (test->debug) {
printf("Setting application pacing to %u\n", rate);
}
}
}
/* Set common socket options */
if (iperf_common_sockopts(test, s) < 0) {
return -1;
}
// 创建一个新的socket并赋值给prot_listener，因为刚才的prot_listener已经作为数据通道接收或者发送数据了。
/*
* Create a new "listening" socket to replace the one we were using before.
*/
FD_CLR(test->prot_listener, &test->read_set); // No control messages from old listener
test->prot_listener = netannounce(test->settings->domain, Pudp, test->bind_address, test->bind_dev, test->server_port);
if (test->prot_listener < 0) {
i_errno = IESTREAMLISTEN;
return -1;
}
// 将新的socket加入fd_set并更新到max_fd中
FD_SET(test->prot_listener, &test->read_set);
test->max_fd = (test->max_fd < test->prot_listener) ? test->prot_listener : test->max_fd;
// 发送消息UDP_CONNECT_REPLY给客户端，告诉客户端可以接收另外一个udp流量了。
/* Let the client know we're ready "accept" another UDP "stream" */
buf = UDP_CONNECT_REPLY;
if (write(s, &buf, sizeof(buf)) < 0) {
i_errno = IESTREAMWRITE;
return -1;
}
// 返回的是上一次的listener，也就是当前的数据通道fd
return s;
}
/*
* iperf_udp_listen
*
* Start up a listener for UDP stream connections. Unlike for TCP,
* there is no listen(2) for UDP. This socket will however accept
* a UDP datagram from a client (indicating the client's presence).
*/
int
iperf_udp_listen(struct iperf_test *test)
{
int s;
// 调用 netannounce声明socket,内部是创建socket并bind到对应地址和端口
if ((s = netannounce(test->settings->domain, Pudp, test->bind_address, test->bind_dev, test->server_port)) < 0) {
i_errno = IESTREAMLISTEN;
return -1;
}
/*
* The caller will put this value into test->prot_listener.
*/
return s;
}
/*
* iperf_udp_connect
*
* "Connect" to a UDP stream listener.
*/
int
iperf_udp_connect(struct iperf_test *test)
{
int s, sz;
unsigned int buf;
#ifdef SO_RCVTIMEO
struct timeval tv;
#endif
int rc;
int i, max_len_wait_for_reply;
// 客户端发起到服务器的连接请求
/* Create and bind our local socket. */
if ((s = netdial(test->settings->domain, Pudp, test->bind_address, test->bind_dev, test->bind_port, test->server_hostname, test->server_port, -1)) < 0) {
i_errno = IESTREAMCONNECT;
return -1;
}
// 下面这部分和iperf_udp_accept中部分代码其实是一样的，都是检查buffer配置，设置gso和gro,设置发送速率配置，设置socket选项。
/* Check and set socket buffer sizes */
rc = iperf_udp_buffercheck(test, s);
if (rc < 0)
/* error */
return rc;
if (test->settings->gso)
iperf_udp_gso(test, s);
if (test->settings->gro)
iperf_udp_gro(test, s);
/*
* If the socket buffer was too small, but it was the default
* size, then try explicitly setting it to something larger.
*/
if (rc > 0) {
if (test->settings->socket_bufsize == 0) {
char str[WARN_STR_LEN];
int bufsize = test->settings->blksize + UDP_BUFFER_EXTRA;
snprintf(str, sizeof(str), "Increasing socket buffer size to %d",
bufsize);
warning(str);
test->settings->socket_bufsize = bufsize;
rc = iperf_udp_buffercheck(test, s);
if (rc < 0)
return rc;
}
}
#if defined(HAVE_SO_MAX_PACING_RATE)
/* If socket pacing is available and not disabled, try it. */
if (test->settings->fqrate) {
/* Convert bits per second to bytes per second */
uint64_t fqrate = test->settings->fqrate / 8;
if (fqrate > 0) {
if (test->debug) {
printf("Setting fair-queue socket pacing to %"PRIu64"\n", fqrate);
}
if (setsockopt(s, SOL_SOCKET, SO_MAX_PACING_RATE, &fqrate, sizeof(fqrate)) < 0) {
warning("Unable to set socket pacing");
}
}
}
#endif /* HAVE_SO_MAX_PACING_RATE */
{
unsigned int rate = test->settings->rate / 8;
if (rate > 0) {
if (test->debug) {
printf("Setting application pacing to %u\n", rate);
}
}
}
/* Set common socket options */
if (iperf_common_sockopts(test, s) < 0) {
return -1;
}
// 设置接收超时时间
#ifdef SO_RCVTIMEO
/* 30 sec timeout for a case when there is a network problem. */
tv.tv_sec = 30;
tv.tv_usec = 0;
setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (struct timeval *)&tv, sizeof(struct timeval));
#endif
// 发送一个连接消息给服务器端，以便上面的iperf_udp_accpet可以获取客户端的地址信息。
/*
* Write a datagram to the UDP stream to let the server know we're here.
* The server learns our address by obtaining its peer's address.
*/
buf = UDP_CONNECT_MSG;
if (test->debug) {
printf("Sending Connect message to Socket %d\n", s);
}
if (write(s, &buf, sizeof(buf)) < 0) {
// XXX: Should this be changed to IESTREAMCONNECT?
i_errno = IESTREAMWRITE;
return -1;
}
// 等待服务器端返回UDP_CONNECT_REPLY消息，支持完成udp的连接建立
/*
* Wait until the server replies back to us with the "accept" response.
*/
i = 0;
max_len_wait_for_reply = sizeof(buf);
if (test->reverse) /* In reverse mode allow few packets to have the "accept" response - to handle out of order packets */
max_len_wait_for_reply += MAX_REVERSE_OUT_OF_ORDER_PACKETS * test->settings->blksize;
do {
if ((sz = recv(s, &buf, sizeof(buf), 0)) < 0) {
i_errno = IESTREAMREAD;
return -1;
}
if (test->debug) {
printf("Connect received for Socket %d, sz=%d, buf=%x, i=%d, max_len_wait_for_reply=%d\n", s, sz, buf, i, max_len_wait_for_reply);
}
i += sz;
} while (buf != UDP_CONNECT_REPLY && buf != LEGACY_UDP_CONNECT_REPLY && i < max_len_wait_for_reply);
if (buf != UDP_CONNECT_REPLY && buf != LEGACY_UDP_CONNECT_REPLY) {
i_errno = IESTREAMREAD;
return -1;
}
return s;
}
/* iperf_udp_init
*
* initializer for UDP streams in TEST_START
*/
int
iperf_udp_init(struct iperf_test *test)
{
return 0;
}
至此，我们完成了iperf主要状态流程的处理，下一篇文章我们看看数据发送和接收中的统计和计算流程，以及最后的结果结算和展示过程。