#include "PL_Queue.h"
|
#include <set>
|
#include <queue>
|
#include <string.h>
|
|
struct QBlock
|
{
|
uint8_t* data;
|
size_t maxSize;
|
size_t size;
|
|
QBlock() : data(nullptr), maxSize(0), size(0) { }
|
|
QBlock(uint8_t* _data, size_t _maxSize) : data(_data), maxSize(_maxSize), size(0) { }
|
|
~QBlock()
|
{
|
data = nullptr;
|
maxSize = 0;
|
size = 0;
|
}
|
};
|
|
struct QBlockPool
|
{
|
uint8_t* blocksDataPool;
|
|
typedef std::set<QBlock*> qpool_t;
|
qpool_t staticPool;
|
qpool_t freePool;
|
|
QBlockPool() : blocksDataPool(nullptr)
|
{
|
}
|
|
QBlockPool(size_t maxBlockCount, size_t maxBlockSize) :
|
blocksDataPool(nullptr)
|
{
|
bp_init(maxBlockCount, maxBlockSize);
|
}
|
|
void bp_init(size_t maxBlockCount, size_t maxBlockSize)
|
{
|
const size_t totalBytes = maxBlockCount * maxBlockSize;
|
blocksDataPool = new uint8_t[totalBytes];
|
printf("QBlockPool allocate byte: %u\n", totalBytes);
|
|
for (size_t i = 0; i < maxBlockCount; i++)
|
{
|
uint8_t* qbData = new (blocksDataPool + i * maxBlockSize) uint8_t[maxBlockSize];
|
QBlock* qb = new QBlock(qbData, maxBlockSize);
|
staticPool.insert(qb);
|
freePool.insert(qb);
|
}
|
}
|
|
~QBlockPool()
|
{
|
if (freePool.size() != staticPool.size())
|
{
|
printf("QBlockPool memory leakage\n");
|
}
|
|
for(qpool_t::iterator iter = staticPool.begin(); iter != staticPool.end(); ++iter)
|
{
|
delete *iter;
|
}
|
|
delete[] blocksDataPool;
|
blocksDataPool = nullptr;
|
}
|
|
QBlock* bp_alloc()
|
{
|
if (freePool.empty())
|
{
|
printf("QBlockPool bp_alloc empty\n");
|
return nullptr;
|
}
|
|
qpool_t::iterator iter = freePool.begin();
|
QBlock* qb = *iter;
|
freePool.erase(iter);
|
return qb;
|
}
|
|
void bp_free(QBlock* qb)
|
{
|
if (qb == nullptr)
|
return;
|
|
qpool_t::iterator iter = staticPool.find(qb);
|
if (iter == staticPool.end())
|
{
|
printf("QBlockPool bp_free not in pool\n");
|
return;
|
}
|
|
//iter = freePool.find(qb);
|
//if (iter != freePool.end())
|
//{
|
// printf("QBlockPool bp_free block is free");
|
// return;
|
//}
|
|
qb->size = 0;
|
freePool.insert(qb);
|
}
|
|
bool bp_empty() const
|
{
|
return freePool.empty();
|
}
|
};
|
|
struct PL_Queue_Internal
|
{
|
PL_Queue_Config config;
|
QBlockPool pool;
|
|
typedef std::queue<QBlock*> queue_t;
|
queue_t que;
|
|
pthread_mutex_t* queue_pool_mutex;
|
pthread_mutex_t* sync_full_mutex;
|
pthread_mutex_t* sync_empty_mutex;
|
|
PL_Queue_Internal() :
|
config(), pool(), que(),
|
queue_pool_mutex(new pthread_mutex_t),
|
sync_full_mutex(new pthread_mutex_t), sync_empty_mutex(new pthread_mutex_t)
|
{
|
pthread_mutex_init(queue_pool_mutex, NULL);
|
pthread_mutex_init(sync_full_mutex, NULL);
|
pthread_mutex_init(sync_empty_mutex, NULL);
|
}
|
|
~PL_Queue_Internal()
|
{
|
if (queue_pool_mutex != nullptr)
|
{
|
pthread_mutex_destroy(queue_pool_mutex);
|
delete queue_pool_mutex;
|
queue_pool_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_pool_mutex != nullptr)
|
{
|
pthread_mutex_destroy(queue_pool_mutex);
|
delete queue_pool_mutex;
|
queue_pool_mutex = nullptr;
|
}
|
|
queue_pool_mutex = new pthread_mutex_t;
|
pthread_mutex_init(queue_pool_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);
|
|
//#todo free que
|
}
|
};
|
|
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;
|
PL_Queue_Config* config = (PL_Queue_Config*)args;
|
|
in->reset();
|
in->config = *config;
|
in->pool.bp_init(config->maxBlockCount, config->maxBlockSize);
|
|
if (config->syncQueueFull)
|
{
|
int ret = pthread_mutex_lock(in->sync_full_mutex);
|
if(ret != 0)
|
{
|
printf("pthread_mutex_lock sync_full_mutex: %d\n", ret);
|
return false;
|
}
|
}
|
|
if (config->syncQueueEmpty)
|
{
|
int ret = pthread_mutex_lock(in->sync_empty_mutex);
|
if(ret != 0)
|
{
|
printf("pthread_mutex_lock sync_empty_mutex: %d\n", 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->que.size() >= in->config.maxBlockCount || in->pool.bp_empty())
|
{
|
// there is no available qb
|
|
if (in->config.queueFullDropBlock)
|
{
|
printf("PL_Queue::pay queueFullDropBlock\n");
|
|
pthread_mutex_lock(in->queue_pool_mutex);
|
qb = in->que.front();
|
in->que.pop();
|
|
in->pool.bp_free(qb);
|
qb = nullptr;
|
pthread_mutex_unlock(in->queue_pool_mutex);
|
}
|
else
|
{
|
if (in->config.syncQueueFull)
|
{
|
printf("PL_Queue::pay sync by sync_full_mutex\n");
|
pthread_mutex_lock(in->sync_full_mutex);
|
}
|
}
|
}
|
|
if (in->que.size() >= in->config.maxBlockCount || in->pool.bp_empty())
|
{
|
printf("PL_Queue::pay full\n");
|
return false;
|
}
|
|
pthread_mutex_lock(in->queue_pool_mutex);
|
qb = in->pool.bp_alloc();
|
if (qb != nullptr)
|
in->que.push(qb);
|
pthread_mutex_unlock(in->queue_pool_mutex);
|
|
if (qb == nullptr)
|
{
|
printf("PL_Queue::pay bp_alloc nullptr\n");
|
return false;
|
}
|
|
if (in->config.copyData)
|
{
|
memcpy(qb->data, pm.buffer, pm.buffSize);
|
qb->size = pm.buffSize;
|
}
|
|
if (in->config.syncQueueEmpty)
|
pthread_mutex_unlock(in->sync_empty_mutex);
|
|
return true;
|
}
|
|
void PL_Queue::pm_deleter_qb(PipeMaterial* pm)
|
{
|
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_pool_mutex);
|
in->pool.bp_free(qb);
|
pthread_mutex_unlock(in->queue_pool_mutex);
|
}
|
|
bool PL_Queue::gain(PipeMaterial& pm)
|
{
|
PL_Queue_Internal* in = (PL_Queue_Internal*)internal;
|
|
if (in->que.empty())
|
{
|
if (in->config.syncQueueEmpty)
|
pthread_mutex_lock(in->sync_empty_mutex);
|
}
|
|
if (in->que.empty())
|
{
|
printf("PL_Queue::gain empty\n");
|
pm.buffer = nullptr;
|
pm.buffSize = 0;
|
pm.former = this;
|
return false;
|
}
|
|
QBlock* qb = nullptr;
|
pthread_mutex_lock(in->queue_pool_mutex);
|
qb = in->que.front();
|
in->que.pop();
|
pthread_mutex_unlock(in->queue_pool_mutex);
|
|
if (in->config.syncQueueFull)
|
pthread_mutex_unlock(in->sync_full_mutex);
|
|
pm.type = PMT_BYTES;
|
pm.buffer = qb->data;
|
pm.buffSize = qb->size;
|
pm.former = this;
|
pm.deleter = pm_deleter_qb;
|
pm.args = qb;
|
return true;
|
}
|
