Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added CSync class as high level CV wrapper #1067

Merged
merged 8 commits into from
Jan 21, 2020
Merged

Added CSync class as high level CV wrapper #1067

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
edf09f9
[core] Renamed CRcvQueue::m_PassLock
maxsharabayko Jan 17, 2020
98f2ad7
[core] Replaced CTimer::sleep()
maxsharabayko Jan 17, 2020
77f0fff
[core] Major! Added mutex wrapper class
maxsharabayko Jan 17, 2020
84c2110
[tests] Updated sync tests by using sync::Mutex
maxsharabayko Jan 17, 2020
de175ff
Added CSync changes
Jan 17, 2020
e562d32
Updated from upstream
Jan 20, 2020
6ea5987
Updated from upstream
Jan 20, 2020
ba217f8
Post-review fixes (maxlovic)
Jan 20, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions srtcore/api.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@ int CUDTUnited::cleanup()
return 0;

m_bClosing = true;
pthread_cond_signal(&m_GCStopCond);
CSync::signal_relaxed(m_GCStopCond);
pthread_join(m_GCThread, NULL);

// XXX There's some weird bug here causing this
Expand Down Expand Up @@ -500,9 +500,7 @@ int CUDTUnited::newConnection(const SRTSOCKET listen, const sockaddr_any& peer,
}

// wake up a waiting accept() call
enterCS(ls->m_AcceptLock);
pthread_cond_signal(&(ls->m_AcceptCond));
leaveCS(ls->m_AcceptLock);
CSync::lock_signal(ls->m_AcceptCond, ls->m_AcceptLock);

return 1;
}
Expand Down Expand Up @@ -706,7 +704,8 @@ SRTSOCKET CUDTUnited::accept(const SRTSOCKET listen, sockaddr* pw_addr, int* pw_
// !!only one conection can be set up each time!!
while (!accepted)
{
CGuard cg(ls->m_AcceptLock);
CGuard accept_lock(ls->m_AcceptLock);
CSync accept_sync(ls->m_AcceptCond, accept_lock);

if ((ls->m_Status != SRTS_LISTENING) || ls->m_pUDT->m_bBroken)
{
Expand Down Expand Up @@ -745,7 +744,7 @@ SRTSOCKET CUDTUnited::accept(const SRTSOCKET listen, sockaddr* pw_addr, int* pw_
}

if (!accepted && (ls->m_Status == SRTS_LISTENING))
pthread_cond_wait(&(ls->m_AcceptCond), &ls->m_AcceptLock.ref());
accept_sync.wait();

if (ls->m_pQueuedSockets->empty())
m_EPoll.update_events(listen, ls->m_pUDT->m_sPollID, UDT_EPOLL_IN, false);
Expand Down
101 changes: 49 additions & 52 deletions srtcore/core.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4712,7 +4712,10 @@ void *CUDT::tsbpd(void *param)

THREAD_STATE_INIT("SRT:TsbPd");

CGuard recv_lock(self->m_RecvLock);
CGuard recv_lock (self->m_RecvLock);
CSync recvdata_cc (self->m_RecvDataCond, recv_lock);
CSync tsbpd_cc (self->m_RcvTsbPdCond, recv_lock);

self->m_bTsbPdAckWakeup = true;
while (!self->m_bClosing)
{
Expand Down Expand Up @@ -4814,7 +4817,7 @@ void *CUDT::tsbpd(void *param)
*/
if (self->m_bSynRecving)
{
pthread_cond_signal(&self->m_RecvDataCond);
recvdata_cc.signal_locked(recv_lock);
}
/*
* Set EPOLL_IN to wakeup any thread waiting on epoll
Expand All @@ -4832,12 +4835,10 @@ void *CUDT::tsbpd(void *param)
* Schedule wakeup when it will be.
*/
self->m_bTsbPdAckWakeup = false;
THREAD_PAUSED();
HLOGC(tslog.Debug,
log << self->CONID() << "tsbpd: FUTURE PACKET seq=" << current_pkt_seq
<< " T=" << FormatTime(tsbpdtime) << " - waiting " << count_milliseconds(timediff) << "ms");
SyncEvent::wait_for(&self->m_RcvTsbPdCond, &(self->m_RecvLock.ref()), timediff);
THREAD_RESUMED();
tsbpd_cc.wait_for(timediff);
}
else
{
Expand All @@ -4854,12 +4855,9 @@ void *CUDT::tsbpd(void *param)
*/
HLOGC(tslog.Debug, log << self->CONID() << "tsbpd: no data, scheduling wakeup at ack");
self->m_bTsbPdAckWakeup = true;
THREAD_PAUSED();
pthread_cond_wait(&self->m_RcvTsbPdCond, &(self->m_RecvLock.ref()));
THREAD_RESUMED();
tsbpd_cc.wait();
}
}
recv_lock.unlock();
THREAD_EXIT();
HLOGC(tslog.Debug, log << self->CONID() << "tsbpd: EXITING");
return NULL;
Expand Down Expand Up @@ -5475,6 +5473,8 @@ int CUDT::receiveBuffer(char *data, int len)
throw CUDTException(MJ_CONNECTION, MN_CONNLOST, 0);
}

CSync rcond (m_RecvDataCond, recvguard);
CSync tscond (m_RcvTsbPdCond, recvguard);
if (!m_pRcvBuffer->isRcvDataReady())
{
if (!m_bSynRecving)
Expand All @@ -5489,15 +5489,15 @@ int CUDT::receiveBuffer(char *data, int len)
while (stillConnected() && !m_pRcvBuffer->isRcvDataReady())
{
// Do not block forever, check connection status each 1 sec.
SyncEvent::wait_for(&m_RecvDataCond, &m_RecvLock.ref(), seconds_from(1));
rcond.wait_for(seconds_from(1));
}
}
else
{
const steady_clock::time_point exptime = steady_clock::now() + milliseconds_from(m_iRcvTimeOut);
while (stillConnected() && !m_pRcvBuffer->isRcvDataReady())
{
SyncEvent::wait_for(&m_RecvDataCond, &m_RecvLock.ref(), milliseconds_from(m_iRcvTimeOut));
rcond.wait_for(milliseconds_from(m_iRcvTimeOut));
if (steady_clock::now() >= exptime)
break;
}
Expand Down Expand Up @@ -5530,7 +5530,7 @@ int CUDT::receiveBuffer(char *data, int len)
if (m_bTsbPd)
{
HLOGP(tslog.Debug, "Ping TSBPD thread to schedule wakeup");
pthread_cond_signal(&m_RcvTsbPdCond);
tscond.signal_locked(recvguard);
}

if (!m_pRcvBuffer->isRcvDataReady())
Expand Down Expand Up @@ -5733,20 +5733,21 @@ int CUDT::sendmsg2(const char *data, int len, SRT_MSGCTRL& w_mctrl)

{
// wait here during a blocking sending
CGuard sendblock_lock(m_SendBlockLock);
CGuard sendblock_lock (m_SendBlockLock);
CSync sendcond (m_SendBlockCond, sendblock_lock);

if (m_iSndTimeOut < 0)
{
while (stillConnected() && sndBuffersLeft() < minlen && m_bPeerHealth)
pthread_cond_wait(&m_SendBlockCond, &m_SendBlockLock.ref());
sendcond.wait();
}
else
{
const steady_clock::time_point exptime = steady_clock::now() + milliseconds_from(m_iSndTimeOut);

while (stillConnected() && sndBuffersLeft() < minlen && m_bPeerHealth && exptime > steady_clock::now())
{
SyncEvent::wait_for(&m_SendBlockCond, &m_SendBlockLock.ref(), milliseconds_from(m_iSndTimeOut));
sendcond.wait_for(milliseconds_from(m_iSndTimeOut));
}
}
}
Expand Down Expand Up @@ -5905,7 +5906,8 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)
if (!m_CongCtl->checkTransArgs(SrtCongestion::STA_MESSAGE, SrtCongestion::STAD_RECV, data, len, -1, false))
throw CUDTException(MJ_NOTSUP, MN_INVALMSGAPI, 0);

CGuard recvguard(m_RecvLock);
CGuard recvguard (m_RecvLock);
CSync tscond (m_RcvTsbPdCond, recvguard);

/* XXX DEBUG STUFF - enable when required
char charbool[2] = {'0', '1'};
Expand All @@ -5927,7 +5929,7 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)

/* Kick TsbPd thread to schedule next wakeup (if running) */
if (m_bTsbPd)
pthread_cond_signal(&m_RcvTsbPdCond);
tscond.signal_locked(recvguard);

if (!m_pRcvBuffer->isRcvDataReady())
{
Expand Down Expand Up @@ -5955,7 +5957,7 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)

// Kick TsbPd thread to schedule next wakeup (if running)
if (m_bTsbPd)
pthread_cond_signal(&m_RcvTsbPdCond);
tscond.signal_locked(recvguard);

// Shut up EPoll if no more messages in non-blocking mode
s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_IN, false);
Expand All @@ -5967,7 +5969,7 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)
{
// Kick TsbPd thread to schedule next wakeup (if running)
if (m_bTsbPd)
pthread_cond_signal(&m_RcvTsbPdCond);
tscond.signal_locked(recvguard);

// Shut up EPoll if no more messages in non-blocking mode
s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_IN, false);
Expand All @@ -5987,20 +5989,22 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)
// Do not block forever, check connection status each 1 sec.
const steady_clock::duration recv_timeout = m_iRcvTimeOut < 0 ? seconds_from(1) : milliseconds_from(m_iRcvTimeOut);

CSync recv_cond (m_RecvDataCond, recvguard);

do
{
if (stillConnected() && !timeout && (!m_pRcvBuffer->isRcvDataReady()))
{
/* Kick TsbPd thread to schedule next wakeup (if running) */
if (m_bTsbPd)
{
HLOGP(tslog.Debug, "recvmsg: KICK tsbpd()");
pthread_cond_signal(&m_RcvTsbPdCond);
HLOGP(tslog.Debug, "receiveMessage: KICK tsbpd");
tscond.signal_locked(recvguard);
}

do
{
if (SyncEvent::wait_for(&m_RecvDataCond, &m_RecvLock.ref(), recv_timeout) == ETIMEDOUT)
if (!recv_cond.wait_for(recv_timeout))
{
if (!(m_iRcvTimeOut < 0))
timeout = true;
Expand Down Expand Up @@ -6044,7 +6048,7 @@ int CUDT::receiveMessage(char *data, int len, SRT_MSGCTRL& w_mctrl)
if (m_bTsbPd)
{
HLOGP(tslog.Debug, "recvmsg: KICK tsbpd() (buffer empty)");
pthread_cond_signal(&m_RcvTsbPdCond);
tscond.signal_locked(recvguard);
}

// Shut up EPoll if no more messages in non-blocking mode
Expand Down Expand Up @@ -6133,9 +6137,10 @@ int64_t CUDT::sendfile(fstream &ifs, int64_t &offset, int64_t size, int block)

{
CGuard lock(m_SendBlockLock);
CSync sendcond (m_SendBlockCond, lock);

while (stillConnected() && (sndBuffersLeft() <= 0) && m_bPeerHealth)
pthread_cond_wait(&m_SendBlockCond, &m_SendBlockLock.ref());
sendcond.wait();
}

if (m_bBroken || m_bClosing)
Expand Down Expand Up @@ -6262,10 +6267,13 @@ int64_t CUDT::recvfile(fstream &ofs, int64_t &offset, int64_t size, int block)
throw CUDTException(MJ_FILESYSTEM, MN_WRITEFAIL);
}

m_RecvDataLock.lock();
while (stillConnected() && !m_pRcvBuffer->isRcvDataReady())
pthread_cond_wait(&m_RecvDataCond, &m_RecvDataLock.ref());
m_RecvDataLock.unlock();
{
CGuard gl (m_RecvDataLock);
CSync rcond (m_RecvDataCond, gl);

while (stillConnected() && !m_pRcvBuffer->isRcvDataReady())
rcond.wait();
}

if (!m_bConnected)
throw CUDTException(MJ_CONNECTION, MN_NOCONN, 0);
Expand Down Expand Up @@ -6642,20 +6650,13 @@ void CUDT::destroySynch()
void CUDT::releaseSynch()
{
// wake up user calls
enterCS(m_SendBlockLock);
pthread_cond_signal(&m_SendBlockCond);
leaveCS(m_SendBlockLock);
CSync::lock_signal(m_SendBlockCond, m_SendBlockLock);

enterCS(m_SendLock);
leaveCS(m_SendLock);

enterCS(m_RecvDataLock);
pthread_cond_signal(&m_RecvDataCond);
leaveCS(m_RecvDataLock);

enterCS(m_RecvLock);
pthread_cond_signal(&m_RcvTsbPdCond);
leaveCS(m_RecvLock);
CSync::lock_signal(m_RecvDataCond, m_RecvDataLock);
CSync::lock_signal(m_RcvTsbPdCond, m_RecvLock);

enterCS(m_RecvDataLock);
if (!pthread_equal(m_RcvTsbPdThread, pthread_t()))
Expand Down Expand Up @@ -6782,19 +6783,17 @@ void CUDT::sendCtrl(UDTMessageType pkttype, const void *lparam, void *rparam, in
if (m_bTsbPd)
{
/* Newly acknowledged data, signal TsbPD thread */
enterCS(m_RecvLock);
CGuard rcvlock (m_RecvLock);
CSync tscond (m_RcvTsbPdCond, rcvlock);
if (m_bTsbPdAckWakeup)
pthread_cond_signal(&m_RcvTsbPdCond);
leaveCS(m_RecvLock);
tscond.signal_locked(rcvlock);
}
else
{
if (m_bSynRecving)
{
// signal a waiting "recv" call if there is any data available
enterCS(m_RecvDataLock);
pthread_cond_signal(&m_RecvDataCond);
leaveCS(m_RecvDataLock);
CSync::lock_signal(m_RecvDataCond, m_RecvDataLock);
}
// acknowledge any waiting epolls to read
s_UDTUnited.m_EPoll.update_events(m_SocketID, m_sPollID, UDT_EPOLL_IN, true);
Expand Down Expand Up @@ -7042,8 +7041,7 @@ void CUDT::updateSndLossListOnACK(int32_t ackdata_seqno)

if (m_bSynSending)
{
CGuard lk(m_SendBlockLock);
pthread_cond_signal(&m_SendBlockCond);
CSync::lock_signal(m_SendBlockCond, m_SendBlockLock);
}

const steady_clock::time_point currtime = steady_clock::now();
Expand Down Expand Up @@ -7971,8 +7969,7 @@ void CUDT::processClose()
if (m_bTsbPd)
{
HLOGP(mglog.Debug, "processClose: lock-and-signal TSBPD");
CGuard rl(m_RecvLock);
pthread_cond_signal(&m_RcvTsbPdCond);
CSync::lock_signal(m_RcvTsbPdCond, m_RecvLock);
}

// Signal the sender and recver if they are waiting for data.
Expand Down Expand Up @@ -8407,8 +8404,8 @@ int CUDT::processData(CUnit *in_unit)

if (m_bTsbPd)
{
CGuard lock(m_RecvLock);
pthread_cond_signal(&m_RcvTsbPdCond);
HLOGC(mglog.Debug, log << "loss: signaling TSBPD cond");
CSync::lock_signal(m_RcvTsbPdCond, m_RecvLock);
}
}

Expand All @@ -8424,8 +8421,8 @@ int CUDT::processData(CUnit *in_unit)

if (m_bTsbPd)
{
CGuard lock(m_RecvLock);
pthread_cond_signal(&m_RcvTsbPdCond);
HLOGC(mglog.Debug, log << "loss: signaling TSBPD cond");
CSync::lock_signal(m_RcvTsbPdCond, m_RecvLock);
}
}

Expand Down
Loading