@@ -6275,8 +6275,8 @@ int srt::CUDT::receiveBuffer(char *data, int len)
6275 6275
return 0;
6276 6276
}
6277 6277
HLOGC(arlog.Debug,
6278 -
log << (m_config.bMessageAPI ? "MESSAGE" : "STREAM") << " API, " << (m_bShutdown ? "" : "no")
6279 -
<< " SHUTDOWN. Reporting as BROKEN.");
6278 +
log << (m_config.bMessageAPI ? "MESSAGE" : "STREAM") << " API, " << (m_bShutdown ? "" : "no")
6279 +
<< " SHUTDOWN. Reporting as BROKEN.");
6280 6280
throw CUDTException(MJ_CONNECTION, MN_CONNLOST, 0);
6281 6281
}
6282 6282
@@ -6288,31 +6288,29 @@ int srt::CUDT::receiveBuffer(char *data, int len)
6288 6288
{
6289 6289
throw CUDTException(MJ_AGAIN, MN_RDAVAIL, 0);
6290 6290
}
6291 -
else
6291 + 6292 +
// Kick TsbPd thread to schedule next wakeup (if running)
6293 +
if (m_config.iRcvTimeOut < 0)
6292 6294
{
6293 -
/* Kick TsbPd thread to schedule next wakeup (if running) */
6294 -
if (m_config.iRcvTimeOut < 0)
6295 +
THREAD_PAUSED();
6296 +
while (stillConnected() && !isRcvBufferReady())
6295 6297
{
6296 -
THREAD_PAUSED();
6297 -
while (stillConnected() && !isRcvBufferReady())
6298 -
{
6299 -
// Do not block forever, check connection status each 1 sec.
6300 -
rcond.wait_for(seconds_from(1));
6301 -
}
6302 -
THREAD_RESUMED();
6298 +
// Do not block forever, check connection status each 1 sec.
6299 +
rcond.wait_for(seconds_from(1));
6303 6300
}
6304 -
else
6301 +
THREAD_RESUMED();
6302 +
}
6303 +
else
6304 +
{
6305 +
const steady_clock::time_point exptime =
6306 +
steady_clock::now() + milliseconds_from(m_config.iRcvTimeOut);
6307 +
THREAD_PAUSED();
6308 +
while (stillConnected() && !isRcvBufferReady())
6305 6309
{
6306 -
const steady_clock::time_point exptime =
6307 -
steady_clock::now() + milliseconds_from(m_config.iRcvTimeOut);
6308 -
THREAD_PAUSED();
6309 -
while (stillConnected() && !isRcvBufferReady())
6310 -
{
6311 -
if (!rcond.wait_until(exptime)) // NOT means "not received a signal"
6312 -
break; // timeout
6313 -
}
6314 -
THREAD_RESUMED();
6310 +
if (!rcond.wait_until(exptime)) // NOT means "not received a signal"
6311 +
break; // timeout
6315 6312
}
6313 +
THREAD_RESUMED();
6316 6314
}
6317 6315
}
6318 6316
@@ -8066,23 +8064,31 @@ int srt::CUDT::sendCtrlAck(CPacket& ctrlpkt, int size)
8066 8064
{
8067 8065
UniqueLock rdlock (m_RecvLock);
8068 8066
CSync rdcond (m_RecvDataCond, rdlock);
8069 -
if (m_config.bSynRecving)
8067 + 8068 +
#if ENABLE_NEW_RCVBUFFER
8069 +
// Locks m_RcvBufferLock, which is unlocked above by InvertedLock un_bufflock.
8070 +
// Must check read-readiness under m_RecvLock to protect the epoll from concurrent changes in readBuffer()
8071 +
if (isRcvBufferReady())
8072 +
#endif
8070 8073
{
8071 -
// signal a waiting "recv" call if there is any data available
8072 -
rdcond.signal_locked(rdlock);
8074 +
if (m_config.bSynRecving)
8075 +
{
8076 +
// signal a waiting "recv" call if there is any data available
8077 +
rdcond.signal_locked(rdlock);
8078 +
}
8079 +
// acknowledge any waiting epolls to read
8080 +
// fix SRT_EPOLL_IN event loss but rcvbuffer still have data：
8081 +
// 1. user call receive/receivemessage(about line number:6482)
8082 +
// 2. after read/receive, if rcvbuffer is empty, will set SRT_EPOLL_IN event to false
8083 +
// 3. but if we do not do some lock work here, will cause some sync problems between threads:
8084 +
// (1) user thread: call receive/receivemessage
8085 +
// (2) user thread: read data
8086 +
// (3) user thread: no data in rcvbuffer, set SRT_EPOLL_IN event to false
8087 +
// (4) receive thread: receive data and set SRT_EPOLL_IN to true
8088 +
// (5) user thread: set SRT_EPOLL_IN to false
8089 +
// 4. so , m_RecvLock must be used here to protect epoll event
8090 +
uglobal().m_EPoll.update_events(m_SocketID, m_sPollID, SRT_EPOLL_IN, true);
8073 8091
}
8074 -
// acknowledge any waiting epolls to read
8075 -
// fix SRT_EPOLL_IN event loss but rcvbuffer still have data：
8076 -
// 1. user call receive/receivemessage(about line number:6482)
8077 -
// 2. after read/receive, if rcvbuffer is empty, will set SRT_EPOLL_IN event to false
8078 -
// 3. but if we do not do some lock work here, will cause some sync problems between threads:
8079 -
// (1) user thread: call receive/receivemessage
8080 -
// (2) user thread: read data
8081 -
// (3) user thread: no data in rcvbuffer, set SRT_EPOLL_IN event to false
8082 -
// (4) receive thread: receive data and set SRT_EPOLL_IN to true
8083 -
// (5) user thread: set SRT_EPOLL_IN to false
8084 -
// 4. so , m_RecvLock must be used here to protect epoll event
8085 -
uglobal().m_EPoll.update_events(m_SocketID, m_sPollID, SRT_EPOLL_IN, true);
8086 8092
}
8087 8093
#if ENABLE_EXPERIMENTAL_BONDING
8088 8094
if (m_parent->m_GroupOf)
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