UDT API的用法

在分析连接的建立过程之前，先来看一下UDT API的用法。在UDT网络中，通常要有一个UDT Server监听在某台机器的某个UDP端口上，等待客户端的连接；有一个或多个客户端连接UDT Server；UDT Server接收到来自客户端的连接请求后，创建另外一个单独的UDT Socket用于与该客户端进行通信。

先来看一下UDT Server的简单的实现，UDT的开发者已经提供了一些demo程序可供参考，位于app/目录下。

1#include <unistd.h>
2#include <cstdlib>
3#include <cstring>
4#include <netdb.h>
5
6#include <iostream>
7#include <udt.h>
8
9using namespace std;
10
11void* recvdata(void*);
12
13struct UDTUpDown {
14    UDTUpDown() {
15        // use this function to initialize the UDT library
16        UDT::startup();
17    }
18    ~UDTUpDown() {
19        // use this function to release the UDT library
20        UDT::cleanup();
21    }
22};
23
24int main(int argc, char* argv[]) {
25    if ((1 != argc) && ((2 != argc) || (0 == atoi(argv[1])))) {
26        cout << "usage: appserver [server_port]" << endl;
27        return 0;
28    }
29
30    // Automatically start up and clean up UDT module.
31    UDTUpDown _udt_;
32
33    addrinfo hints;
34    addrinfo* res;
35
36    memset(&hints, 0, sizeof(struct addrinfo));
37
38    hints.ai_flags = AI_PASSIVE;
39    hints.ai_family = AF_INET;
40    hints.ai_socktype = SOCK_STREAM;
41    //hints.ai_socktype = SOCK_DGRAM;
42
43    string service("9000");
44    if (2 == argc)
45        service = argv[1];
46
47    if (0 != getaddrinfo(NULL, service.c_str(), &hints, &res)) {
48        cout << "illegal port number or port is busy.\n" << endl;
49        return 0;
50    }
51
52    UDTSOCKET serv = UDT::socket(res->ai_family, res->ai_socktype, res->ai_protocol);
53
54    // UDT Options
55    //UDT::setsockopt(serv, 0, UDT_CC, new CCCFactory<CUDPBlast>, sizeof(CCCFactory<CUDPBlast>));
56    //UDT::setsockopt(serv, 0, UDT_MSS, new int(9000), sizeof(int));
57    //UDT::setsockopt(serv, 0, UDT_RCVBUF, new int(10000000), sizeof(int));
58    //UDT::setsockopt(serv, 0, UDP_RCVBUF, new int(10000000), sizeof(int));
59
60    if (UDT::ERROR == UDT::bind(serv, res->ai_addr, res->ai_addrlen)) {
61        cout << "bind: " << UDT::getlasterror().getErrorMessage() << endl;
62        return 0;
63    }
64
65    freeaddrinfo(res);
66
67    cout << "server is ready at port: " << service << endl;
68
69    if (UDT::ERROR == UDT::listen(serv, 10)) {
70        cout << "listen: " << UDT::getlasterror().getErrorMessage() << endl;
71        return 0;
72    }
73
74    sockaddr_storage clientaddr;
75    int addrlen = sizeof(clientaddr);
76
77    UDTSOCKET recver;
78
79    while (true) {
80        if (UDT::INVALID_SOCK == (recver = UDT::accept(serv, (sockaddr*) &clientaddr, &addrlen))) {
81            cout << "accept: " << UDT::getlasterror().getErrorMessage() << endl;
82            return 0;
83        }
84
85        char clienthost[NI_MAXHOST];
86        char clientservice[NI_MAXSERV];
87        getnameinfo((sockaddr *) &clientaddr, addrlen, clienthost, sizeof(clienthost), clientservice,
88                    sizeof(clientservice), NI_NUMERICHOST | NI_NUMERICSERV);
89        cout << "new connection: " << clienthost << ":" << clientservice << endl;
90
91        pthread_t rcvthread;
92        pthread_create(&rcvthread, NULL, recvdata, new UDTSOCKET(recver));
93        pthread_detach(rcvthread);
94    }
95
96    UDT::close(serv);
97
98    return 0;
99}
100
101void* recvdata(void* usocket) {
102    UDTSOCKET recver = *(UDTSOCKET*) usocket;
103    delete (UDTSOCKET*) usocket;
104
105    char* data;
106    int size = 100000;
107    data = new char[size];
108
109    while (true) {
110        int rsize = 0;
111        int rs;
112        while (rsize < size) {
113            int rcv_size;
114            int var_size = sizeof(int);
115            UDT::getsockopt(recver, 0, UDT_RCVDATA, &rcv_size, &var_size);
116            if (UDT::ERROR == (rs = UDT::recv(recver, data + rsize, size - rsize, 0))) {
117                cout << "recv:" << UDT::getlasterror().getErrorMessage() << endl;
118                break;
119            }
120
121            rsize += rs;
122        }
123
124        if (rsize < size)
125            break;
126    }
127
128    delete[] data;
129
130    UDT::close(recver);
131
132    return NULL;
133}

1. 如在 UDT协议实现分析——UDT初始化和销毁一文中提到的，在调用任何UDT API之前，需要首先调用UDT::startup()来对库做一个初始化，并在结束之后执行UDT::cleanup()做最后的清理。在这个示例中，是创建了一个helper类UDTUpDown来帮助做这些事情的。利用编程语言本身提供的构造-析够机制来对UDT进行初始化和销毁要比手动调用这些函数要可靠得多。

2. 获得本地UDP端口的网络地址。

3. 调用UDT::socket()函数创建一个UDT Socket。这个Socket是UDT抽象出来的一个逻辑的Socket，我们并不能利用这个Socket本身来收发数据。

4. 调用UDT::bind()将创建的UDT Socket绑定到本地端口的网络地址。这一步将会把UDT的逻辑Socket与能够进行数据收发的系统UDP socket进行关联，自此我们就可以利用UDT Socket进行收发数据了。

5. 调用UDT::listen()告诉UDT，把这个UDT Socket做为这个端口上的Listening Socket。一个UDP端口可以被多个UDT Socket复用，在调用UDT::listen()之后，UDT就知道，在有其它节点连接这个UDP端口时，需要把相关的连接请求消息发送到哪个UDT Socket的接收缓冲区了。

对绑定到相同UDP端口的多个不同UDT Socket调用UDT::listen()时，UDT是如何处理的？我们知道，一个UDP端口上最多只能有一个listening Socket。

6. 调用UDT::accept()函数等待其它节点的连接。其它节点连接时，这个函数返回另外一个单独的UDT Socket，以用于与发起连接的节点进行通信。

UDT::accept()函数返回的UDT Socket会被绑定到另外的一个不同的UDP端口，还是会被绑定到listening Socket所绑定的UDP端口？

7. 使用UDT::accept()返回的UDT Socket，利用UDT::recv()与UDT::send()等函数同发起连接的节点进行数据传输。

8. 在不再需要UDT Socket时，调用UDT::close()函数关掉它，以释放资源。

然后再来看下UDT Client的实现：

1#include <unistd.h>
2#include <cstdlib>
3#include <cstring>
4#include <netdb.h>
5
6#include <iostream>
7#include <udt.h>
8
9using namespace std;
10
11void* monitor(void*);
12
13struct UDTUpDown {
14    UDTUpDown() {
15        // use this function to initialize the UDT library
16        UDT::startup();
17    }
18    ~UDTUpDown() {
19        // use this function to release the UDT library
20        UDT::cleanup();
21    }
22};
23
24int main(int argc, char* argv[]) {
25    if ((3 != argc) || (0 == atoi(argv[2]))) {
26        cout << "usage: appclient server_ip server_port" << endl;
27        return 0;
28    }
29
30    // Automatically start up and clean up UDT module.
31    UDTUpDown _udt_;
32
33    struct addrinfo hints, *local, *peer;
34
35    memset(&hints, 0, sizeof(struct addrinfo));
36
37    hints.ai_flags = AI_PASSIVE;
38    hints.ai_family = AF_INET;
39    hints.ai_socktype = SOCK_STREAM;
40    //hints.ai_socktype = SOCK_DGRAM;
41
42    if (0 != getaddrinfo(NULL, "9000", &hints, &local)) {
43        cout << "incorrect network address.\n" << endl;
44        return 0;
45    }
46
47    UDTSOCKET client = UDT::socket(local->ai_family, local->ai_socktype, local->ai_protocol);
48
49    // UDT Options
50    //UDT::setsockopt(client, 0, UDT_CC, new CCCFactory<CUDPBlast>, sizeof(CCCFactory<CUDPBlast>));
51    //UDT::setsockopt(client, 0, UDT_MSS, new int(9000), sizeof(int));
52    //UDT::setsockopt(client, 0, UDT_SNDBUF, new int(10000000), sizeof(int));
53    //UDT::setsockopt(client, 0, UDP_SNDBUF, new int(10000000), sizeof(int));
54    //UDT::setsockopt(client, 0, UDT_MAXBW, new int64_t(12500000), sizeof(int));
55
56    // for rendezvous connection, enable the code below
57    /*
58     UDT::setsockopt(client, 0, UDT_RENDEZVOUS, new bool(true), sizeof(bool));
59     if (UDT::ERROR == UDT::bind(client, local->ai_addr, local->ai_addrlen))
60     {
61     cout << "bind: " << UDT::getlasterror().getErrorMessage() << endl;
62     return 0;
63     }
64     */
65
66    freeaddrinfo(local);
67
68    if (0 != getaddrinfo(argv[1], argv[2], &hints, &peer)) {
69        cout << "incorrect server/peer address. " << argv[1] << ":" << argv[2] << endl;
70        return 0;
71    }
72
73    // connect to the server, implict bind
74    if (UDT::ERROR == UDT::connect(client, peer->ai_addr, peer->ai_addrlen)) {
75        cout << "connect: " << UDT::getlasterror().getErrorMessage() << endl;
76        return 0;
77    }
78
79    freeaddrinfo(peer);
80
81    int size = 100000;
82    char* data = new char[size];
83
84    pthread_create(new pthread_t, NULL, monitor, &client);
85
86    for (int i = 0; i < 1000000; i++) {
87        int ssize = 0;
88        int ss;
89        while (ssize < size) {
90            if (UDT::ERROR == (ss = UDT::send(client, data + ssize, size - ssize, 0))) {
91                cout << "send:" << UDT::getlasterror().getErrorMessage() << endl;
92                break;
93            }
94
95            ssize += ss;
96        }
97
98        if (ssize < size)
99            break;
100    }
101
102    UDT::close(client);
103    delete[] data;
104    return 0;
105}
106
107void* monitor(void* s) {
108    UDTSOCKET u = *(UDTSOCKET*) s;
109    UDT::TRACEINFO perf;
110    cout << "SendRate(Mb/s)\tRTT(ms)\tCWnd\tPktSndPeriod(us)\tRecvACK\tRecvNAK" << endl;
111
112    while (true) {
113        sleep(1);
114        if (UDT::ERROR == UDT::perfmon(u, &perf)) {
115            cout << "perfmon: " << UDT::getlasterror().getErrorMessage() << endl;
116            break;
117        }
118
119        cout << perf.mbpsSendRate << "\t\t" << perf.msRTT << "\t" << perf.pktCongestionWindow << "\t"
120             << perf.usPktSndPeriod << "\t\t\t" << perf.pktRecvACK << "\t" << perf.pktRecvNAK << endl;
121    }
122
123    return NULL;
124}

可以看到UDT Client连接UDT Server并发送数据的过程大体如下：

1. 同样需要在调用任何UDT API之前，先调用UDT::startup()来对库做初始化，并在结束之后执行UDT::cleanup()做最后的清理。这里同样使用helper类UDTUpDown来帮助做这些事情。

2. 获得UDT Server的网络地址。

3. 调用UDT::socket()函数创建一个UDT Socket。这个Socket可以与特定的本地地址绑定，也可以不绑定。如果绑定，则发送数据时的出口地址就是该端口，如果不绑定，出口地址则是一个不确定的值。

4. 调用UDT::connect()连接UDT Server。

5. 使用UDT Socket，利用UDT::recv()与UDT::send()等函数同UDT Server进行数据传输。

6. 在不再需要UDT Socket时，调用UDT::close()函数关掉它，以释放资源。

UDT基本的收发数据的API的用法大体如上面所示。接着我们来看，这些函数是如何实现的。

UDT Socket的创建

无论是UDT Server要listening，还是UDT client要连接UDT Server，在调用UDT::startup()初始化UDT之后，首先要做的事情都是调用UDT::socket()创建UDT Socket了。我们就来看一下创建UDT Socket的过程：

1UDTSOCKET CUDTUnited::newSocket(int af, int type) {
2    if ((type != SOCK_STREAM) && (type != SOCK_DGRAM))
3        throw CUDTException(5, 3, 0);
4
5    CUDTSocket* ns = NULL;
6
7    try {
8        ns = new CUDTSocket;
9        ns->m_pUDT = new CUDT;
10        if (AF_INET == af) {
11            ns->m_pSelfAddr = (sockaddr*) (new sockaddr_in);
12            ((sockaddr_in*) (ns->m_pSelfAddr))->sin_port = 0;
13        } else {
14            ns->m_pSelfAddr = (sockaddr*) (new sockaddr_in6);
15            ((sockaddr_in6*) (ns->m_pSelfAddr))->sin6_port = 0;
16        }
17    } catch (...) {
18        delete ns;
19        throw CUDTException(3, 2, 0);
20    }
21
22    CGuard::enterCS(m_IDLock);
23    ns->m_SocketID = --m_SocketID;
24    CGuard::leaveCS(m_IDLock);
25
26    ns->m_Status = INIT;
27    ns->m_ListenSocket = 0;
28    ns->m_pUDT->m_SocketID = ns->m_SocketID;
29    ns->m_pUDT->m_iSockType = (SOCK_STREAM == type) ? UDT_STREAM : UDT_DGRAM;
30    ns->m_pUDT->m_iIPversion = ns->m_iIPversion = af;
31    ns->m_pUDT->m_pCache = m_pCache;
32
33    // protect the m_Sockets structure.
34    CGuard::enterCS(m_ControlLock);
35    try {
36        m_Sockets[ns->m_SocketID] = ns;
37    } catch (...) {
38        //failure and rollback
39        CGuard::leaveCS(m_ControlLock);
40        delete ns;
41        ns = NULL;
42    }
43    CGuard::leaveCS(m_ControlLock);
44
45    if (NULL == ns)
46        throw CUDTException(3, 2, 0);
47
48    return ns->m_SocketID;
49}
50
51
52UDTSOCKET CUDT::socket(int af, int type, int) {
53    if (!s_UDTUnited.m_bGCStatus)
54        s_UDTUnited.startup();
55
56    try {
57        return s_UDTUnited.newSocket(af, type);
58    } catch (CUDTException& e) {
59        s_UDTUnited.setError(new CUDTException(e));
60        return INVALID_SOCK;
61    } catch (bad_alloc&) {
62        s_UDTUnited.setError(new CUDTException(3, 2, 0));
63        return INVALID_SOCK;
64    } catch (...) {
65        s_UDTUnited.setError(new CUDTException(-1, 0, 0));
66        return INVALID_SOCK;
67    }
68}
69
70
71
72UDTSOCKET socket(int af, int type, int protocol) {
73    return CUDT::socket(af, type, protocol);
74}

调用流程为，UDT::socket() -> CUDT::socket() -> CUDTUnited::newSocket()。

在CUDT::socket()中，我们看到，它会首先检查s_UDTUnited.m_bGCStatus，若发现UDT还没有初始化完成的话，则会调用s_UDTUnited.startup()进行初始化。这个地方对UDT状态s_UDTUnited.m_bGCStatus的检查没有问题，但在发现UDT没有初始化完成时调用s_UDTUnited.startup()似乎并不恰当。这个地方调用了s_UDTUnited.startup()，那与这次调用相对应的cleanup()又在哪调用了呢？显然在UDT内部是没有。若是调用cleanup()的职责总是在UDT的使用者，那倒不如在这个地方返回错误给调用者，完全让调用者来管理UDT的生命周期，以尽可能地避免资源泄漏。

创建UDT Socket的工作实际都在CUDTUnited::newSocket()函数中完成。可以看到，在这个函数中主要做了如下的这样一些事情：

1. 创建一个CUDTSocket对象ns，并创建一个CUDT对象被ns->m_pUDT引用。初始化ns对象的Self网络地址，端口会被设置为0。在UDT中使用CUDTSocket和CUDT共同来描述一个Socket。每个UDT Socket都会有其相应的CUDTSocket对象和CUDT对象。

可以看一下CUDTSocket类的定义，来了解它都描述了UDT Socket的哪些属性(src/api.h)：

1class CUDTSocket {
2 public:
3    CUDTSocket();
4    ~CUDTSocket();
5
6    UDTSTATUS m_Status;                       // current socket state
7
8    uint64_t m_TimeStamp;                     // time when the socket is closed
9
10    int m_iIPversion;                         // IP version
11    sockaddr* m_pSelfAddr;                    // pointer to the local address of the socket
12    sockaddr* m_pPeerAddr;                    // pointer to the peer address of the socket
13
14    UDTSOCKET m_SocketID;                     // socket ID
15    UDTSOCKET m_ListenSocket;                 // ID of the listener socket; 0 means this is an independent socket
16
17    UDTSOCKET m_PeerID;                       // peer socket ID
18    int32_t m_iISN;                      // initial sequence number, used to tell different connection from same IP:port
19
20    CUDT* m_pUDT;                             // pointer to the UDT entity
21
22    std::set<UDTSOCKET>* m_pQueuedSockets;    // set of connections waiting for accept()
23    std::set<UDTSOCKET>* m_pAcceptSockets;    // set of accept()ed connections
24
25    pthread_cond_t m_AcceptCond;              // used to block "accept" call
26    pthread_mutex_t m_AcceptLock;             // mutex associated to m_AcceptCond
27
28    unsigned int m_uiBackLog;                 // maximum number of connections in queue
29
30    int m_iMuxID;                             // multiplexer ID
31
32    pthread_mutex_t m_ControlLock;            // lock this socket exclusively for control APIs: bind/listen/connect
33
34 private:
35    CUDTSocket(const CUDTSocket&);
36    CUDTSocket& operator=(const CUDTSocket&);
37};

这个类只提供了构造和析构两个成员函数。还声明了私有的copy构造函数和赋值操作符函数，但没有定义它们，以避免类对象的复制。

可以再来看一下CUDTSocket的构造函数实现(src/api.h)：

1CUDTSocket::CUDTSocket()
2        : m_Status(INIT),
3          m_TimeStamp(0),
4          m_iIPversion(0),
5          m_pSelfAddr(NULL),
6          m_pPeerAddr(NULL),
7          m_SocketID(0),
8          m_ListenSocket(0),
9          m_PeerID(0),
10          m_iISN(0),
11          m_pUDT(NULL),
12          m_pQueuedSockets(NULL),
13          m_pAcceptSockets(NULL),
14          m_AcceptCond(),
15          m_AcceptLock(),
16          m_uiBackLog(0),
17          m_iMuxID(-1) {
18#ifndef WIN32
19    pthread_mutex_init(&m_AcceptLock, NULL);
20    pthread_cond_init(&m_AcceptCond, NULL);
21    pthread_mutex_init(&m_ControlLock, NULL);
22#else
23    m_AcceptLock = CreateMutex(NULL, false, NULL);
24    m_AcceptCond = CreateEvent(NULL, false, false, NULL);
25    m_ControlLock = CreateMutex(NULL, false, NULL);
26#endif
27}

特别注意m_Status的初始化，该值被初始化为了INIT。从状态机的角度来看CUDTSocket，在它刚被new出来时，它处于INIT状态。

CUDT类有两个主要的职责，一是描述UDT Socket，包括所有的非静态成员变量和非静态成员函数，定义UDT Socket的大部分属性和所能提供操作；二是提供API，包括绝大部分的static成员函数，这些函数将调用者与UDT内部的实现连接起来。CUDT类这样的设计，明显违背了OO的SRP单一职责原则，这多少还是给代码的阅读带来了一定的障碍。再来看一下CUDT的构造函数实现(src/core.cpp)：

1CUDT::CUDT() {
2    m_pSndBuffer = NULL;
3    m_pRcvBuffer = NULL;
4    m_pSndLossList = NULL;
5    m_pRcvLossList = NULL;
6    m_pACKWindow = NULL;
7    m_pSndTimeWindow = NULL;
8    m_pRcvTimeWindow = NULL;
9
10    m_pSndQueue = NULL;
11    m_pRcvQueue = NULL;
12    m_pPeerAddr = NULL;
13    m_pSNode = NULL;
14    m_pRNode = NULL;
15
16    // Initilize mutex and condition variables
17    initSynch();
18
19    // Default UDT configurations
20    m_iMSS = 1500;
21    m_bSynSending = true;
22    m_bSynRecving = true;
23    m_iFlightFlagSize = 25600;
24    m_iSndBufSize = 8192;
25    m_iRcvBufSize = 8192;  //Rcv buffer MUST NOT be bigger than Flight Flag size
26    m_Linger.l_onoff = 1;
27    m_Linger.l_linger = 180;
28    m_iUDPSndBufSize = 65536;
29    m_iUDPRcvBufSize = m_iRcvBufSize * m_iMSS;
30    m_iSockType = UDT_STREAM;
31    m_iIPversion = AF_INET;
32    m_bRendezvous = false;
33    m_iSndTimeOut = -1;
34    m_iRcvTimeOut = -1;
35    m_bReuseAddr = true;
36    m_llMaxBW = -1;
37
38    m_pCCFactory = new CCCFactory<CUDTCC>;
39    m_pCC = NULL;
40    m_pCache = NULL;
41
42    // Initial status
43    m_bOpened = false;
44    m_bListening = false;
45    m_bConnecting = false;
46    m_bConnected = false;
47    m_bClosing = false;
48    m_bShutdown = false;
49    m_bBroken = false;
50    m_bPeerHealth = true;
51    m_ullLingerExpiration = 0;
52}
53
54
55void CUDT::initSynch() {
56#ifndef WIN32
57    pthread_mutex_init(&m_SendBlockLock, NULL);
58    pthread_cond_init(&m_SendBlockCond, NULL);
59    pthread_mutex_init(&m_RecvDataLock, NULL);
60    pthread_cond_init(&m_RecvDataCond, NULL);
61    pthread_mutex_init(&m_SendLock, NULL);
62    pthread_mutex_init(&m_RecvLock, NULL);
63    pthread_mutex_init(&m_AckLock, NULL);
64    pthread_mutex_init(&m_ConnectionLock, NULL);
65#else
66    m_SendBlockLock = CreateMutex(NULL, false, NULL);
67    m_SendBlockCond = CreateEvent(NULL, false, false, NULL);
68    m_RecvDataLock = CreateMutex(NULL, false, NULL);
69    m_RecvDataCond = CreateEvent(NULL, false, false, NULL);
70    m_SendLock = CreateMutex(NULL, false, NULL);
71    m_RecvLock = CreateMutex(NULL, false, NULL);
72    m_AckLock = CreateMutex(NULL, false, NULL);
73    m_ConnectionLock = CreateMutex(NULL, false, NULL);
74#endif
75}

都是成员变量的初始化，后续再来详细了解这些成员变量的作用。

2. 为Socket分配SocketID，其值为CUDTUnited的m_SocketID递减的结果。m_SocketID在CUDTUnited的构造函数中初始化：

1CUDTUnited::CUDTUnited()
2        : m_Sockets(),
3          m_ControlLock(),
4          m_IDLock(),
5          m_SocketID(0),
6          m_TLSError(),
7          m_mMultiplexer(),
8          m_MultiplexerLock(),
9          m_pCache(NULL),
10          m_bClosing(false),
11          m_GCStopLock(),
12          m_GCStopCond(),
13          m_InitLock(),
14          m_iInstanceCount(0),
15          m_bGCStatus(false),
16          m_GCThread(),
17          m_ClosedSockets() {
18    // Socket ID MUST start from a random value
19    srand((unsigned int) CTimer::getTime());
20    m_SocketID = 1 + (int) ((1 << 30) * (double(rand()) / RAND_MAX));
21
22#ifndef WIN32
23    pthread_mutex_init(&m_ControlLock, NULL);
24    pthread_mutex_init(&m_IDLock, NULL);
25    pthread_mutex_init(&m_InitLock, NULL);
26#else
27    m_ControlLock = CreateMutex(NULL, false, NULL);
28    m_IDLock = CreateMutex(NULL, false, NULL);
29    m_InitLock = CreateMutex(NULL, false, NULL);
30#endif
31
32#ifndef WIN32
33    pthread_key_create(&m_TLSError, TLSDestroy);
34#else
35    m_TLSError = TlsAlloc();
36    m_TLSLock = CreateMutex(NULL, false, NULL);
37#endif
38
39    m_pCache = new CCache<CInfoBlock>;
40}

m_SocketID的初始值是一个随机数。

3. 初始化ns及它的CUDT对象的一些成员变量。特别注意ns->m_Status的赋值，这里该值被赋为了INIT。从状态机的角度来看待CUDTSocket，在执行UDT Socket创建结束执行时，它处于INIT状态下。

4. 将ns放在std::map<UDTSOCKET, CUDTSocket*> m_Sockets中。

5. 将UDT socket的SocketID返回给调用者。

这个接口直接返回一个表示UDT Socket的类对象，而不是一个handle，以便于调用者在调用UDT API时无需每次都输入UDT Socket handle参数，这样的API设计相对于UDT的这种API设计，有什么样的优缺点？

UDT的错误处理机制

在前面UDT Server和UDT Client的demo程序中，我们有看到，所有的UDT API都会在出错时返回一个错误码，比如UDT::bind()、UDT::bind2()、UDT::listen()和UDT::connect()等返回值类型为int的函数，在出错时返回UDT::ERROR，UDT::socket()和UDT::accept()等返回值类型为UDTSOCKET的函数在出错时返回UDT::INVALID_SOCK。

UDT API的调用者在检测到它们返回了错误值时，通过调用UDT::getlasterror()函数来获取关于异常更加详细的信息。这里我们就来看一下这套机制的实现。

注意看CUDT::socket()函数的实现。在这个函数中，会将实际创建UDT Socket的任务委托给s_UDTUnited.newSocket()，但这个调用会被包在一个try-catch块中。在s_UDTUnited.newSocket()函数执行的过程中发生任何的异常，都会先被CUDT::socket()捕获。CUDT::socket()函数在捕获这些异常之后，会根据捕获的异常的类型创建不同的CUDTException对象，并通过调用CUDTUnited::setError()函数将该CUDTException对象设置给s_UDTUnited。

我们来看一下CUDTUnited::setError()函数的实现(src/api.cpp)：

1void CUDTUnited::setError(CUDTException* e) {
2#ifndef WIN32
3    delete (CUDTException*) pthread_getspecific(m_TLSError);
4    pthread_setspecific(m_TLSError, e);
5#else
6    CGuard tg(m_TLSLock);
7    delete (CUDTException*)TlsGetValue(m_TLSError);
8    TlsSetValue(m_TLSError, e);
9    m_mTLSRecord[GetCurrentThreadId()] = e;
10#endif
11}

在这个函数中，会首先取出线程局部存储变量m_TLSError中保存的本线程上一次创建的 CUDTException对象，将其delete掉，并将本次创建的CUDTException对象设置进去。CUDTUnited类定义中m_TLSError的声明(src/api.h)：

1private:
2    pthread_key_t m_TLSError;                         // thread local error record (last error)
3#ifndef WIN32
4    static void TLSDestroy(void* e) {
5        if (NULL != e)
6            delete (CUDTException*) e;
7    }
8#else
9    std::map<DWORD, CUDTException*> m_mTLSRecord;
10    void checkTLSValue();
11    pthread_mutex_t m_TLSLock;
12#endif

在CUDTUnited构造函数中也可以看到对这个对象的初始化。 CUDTUnited::TLSDestroy()函数是m_TLSError的析构函数，在m_TLSError最后被销毁时，这个函数被调用，以便于释放搜有还未释放的资源。

再来看UDT API的调用者获取上一次发生的异常的函数UDT::getlasterror()：

1CUDTException* CUDTUnited::getError() {
2#ifndef WIN32
3    if (NULL == pthread_getspecific(m_TLSError))
4        pthread_setspecific(m_TLSError, new CUDTException);
5    return (CUDTException*) pthread_getspecific(m_TLSError);
6#else
7    CGuard tg(m_TLSLock);
8    if(NULL == TlsGetValue(m_TLSError))
9    {
10        CUDTException* e = new CUDTException;
11        TlsSetValue(m_TLSError, e);
12        m_mTLSRecord[GetCurrentThreadId()] = e;
13    }
14    return (CUDTException*)TlsGetValue(m_TLSError);
15#endif
16}
17
18
19
20CUDTException& CUDT::getlasterror() {
21    return *s_UDTUnited.getError();
22}
23
24
25ERRORINFO& getlasterror() {
26    return CUDT::getlasterror();
27}

调用过程为UDT::getlasterror() -> CUDT::getlasterror() -> CUDTUnited::getError()。

主要就是从s_UDTUnited的线程局部存储变量m_TLSError中取出前面设置的本线程上次创建的CUDTException对象返回给调用者。

总结一下，UDT的使用者在调用UDT API时，UDT API会直接调用CUDT类对应的static API函数，在CUDT类的这些static API函数中会将做实际事情的工作委托给s_UDTUnited的相应函数，但这个委托调用会被包在一个try-catch block中。s_UDTUnited的函数在遇到异常情况时抛出异常，CUDT类的static API函数捕获异常，根据捕获到的异常的具体类型，创建不同的CUDTException对象设置给s_UDTUnited的线程局部存储变量m_TLSError中并向UDT API调用者返回错误码，UDT API的调用者检测到错误码后，通过UDT::getlasterror()获取存储在m_TLSError中的异常。

此处可以看到，CUDT提供的这一层API，一个比较重要的作用大概就是做异常处理了。

在UDT中，使用CUDTException来描述所有出现的异常。可以看一下这个类的定义(src/udt.h)：

1class UDT_API CUDTException {
2 public:
3    CUDTException(int major = 0, int minor = 0, int err = -1);
4    CUDTException(const CUDTException& e);
5    virtual ~CUDTException();
6
7    // Functionality:
8    //    Get the description of the exception.
9    // Parameters:
10    //    None.
11    // Returned value:
12    //    Text message for the exception description.
13
14    virtual const char* getErrorMessage();
15
16    // Functionality:
17    //    Get the system errno for the exception.
18    // Parameters:
19    //    None.
20    // Returned value:
21    //    errno.
22
23    virtual int getErrorCode() const;
24
25    // Functionality:
26    //    Clear the error code.
27    // Parameters:
28    //    None.
29    // Returned value:
30    //    None.
31
32    virtual void clear();
33
34 private:
35    int m_iMajor;        // major exception categories
36
37// 0: correct condition
38// 1: network setup exception
39// 2: network connection broken
40// 3: memory exception
41// 4: file exception
42// 5: method not supported
43// 6+: undefined error
44
45    int m_iMinor;		// for specific error reasons
46    int m_iErrno;		// errno returned by the system if there is any
47    std::string m_strMsg;	// text error message
48
49    std::string m_strAPI;	// the name of UDT function that returns the error
50    std::string m_strDebug;  // debug information, set to the original place that causes the error
51
52 public:
53    // Error Code
54    static const int SUCCESS;
55    static const int ECONNSETUP;
56    static const int ENOSERVER;
57    static const int ECONNREJ;
58    static const int ESOCKFAIL;
59    static const int ESECFAIL;
60    static const int ECONNFAIL;
61    static const int ECONNLOST;
62    static const int ENOCONN;
63    static const int ERESOURCE;
64    static const int ETHREAD;
65    static const int ENOBUF;
66    static const int EFILE;
67    static const int EINVRDOFF;
68    static const int ERDPERM;
69    static const int EINVWROFF;
70    static const int EWRPERM;
71    static const int EINVOP;
72    static const int EBOUNDSOCK;
73    static const int ECONNSOCK;
74    static const int EINVPARAM;
75    static const int EINVSOCK;
76    static const int EUNBOUNDSOCK;
77    static const int ENOLISTEN;
78    static const int ERDVNOSERV;
79    static const int ERDVUNBOUND;
80    static const int ESTREAMILL;
81    static const int EDGRAMILL;
82    static const int EDUPLISTEN;
83    static const int ELARGEMSG;
84    static const int EINVPOLLID;
85    static const int EASYNCFAIL;
86    static const int EASYNCSND;
87    static const int EASYNCRCV;
88    static const int ETIMEOUT;
89    static const int EPEERERR;
90    static const int EUNKNOWN;
91};

这个class主要通过Major错误码和Minor错误码来描述异常情况，如果是调用系统调用出错了，还会用Errno值。

具体看一下这个class的实现，特别是CUDTException::getErrorMessage()函数，来了解每一个错误码所代表的含义：

1CUDTException::CUDTException(int major, int minor, int err)
2        : m_iMajor(major),
3          m_iMinor(minor) {
4    if (-1 == err)
5#ifndef WIN32
6        m_iErrno = errno;
7#else
8        m_iErrno = GetLastError();
9#endif
10        else
11        m_iErrno = err;
12}
13
14CUDTException::CUDTException(const CUDTException& e)
15        : m_iMajor(e.m_iMajor),
16          m_iMinor(e.m_iMinor),
17          m_iErrno(e.m_iErrno),
18          m_strMsg() {
19}
20
21CUDTException::~CUDTException() {
22}
23
24const char* CUDTException::getErrorMessage() {
25    // translate "Major:Minor" code into text message.
26
27    switch (m_iMajor) {
28        case 0:
29            m_strMsg = "Success";
30            break;
31
32        case 1:
33            m_strMsg = "Connection setup failure";
34
35            switch (m_iMinor) {
36                case 1:
37                    m_strMsg += ": connection time out";
38                    break;
39
40                case 2:
41                    m_strMsg += ": connection rejected";
42                    break;
43
44                case 3:
45                    m_strMsg += ": unable to create/configure UDP socket";
46                    break;
47
48                case 4:
49                    m_strMsg += ": abort for security reasons";
50                    break;
51
52                default:
53                    break;
54            }
55
56            break;
57
58        case 2:
59            switch (m_iMinor) {
60                case 1:
61                    m_strMsg = "Connection was broken";
62                    break;
63
64                case 2:
65                    m_strMsg = "Connection does not exist";
66                    break;
67
68                default:
69                    break;
70            }
71
72            break;
73
74        case 3:
75            m_strMsg = "System resource failure";
76
77            switch (m_iMinor) {
78                case 1:
79                    m_strMsg += ": unable to create new threads";
80                    break;
81
82                case 2:
83                    m_strMsg += ": unable to allocate buffers";
84                    break;
85
86                default:
87                    break;
88            }
89
90            break;
91
92        case 4:
93            m_strMsg = "File system failure";
94
95            switch (m_iMinor) {
96                case 1:
97                    m_strMsg += ": cannot seek read position";
98                    break;
99
100                case 2:
101                    m_strMsg += ": failure in read";
102                    break;
103
104                case 3:
105                    m_strMsg += ": cannot seek write position";
106                    break;
107
108                case 4:
109                    m_strMsg += ": failure in write";
110                    break;
111
112                default:
113                    break;
114            }
115
116            break;
117
118        case 5:
119            m_strMsg = "Operation not supported";
120
121            switch (m_iMinor) {
122                case 1:
123                    m_strMsg += ": Cannot do this operation on a BOUND socket";
124                    break;
125
126                case 2:
127                    m_strMsg += ": Cannot do this operation on a CONNECTED socket";
128                    break;
129
130                case 3:
131                    m_strMsg += ": Bad parameters";
132                    break;
133
134                case 4:
135                    m_strMsg += ": Invalid socket ID";
136                    break;
137
138                case 5:
139                    m_strMsg += ": Cannot do this operation on an UNBOUND socket";
140                    break;
141
142                case 6:
143                    m_strMsg += ": Socket is not in listening state";
144                    break;
145
146                case 7:
147                    m_strMsg += ": Listen/accept is not supported in rendezous connection setup";
148                    break;
149
150                case 8:
151                    m_strMsg += ": Cannot call connect on UNBOUND socket in rendezvous connection setup";
152                    break;
153
154                case 9:
155                    m_strMsg += ": This operation is not supported in SOCK_STREAM mode";
156                    break;
157
158                case 10:
159                    m_strMsg += ": This operation is not supported in SOCK_DGRAM mode";
160                    break;
161
162                case 11:
163                    m_strMsg += ": Another socket is already listening on the same port";
164                    break;
165
166                case 12:
167                    m_strMsg += ": Message is too large to send (it must be less than the UDT send buffer size)";
168                    break;
169
170                case 13:
171                    m_strMsg += ": Invalid epoll ID";
172                    break;
173
174                default:
175                    break;
176            }
177
178            break;
179
180        case 6:
181            m_strMsg = "Non-blocking call failure";
182
183            switch (m_iMinor) {
184                case 1:
185                    m_strMsg += ": no buffer available for sending";
186                    break;
187
188                case 2:
189                    m_strMsg += ": no data available for reading";
190                    break;
191
192                default:
193                    break;
194            }
195
196            break;
197
198        case 7:
199            m_strMsg = "The peer side has signalled an error";
200
201            break;
202
203        default:
204            m_strMsg = "Unknown error";
205    }
206
207    // Adding "errno" information
208    if ((0 != m_iMajor) && (0 < m_iErrno)) {
209        m_strMsg += ": ";
210#ifndef WIN32
211        char errmsg[1024];
212        if (strerror_r(m_iErrno, errmsg, 1024) == 0)
213            m_strMsg += errmsg;
214#else
215        LPVOID lpMsgBuf;
216        FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS, NULL, m_iErrno, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), (LPTSTR)&lpMsgBuf, 0, NULL);
217        m_strMsg += (char*)lpMsgBuf;
218        LocalFree(lpMsgBuf);
219#endif
220    }
221
222    // period
223#ifndef WIN32
224    m_strMsg += ".";
225#endif
226
227    return m_strMsg.c_str();
228}
229
230int CUDTException::getErrorCode() const {
231    return m_iMajor * 1000 + m_iMinor;
232}
233
234void CUDTException::clear() {
235    m_iMajor = 0;
236    m_iMinor = 0;
237    m_iErrno = 0;
238}
239
240const int CUDTException::SUCCESS = 0;
241const int CUDTException::ECONNSETUP = 1000;
242const int CUDTException::ENOSERVER = 1001;
243const int CUDTException::ECONNREJ = 1002;
244const int CUDTException::ESOCKFAIL = 1003;
245const int CUDTException::ESECFAIL = 1004;
246const int CUDTException::ECONNFAIL = 2000;
247const int CUDTException::ECONNLOST = 2001;
248const int CUDTException::ENOCONN = 2002;
249const int CUDTException::ERESOURCE = 3000;
250const int CUDTException::ETHREAD = 3001;
251const int CUDTException::ENOBUF = 3002;
252const int CUDTException::EFILE = 4000;
253const int CUDTException::EINVRDOFF = 4001;
254const int CUDTException::ERDPERM = 4002;
255const int CUDTException::EINVWROFF = 4003;
256const int CUDTException::EWRPERM = 4004;
257const int CUDTException::EINVOP = 5000;
258const int CUDTException::EBOUNDSOCK = 5001;
259const int CUDTException::ECONNSOCK = 5002;
260const int CUDTException::EINVPARAM = 5003;
261const int CUDTException::EINVSOCK = 5004;
262const int CUDTException::EUNBOUNDSOCK = 5005;
263const int CUDTException::ENOLISTEN = 5006;
264const int CUDTException::ERDVNOSERV = 5007;
265const int CUDTException::ERDVUNBOUND = 5008;
266const int CUDTException::ESTREAMILL = 5009;
267const int CUDTException::EDGRAMILL = 5010;
268const int CUDTException::EDUPLISTEN = 5011;
269const int CUDTException::ELARGEMSG = 5012;
270const int CUDTException::EINVPOLLID = 5013;
271const int CUDTException::EASYNCFAIL = 6000;
272const int CUDTException::EASYNCSND = 6001;
273const int CUDTException::EASYNCRCV = 6002;
274const int CUDTException::ETIMEOUT = 6003;
275const int CUDTException::EPEERERR = 7000;
276const int CUDTException::EUNKNOWN = -1;

Done。

UDT协议实现分析——UDT Socket的创建

UDT API的用法

UDT Socket的创建

UDT的错误处理机制

评论区

相关推荐