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 want since not all consumer 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>;
