UDT协议实现分析——close过程

最后再来看一下close的过程(src/api.cpp)：

1int CUDTUnited::close(const UDTSOCKET u) {
2    CUDTSocket* s = locate(u);
3    if (NULL == s)
4        throw CUDTException(5, 4, 0);
5
6    CGuard socket_cg(s->m_ControlLock);
7
8    if (s->m_Status == LISTENING) {
9        if (s->m_pUDT->m_bBroken)
10            return 0;
11
12        s->m_TimeStamp = CTimer::getTime();
13        s->m_pUDT->m_bBroken = true;
14
15        // broadcast all "accept" waiting
16#ifndef WIN32
17        pthread_mutex_lock(&(s->m_AcceptLock));
18        pthread_cond_broadcast(&(s->m_AcceptCond));
19        pthread_mutex_unlock(&(s->m_AcceptLock));
20#else
21        SetEvent(s->m_AcceptCond);
22#endif
23
24        return 0;
25    }
26
27    s->m_pUDT->close();
28
29    // synchronize with garbage collection.
30    CGuard manager_cg(m_ControlLock);
31
32    // since "s" is located before m_ControlLock, locate it again in case it became invalid
33    map<UDTSOCKET, CUDTSocket*>::iterator i = m_Sockets.find(u);
34    if ((i == m_Sockets.end()) || (i->second->m_Status == CLOSED))
35        return 0;
36    s = i->second;
37
38    s->m_Status = CLOSED;
39
40    // a socket will not be immediated removed when it is closed
41    // in order to prevent other methods from accessing invalid address
42    // a timer is started and the socket will be removed after approximately 1 second
43    s->m_TimeStamp = CTimer::getTime();
44
45    m_Sockets.erase(s->m_SocketID);
46    m_ClosedSockets.insert(pair<UDTSOCKET, CUDTSocket*>(s->m_SocketID, s));
47
48    CTimer::triggerEvent();
49
50    return 0;
51}
52
53
54
55int CUDT::close(UDTSOCKET u) {
56    try {
57        return s_UDTUnited.close(u);
58    } catch (CUDTException &e) {
59        s_UDTUnited.setError(new CUDTException(e));
60        return ERROR;
61    } catch (...) {
62        s_UDTUnited.setError(new CUDTException(-1, 0, 0));
63        return ERROR;
64    }
65}
66
67
68int close(UDTSOCKET u) {
69    return CUDT::close(u);
70}

这个API的实现结构并没有什么特别值得关注的地方。直接来看CUDTUnited::close()。在CUDTUnited::close()函数中，主要是分两种情况来处理的：一种是Server端用于接受连接的Listening socket；另一种是常规的用于进行数据收发的socket。

对于第一种情况，可以看到，这里主要是设置了对应的CUDTSocket s的m_TimeStamp为当前时间，并将s->m_pUDT->m_bBroken置为true，然后将等待在accept的线程唤醒就结束了。不管是在CUDTUnited::close()中，还是在被它唤醒的执行CUDTUnited::accept()的线程中，都没有看到有实际做最后的清理的动作，比如被加入队列的为新连接请求创建的UDT Socket的清理，将当前UDT Socket从RcvQueue的listener移除。这些清理的动作只有在UDT的垃圾回收线程里做了。

接着是第二种情况，可以看到首先是执行了s->m_pUDT->close()，不难想像这个close中做的事情一定特别多；做状态的切换，将Socket的状态切换到CLOSED状态；更新对应的CUDTSocket s的m_TimeStamp为当前时间；将UDT Socket从总的打开socket表m_Sockets中移除，并加入到已关闭socket表m_ClosedSockets中，然后trigger一个Timer event并返回。

然后来看CUDT::close()(src/core.cpp)：

1void CUDT::close() {
2    if (!m_bOpened)
3        return;
4
5    if (0 != m_Linger.l_onoff) {
6        uint64_t entertime = CTimer::getTime();
7
8        while (!m_bBroken && m_bConnected && (m_pSndBuffer->getCurrBufSize() > 0)
9                && (CTimer::getTime() - entertime < m_Linger.l_linger * 1000000ULL)) {
10            // linger has been checked by previous close() call and has expired
11            if (m_ullLingerExpiration >= entertime)
12                break;
13
14            if (!m_bSynSending) {
15                // if this socket enables asynchronous sending, return immediately and let GC to close it later
16                if (0 == m_ullLingerExpiration)
17                    m_ullLingerExpiration = entertime + m_Linger.l_linger * 1000000ULL;
18
19                return;
20            }
21
22#ifndef WIN32
23            timespec ts;
24            ts.tv_sec = 0;
25            ts.tv_nsec = 1000000;
26            nanosleep(&ts, NULL);
27#else
28            Sleep(1);
29#endif
30        }
31    }
32
33    // remove this socket from the snd queue
34    if (m_bConnected)
35        m_pSndQueue->m_pSndUList->remove(this);
36
37    // trigger any pending IO events.
38    s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_ERR, true);
39    // then remove itself from all epoll monitoring
40    try {
41        for (set<int>::iterator i = m_sPollID.begin(); i != m_sPollID.end(); ++i)
42            s_UDTUnited.m_EPoll.remove_usock(*i, m_SocketID);
43    } catch (...) {
44    }
45
46    if (!m_bOpened)
47        return;
48
49    // Inform the threads handler to stop.
50    m_bClosing = true;
51
52    CGuard cg(m_ConnectionLock);
53
54    // Signal the sender and recver if they are waiting for data.
55    releaseSynch();
56
57    if (m_bListening) {
58        m_bListening = false;
59        m_pRcvQueue->removeListener(this);
60    } else if (m_bConnecting) {
61        m_pRcvQueue->removeConnector(m_SocketID);
62    }
63
64    if (m_bConnected) {
65        if (!m_bShutdown)
66            sendCtrl(5);
67
68        m_pCC->close();
69
70        // Store current connection information.
71        CInfoBlock ib;
72        ib.m_iIPversion = m_iIPversion;
73        CInfoBlock::convert(m_pPeerAddr, m_iIPversion, ib.m_piIP);
74        ib.m_iRTT = m_iRTT;
75        ib.m_iBandwidth = m_iBandwidth;
76        m_pCache->update(&ib);
77
78        m_bConnected = false;
79    }
80
81    // waiting all send and recv calls to stop
82    CGuard sendguard(m_SendLock);
83    CGuard recvguard(m_RecvLock);
84
85    // CLOSED.
86    m_bOpened = false;
87}

可以看到，这个函数中做的事情大体如下：

1. 检查Open状态m_bOpened，若m_bOpened为false，则直接返回，否则继续执行。

2. 等待一段时间，以使在发送缓冲区中还没有发送完成的数据能够可靠地发送完成，当然过了一定时间之后，发送缓冲区中还是存在没有可靠地发送完成的数据，则那些数据会被直接丢弃掉。这里的等待，是一种比较高频率地轮询。

3. 如果当前处于Connected状态，则将当前UDT Socket从发送队列的发送者列表m_pSndUList中移除出去。

4. 这里再一次检查了m_bOpened的值。

5. 将m_bClosing置为true。

如我们前面看到的，连接成功建立之后，UDT Socket会被加入到RcvQueue的数据接收者列表m_pRcvUList和m_pHash中，但这里在关闭UDT Socket时，却没有看到有将当前UDT Socket从那些列表中移除的code。这到底是怎么回事呢？

来看一下CRcvQueue::worker()中的这段code：

1CRNode* ul = self->m_pRcvUList->m_pUList;
2        uint64_t ctime = currtime - 100000 * CTimer::getCPUFrequency();
3        while ((NULL != ul) && (ul->m_llTimeStamp < ctime)) {
4            CUDT* u = ul->m_pUDT;
5
6            if (u->m_bConnected && !u->m_bBroken && !u->m_bClosing) {
7                u->checkTimers();
8                self->m_pRcvUList->update(u);
9            } else {
10                // the socket must be removed from Hash table first, then RcvUList
11                self->m_pHash->remove(u->m_SocketID);
12                self->m_pRcvUList->remove(u);
13                u->m_pRNode->m_bOnList = false;
14            }
15
16            ul = self->m_pRcvUList->m_pUList;
17        }

将一个UDT Socket加入到RcvQueue的数据接收者列表m_pRcvUList和m_pHash中的动作是由RcvQueue的worker线程自动完成的，那么将一个UDT Socket从RcvQueue的数据接收者列表m_pRcvUList和m_pHash中移除的动作自然也是有 RcvQueue的worker线程自动完成的。

在这里可以看到，将UDT Socket的m_bClosing置为true之后，RcvQueue的worker线程自会将UDT Socket从RcvQueue的数据接收者列表m_pRcvUList和m_pHash中移除。

6. 唤醒所有等待在这个UDT Socket的condition上的线程。

7. 处理处于listening状态的socket，主要是将m_bListening置为false，并将当前UDT Socket从RcvQueue的listener移除。

这里倒是看到了对于listener的处理了，但是还是没有看到对于处在队列中，但还没有被accept返回的UDT Socket的处理。

8. 处理处于m_bConnecting状态的socket，主要是将当前UDT Socket从connector列表中移除。

9. 处理处于m_bConnected状态的socket，主要是发送shutdown消息给peer端，并将m_bConnected置为false。

10. 将m_bOpened置为false。

可以看到，7、8、9的处理应该是互斥的。

Done。

UDT协议实现分析——close过程

评论区

相关推荐