diff --git a/core/AmRtpReceiver.cpp b/core/AmRtpReceiver.cpp index 5fccc7ed..8ee42c29 100644 --- a/core/AmRtpReceiver.cpp +++ b/core/AmRtpReceiver.cpp @@ -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(); } } } diff --git a/core/AmRtpStream.cpp b/core/AmRtpStream.cpp index ab0147f6..ce5f6a35 100644 --- a/core/AmRtpStream.cpp +++ b/core/AmRtpStream.cpp @@ -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 ReceiveBuffer; +typedef std::map 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 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