libutp收发机制分析

悠悠 2022-03-10 05:16 275阅读 0赞

### 一. 前言 ###

本文旨在通过libutp收发部分源码的分析了解libutp内部的实现机制。

### 二. 基本结构体 ###

无论是了解源码还是使用libutp，都离不开libutp的两个基本结构体：

typedef struct UTPSocket					utp_socket;
    typedef struct struct_utp_context			utp_context;

utp\_socket和utp\_context是对用户开放的外部接口，而UTPSocket和struct\_utp\_context则是内部的实现。这两个结构体的关系有点类似于libevent中的event和event\_base，即utp\_socket负责对一个上层传递的套接字的封装、发送、接受、重传等的判断和处理，而utp\_context负责对utp\_socket进行管理。下面分别分析一下两个结构体：

struct struct_utp_context { 
    	void *userdata;              //常见的存储用户数据的结构体成员
    	utp_callback_t* callbacks[UTP_ARRAY_SIZE];     //使用者定义的回调函数
    
    	uint64 current_ms;                 //时间戳
    	utp_context_stats context_stats;   //当前状态
    	UTPSocket *last_utp_socket;        //上一个连入的utp套接字
    	Array<UTPSocket*> ack_sockets;     //收到的ack
    	Array<RST_Info> rst_info;          //rst信息
    	UTPSocketHT *utp_sockets;          //存储所有utp套接字的哈希表
    	size_t target_delay;               //延迟情况
    	size_t opt_sndbuf;                 //用户设置的发送缓存
    	size_t opt_rcvbuf;                 //用户设置的接收缓存
    	uint64 last_check;                 //上次接入事件
    
    	struct_utp_context();
    	~struct_utp_context();
    
    	void log(int level, utp_socket *socket, char const *fmt, ...);
    	void log_unchecked(utp_socket *socket, char const *fmt, ...);
    	bool would_log(int level);
    
    	bool log_normal:1;	// log normal events?
    	bool log_mtu:1;		// log MTU related events?
    	bool log_debug:1;	// log debugging events? (Must also compile with UTP_DEBUG_LOGGING defined)
    };

注释部分已经说明了各个成员变量的用处，其中最关键的在于存储utp\_socket的哈希表，该表上限在源码中定为3000，若不够用可自行修改（一般都够用）。

utp\_socket结构体包括了对单个套接字的所有处理流程，较为复杂，在此处仅分析少量相关的变量，具体函数在第三节中介绍。

struct UTPSocket { 
    	~UTPSocket();
    
    	PackedSockAddr addr;     //IP地址
    	utp_context *ctx;        //对应的utp_context
    
    	int ida; //for ack socket list
    
    	uint16 retransmit_count;   //重传统计
    
    	uint16 reorder_count;      //乱序统计
    	byte duplicate_ack;        //重复ack
    
    	// the number of packets in the send queue. Packets that haven't
    	// yet been sent count as well as packets marked as needing resend
    	// the oldest un-acked packet in the send queue is seq_nr - cur_window_packets
    	uint16 cur_window_packets;
    
    	// how much of the window is used, number of bytes in-flight
    	// packets that have not yet been sent do not count, packets
    	// that are marked as needing to be re-sent (due to a timeout)
    	// don't count either
    	size_t cur_window;
    	// maximum window size, in bytes
    	size_t max_window;

和收发相关最重要的就是这里的窗口，即cur\_window\_packets，cur\_window和max\_window。其中cur\_window\_packets表示数据包的窗口，包含所有在发送队列的数据包，无论是否已经被发送，或者是否需要重传。cur\_window就是按字节算的通常意义上的窗口，在计算时不包含需要重传的包。max\_window表示最大的窗口。

### 三. 关键函数分析 ###

libutp的发送API是utp\_write()，该函数会调用utp\_writev()，源码如下所示（这里为了突出重点删除了源码中的一部分，用省略号表示）

ssize_t utp_write(utp_socket *socket, void *buf, size_t len) { 
    	struct utp_iovec iovec = {  buf, len };
    	return utp_writev(socket, &iovec, 1);
    }
    
    // Write bytes to the UTP socket. Returns the number of bytes written.
    // 0 indicates the socket is no longer writable, -1 indicates an error
    ssize_t utp_writev(utp_socket *conn, struct utp_iovec *iovec_input, size_t num_iovecs)
    { 
    	....
    	while (!conn->is_full(num_to_send)) { 
    		// Send an outgoing packet.
    		// Also add it to the outgoing of packets that have been sent but not ACKed.
    
    		bytes -= num_to_send;
    		sent  += num_to_send;
            ....
            //关键所在：调用write_outgoing_packet函数发包
    		conn->write_outgoing_packet(num_to_send, ST_DATA, iovec, num_iovecs);
    		num_to_send = min<size_t>(bytes, packet_size);
    
    		if (num_to_send == 0) { 
    			return sent;
    		}
    	}
    
    	bool full = conn->is_full();
    	if (full) { 
    		// mark the socket as not being writable.
    		conn->state = CS_CONNECTED_FULL;
    	}
    	// returns whether or not the socket is still writable
    	// if the congestion window is not full, we can still write to it
    	//return !full;
    	return sent;
    }

这个函数主要进行了众多的异常处理，并确认是否有窗口发包，确认成功则调用write\_outgoing\_packet()函数将数据发送出去。而该函数主要是对数据包进行utp头部的封装，并最终调用 flush\_packets()函数发包。

bool UTPSocket::flush_packets()
    { 
    	size_t packet_size = get_packet_size();
    
    	// send packets that are waiting on the pacer to be sent
    	// i has to be an unsigned 16 bit counter to wrap correctly
    	// signed types are not guaranteed to wrap the way you expect
    	for (uint16 i = seq_nr - cur_window_packets; i != seq_nr; ++i) { 
    		OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(i);
    		if (pkt == 0 || (pkt->transmissions > 0 && pkt->need_resend == false)) continue;
    		// have we run out of quota?
    		if (is_full()) return true;
    
    		// Nagle check
    		// don't send the last packet if we have one packet in-flight
    		// and the current packet is still smaller than packet_size.
    		if (i != ((seq_nr - 1) & ACK_NR_MASK) ||
    			cur_window_packets == 1 ||
    			pkt->payload >= packet_size) { 
    			send_packet(pkt);
    		}
    	}
    	return false;
    }

最终send\_packet()调用send\_data()函数发送数据

void UTPSocket::send_data(byte* b, size_t length, bandwidth_type_t type, uint32 flags)
    { 
    	// time stamp this packet with local time, the stamp goes into
    	// the header of every packet at the 8th byte for 8 bytes :
    	// two integers, check packet.h for more
    	uint64 time = utp_call_get_microseconds(ctx, this);
    
    	PacketFormatV1* b1 = (PacketFormatV1*)b;
    	b1->tv_usec = (uint32)time;
    	b1->reply_micro = reply_micro;
    
    	last_sent_packet = ctx->current_ms;
    
        //若定义了回调函数则调用回调函数发包
    	if (ctx->callbacks[UTP_ON_OVERHEAD_STATISTICS]) { 
    		size_t n;
    		if (type == payload_bandwidth) { 
    			// if this packet carries payload, just
    			// count the header as overhead
    			type = header_overhead;
    			n = get_overhead();
    		} else { 
    			n = length + get_udp_overhead();
    		}
    		utp_call_on_overhead_statistics(ctx, this, true, n, type);
    	}
    	//否则使用自带的send_to
    	send_to_addr(ctx, b, length, addr, flags);
    	removeSocketFromAckList(this);
    }

看到这里大家应该明白了自己设置回调函数是做什么的、什么时候被调用的了吧。这也就是libutp设计者“贴心”的地方，给了大家自己实现发送函数的机会，当然要是自己懒得动手，他也可以帮你直接发出去。

下面再看看接收端。开头依旧是开放给用户的API，utp\_process\_udp()函数。接收端的过程是发送端的镜像：对封装的包头解封，分别记录、计算窗口余量，RTT，是否重传，收到的包是SYN/DATA还是FIN等等。其中utp\_process\_udp()函数主要负责对包种类的分析，依据不同的种类调用不同的函数：

（1）对新来的套接字，调用防火墙回调，accept回调依次检验，检验通过则创建新的utp\_socket并握手  
（2）对DATA数据，调用utp\_process\_incoming()进行解包操作，并且根据ACK信息重新计算该utp\_socket对应的cur\_window的值，再调用用户设置的回调函数进行进一步处理  
（3）对FIN信息进行断开处理，删除utp\_socket等操作

由于utp\_process\_dup()和utp\_process\_incoming()函数比较长而且源码注释很详尽，这里不再列出来，仅仅说明功能，有兴趣的可以自行阅读，源码位于utp\_internal.cpp中。

### 四. 总结 ###

本文简要分析了libutp的收发两端功能的实现，算是对libutp源码阅读和使用的一点点总结和感悟。