iqueue.h

Go to the documentation of this file.

// Copyright (C) 2001-2015 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 Lesser General Public License
// along with GNU ccRTP.  If not, see <http://www.gnu.org/licenses/>.
//
// 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>

NAMESPACE_COMMONCPP

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)
    { SysTime::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 void setBaseSeqNum(uint16 seqnum)
        { baseSeqNum = seqnum; }

        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 std::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

END_NAMESPACE

#endif  //CCXX_RTP_IQUEUE_H_