sipwise
/
sems
mirror of https://github.com/sipwise/sems.git
1
0
Fork 0
Code Issues Releases Wiki Activity

optimization: the RTP packet buffer which is used in recvfrom() is used throughout the RTP packet

Browse Source 
processing, avoiding unnecessary copying of the received RTP packet


git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@787 8eb893ce-cfd4-0310-b710-fb5ebe64c474
sayer/1.4-spce2.6
Stefan Sayer 19 years ago
parent 799ae8c9ae
commit 407b31541f
3 changed files with 103 additions and 39 deletions
Show Stats Download Patch File Download Diff File

27
core/AmRtpReceiver.cpp
Unescape View File

@ -76,7 +76,6 @@ void AmRtpReceiver::run()
{
unsigned int tmp_nfds = 0;
struct pollfd* tmp_fds = new struct pollfd[MAX_RTP_SESSIONS];
AmRtpPacket p;
while(true){
@ -97,23 +96,29 @@ void AmRtpReceiver::run()
if(!(tmp_fds[i].revents & POLLIN))
continue;
if(p.recv(tmp_fds[i].fd) > 0){
int parse_res = p.parse();
gettimeofday(&p.recv_time,NULL);
streams_mut.lock();
Streams::iterator it = streams.find(tmp_fds[i].fd);
if(it != streams.end()) {
AmRtpPacket* p = it->second->newPacket();
if (!p)
continue;
if(p->recv(tmp_fds[i].fd) > 0){
int parse_res = p->parse();
gettimeofday(&p->recv_time,NULL);
streams_mut.lock();
Streams::iterator it = streams.find(tmp_fds[i].fd);
if(it != streams.end()) {
if (parse_res == -1) {
DBG("error while parsing RTP packet.\n");
it->second->clearRTPTimeout(&p.recv_time);
it->second->clearRTPTimeout(&p->recv_time);
it->second->freePacket(p);
} else {
it->second->bufferPacket(&p);
it->second->bufferPacket(p);
}
} else {
it->second->freePacket(p);
}
streams_mut.unlock();
}
streams_mut.unlock();
}
}
}

87
core/AmRtpStream.cpp
Unescape View File

@ -219,9 +219,12 @@ int AmRtpStream::send( unsigned int ts, unsigned char* buffer, unsigned int size
int AmRtpStream::receive( unsigned char* buffer, unsigned int size,
unsigned int& ts, int &out_payload)
{
AmRtpPacket rp;
AmRtpPacket* rp = NULL;
int err = nextPacket(rp);
if (!rp)
return 0;
if(err <= 0)
return err;
@ -231,7 +234,7 @@ int AmRtpStream::receive( unsigned char* buffer, unsigned int size,
// struct sockaddr_storage recv_addr;
// #else
struct sockaddr_in recv_addr;
rp.getAddr(&recv_addr);
rp->getAddr(&recv_addr);
// symmetric RTP
if ((recv_addr.sin_port != r_saddr.sin_port)
@ -252,38 +255,39 @@ int AmRtpStream::receive( unsigned char* buffer, unsigned int size,
#endif
/* do we have a new talk spurt? */
begin_talk = ((last_payload == 13) || rp.marker);
last_payload = rp.payload;
begin_talk = ((last_payload == 13) || rp->marker);
last_payload = rp->payload;
if(!rp.getDataSize())
if(!rp->getDataSize()) {
mem.freePacket(rp);
return RTP_EMPTY;
}
if (telephone_event_pt.get() && rp.payload == telephone_event_pt->payload_type)
if (telephone_event_pt.get() && rp->payload == telephone_event_pt->payload_type)
{
dtmf_payload_t* dpl = (dtmf_payload_t*)rp.getData();
dtmf_payload_t* dpl = (dtmf_payload_t*)rp->getData();
DBG("DTMF: event=%i; e=%i; r=%i; volume=%i; duration=%i\n",
dpl->event,dpl->e,dpl->r,dpl->volume,ntohs(dpl->duration));
session->postDtmfEvent(new AmRtpDtmfEvent(dpl, getTelephoneEventRate()));
mem.freePacket(rp);
return RTP_DTMF;
}
/*
if (payload != rp.payload){
return RTP_UNKNOWN_PL;
}
*/
assert(rp.getData());
if(rp.getDataSize() > size){
assert(rp->getData());
if(rp->getDataSize() > size){
ERROR("received too big RTP packet\n");
mem.freePacket(rp);
return RTP_BUFFER_SIZE;
}
memcpy(buffer,rp.getData(),rp.getDataSize());
ts = rp.timestamp;
out_payload = rp.payload;
return rp.getDataSize();
memcpy(buffer,rp->getData(),rp->getDataSize());
ts = rp->timestamp;
out_payload = rp->payload;
int res = rp->getDataSize();
mem.freePacket(rp);
return res;
}
AmRtpStream::AmRtpStream(AmSession* _s)
@ -427,15 +431,29 @@ void AmRtpStream::icmpError()
}
}
void AmRtpStream::bufferPacket(const AmRtpPacket* p)
AmRtpPacket* AmRtpStream::newPacket() {
return mem.newPacket();
}
void AmRtpStream::freePacket(AmRtpPacket* p) {
return mem.freePacket(p);
}
void AmRtpStream::bufferPacket(AmRtpPacket* p)
{
memcpy(&last_recv_time, &p->recv_time, sizeof(struct timeval));
if (!receiving && !passive)
if (!receiving && !passive) {
mem.freePacket(p);
return;
}
receive_mut.lock();
receive_buf[p->timestamp].copy(p);
// free packet on double packet for TS received
if (receive_buf.find(p->timestamp) != receive_buf.end())
mem.freePacket(receive_buf[p->timestamp]);
receive_buf[p->timestamp] = p;
receive_mut.unlock();
}
@ -443,7 +461,7 @@ void AmRtpStream::clearRTPTimeout(struct timeval* recv_time) {
memcpy(&last_recv_time, recv_time, sizeof(struct timeval));
}
int AmRtpStream::nextPacket(AmRtpPacket& p)
int AmRtpStream::nextPacket(AmRtpPacket*& p)
{
if (!receiving && !passive)
return RTP_EMPTY;
@ -469,8 +487,7 @@ int AmRtpStream::nextPacket(AmRtpPacket& p)
return RTP_EMPTY;
}
AmRtpPacket& pp = receive_buf.begin()->second;
p.copy(&pp);
p = receive_buf.begin()->second;
receive_buf.erase(receive_buf.begin());
receive_mut.unlock();
@ -484,3 +501,23 @@ int AmRtpStream::getTelephoneEventRate()
retval = telephone_event_pt->clock_rate;
return retval;
}
PacketMem::PacketMem() {
memset(used, 0, sizeof(used));
}
inline AmRtpPacket* PacketMem::newPacket() {
for (int i=0;i<MAX_PACKETS;i++)
if (!used[i]) {
used[i]=true;
return &packets[i];
}
return NULL;
}
inline void PacketMem::freePacket(AmRtpPacket* p) {
if (!p) return;
used[p-packets] = false;
}

28
core/AmRtpStream.h
Unescape View File

@ -57,7 +57,24 @@ struct amci_payload_t;
class AmAudio;
class AmSession;
class SdpPayload;
typedef std::map<unsigned int, AmRtpPacket, ts_less> ReceiveBuffer;
typedef std::map<unsigned int, AmRtpPacket*, ts_less> ReceiveBuffer;
/**
* This provides the memory for the receive buffer.
*/
struct PacketMem {
#define MAX_PACKETS 32
AmRtpPacket packets[MAX_PACKETS];
bool used[MAX_PACKETS];
PacketMem();
inline AmRtpPacket* newPacket();
inline void freePacket(AmRtpPacket* p);
};
/**
* \brief RTP implementation
@ -120,11 +137,13 @@ protected:
auto_ptr<const SdpPayload> telephone_event_pt;
PacketMem mem;
ReceiveBuffer receive_buf;
AmMutex receive_mut;
/* get next packet in buffer */
int nextPacket(AmRtpPacket& p);
int nextPacket(AmRtpPacket*& p);
AmSession* session;
@ -134,6 +153,9 @@ protected:
public:
AmRtpPacket* newPacket();
void freePacket(AmRtpPacket* p);
/** Mute */
bool mute;
/** mute && port == 0 */
@ -246,7 +268,7 @@ public:
* Insert an RTP packet to the buffer.
* Note: memory is owned by this instance.
*/
void bufferPacket(const AmRtpPacket* p);
void bufferPacket(AmRtpPacket* p);
/*
* clear RTP timeout at time recv_time

Write Preview
Loading…
Cancel
Save