6
\$\begingroup\$

I'm working on a design for a video buffer in C++ that will be used by many consumers both in the native library space(C++) as well as Java side using JNI. My Idea is the following:

Have a buffer manager that will receive frames directly from the hardware and allocate each frame only once on the heap using shared pointer if it is aware of at least one consumer for this frame. When a consumer would like to to receive video frame it will call the bind method and specify the buffer size it wants since not all consumers will process data at the same rate. I think it is better to have a dedicated buffer for each client/consumer. After binding the buffer manager returns a DataBufferStream object that will be used by the consumer to read/fetch video frames. The DataBufferStram class is nothing more than a circular buffer of certain type and size. Using JNI im planning to pass a pointer to the buffer or copy the data to a byte array

I would like to get some feedback from people here to see if there is anything I can improve on. is there a better solution that you can recommend

    class BufferManager {

    public:
       std::shared_ptr<DataBufferStream<SHARED_ARRAY>> bind(const unsigned int bufferSize,const int 
       requestorID);
       void unbind(const int requestorID);
       std::shared_ptr<BufferManager> getInstance();

    private:
       std::map<int,std::shared_ptr<DataBufferStream<SHARED_ARRAY>>> mDataBufferMap;
       std::mutex mMutex;
    };

    std::shared_ptr<DataBufferStream<SHARED_ARRAY>> BufferManager::bind(const unsigned int 
    bufferSize,const int requestorID) {
    std::lock_guard<std::mutex> lockGuard(mMutex);
    auto it = mDataBufferMap.find(requestorID);
    if(it == mDataBufferMap.end()){
        mDataBufferMap[requestorID] = std::make_shared<DataBufferStream<SHARED_ARRAY>>(bufferSize);
    }
     return mDataBufferMap[requestorID];
   }

   void BufferManager::unbind(const int requestorID) {
        std::lock_guard<std::mutex> lock(mMutex);
        mDataBufferMap.erase(requestorID);
   }

`

    #include "../include/boost/circular_buffer.hpp"
    #include <mutex>
    #include <array>
    #include <condition_variable>
    #include <boost/shared_array.hpp>

    typedef boost::shared_array<char> SHARED_ARRAY;

    template <class T>
    class DataBufferStream {

    private:
        boost::circular_buffer<T> mCircularBuffer;
        std::mutex mMutex;
        std::condition_variable mConditionalVariable;
        unsigned int mIndex;
    public:
        DataBufferStream(const unsigned int bufferSize);
        DataBufferStream(DataBufferStream& other);
        DataBufferStream() = delete;
        virtual ~DataBufferStream();
        void pushData(T data);
        T fetchData();
        T readData(unsigned int duration);
        T operator[](const unsigned int index);
        T operator*();
        void operator++();
        void operator--();
        DataBufferStream<T> &operator=(DataBufferStream<T>& other);
        void clear();
        unsigned int size();
     };

` #include "DataBuffer.h"

template <class T>
DataBufferStream<T>::DataBufferStream(const unsigned int bufferSize):
mCircularBuffer(bufferSize),
mMutex(),
mConditionalVariable(),
mIndex(0)
{
}

template <class T>
T DataBufferStream<T>::fetchData() {
std::lock_guard<std::mutex> lock (mMutex);
if(mCircularBuffer.size()>0) {
    auto ptr = mCircularBuffer.front();
    mCircularBuffer.pop_front();
    return ptr;
}
return nullptr;
}

  template <class T>
  void DataBufferStream<T>::pushData(T data) {
       std::lock_guard<std::mutex> lock (mMutex);
       mCircularBuffer.push_back(data);
       mConditionalVariable.notify_all();
  }

    template<class T>
    T DataBufferStream<T>::operator[](const unsigned int index) {
           std::lock_guard<std::mutex> lock (mMutex);
           return mCircularBuffer.size()>index ? mCircularBuffer[index] : nullptr;
    }

  template<class T>
  DataBufferStream<T>::~DataBufferStream() {

  }

  template<class T>
  T DataBufferStream<T>::operator*() {
        std::lock_guard<std::mutex> lock(mMutex);
        return mCircularBuffer.size() > mIndex ? mCircularBuffer[mIndex] : nullptr;
  }

   template<class T>
   void DataBufferStream<T>::operator++() {
        std::lock_guard<std::mutex> lock(mMutex);
        mIndex = mCircularBuffer.size() < mIndex+1 ? 0 : mIndex++;
   }

   template<class T>
   void DataBufferStream<T>::operator--() {
        std::lock_guard<std::mutex> lock(mMutex);
        mIndex = mIndex > 0 ? mIndex-- : 0;
   }

    template<class T>
    void DataBufferStream<T>::clear() {
         std::lock_guard<std::mutex> lock(mMutex);
         mCircularBuffer.clear();
         mIndex = mCircularBuffer.size();
    }

    template<class T>
    DataBufferStream<T>::DataBufferStream(DataBufferStream &other):
         mMutex(),
         mConditionalVariable(){
         std::lock_guard<std::mutex> lock(other.mMutex);
         this->mCircularBuffer = other.mCircularBuffer;
         this->mIndex = other.mIndex;
    }

    template<class T>
    DataBufferStream<T> &DataBufferStream<T>::operator=(DataBufferStream<T> &other) {
          if(this!=&other){
                std::unique_lock<std::mutex> myLock (mMutex,std::defer_lock);
                std::unique_lock<std::mutex> otherLock(other.mMutex,std::defer_lock);
                std::lock(myLock,otherLock);
                mCircularBuffer = other.mCircularBuffer;
                mIndex = other.mIndex;
          }
        return *this;
     }

    template<class T>
    unsigned int DataBufferStream<T>::size() {
           std::lock_guard<std::mutex> lock(mMutex);
           return mCircularBuffer.size();
    }

    template<class T>
    T DataBufferStream<T>::readData(unsigned int duration) {
        auto data = fetchData();
        std::unique_lock<std::mutex> lock(mMutex);
        if((data == nullptr) && 
               (mConditionalVariable.wait_for(lock,std::chrono::milliseconds(duration)) == 
               std::cv_status::no_timeout)){
               if(mCircularBuffer.size()>0) {
                  auto data = mCircularBuffer.front();
                  mCircularBuffer.pop_front();
              }
        }
     return data;
    }

    template class DataBufferStream<SHARED_ARRAY>;
\$\endgroup\$
2
\$\begingroup\$

If your system is going to be impacted by high rate operations, looks like, I recommend you to preallocate the memory on boot time of the application, so during the execution of your program you dont do any allocations, for example:

auto it = mDataBufferMap.find(requestorID);
if(it == mDataBufferMap.end()){
    mDataBufferMap[requestorID] = std::make_shared<DataBufferStream<SHARED_ARRAY>>(bufferSize);
}

will be like

auto it = mDataBufferMap.find(requestorID);
if(it == mDataBufferMap.end()){
    mDataBufferMap[requestorID] = getDataBufferStreamFromPool();
}

And on the erase release the block from the map but without free the memory. Hope it helps

\$\endgroup\$
  • \$\begingroup\$ () thanks for the feedback \$\endgroup\$ – eliranno Nov 15 '19 at 20:37

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.