added mutex for protocol

This commit is contained in:
Volker Fischer 2006-03-06 17:04:07 +00:00
parent 1595a7446b
commit fc816a7f09
8 changed files with 318 additions and 379 deletions

View file

@ -112,11 +112,11 @@ bool CChannelSet::PutData ( const CVector<unsigned char>& vecbyRecBuf,
const int iNumBytesRead, const int iNumBytesRead,
const CHostAddress& HostAdr ) const CHostAddress& HostAdr )
{ {
Mutex.lock ();
/* get channel ID ------------------------------------------------------- */
bool bChanOK = true; bool bChanOK = true;
Mutex.lock ();
{
/* get channel ID --------------------------------------------------- */
/* check address */ /* check address */
int iCurChanID = CheckAddr ( HostAdr ); int iCurChanID = CheckAddr ( HostAdr );
@ -136,20 +136,26 @@ bool CChannelSet::PutData ( const CVector<unsigned char>& vecbyRecBuf,
} }
/* put received data in jitter buffer ----------------------------------- */ /* put received data in jitter buffer ------------------------------- */
if ( bChanOK ) if ( bChanOK )
{ {
/* put packet in socket buffer */ /* put packet in socket buffer */
if ( vecChannels[iCurChanID].PutData ( vecbyRecBuf, iNumBytesRead ) ) switch ( vecChannels[iCurChanID].PutData ( vecbyRecBuf, iNumBytesRead ) )
{ {
case PS_AUDIO_OK:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_GREEN, iCurChanID ); PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_GREEN, iCurChanID );
} break;
else
{
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_RED, iCurChanID );
}
}
case PS_AUDIO_ERR:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_RED, iCurChanID );
break;
case PS_PROT_ERR:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_YELLOW, iCurChanID );
break;
}
}
}
Mutex.unlock (); Mutex.unlock ();
return !bChanOK; /* return 1 if error */ return !bChanOK; /* return 1 if error */
@ -167,7 +173,7 @@ void CChannelSet::GetBlockAllConC ( CVector<int>& vecChanID,
/* make put and get calls thread safe. Do not forget to unlock mutex /* make put and get calls thread safe. Do not forget to unlock mutex
afterwards! */ afterwards! */
Mutex.lock (); Mutex.lock ();
{
/* Check all possible channels */ /* Check all possible channels */
for ( int i = 0; i < MAX_NUM_CHANNELS; i++ ) for ( int i = 0; i < MAX_NUM_CHANNELS; i++ )
{ {
@ -196,7 +202,7 @@ void CChannelSet::GetBlockAllConC ( CVector<int>& vecChanID,
} }
} }
} }
}
Mutex.unlock (); /* release mutex */ Mutex.unlock (); /* release mutex */
} }
@ -221,20 +227,6 @@ void CChannelSet::GetConCliParam ( CVector<CHostAddress>& vecHostAddresses,
} }
} }
void CChannelSet::SetSockBufSize ( const int iNewBlockSize, const int iNumBlocks )
{
/* this opperation must be done with mutex */
Mutex.lock ();
/* as a test we adjust the buffers of all channels to the new value. Maybe later
do change only for some channels -> take care to set value back to default if
channel is disconnected, afterwards! */
for ( int i = 0; i < MAX_NUM_CHANNELS; i++ )
vecChannels[i].SetSockBufSize ( iNewBlockSize, iNumBlocks );
Mutex.unlock ();
}
/******************************************************************************\ /******************************************************************************\
* CChannel * * CChannel *
@ -286,9 +278,9 @@ void CChannel::SetSockBufSize ( const int iNewBlockSize, const int iNumBlocks )
{ {
/* this opperation must be done with mutex */ /* this opperation must be done with mutex */
Mutex.lock (); Mutex.lock ();
{
SockBuf.Init ( iNewBlockSize, iNumBlocks ); SockBuf.Init ( iNewBlockSize, iNumBlocks );
}
Mutex.unlock (); Mutex.unlock ();
} }
@ -314,10 +306,10 @@ bool CChannel::GetAddress(CHostAddress& RetAddr)
} }
} }
bool CChannel::PutData ( const CVector<unsigned char>& vecbyData, EPutDataStat CChannel::PutData ( const CVector<unsigned char>& vecbyData,
int iNumBytes ) int iNumBytes )
{ {
bool bRet = true; EPutDataStat eRet = PS_GEN_ERROR;
/* only process if packet has correct size */ /* only process if packet has correct size */
if ( iNumBytes == iAudComprSize ) if ( iNumBytes == iAudComprSize )
@ -342,31 +334,51 @@ for (int i = 0; i < BLOCK_SIZE_SAMPLES; i++)
Mutex.lock (); /* put mutex lock */ Mutex.lock (); /* put mutex lock */
{
bRet = SockBuf.Put ( vecdResOutData ); if ( SockBuf.Put ( vecdResOutData ) )
{
eRet = PS_AUDIO_OK;
}
else
{
eRet = PS_AUDIO_ERR;
}
}
Mutex.unlock (); /* put mutex unlock */ Mutex.unlock (); /* put mutex unlock */
/* reset time-out counter */ // if channel was not connected, emit signal to inform that new connection
// was established
if ( iConTimeOut == 0 )
{
emit NewConnection();
}
// reset time-out counter
iConTimeOut = CON_TIME_OUT_CNT_MAX; iConTimeOut = CON_TIME_OUT_CNT_MAX;
} }
else else
{ {
// this seems not to be an audio block, parse the message // this seems not to be an audio block, parse the message
if ( Protocol.ParseMessage ( vecbyData, iNumBytes ) )
// TODO: different return code for protocol {
eRet = PS_PROT_OK;
bRet = Protocol.ParseMessage ( vecbyData, iNumBytes ); }
else
{
eRet = PS_PROT_ERR;
}
} }
return bRet; return eRet;
} }
bool CChannel::GetData ( CVector<double>& vecdData ) bool CChannel::GetData ( CVector<double>& vecdData )
{ {
Mutex.lock (); /* get mutex lock */ bool bGetOK = false;
const bool bGetOK = SockBuf.Get ( vecdData ); Mutex.lock (); /* get mutex lock */
{
bGetOK = SockBuf.Get ( vecdData );
if ( !bGetOK ) if ( !bGetOK )
{ {
@ -376,7 +388,7 @@ bool CChannel::GetData ( CVector<double>& vecdData )
iConTimeOut--; iConTimeOut--;
} }
} }
}
Mutex.unlock (); /* get mutex unlock */ Mutex.unlock (); /* get mutex unlock */
return bGetOK; return bGetOK;

View file

@ -48,28 +48,17 @@
/* no valid channel number */ /* no valid channel number */
#define INVALID_CHANNEL_ID (MAX_NUM_CHANNELS + 1) #define INVALID_CHANNEL_ID (MAX_NUM_CHANNELS + 1)
enum EPutDataStat
/* Classes ********************************************************************/
class CSampleOffsetEst
{ {
public: PS_GEN_ERROR,
CSampleOffsetEst() {Init();} PS_AUDIO_OK,
virtual ~CSampleOffsetEst() {} PS_AUDIO_ERR,
PS_PROT_OK,
void Init(); PS_PROT_ERR
void AddTimeStampIdx(const int iTimeStampIdx);
double GetSamRate() {return dSamRateEst;}
protected:
QTime RefTime;
int iAccTiStVal;
double dSamRateEst;
CVector<long int> veciTimeElapsed;
CVector<long int> veciTiStIdx;
int iInitCnt;
}; };
/* Classes ********************************************************************/
/* CChannel ----------------------------------------------------------------- */ /* CChannel ----------------------------------------------------------------- */
class CChannel : public QObject class CChannel : public QObject
{ {
@ -79,7 +68,7 @@ public:
CChannel(); CChannel();
virtual ~CChannel() {} virtual ~CChannel() {}
bool PutData ( const CVector<unsigned char>& vecbyData, EPutDataStat PutData ( const CVector<unsigned char>& vecbyData,
int iNumBytes ); int iNumBytes );
bool GetData ( CVector<double>& vecdData ); bool GetData ( CVector<double>& vecdData );
@ -139,6 +128,7 @@ public slots:
signals: signals:
void MessReadyForSending ( CVector<uint8_t> vecMessage ); void MessReadyForSending ( CVector<uint8_t> vecMessage );
void NewConnection();
}; };
@ -155,25 +145,26 @@ public:
const int iNumBytesRead, const CHostAddress& HostAdr ); const int iNumBytesRead, const CHostAddress& HostAdr );
int GetFreeChan(); int GetFreeChan();
int CheckAddr ( const CHostAddress& Addr ); int CheckAddr ( const CHostAddress& Addr );
void GetBlockAllConC ( CVector<int>& vecChanID, void GetBlockAllConC ( CVector<int>& vecChanID,
CVector<CVector<double> >& vecvecdData ); CVector<CVector<double> >& vecvecdData );
void GetConCliParam( CVector<CHostAddress>& vecHostAddresses, void GetConCliParam( CVector<CHostAddress>& vecHostAddresses,
CVector<int>& veciJitBufSize ); CVector<int>& veciJitBufSize );
/* access functions for actual channels */ /* access functions for actual channels */
bool IsConnected ( const int iChanNum ) bool IsConnected ( const int iChanNum )
{ return vecChannels[iChanNum].IsConnected(); } { return vecChannels[iChanNum].IsConnected(); }
CVector<unsigned char> PrepSendPacket ( const int iChanNum, CVector<unsigned char> PrepSendPacket ( const int iChanNum,
const CVector<short>& vecsNPacket ) const CVector<short>& vecsNPacket )
{ return vecChannels[iChanNum].PrepSendPacket ( vecsNPacket ); } { return vecChannels[iChanNum].PrepSendPacket ( vecsNPacket ); }
CHostAddress GetAddress ( const int iChanNum ) CHostAddress GetAddress ( const int iChanNum )
{ return vecChannels[iChanNum].GetAddress(); } { return vecChannels[iChanNum].GetAddress(); }
void SetSockBufSize ( const int iNewBlockSize, const int iNumBlocks);
int GetSockBufSize() {return vecChannels[0].GetSockBufSize();}
protected: protected:
/* do not use the vector class since CChannel does not have appropriate /* do not use the vector class since CChannel does not have appropriate
copy constructor/operator */ copy constructor/operator */
@ -202,4 +193,25 @@ signals:
}; };
/* Sample rate offset estimation -------------------------------------------- */
class CSampleOffsetEst
{
public:
CSampleOffsetEst() { Init(); }
virtual ~CSampleOffsetEst() {}
void Init();
void AddTimeStampIdx ( const int iTimeStampIdx );
double GetSamRate() { return dSamRateEst; }
protected:
QTime RefTime;
int iAccTiStVal;
double dSamRateEst;
CVector<long int> veciTimeElapsed;
CVector<long int> veciTiStIdx;
int iInitCnt;
};
#endif /* !defined(CHANNEL_HOIH9345KJH98_3_4344_BB23945IUHF1912__INCLUDED_) */ #endif /* !defined(CHANNEL_HOIH9345KJH98_3_4344_BB23945IUHF1912__INCLUDED_) */

View file

@ -69,12 +69,6 @@ CLlconServerDlg::CLlconServerDlg ( CServer* pNServP, QWidget* parent,
#endif #endif
} }
/* Init slider control */
SliderNetBuf->setRange(1, MAX_NET_BUF_SIZE_NUM_BL);
const int iCurNumNetBuf = pServer->GetChannelSet()->GetSockBufSize();
SliderNetBuf->setValue(iCurNumNetBuf);
TextNetBuf->setText("Size: " + QString().setNum(iCurNumNetBuf));
/* Init timing jitter text label */ /* Init timing jitter text label */
TextLabelResponseTime->setText(""); TextLabelResponseTime->setText("");
@ -93,10 +87,6 @@ CLlconServerDlg::CLlconServerDlg ( CServer* pNServP, QWidget* parent,
/* timers */ /* timers */
QObject::connect(&Timer, SIGNAL(timeout()), this, SLOT(OnTimer())); QObject::connect(&Timer, SIGNAL(timeout()), this, SLOT(OnTimer()));
/* sliders */
QObject::connect(SliderNetBuf, SIGNAL(valueChanged(int)),
this, SLOT(OnSliderNetBuf(int)));
/* timers --------------------------------------------------------------- */ /* timers --------------------------------------------------------------- */
/* start timer for GUI controls */ /* start timer for GUI controls */
@ -158,12 +148,6 @@ void CLlconServerDlg::OnTimer()
} }
} }
void CLlconServerDlg::OnSliderNetBuf(int value)
{
pServer->GetChannelSet()->SetSockBufSize( MIN_BLOCK_SIZE_SAMPLES, value );
TextNetBuf->setText("Size: " + QString().setNum(value));
}
void CLlconServerDlg::customEvent(QCustomEvent* Event) void CLlconServerDlg::customEvent(QCustomEvent* Event)
{ {
if (Event->type() == QEvent::User + 11) if (Event->type() == QEvent::User + 11)

View file

@ -74,5 +74,4 @@ protected:
public slots: public slots:
void OnTimer(); void OnTimer();
void OnSliderNetBuf(int value);
}; };

View file

@ -11,7 +11,7 @@
<rect> <rect>
<x>0</x> <x>0</x>
<y>0</y> <y>0</y>
<width>633</width> <width>629</width>
<height>240</height> <height>240</height>
</rect> </rect>
</property> </property>
@ -36,21 +36,6 @@
<name>spacing</name> <name>spacing</name>
<number>6</number> <number>6</number>
</property> </property>
<widget>
<class>QLayoutWidget</class>
<property stdset="1">
<name>name</name>
<cstring>Layout4</cstring>
</property>
<hbox>
<property stdset="1">
<name>margin</name>
<number>0</number>
</property>
<property stdset="1">
<name>spacing</name>
<number>6</number>
</property>
<widget> <widget>
<class>QListView</class> <class>QListView</class>
<column> <column>
@ -82,66 +67,6 @@
<cstring>ListViewClients</cstring> <cstring>ListViewClients</cstring>
</property> </property>
</widget> </widget>
<widget>
<class>QGroupBox</class>
<property stdset="1">
<name>name</name>
<cstring>GroupBox3</cstring>
</property>
<property stdset="1">
<name>title</name>
<string>Jitter Buffer</string>
</property>
<vbox>
<property stdset="1">
<name>margin</name>
<number>11</number>
</property>
<property stdset="1">
<name>spacing</name>
<number>6</number>
</property>
<widget>
<class>QLabel</class>
<property stdset="1">
<name>name</name>
<cstring>TextNetBuf</cstring>
</property>
<property stdset="1">
<name>text</name>
<string>Size</string>
</property>
<property stdset="1">
<name>alignment</name>
<set>AlignCenter</set>
</property>
<property>
<name>hAlign</name>
</property>
</widget>
<widget>
<class>QSlider</class>
<property stdset="1">
<name>name</name>
<cstring>SliderNetBuf</cstring>
</property>
<property stdset="1">
<name>pageStep</name>
<number>1</number>
</property>
<property stdset="1">
<name>orientation</name>
<enum>Vertical</enum>
</property>
<property stdset="1">
<name>tickmarks</name>
<enum>Both</enum>
</property>
</widget>
</vbox>
</widget>
</hbox>
</widget>
<widget> <widget>
<class>QLayoutWidget</class> <class>QLayoutWidget</class>
<property stdset="1"> <property stdset="1">

View file

@ -72,14 +72,6 @@ MESSAGES
/* Implementation *************************************************************/ /* Implementation *************************************************************/
// TODO take care of mutexing ressources!!!!!!
CProtocol::CProtocol() : iCounter ( 0 ), iOldRecID ( PROTMESSID_ILLEGAL ), CProtocol::CProtocol() : iCounter ( 0 ), iOldRecID ( PROTMESSID_ILLEGAL ),
iOldRecCnt ( 0 ) iOldRecCnt ( 0 )
{ {
@ -132,11 +124,45 @@ void CProtocol::SendMessage()
} }
} }
void CProtocol::CreateAndSendAcknMess ( const int& iID, const int& iCnt )
{
CVector<uint8_t> vecAcknMessage;
CVector<uint8_t> vecData ( 2 ); // 2 bytes of data
unsigned int iPos = 0; // init position pointer
// build data vector
PutValOnStream ( vecData, iPos, static_cast<uint32_t> ( iID ), 2 );
// build complete message
GenMessageFrame ( vecAcknMessage, iCnt, PROTMESSID_ACKN, vecData );
// immediately send acknowledge message
emit MessReadyForSending ( vecAcknMessage );
}
/*
The following functions are access functions from different threads. These
functions have to be secured by a mutex to avoid data corruption
*/
void CProtocol::DeleteSendMessQueue() void CProtocol::DeleteSendMessQueue()
{
Mutex.lock();
{ {
// delete complete "send message queue" // delete complete "send message queue"
SendMessQueue.clear(); SendMessQueue.clear();
} }
Mutex.unlock();
}
void CProtocol::OnTimerSendMess()
{
Mutex.lock();
{
SendMessage();
}
Mutex.unlock();
}
bool CProtocol::ParseMessage ( const CVector<unsigned char>& vecbyData, bool CProtocol::ParseMessage ( const CVector<unsigned char>& vecbyData,
const int iNumBytes ) const int iNumBytes )
@ -144,6 +170,8 @@ bool CProtocol::ParseMessage ( const CVector<unsigned char>& vecbyData,
/* /*
return code: true -> ok; false -> error return code: true -> ok; false -> error
*/ */
Mutex.lock();
{
int iRecCounter, iRecID, iData; int iRecCounter, iRecID, iData;
unsigned int iPos; unsigned int iPos;
CVector<uint8_t> vecData; CVector<uint8_t> vecData;
@ -223,24 +251,12 @@ for ( int i = 0; i < iNumBytes; i++ ) {
return false; // return error code return false; // return error code
} }
} }
Mutex.unlock();
void CProtocol::CreateAndSendAcknMess ( const int& iID, const int& iCnt )
{
CVector<uint8_t> vecAcknMessage;
CVector<uint8_t> vecData ( 2 ); // 2 bytes of data
unsigned int iPos = 0; // init position pointer
// build data vector
PutValOnStream ( vecData, iPos, static_cast<uint32_t> ( iID ), 2 );
// build complete message
GenMessageFrame ( vecAcknMessage, iCnt, PROTMESSID_ACKN, vecData );
// immediately send acknowledge message
emit MessReadyForSending ( vecAcknMessage );
} }
void CProtocol::CreateJitBufMes ( const int iJitBufSize ) void CProtocol::CreateJitBufMes ( const int iJitBufSize )
{
Mutex.lock();
{ {
CVector<uint8_t> vecNewMessage; CVector<uint8_t> vecNewMessage;
CVector<uint8_t> vecData ( 2 ); // 2 bytes of data CVector<uint8_t> vecData ( 2 ); // 2 bytes of data
@ -250,7 +266,6 @@ void CProtocol::CreateJitBufMes ( const int iJitBufSize )
const int iCurCounter = iCounter; const int iCurCounter = iCounter;
// increase counter (wraps around automatically) // increase counter (wraps around automatically)
// TODO: make it thread safe!!!!!!!!!!!!
iCounter++; iCounter++;
// build data vector // build data vector
@ -262,24 +277,8 @@ iCounter++;
// enqueue message // enqueue message
EnqueueMessage ( vecNewMessage, iCurCounter, PROTMESSID_JITT_BUF_SIZE ); EnqueueMessage ( vecNewMessage, iCurCounter, PROTMESSID_JITT_BUF_SIZE );
} }
Mutex.unlock();
}
/******************************************************************************\ /******************************************************************************\

View file

@ -118,11 +118,12 @@ protected:
uint8_t iCounter; uint8_t iCounter;
int iOldRecID, iOldRecCnt; int iOldRecID, iOldRecCnt;
std::list<CSendMessage> SendMessQueue; std::list<CSendMessage> SendMessQueue;
QTimer TimerSendMess; QTimer TimerSendMess;
QMutex Mutex; QMutex Mutex;
public slots: public slots:
void OnTimerSendMess() { SendMessage(); } void OnTimerSendMess();
signals: signals:
// transmitting // transmitting

View file

@ -105,13 +105,20 @@ void CSocket::OnDataReceived ()
return; return;
} }
if ( pChannel->PutData( vecbyRecBuf, iNumBytesRead ) ) switch ( pChannel->PutData( vecbyRecBuf, iNumBytesRead ) )
{ {
case PS_AUDIO_OK:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_GREEN ); PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_GREEN );
} break;
else
{ case PS_AUDIO_ERR:
case PS_GEN_ERROR:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_RED ); PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_RED );
break;
case PS_PROT_ERR:
PostWinMessage ( MS_JIT_BUF_PUT, MUL_COL_LED_YELLOW );
break;
} }
} }
else else