@@ -2572,10 +2572,6 @@ void srt::CUDTUnited::checkBrokenSockets()
2572 2572
{
2573 2573
ScopedLock cg(m_GlobControlLock);
2574 2574 2575 -
// set of sockets To Be Closed and To Be Removed
2576 -
vector<SRTSOCKET> tbc;
2577 -
vector<SRTSOCKET> tbr;
2578 - 2579 2575
#if ENABLE_EXPERIMENTAL_BONDING
2580 2576
vector<SRTSOCKET> delgids;
2581 2577
@@ -2600,74 +2596,70 @@ void srt::CUDTUnited::checkBrokenSockets()
2600 2596
{
2601 2597
m_ClosedGroups.erase(*di);
2602 2598
}
2603 - 2604 2599
#endif
2605 2600 2606 -
for (sockets_t::iterator i = m_Sockets.begin();
2607 -
i != m_Sockets.end(); ++ i)
2601 +
// set of sockets To Be Closed and To Be Removed
2602 +
vector<SRTSOCKET> tbc;
2603 +
vector<SRTSOCKET> tbr;
2604 + 2605 +
for (sockets_t::iterator i = m_Sockets.begin(); i != m_Sockets.end(); ++ i)
2608 2606
{
2609 -
CUDTSocket* s = i->second;
2607 +
CUDTSocket* s = i->second;
2608 +
if (!s->core().m_bBroken)
2609 +
continue;
2610 2610 2611 -
// check broken connection
2612 -
if (s->core().m_bBroken)
2611 +
if (s->m_Status == SRTS_LISTENING)
2613 2612
{
2614 -
if (s->m_Status == SRTS_LISTENING)
2615 -
{
2616 -
const steady_clock::duration elapsed = steady_clock::now() - s->m_tsClosureTimeStamp;
2617 -
// for a listening socket, it should wait an extra 3 seconds
2618 -
// in case a client is connecting
2619 -
if (elapsed < milliseconds_from(CUDT::COMM_CLOSE_BROKEN_LISTENER_TIMEOUT_MS))
2613 +
const steady_clock::duration elapsed = steady_clock::now() - s->m_tsClosureTimeStamp;
2614 +
// A listening socket should wait an extra 3 seconds
2615 +
// in case a client is connecting.
2616 +
if (elapsed < milliseconds_from(CUDT::COMM_CLOSE_BROKEN_LISTENER_TIMEOUT_MS))
2617 +
continue;
2618 +
}
2619 +
else if ((s->core().m_pRcvBuffer != NULL)
2620 +
// FIXED: calling isRcvDataAvailable() just to get the information
2621 +
// whether there are any data waiting in the buffer,
2622 +
// NOT WHETHER THEY ARE ALSO READY TO PLAY at the time when
2623 +
// this function is called (isRcvDataReady also checks if the
2624 +
// available data is "ready to play").
2625 +
&& s->core().m_pRcvBuffer->isRcvDataAvailable())
2626 +
{
2627 +
const int bc = s->core().m_iBrokenCounter.load();
2628 +
if (bc > 0)
2620 2629
{
2630 +
// if there is still data in the receiver buffer, wait longer
2631 +
s->core().m_iBrokenCounter.store(bc - 1);
2621 2632
continue;
2622 2633
}
2623 -
}
2624 -
else if ((s->core().m_pRcvBuffer != NULL)
2625 -
// FIXED: calling isRcvDataAvailable() just to get the information
2626 -
// whether there are any data waiting in the buffer,
2627 -
// NOT WHETHER THEY ARE ALSO READY TO PLAY at the time when
2628 -
// this function is called (isRcvDataReady also checks if the
2629 -
// available data is "ready to play").
2630 -
&& s->core().m_pRcvBuffer->isRcvDataAvailable())
2631 -
{
2632 -
const int bc = s->core().m_iBrokenCounter.load();
2633 -
if (bc > 0)
2634 -
{
2635 -
// HLOGF(smlog.Debug, "STILL KEEPING socket (still have data):
2636 -
// %d\n", i->first);
2637 -
// if there is still data in the receiver buffer, wait longer
2638 -
s->core().m_iBrokenCounter.store(bc - 1);
2639 -
continue;
2640 -
}
2641 -
}
2634 +
}
2642 2635 2643 2636
#if ENABLE_EXPERIMENTAL_BONDING
2644 -
if (s->m_GroupOf)
2645 -
{
2646 -
LOGC(smlog.Note, log << "@" << s->m_SocketID << " IS MEMBER OF $" << s->m_GroupOf->id() << " - REMOVING FROM GROUP");
2647 -
s->removeFromGroup(true);
2648 -
}
2637 +
if (s->m_GroupOf)
2638 +
{
2639 +
LOGC(smlog.Note, log << "@" << s->m_SocketID << " IS MEMBER OF $" << s->m_GroupOf->id() << " - REMOVING FROM GROUP");
2640 +
s->removeFromGroup(true);
2641 +
}
2649 2642
#endif
2650 2643 2651 -
HLOGC(smlog.Debug, log << "checkBrokenSockets: moving BROKEN socket to CLOSED: @" << i->first);
2644 +
HLOGC(smlog.Debug, log << "checkBrokenSockets: moving BROKEN socket to CLOSED: @" << i->first);
2652 2645 2653 -
//close broken connections and start removal timer
2654 -
s->setClosed();
2655 -
tbc.push_back(i->first);
2656 -
m_ClosedSockets[i->first] = s;
2646 +
//close broken connections and start removal timer
2647 +
s->setClosed();
2648 +
tbc.push_back(i->first);
2649 +
m_ClosedSockets[i->first] = s;
2657 2650 2658 -
// remove from listener's queue
2659 -
sockets_t::iterator ls = m_Sockets.find(s->m_ListenSocket);
2660 -
if (ls == m_Sockets.end())
2661 -
{
2662 -
ls = m_ClosedSockets.find(s->m_ListenSocket);
2663 -
if (ls == m_ClosedSockets.end())
2664 -
continue;
2665 -
}
2666 - 2667 -
enterCS(ls->second->m_AcceptLock);
2668 -
ls->second->m_QueuedSockets.erase(s->m_SocketID);
2669 -
leaveCS(ls->second->m_AcceptLock);
2651 +
// remove from listener's queue
2652 +
sockets_t::iterator ls = m_Sockets.find(s->m_ListenSocket);
2653 +
if (ls == m_Sockets.end())
2654 +
{
2655 +
ls = m_ClosedSockets.find(s->m_ListenSocket);
2656 +
if (ls == m_ClosedSockets.end())
2657 +
continue;
2670 2658
}
2659 + 2660 +
enterCS(ls->second->m_AcceptLock);
2661 +
ls->second->m_QueuedSockets.erase(s->m_SocketID);
2662 +
leaveCS(ls->second->m_AcceptLock);
2671 2663
}
2672 2664 2673 2665
for (sockets_t::iterator j = m_ClosedSockets.begin();
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4