iqueue.h

Go to the documentation of this file.

// Copyright (C) 2001,2002,2004 Federico Montesino Pouzols <fedemp@altern.org>.
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//
// As a special exception, you may use this file as part of a free software
// library without restriction.  Specifically, if other files instantiate
// templates or use macros or inline functions from this file, or you compile
// this file and link it with other files to produce an executable, this
// file does not by itself cause the resulting executable to be covered by
// the GNU General Public License.  This exception does not however
// invalidate any other reasons why the executable file might be covered by
// the GNU General Public License.
//
// This exception applies only to the code released under the name GNU
// ccRTP.  If you copy code from other releases into a copy of GNU
// ccRTP, as the General Public License permits, the exception does
// not apply to the code that you add in this way.  To avoid misleading
// anyone as to the status of such modified files, you must delete
// this exception notice from them.
//
// If you write modifications of your own for GNU ccRTP, it is your choice
// whether to permit this exception to apply to your modifications.
// If you do not wish that, delete this exception notice.
//

#ifndef CCXX_RTP_IQUEUE_H_
#define CCXX_RTP_IQUEUE_H_

#include <ccrtp/queuebase.h>
#include <ccrtp/CryptoContext.h>

#include <list>

#ifdef  CCXX_NAMESPACES
namespace ost {
#endif

class __EXPORT Members
{
public:
        inline void
        setMembersCount(uint32 n)
        { members = n; }

        inline void
        increaseMembersCount()
        { members++; }

        inline void
        decreaseMembersCount()
        { members--; }

        inline uint32
        getMembersCount() const
        { return members; }

        inline void
        setSendersCount(uint32 n)
        { activeSenders = n; }

        inline void
        increaseSendersCount()
        { activeSenders++; }

        inline void
        decreaseSendersCount()
        { activeSenders--; }

        inline uint32
        getSendersCount() const
        { return activeSenders; }

protected:
        Members() :
                members(0),
                activeSenders(0)
        { }

        inline virtual ~Members()
        { }

private:
        uint32 members;
        uint32 activeSenders;
};

class __EXPORT SyncSourceHandler
{
public:
        inline void*
        getLink(const SyncSource& source) const
        { return source.getLink(); }

        inline void
        setLink(SyncSource& source, void* link)
        { source.setLink(link); }

        inline void
        setParticipant(SyncSource& source, Participant& p)
        { source.setParticipant(p); }

        inline void
        setState(SyncSource& source, SyncSource::State ns)
        { source.setState(ns); }

        inline void
        setSender(SyncSource& source, bool active)
        { source.setSender(active); }

        inline void
        setDataTransportPort(SyncSource& source, tpport_t p)
        { source.setDataTransportPort(p); }

        inline void
        setControlTransportPort(SyncSource& source, tpport_t p)
        { source.setControlTransportPort(p); }

        inline void
        setNetworkAddress(SyncSource& source, InetAddress addr)
        { source.setNetworkAddress(addr); }

protected:
        SyncSourceHandler()
        { }

        inline virtual ~SyncSourceHandler()
        { }
};

class __EXPORT ParticipantHandler
{
public:
        inline void
        setSDESItem(Participant* part, SDESItemType item,
                    const std::string& val)
        { part->setSDESItem(item,val); }

        inline void
        setPRIVPrefix(Participant* part, const std::string val)
        { part->setPRIVPrefix(val); }

protected:
        ParticipantHandler()
        { }

        inline virtual ~ParticipantHandler()
        { }
};

class __EXPORT ApplicationHandler
{
public:
        inline void
        addParticipant(RTPApplication& app, Participant& part)
        { app.addParticipant(part); }

        inline void
        removeParticipant(RTPApplication& app,
                          RTPApplication::ParticipantLink* pl)
        { app.removeParticipant(pl); }

protected:
        ApplicationHandler()
        { }

        inline virtual ~ApplicationHandler()
        { }
};

class __EXPORT ConflictHandler
{
public:
        struct ConflictingTransportAddress
        {
                ConflictingTransportAddress(InetAddress na,
                                            tpport_t dtp, tpport_t ctp);

                void setNext(ConflictingTransportAddress* nc)
                { next = nc; }

                inline const InetAddress& getNetworkAddress( ) const
                { return networkAddress; }

                inline tpport_t getDataTransportPort() const
                { return dataTransportPort; }

                inline tpport_t getControlTransportPort() const
                { return controlTransportPort; }

                InetAddress networkAddress;
                tpport_t dataTransportPort;
                tpport_t controlTransportPort;
                ConflictingTransportAddress* next;
                // arrival time of last data or control packet.
                timeval lastPacketTime;
        };

        ConflictingTransportAddress* searchDataConflict(InetAddress na,
                                                        tpport_t dtp);
        ConflictingTransportAddress* searchControlConflict(InetAddress na,
                                                           tpport_t ctp);

        void updateConflict(ConflictingTransportAddress& ca)
        { gettimeofday(&(ca.lastPacketTime),NULL); }

        void addConflict(const InetAddress& na, tpport_t dtp, tpport_t ctp);

protected:
        ConflictHandler()
        { firstConflict = lastConflict = NULL; }

        inline virtual ~ConflictHandler()
        { }

        ConflictingTransportAddress* firstConflict, * lastConflict;
};

class __EXPORT MembershipBookkeeping :
        public SyncSourceHandler,
        public ParticipantHandler,
        public ApplicationHandler,
        public ConflictHandler,
        private Members
{
public:
        inline size_t getDefaultMembersHashSize()
        { return defaultMembersHashSize; }

protected:

        MembershipBookkeeping(uint32 initialSize = defaultMembersHashSize);

        inline virtual
        ~MembershipBookkeeping()
        { endMembers(); }

        struct SyncSourceLink;

        inline SyncSourceLink* getLink(const SyncSource& source) const
        { return static_cast<SyncSourceLink*>(SyncSourceHandler::getLink(source)); }
        inline bool isMine(const SyncSource& source) const
        { return getLink(source)->getMembership() == this; }

        struct IncomingRTPPktLink
        {
                IncomingRTPPktLink(IncomingRTPPkt* pkt, SyncSourceLink* sLink,
                                   struct timeval& recv_ts,
                                   uint32 shifted_ts,
                                   IncomingRTPPktLink* sp,
                                   IncomingRTPPktLink* sn,
                                   IncomingRTPPktLink* p,
                                   IncomingRTPPktLink* n) :
                        packet(pkt),
                        sourceLink(sLink),
                        prev(p), next(n),
                        srcPrev(sp), srcNext(sn),
                        receptionTime(recv_ts),
                        shiftedTimestamp(shifted_ts)
                { }

                ~IncomingRTPPktLink()
                { }

                inline SyncSourceLink* getSourceLink() const
                { return sourceLink; }

                inline void setSourceLink(SyncSourceLink* src)
                { sourceLink = src; }

                inline IncomingRTPPktLink* getNext() const
                { return next; }

                inline void setNext(IncomingRTPPktLink* nl)
                { next = nl; }

                inline IncomingRTPPktLink* getPrev() const
                { return prev; }

                inline void setPrev(IncomingRTPPktLink* pl)
                { prev = pl; }

                inline IncomingRTPPktLink* getSrcNext() const
                { return srcNext; }

                inline void setSrcNext(IncomingRTPPktLink* sn)
                { srcNext = sn; }

                inline IncomingRTPPktLink* getSrcPrev() const
                { return srcPrev; }

                inline void setSrcPrev(IncomingRTPPktLink* sp)
                { srcPrev = sp; }

                inline IncomingRTPPkt* getPacket() const
                { return packet; }

                inline void setPacket(IncomingRTPPkt* pkt)
                { packet = pkt; }

                inline void setRecvTime(const timeval &t)
                { receptionTime = t; }

                inline timeval getRecvTime() const
                { return receptionTime; }

                inline uint32 getTimestamp() const
                { return shiftedTimestamp; }

                inline void setTimestamp(uint32 ts)
                { shiftedTimestamp = ts;}

                // the packet this link refers to.
                IncomingRTPPkt* packet;
                // the synchronization source this packet comes from.
                SyncSourceLink* sourceLink;
                // global incoming packet queue links.
                IncomingRTPPktLink* prev, * next;
                // source specific incoming packet queue links.
                IncomingRTPPktLink* srcPrev, * srcNext;
                // time this packet was received at
                struct timeval receptionTime;
                // timestamp of the packet in host order and after
                // substracting the initial timestamp for its source
                // (it is an increment from the initial timestamp).
                uint32 shiftedTimestamp;
        };

        struct SyncSourceLink
        {
                // 2^16
                static const uint32 SEQNUMMOD;

                SyncSourceLink(MembershipBookkeeping* m,
                               SyncSource* s,
                               IncomingRTPPktLink* fp = NULL,
                               IncomingRTPPktLink* lp = NULL,
                               SyncSourceLink* ps = NULL,
                               SyncSourceLink* ns = NULL,
                               SyncSourceLink* ncollis = NULL) :
                        membership(m), source(s), first(fp), last(lp),
                        prev(ps), next(ns), nextCollis(ncollis),
                        prevConflict(NULL)
                { m->setLink(*s,this); // record that the source is associated
                  initStats();         // to this link.
                }

                ~SyncSourceLink();

                inline MembershipBookkeeping* getMembership()
                { return membership; }

                inline SyncSource* getSource() { return source; }

                inline IncomingRTPPktLink* getFirst()
                { return first; }

                inline void setFirst(IncomingRTPPktLink* fp)
                { first = fp; }

                inline IncomingRTPPktLink* getLast()
                { return last; }

                inline void setLast(IncomingRTPPktLink* lp)
                { last = lp; }

                inline SyncSourceLink* getPrev()
                { return prev; }

                inline void setPrev(SyncSourceLink* ps)
                { prev = ps; }

                inline SyncSourceLink* getNext()
                { return next; }

                inline void setNext(SyncSourceLink *ns)
                { next = ns; }

                inline SyncSourceLink* getNextCollis()
                { return nextCollis; }

                inline void setNextCollis(SyncSourceLink* ns)
                { nextCollis = ns; }

                inline ConflictingTransportAddress* getPrevConflict() const
                { return prevConflict; }

                void setPrevConflict(InetAddress& addr, tpport_t dataPort,
                                     tpport_t controlPort);

                unsigned char* getSenderInfo()
                { return senderInfo; }

                void setSenderInfo(unsigned char* si);

                unsigned char* getReceiverInfo()
                { return receiverInfo; }

                void setReceiverInfo(unsigned char* ri);

                inline timeval getLastPacketTime() const
                { return lastPacketTime; }

                inline timeval getLastRTCPPacketTime() const
                { return lastRTCPPacketTime; }

                inline timeval getLastRTCPSRTime() const
                { return lastRTCPSRTime; }

                inline uint32 getObservedPacketCount() const
                { return obsPacketCount; }

                inline void incObservedPacketCount()
                { obsPacketCount++; }

                inline uint32 getObservedOctetCount() const
                { return obsOctetCount; }

                inline void incObservedOctetCount(uint32 n)
                { obsOctetCount += n; }

                uint16
                getMaxSeqNum() const
                { return maxSeqNum; }

                void
                setMaxSeqNum(uint16 max)
                { maxSeqNum = max; }

                inline uint32
                getExtendedMaxSeqNum() const
                { return extendedMaxSeqNum; }

                inline void
                setExtendedMaxSeqNum(uint32 seq)
                { extendedMaxSeqNum = seq; }

                inline uint32 getCumulativePacketLost() const
                { return cumulativePacketLost; }

                inline void setCumulativePacketLost(uint32 pl)
                { cumulativePacketLost = pl; }

                inline uint8 getFractionLost() const
                { return fractionLost; }

                inline void setFractionLost(uint8 fl)
                { fractionLost = fl; }

                inline uint32 getLastPacketTransitTime()
                { return lastPacketTransitTime; }

                inline void setLastPacketTransitTime(uint32 time)
                { lastPacketTransitTime = time; }

                inline float getJitter() const
                { return jitter; }

                inline void setJitter(float j)
                { jitter = j; }

                inline uint32 getInitialDataTimestamp() const
                { return initialDataTimestamp; }

                inline void setInitialDataTimestamp(uint32 ts)
                { initialDataTimestamp = ts; }

                inline timeval getInitialDataTime() const
                { return initialDataTime; }

                inline void setInitialDataTime(timeval it)
                { initialDataTime = it; }

                bool getGoodbye()
                {
                        if(!flag)
                                return false;
                        flag = false;
                        return true;
                }

                bool getHello() {
                        if(flag)
                                return false;
                        flag = true;
                        return true;
                }

                inline uint32 getBadSeqNum() const
                { return badSeqNum; }

                inline void setBadSeqNum(uint32 seq)
                { badSeqNum = seq; }

                uint8 getProbation() const
                { return probation; }

                inline void setProbation(uint8 p)
                { probation = p; }

                inline void decProbation()
                { --probation; }

                bool isValid() const
                { return 0 == probation; }

                inline uint16 getBaseSeqNum() const
                { return baseSeqNum; }

                inline uint32 getSeqNumAccum() const
                { return seqNumAccum; }

                inline void incSeqNumAccum()
                { seqNumAccum += SEQNUMMOD; }

                inline void initSequence(uint16 seqnum)
                { maxSeqNum = seqNumAccum = seqnum; }

                void recordInsertion(const IncomingRTPPktLink& pl);

                void initStats();

                void computeStats();

                MembershipBookkeeping* membership;
                // The source this link object refers to.
                SyncSource* source;
                // first/last packets from this source in the queue.
                IncomingRTPPktLink* first, * last;
                // Links for synchronization sources located before
                // and after this one in the list of sources.
                SyncSourceLink* prev, * next;
                // Prev and next inside the hash table collision list.
                SyncSourceLink* nextCollis;
                ConflictingTransportAddress* prevConflict;
                unsigned char* senderInfo;
                unsigned char* receiverInfo;
                // time the last RTP packet from this source was
                // received at.
                timeval lastPacketTime;
                // time the last RTCP packet was received.
                timeval lastRTCPPacketTime;
                // time the lasrt RTCP SR was received. Required for
                // DLSR computation.
                timeval lastRTCPSRTime;

                // for outgoing RR reports.
                // number of packets received from this source.
                uint32 obsPacketCount;
                // number of octets received from this source.
                uint32 obsOctetCount;
                // the higher sequence number seen from this source
                uint16 maxSeqNum;
                uint32 extendedMaxSeqNum;
                uint32 cumulativePacketLost;
                uint8 fractionLost;
                // for interarrivel jitter computation
                uint32 lastPacketTransitTime;
                // interarrival jitter of packets from this source.
                float jitter;
                uint32 initialDataTimestamp;
                timeval initialDataTime;

                // this flag assures we only call one gotHello and one
                // gotGoodbye for this src.
                bool flag;

                // for source validation:
                uint32 badSeqNum;
                uint8 probation;  // packets in sequence before valid.
                uint16 baseSeqNum;
                uint32 expectedPrior;
                uint32 receivedPrior;
                uint32 seqNumAccum;
        };

        bool
        isRegistered(uint32 ssrc);

        SyncSourceLink*
        getSourceBySSRC(uint32 ssrc, bool& created);

        bool
        BYESource(uint32 ssrc);

        bool
        removeSource(uint32 ssrc);

        inline SyncSourceLink* getFirst()
        { return first; }

        inline SyncSourceLink* getLast()
        { return last; }

        inline uint32
        getMembersCount()
        { return Members::getMembersCount(); }

        inline void
        setMembersCount(uint32 n)
        { Members::setMembersCount(n); }

        inline uint32
        getSendersCount()
        { return Members::getSendersCount(); }

        static const size_t defaultMembersHashSize;
        static const uint32 SEQNUMMOD;

private:
        MembershipBookkeeping(const MembershipBookkeeping &o);

        MembershipBookkeeping&
        operator=(const MembershipBookkeeping &o);

        void
        endMembers();

        // Hash table with sources of RTP and RTCP packets
        uint32 sourceBucketsNum;
        SyncSourceLink** sourceLinks;
        // List of sources, ordered from older to newer
        SyncSourceLink* first, * last;
};

class __EXPORT IncomingDataQueue: public IncomingDataQueueBase,
        protected MembershipBookkeeping
{
public:
        class SyncSourcesIterator
        {
        public:
                typedef std::forward_iterator_tag iterator_category;
                typedef SyncSource value_type;
                typedef ptrdiff_t difference_type;
                typedef const SyncSource* pointer;
                typedef const SyncSource& reference;

                SyncSourcesIterator(SyncSourceLink* l = NULL) :
                        link(l)
                { }

                SyncSourcesIterator(const SyncSourcesIterator& si) :
                        link(si.link)
                { }

                reference operator*() const
                { return *(link->getSource()); }

                pointer operator->() const
                { return link->getSource(); }

                SyncSourcesIterator& operator++() {
                        link = link->getNext();
                        return *this;
                }

                SyncSourcesIterator operator++(int) {
                        SyncSourcesIterator result(*this);
                        ++(*this);
                        return result;
                }

                friend bool operator==(const SyncSourcesIterator& l,
                                       const SyncSourcesIterator& r)
                { return l.link == r.link; }

                friend bool operator!=(const SyncSourcesIterator& l,
                                       const SyncSourcesIterator& r)
                { return l.link != r.link; }

        private:
                SyncSourceLink *link;
        };

        SyncSourcesIterator begin()
        { return SyncSourcesIterator(MembershipBookkeeping::getFirst()); }

        SyncSourcesIterator end()
        { return SyncSourcesIterator(NULL); }

        const AppDataUnit*
        getData(uint32 stamp, const SyncSource* src = NULL);


        bool
        isWaiting(const SyncSource* src = NULL) const;

        uint32
        getFirstTimestamp(const SyncSource* src = NULL) const;

        void
        setMinValidPacketSequence(uint8 packets)
        { minValidPacketSequence = packets; }

        uint8
        getDefaultMinValidPacketSequence() const
        { return defaultMinValidPacketSequence; }

        uint8
        getMinValidPacketSequence() const
        { return minValidPacketSequence; }

        void
        setMaxPacketMisorder(uint16 packets)
        { maxPacketMisorder = packets; }

        uint16
        getDefaultMaxPacketMisorder() const
        { return defaultMaxPacketMisorder; }

        uint16
        getMaxPacketMisorder() const
        { return maxPacketMisorder; }

        void
        setMaxPacketDropout(uint16 packets) // default: 3000.
        { maxPacketDropout = packets; }

        uint16
        getDefaultMaxPacketDropout() const
        { return defaultMaxPacketDropout; }

        uint16
        getMaxPacketDropout() const
        { return maxPacketDropout; }

        // default value for constructors that allow to specify
        // members table s\ize
        inline static size_t
        getDefaultMembersSize()
        { return defaultMembersSize; }

        void
        setInQueueCryptoContext(CryptoContext* cc);

        void
        removeInQueueCryptoContext(CryptoContext* cc);

        CryptoContext*
        getInQueueCryptoContext(uint32 ssrc);

protected:
        IncomingDataQueue(uint32 size);

        virtual ~IncomingDataQueue()
        { }

        bool checkSSRCInIncomingRTPPkt(SyncSourceLink& sourceLink,
                                       bool is_new, InetAddress& na,
                                       tpport_t tp);

        void setSourceExpirationPeriod(uint8 intervals)
        { sourceExpirationPeriod = intervals; }

        virtual size_t
        takeInDataPacket();

        void renewLocalSSRC();

        IncomingDataQueue::IncomingRTPPktLink*
        getWaiting(uint32 timestamp, const SyncSource *src = NULL);

        bool
        recordReception(SyncSourceLink& srcLink, const IncomingRTPPkt& pkt,
                        const timeval recvtime);

        void
        recordExtraction(const IncomingRTPPkt& pkt);

        void purgeIncomingQueue();

        inline virtual void
        onNewSyncSource(const SyncSource&)
        { }

protected:
        inline virtual bool
        onRTPPacketRecv(IncomingRTPPkt&)
        { return true; }

        inline virtual void onExpireRecv(IncomingRTPPkt&)
        { return; }

        inline virtual bool
        onSRTPPacketError(IncomingRTPPkt& pkt, int32 errorCode)
        { return false; }

        inline virtual bool
        end2EndDelayed(IncomingRTPPktLink&)
        { return false; }

        bool
        insertRecvPacket(IncomingRTPPktLink* packetLink);

        virtual size_t
        recvData(unsigned char* buffer, size_t length,
                 InetHostAddress& host, tpport_t& port) = 0;

        virtual size_t
        getNextDataPacketSize() const = 0;

        mutable ThreadLock recvLock;
        // reception queue
        IncomingRTPPktLink* recvFirst, * recvLast;
        // values for packet validation.
        static const uint8 defaultMinValidPacketSequence;
        static const uint16 defaultMaxPacketMisorder;
        static const uint16 defaultMaxPacketDropout;
        uint8 minValidPacketSequence;
        uint16 maxPacketMisorder;
        uint16 maxPacketDropout;
        static const size_t defaultMembersSize;
        uint8 sourceExpirationPeriod;
        mutable Mutex cryptoMutex;
        std::list<CryptoContext *> cryptoContexts;
};
 // iqueue

#ifdef  CCXX_NAMESPACES
}
#endif

#endif  //CCXX_RTP_IQUEUE_H_