#include "PL_Queue.h"
|
#include "logger.h"
|
#include "MaterialBuffer.h"
|
#include <set>
|
#include <queue>
|
#include <string.h>
|
|
struct QBlock
|
{
|
PipeMaterial lastPM;
|
std::vector<MB_Frame> lastMbfs;
|
|
uint8_t* data;
|
size_t maxSize;
|
size_t size;
|
|
QBlock() : lastPM(), lastMbfs(), data(nullptr), maxSize(0), size(0) { }
|
|
QBlock(uint8_t* _data, size_t _maxSize) : lastPM(), lastMbfs(), data(_data), maxSize(_maxSize), size(0) { }
|
|
~QBlock()
|
{
|
data = nullptr;
|
maxSize = 0;
|
size = 0;
|
}
|
|
void reset()
|
{
|
size = 0;
|
lastPM.reset();
|
lastMbfs.clear();
|
}
|
};
|
|
struct QBlockPool
|
{
|
uint8_t* blocksDataCache;
|
|
typedef std::set<QBlock*> qblock_set_t;
|
qblock_set_t staticPool;
|
qblock_set_t freePool;
|
|
pthread_mutex_t pool_mutex;
|
|
QBlockPool() : blocksDataCache(nullptr), staticPool(), freePool(), pool_mutex()
|
{
|
}
|
|
QBlockPool(size_t maxBlockCount, size_t maxBlockSize) :
|
blocksDataCache(nullptr)
|
{
|
bp_init(maxBlockCount, maxBlockSize);
|
}
|
|
void bp_init(size_t maxBlockCount, size_t maxBlockSize)
|
{
|
pthread_mutex_init(&pool_mutex, NULL);
|
pthread_mutex_lock(&pool_mutex);
|
|
const size_t totalBytes = maxBlockCount * maxBlockSize;
|
blocksDataCache = new uint8_t[totalBytes];
|
LOGP(WARN, "QBlockPool allocate byte: %u", totalBytes);
|
|
for (size_t i = 0; i < maxBlockCount; i++)
|
{
|
uint8_t* qbData = new (blocksDataCache + i * maxBlockSize) uint8_t[maxBlockSize];
|
QBlock* qb = new QBlock(qbData, maxBlockSize);
|
staticPool.insert(qb);
|
freePool.insert(qb);
|
}
|
|
pthread_mutex_unlock(&pool_mutex);
|
}
|
|
void bp_reset()
|
{
|
pthread_mutex_lock(&pool_mutex);
|
|
if (freePool.size() != staticPool.size())
|
{
|
LOG_WARN << "QBlockPool memory leakage" << LOG_ENDL;
|
}
|
|
freePool.clear();
|
|
for(qblock_set_t::iterator iter = staticPool.begin(); iter != staticPool.end(); ++iter)
|
delete *iter;
|
staticPool.clear();
|
|
delete[] blocksDataCache;
|
blocksDataCache = nullptr;
|
|
pthread_mutex_unlock(&pool_mutex);
|
}
|
|
~QBlockPool()
|
{
|
bp_reset();
|
pthread_mutex_destroy(&pool_mutex);
|
}
|
|
QBlock* bp_alloc()
|
{
|
pthread_mutex_lock(&pool_mutex);
|
|
if (freePool.empty())
|
{
|
pthread_mutex_unlock(&pool_mutex);
|
LOG_WARN << "no free blocks" << LOG_ENDL;
|
return nullptr;
|
}
|
|
qblock_set_t::iterator iter = freePool.begin();
|
QBlock* qb = *iter;
|
freePool.erase(iter);
|
|
pthread_mutex_unlock(&pool_mutex);
|
return qb;
|
}
|
|
void bp_free(QBlock* qb)
|
{
|
if (qb == nullptr)
|
return;
|
|
pthread_mutex_lock(&pool_mutex);
|
|
qblock_set_t::iterator iter = staticPool.find(qb);
|
if (iter == staticPool.end())
|
{
|
pthread_mutex_unlock(&pool_mutex);
|
LOG_WARN << "block not belongs to pool" << LOG_ENDL;
|
return;
|
}
|
|
//iter = freePool.find(qb);
|
//if (iter != freePool.end())
|
//{
|
// printf("QBlockPool bp_free block is free");
|
// return;
|
//}
|
|
qb->reset();
|
freePool.insert(qb);
|
pthread_mutex_unlock(&pool_mutex);
|
}
|
|
bool bp_empty() const
|
{
|
return freePool.empty();
|
}
|
};
|
|
struct PL_Queue_Internal
|
{
|
PL_Queue_Config config;
|
QBlockPool poolBlocks;
|
|
typedef std::queue<QBlock*> qblock_queue_t;
|
qblock_queue_t queBlocks;
|
|
pthread_mutex_t* queue_mutex;
|
pthread_mutex_t* sync_full_mutex;
|
pthread_mutex_t* sync_empty_mutex;
|
|
PL_Queue_Internal() :
|
config(), poolBlocks(), queBlocks(),
|
queue_mutex(new pthread_mutex_t),
|
sync_full_mutex(new pthread_mutex_t), sync_empty_mutex(new pthread_mutex_t)
|
{
|
pthread_mutex_init(queue_mutex, NULL);
|
pthread_mutex_init(sync_full_mutex, NULL);
|
pthread_mutex_init(sync_empty_mutex, NULL);
|
}
|
|
~PL_Queue_Internal()
|
{
|
if (queue_mutex != nullptr)
|
{
|
pthread_mutex_destroy(queue_mutex);
|
delete queue_mutex;
|
queue_mutex = nullptr;
|
}
|
|
if (sync_full_mutex != nullptr)
|
{
|
pthread_mutex_destroy(sync_full_mutex);
|
delete sync_full_mutex;
|
sync_full_mutex = nullptr;
|
}
|
|
if (sync_empty_mutex != nullptr)
|
{
|
pthread_mutex_destroy(sync_empty_mutex);
|
delete sync_empty_mutex;
|
sync_empty_mutex = nullptr;
|
}
|
}
|
|
void reset()
|
{
|
if (queue_mutex != nullptr)
|
{
|
pthread_mutex_destroy(queue_mutex);
|
delete queue_mutex;
|
queue_mutex = nullptr;
|
}
|
|
queue_mutex = new pthread_mutex_t;
|
pthread_mutex_init(queue_mutex, NULL);
|
|
if (sync_full_mutex != nullptr)
|
{
|
pthread_mutex_destroy(sync_full_mutex);
|
delete sync_full_mutex;
|
sync_full_mutex = nullptr;
|
}
|
|
sync_full_mutex = new pthread_mutex_t;
|
pthread_mutex_init(sync_full_mutex, NULL);
|
|
if (sync_empty_mutex != nullptr)
|
{
|
pthread_mutex_destroy(sync_empty_mutex);
|
delete sync_empty_mutex;
|
sync_empty_mutex = nullptr;
|
}
|
|
sync_empty_mutex = new pthread_mutex_t;
|
pthread_mutex_init(sync_empty_mutex, NULL);
|
|
poolBlocks.bp_reset();
|
}
|
};
|
|
PipeLineElem* create_PL_Queue()
|
{
|
return new PL_Queue;
|
}
|
|
PL_Queue::PL_Queue() : internal(new PL_Queue_Internal)
|
{
|
}
|
|
PL_Queue::~PL_Queue()
|
{
|
delete (PL_Queue_Internal*)internal;
|
internal= nullptr;
|
}
|
|
bool PL_Queue::init(void* args)
|
{
|
PL_Queue_Internal* in = (PL_Queue_Internal*)internal;
|
|
in->reset();
|
if (args != nullptr)
|
{
|
PL_Queue_Config* config = (PL_Queue_Config*)args;
|
in->config = *config;
|
}
|
in->poolBlocks.bp_init(in->config.maxBlockCount, in->config.maxBlockSize);
|
|
if (in->config.syncQueueFull)
|
{
|
int ret = pthread_mutex_lock(in->sync_full_mutex);
|
if(ret != 0)
|
{
|
LOGP(WARN, "pthread_mutex_lock sync_full_mutex: %d", ret);
|
return false;
|
}
|
}
|
|
if (in->config.syncQueueEmpty)
|
{
|
int ret = pthread_mutex_lock(in->sync_empty_mutex);
|
if(ret != 0)
|
{
|
LOGP(WARN, "pthread_mutex_lock sync_empty_mutex: %d", ret);
|
return false;
|
}
|
}
|
|
return true;
|
}
|
|
void PL_Queue::finit()
|
{
|
PL_Queue_Internal* in = (PL_Queue_Internal*)internal;//#todo delete
|
}
|
|
bool PL_Queue::pay(const PipeMaterial& pm)
|
{
|
PL_Queue_Internal* in = (PL_Queue_Internal*)internal;
|
|
QBlock* qb = nullptr;
|
if (in->queBlocks.size() >= in->config.maxBlockCount || in->poolBlocks.bp_empty())
|
{
|
// there is no available qb
|
|
if (in->config.queueFullDropFrontBlock)
|
{
|
LOG_WARN << "PL_Queue::pay queueFullDropFrontBlock" << LOG_ENDL;
|
|
pthread_mutex_lock(in->queue_mutex);
|
qb = in->queBlocks.front();
|
in->queBlocks.pop();
|
pthread_mutex_unlock(in->queue_mutex);
|
|
in->poolBlocks.bp_free(qb);
|
qb = nullptr;
|
}
|
else
|
{
|
if (in->config.syncQueueFull)
|
{
|
LOG_WARN << "PL_Queue::pay sync by sync_full_mutex" << LOG_ENDL;
|
pthread_mutex_lock(in->sync_full_mutex);
|
}
|
}
|
}
|
|
if (in->queBlocks.size() >= in->config.maxBlockCount || in->poolBlocks.bp_empty())
|
{
|
LOG_WARN << "PL_Queue::pay full" << LOG_ENDL;
|
return false;
|
}
|
|
qb = in->poolBlocks.bp_alloc();
|
if (qb == nullptr)
|
{
|
LOG_WARN << "PL_Queue::pay bp_alloc nullptr" << LOG_ENDL;
|
return false;
|
}
|
|
bool cacheOk = false;
|
|
if (pm.type == PipeMaterial::PMT_BYTES)
|
{
|
qb->lastPM = pm;
|
|
if (in->config.copyData)
|
{
|
memcpy(qb->data, pm.buffer, pm.buffSize);
|
qb->size = pm.buffSize;
|
qb->lastPM.buffer = qb->data;
|
}
|
|
cacheOk = true;
|
}
|
else if (pm.type == PipeMaterial::PMT_FRAME)
|
{
|
qb->lastPM = pm;
|
|
MB_Frame* mbf = (MB_Frame*)pm.buffer;
|
qb->lastMbfs.push_back(*mbf);
|
|
qb->lastPM.buffer = &(qb->lastMbfs[0]);
|
|
if (in->config.copyData)
|
{
|
memcpy(qb->data, mbf->buffer, mbf->buffSize);
|
qb->size = mbf->buffSize;
|
qb->lastMbfs[0].buffer = qb->data;
|
}
|
|
cacheOk = true;
|
}
|
else if (pm.type == PipeMaterial::PMT_FRAME_LIST)
|
{
|
if (in->config.cacheFrameListFunc != nullptr)
|
{
|
size_t s = qb->maxSize;
|
cacheOk = in->config.cacheFrameListFunc(pm, qb->lastPM, qb->lastMbfs, qb->data, s);
|
if (cacheOk)
|
qb->size = s;
|
}
|
}
|
|
if (! cacheOk)
|
{
|
in->poolBlocks.bp_free(qb);
|
|
LOG_WARN << "pm/mbf type not support" << LOG_ENDL;
|
return false;
|
}
|
|
pthread_mutex_lock(in->queue_mutex);
|
in->queBlocks.push(qb);
|
pthread_mutex_unlock(in->queue_mutex);
|
|
if (in->config.syncQueueEmpty)
|
pthread_mutex_unlock(in->sync_empty_mutex);
|
|
return true;
|
}
|
|
void PL_Queue::pm_deleter_qb(PipeMaterial* pm, bool lastRet)
|
{
|
if (pm->former == nullptr || pm->args == nullptr)
|
return;
|
|
PL_Queue* _this = (PL_Queue*)pm->former;
|
QBlock* qb = (QBlock*)pm->args;
|
PL_Queue_Internal* in = (PL_Queue_Internal*)_this->internal;
|
|
pthread_mutex_lock(in->queue_mutex);//#todo lastRet
|
in->queBlocks.pop();
|
pthread_mutex_unlock(in->queue_mutex);
|
|
in->poolBlocks.bp_free(qb);
|
|
if (in->config.syncQueueFull)
|
pthread_mutex_unlock(in->sync_full_mutex);
|
}
|
|
bool PL_Queue::gain(PipeMaterial& pm)
|
{
|
PL_Queue_Internal* in = (PL_Queue_Internal*)internal;
|
|
if (in->queBlocks.empty())
|
{
|
if (in->config.syncQueueEmpty)
|
pthread_mutex_lock(in->sync_empty_mutex);
|
}
|
|
if (in->queBlocks.empty())
|
{
|
LOG_DEBUG << "PL_Queue::gain empty" << LOG_ENDL;
|
pm.buffer = nullptr;
|
pm.buffSize = 0;
|
pm.former = this;
|
return false;
|
}
|
|
QBlock* qb = nullptr;
|
pthread_mutex_lock(in->queue_mutex);
|
qb = in->queBlocks.front();
|
//in->queBlocks.pop(); // pop in deleter
|
pthread_mutex_unlock(in->queue_mutex);
|
|
pm = qb->lastPM;
|
|
pm.former = this;
|
pm.deleter = pm_deleter_qb;
|
pm.args = qb;
|
return true;
|
}
|
