// Loss retransmission always has higher priority.
if ((datapkt.m_iSeqNo = self->m_pSndLossList->getLostSeq()) >= 0)
{
// protect m_iSndLastDataAck from updating by ACK processing
CGuard ackguard(self->m_AckLock);
offset = CSeqNo::seqoff(self->m_iSndLastDataAck, datapkt.m_iSeqNo) * self->m_iPayloadSize;
if (offset < 0)
continue;
int32_t seqpair[2];
int msglen;
payload = self->m_pSndBuffer->readData(&(datapkt.m_pcData), offset, self->m_iPayloadSize, datapkt.m_iMsgNo, seqpair[0], msglen);
if (-1 == payload)
{
seqpair[1] = CSeqNo::incseq(seqpair[0], msglen / self->m_iPayloadSize);
self->sendCtrl(7, &datapkt.m_iMsgNo, seqpair, 8);
// only one msg drop request is necessary
self->m_pSndLossList->remove(seqpair[1]);
continue;
}
else if (0 == payload)
continue;
++ self->m_iTraceRetrans;
}
else
{
// If no loss, pack a new packet.
newdata = false;
// check congestion/flow window limit
#ifndef CUSTOM_CC
if (self->m_iFlowWindowSize > CSeqNo::seqlen(const_cast<int32_t&>(self->m_iSndLastAck), CSeqNo::incseq(self->m_iSndCurrSeqNo)) - 1)
#else
cwnd = (self->m_iFlowWindowSize < (int)self->m_dCongestionWindow) ? self->m_iFlowWindowSize : (int)self->m_dCongestionWindow;
if (cwnd > CSeqNo::seqlen(const_cast<int32_t&>(self->m_iSndLastAck), CSeqNo::incseq(self->m_iSndCurrSeqNo)) - 1)
#endif
{
if (0 != (payload = self->m_pSndBuffer->readData(&(datapkt.m_pcData), self->m_iPayloadSize, datapkt.m_iMsgNo)))
newdata = true;
else
{
//check if the sender buffer is empty
if (0 == self->m_pSndBuffer->getCurrBufSize())
{
// If yes, sleep here until a signal comes.
#ifndef WIN32
pthread_mutex_lock(&(self->m_SendDataLock));
while ((0 == self->m_pSndBuffer->getCurrBufSize()) && (!self->m_bClosing))
pthread_cond_wait(&(self->m_SendDataCond), &(self->m_SendDataLock));
pthread_mutex_unlock(&(self->m_SendDataLock));
#else
WaitForSingleObject(self->m_SendDataLock, INFINITE);
while ((0 == self->m_pSndBuffer->getCurrBufSize()) && (!self->m_bClosing))
{
ReleaseMutex(self->m_SendDataLock);
WaitForSingleObject(self->m_SendDataCond, INFINITE);
WaitForSingleObject(self->m_SendDataLock, INFINITE);
}
ReleaseMutex(self->m_SendDataLock);
#endif
#ifdef NO_BUSY_WAITING
// the waiting time should not be counted in. clear the time diff to zero.
self->m_ullTimeDiff = 0;
#endif
continue;
}
}
}
if (newdata)
{
self->m_iSndCurrSeqNo = CSeqNo::incseq(self->m_iSndCurrSeqNo);
datapkt.m_iSeqNo = self->m_iSndCurrSeqNo;
// every 16 (0xF) packets, a packet pair is sent
if (0 == (datapkt.m_iSeqNo & 0xF))
probe = true;
}
else
{
//wait here for ACK, NAK, or EXP (i.e, some data to sent)
#ifndef WIN32
gettimeofday(&now, 0);
if (now.tv_usec < 990000)
{
timeout.tv_sec = now.tv_sec;
timeout.tv_nsec = (now.tv_usec + 10000) * 1000;
}
else
{
timeout.tv_sec = now.tv_sec + 1;
timeout.tv_nsec = now.tv_usec * 1000;
}
pthread_cond_timedwait(&self->m_WindowCond, &self->m_WindowLock, &timeout);
#else
WaitForSingleObject(self->m_WindowCond, 1);
#endif
#ifdef NO_BUSY_WAITING
// the waiting time should not be counted in. clear the time diff to zero.
self->m_ullTimeDiff = 0;
#endif
continue;
}
}
gettimeofday(&now, 0);
datapkt.m_iTimeStamp = (now.tv_sec - self->m_StartTime.tv_sec) * 1000000 + now.tv_usec - self->m_StartTime.tv_usec;
self->m_pSndTimeWindow->onPktSent(datapkt.m_iTimeStamp);
// Now sending.
datapkt.setLength(payload);
*(self->m_pChannel) << datapkt;
#ifdef CUSTOM_CC
self->m_pCC->onPktSent(&datapkt);
#endif
++ self->m_llTraceSent;
if (probe)
{
// sends out probing packet pair
self->m_pTimer->rdtsc(targettime);
probe = false;
}
else if (self->m_bFreeze)
{
// sending is fronzen!
targettime = entertime + self->m_iSYNInterval * self->m_ullCPUFrequency + self->m_ullInterval;
self->m_bFreeze = false;
}
else
targettime = entertime + self->m_ullInterval;
// wait for an inter-packet time.
#ifndef NO_BUSY_WAITING
self->m_pTimer->sleepto(targettime);
#else
self->m_pTimer->rdtsc(currtime);
if (currtime >= targettime)
continue;
while (currtime + self->m_ullTimeDiff < targettime)
{
#ifndef WIN32
gettimeofday(&now, 0);
if (now.tv_usec < 990000)
{
timeout.tv_sec = now.tv_sec;
timeout.tv_nsec = (now.tv_usec + 10000) * 1000;
}
else
{
timeout.tv_sec = now.tv_sec + 1;
timeout.tv_nsec = now.tv_usec * 1000;
}
if (0 == pthread_cond_timedwait(&self->m_WindowCond, &self->m_WindowLock, &timeout))
break;
#else
if (WAIT_TIMEOUT != WaitForSingleObject(self->m_WindowCond, 1))
break;
#endif
self->m_pTimer->rdtsc(currtime);
}
self->m_pTimer->rdtsc(currtime);
if (currtime >= targettime)
self->m_ullTimeDiff += currtime - targettime;
else if (self->m_ullTimeDiff > targettime - currtime)
self->m_ullTimeDiff -= targettime - currtime;
else
self->m_ullTimeDiff = 0;
#endif
}