AbstractFifo.hxx

Go to the documentation of this file.

#ifndef RESIP_AbstractFifo_hxx
#define RESIP_AbstractFifo_hxx 
 
#include "rutil/ResipAssert.h"
#include <deque>
 
#include "rutil/Mutex.hxx"
#include "rutil/Condition.hxx"
#include "rutil/Lock.hxx"
#include "rutil/CongestionManager.hxx"
 
#include "rutil/compat.hxx"
#include "rutil/Timer.hxx"
 
namespace resip
{
class FifoStatsInterface
{
   public:
         
     FifoStatsInterface();
     virtual ~FifoStatsInterface();
     
     virtual time_t expectedWaitTimeMilliSec() const =0;
 
      virtual time_t getTimeDepth() const = 0;
     
      virtual size_t getCountDepth() const = 0;
 
     virtual time_t averageServiceTimeMicroSec() const = 0;
 
      inline uint8_t getRole() const {return mRole;}
 
      inline void setRole(uint8_t role) {mRole=role;}
 
      inline void setDescription(const resip::Data& description)
      {
         mDescription=description;
      }
 
      virtual const resip::Data& getDescription() const {return mDescription;}
 
   protected:
      Data mDescription;
      uint8_t mRole;
};
 
#define RESIP_FIFO_NOWAIT   -1
#define RESIP_FIFO_FOREVER  0
 
template <typename T>
class AbstractFifo : public FifoStatsInterface
{
   public:
      AbstractFifo()
         : FifoStatsInterface(),
            mLastSampleTakenMicroSec(0),
            mCounter(0),
            mAverageServiceTimeMicroSec(0),
            mSize(0)
      {}
 
      virtual ~AbstractFifo()
      {
      }
 
      bool empty() const
      {
         Lock lock(mMutex); (void)lock;
         return mFifo.empty();
      }
 
      virtual unsigned int size() const
      {
         Lock lock(mMutex); (void)lock;
         return (unsigned int)mFifo.size();
      }
 
      bool messageAvailable() const
      {
         Lock lock(mMutex); (void)lock;
         return !mFifo.empty();
      }
 
      virtual time_t getTimeDepth() const
      {
         return 0;
      }
 
      virtual size_t getCountDepth() const
      {
         return mSize;
      }
 
      virtual time_t expectedWaitTimeMilliSec() const
      {
         return ((mAverageServiceTimeMicroSec*mSize)+500)/1000;
      }
 
      virtual time_t averageServiceTimeMicroSec() const
      {
         return mAverageServiceTimeMicroSec;
      }
 
      virtual void clear() {};
 
   protected:
      T getNext()
      {
         Lock lock(mMutex); (void)lock;
         onFifoPolled();
 
         // Wait util there are messages available.
         while (mFifo.empty())
         {
            mCondition.wait(lock);
         }
 
         // Return the first message on the fifo.
         //
         T firstMessage(mFifo.front());
         mFifo.pop_front();
         onMessagePopped();
         return firstMessage;
      }
 
 
      bool getNext(int ms, T& toReturn)
      {
         if(ms == 0) 
         {
            toReturn = getNext();
            return true;
         }
 
         if(ms < 0)
         {
            Lock lock(mMutex); (void)lock;
            onFifoPolled();
            if (mFifo.empty())  // WATCHOUT: Do not test mSize instead
              return false;
            toReturn = mFifo.front();
            mFifo.pop_front();
            onMessagePopped();
            return true;
         }
 
         const auto begin = std::chrono::steady_clock::now();
         const auto end = begin + std::chrono::milliseconds(ms);
         Lock lock(mMutex); (void)lock;
         onFifoPolled();
 
         // Wait until there are messages available
         while (mFifo.empty())
         {
            if(ms==0)
            {
               return false;
            }
            auto now = std::chrono::steady_clock::now();
            if(now >= end)
            {
                return false;
            }
      
            // bail if total wait time exceeds limit
            bool signaled = mCondition.wait_until(lock, end) == std::cv_status::no_timeout;
            if (!signaled)
            {
               return false;
            }
         }
      
         // Return the first message on the fifo.
         //
         toReturn=mFifo.front();
         mFifo.pop_front();
         onMessagePopped();
         return true;
      }
 
      typedef std::deque<T> Messages;
 
      void getMultiple(Messages& other, unsigned int max)
      {
         Lock lock(mMutex); (void)lock;
         onFifoPolled();
         resip_assert(other.empty());
         while (mFifo.empty())
         {
            mCondition.wait(lock);
         }
 
         if(mFifo.size() <= max)
         {
            std::swap(mFifo, other);
            onMessagePopped(mSize);
         }
         else
         {
            size_t num=max;
            while( 0 != max-- )
            {
               other.push_back(mFifo.front());
               mFifo.pop_front();
            }
            onMessagePopped((unsigned int)num);
         }
      }
 
      bool getMultiple(int ms, Messages& other, unsigned int max)
      {
         if(ms==0)
         {
            getMultiple(other,max);
            return true;
         }
 
         resip_assert(other.empty());
         const auto begin = std::chrono::steady_clock::now();
         const auto end = begin + std::chrono::milliseconds(ms); // !kh! ms should've been unsigned :(
         Lock lock(mMutex); (void)lock;
         onFifoPolled();
 
         // Wait until there are messages available
         while (mFifo.empty())
         {
            if(ms < 0)
            {
               return false;
            }
            const auto now = std::chrono::steady_clock::now();
            if(now >= end)
            {
                return false;
            }
 
            // bail if total wait time exceeds limit
            bool signaled = mCondition.wait_until(lock, end) == std::cv_status::no_timeout;
            if (!signaled)
            {
               return false;
            }
         }
 
         if(mFifo.size() <= max)
         {
            std::swap(mFifo, other);
            onMessagePopped(mSize);
         }
         else
         {
            size_t num=max;
            while( 0 != max-- )
            {
               other.push_back(mFifo.front());
               mFifo.pop_front();
            }
            onMessagePopped((unsigned int)num);
         }
         return true;
      }
 
      size_t add(const T& item)
      {
         Lock lock(mMutex); (void)lock;
         mFifo.push_back(item);
         mCondition.notify_one();
         onMessagePushed(1);
         return mFifo.size();
      }
 
      size_t addMultiple(Messages& items)
      {
         Lock lock(mMutex); (void)lock;
         size_t size=items.size();
         if(mFifo.empty())
         {
            std::swap(mFifo, items);
         }
         else
         {
            // I suppose it is possible to optimize this as a push_front() from
            // mFifo to items, and then do a swap, if items is larger.
            while(!items.empty())
            {
               mFifo.push_back(items.front());
               items.pop_front();
            }
         }
         mCondition.notify_one();
         onMessagePushed((int)size);
         return mFifo.size();
      }
 
      Messages mFifo;
      mutable Mutex mMutex;
      Condition mCondition;
 
      mutable uint64_t mLastSampleTakenMicroSec;
      mutable uint32_t mCounter;
      mutable uint32_t mAverageServiceTimeMicroSec;
      // std::deque has to perform some amount of traversal to calculate its 
      // size; we maintain this count so that it can be queried without locking, 
      // in situations where it being off by a small amount is ok.
      uint32_t mSize;
 
      virtual void onFifoPolled()
      {
         // !bwc! TODO allow this sampling frequency to be tweaked
         if(mLastSampleTakenMicroSec &&
            mCounter &&
            (mCounter >= 64 || mFifo.empty()))
         {
            uint64_t now(Timer::getTimeMicroSec());
            uint64_t diff = now-mLastSampleTakenMicroSec;
 
            if(mCounter >= 4096)
            {
               mAverageServiceTimeMicroSec=(uint32_t)resipIntDiv(diff, mCounter);
            }
            else // fifo got emptied; merge into a rolling average
            {
               // .bwc. This is a moving average with period 64, round to 
               // nearest int.
               mAverageServiceTimeMicroSec=(uint32_t)resipIntDiv(
                     diff+((4096-mCounter)*mAverageServiceTimeMicroSec),
                     4096U);
            }
            mCounter=0;
            if(mFifo.empty())
            {
               mLastSampleTakenMicroSec=0;
            }
            else
            {
               mLastSampleTakenMicroSec=now;
            }
         }
      }
 
      virtual void onMessagePopped(unsigned int num=1)
      {
         mCounter+=num;
         mSize-=num;
      }
 
      virtual void onMessagePushed(int num)
      {
         if(mSize==0)
         {
            // Fifo went from empty to non-empty. Take a timestamp, and record
            // how long it takes to process some messages.
            mLastSampleTakenMicroSec=Timer::getTimeMicroSec();
         }
         mSize+=num;
      }
   private:
      // no value semantics
      AbstractFifo(const AbstractFifo&);
      AbstractFifo& operator=(const AbstractFifo&);
};
 
} // namespace resip
 
#endif
 
/* ====================================================================
 * The Vovida Software License, Version 1.0 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 * 
 * 3. The names "VOCAL", "Vovida Open Communication Application Library",
 *    and "Vovida Open Communication Application Library (VOCAL)" must
 *    not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact vocal@vovida.org.
 *
 * 4. Products derived from this software may not be called "VOCAL", nor
 *    may "VOCAL" appear in their name, without prior written
 *    permission of Vovida Networks, Inc.
 * 
 * THIS SOFTWARE IS PROVIDED "AS IS" AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, TITLE AND
 * NON-INFRINGEMENT ARE DISCLAIMED.  IN NO EVENT SHALL VOVIDA
 * NETWORKS, INC. OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT DAMAGES
 * IN EXCESS OF $1,000, NOR FOR ANY INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
 * OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 * USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
 * DAMAGE.
 * 
 * ====================================================================
 * 
 * This software consists of voluntary contributions made by Vovida
 * Networks, Inc. and many individuals on behalf of Vovida Networks,
 * Inc.  For more information on Vovida Networks, Inc., please see
 * <http://www.vovida.org/>.
 *
 */