127 lines
2.5 KiB
C++
127 lines
2.5 KiB
C++
#include "shmqueue.h"
|
|
#include <string.h>
|
|
|
|
#ifdef _WIN32
|
|
#define sprintf sprintf_s
|
|
#endif
|
|
|
|
ShmQueue::ShmQueue(Shm *shm, ShmSlot slot, bool owner)
|
|
{
|
|
_shm = shm;
|
|
_owner = owner;
|
|
|
|
if (owner) {
|
|
ShmPlace queue_place = shm->alloc(sizeof(ShmQueueStart));
|
|
shm->setSlot(slot, queue_place);
|
|
ref(shm, queue_place, &_queue);
|
|
_queue->first = SHM_NULL;
|
|
_queue->last = SHM_NULL;
|
|
_queue->count = 0;
|
|
} else {
|
|
ShmPlace queue_start = shm->slot(slot);
|
|
ref(shm, queue_start, &_queue);
|
|
}
|
|
|
|
char buf[1024];
|
|
sprintf(buf, "sem_%s_%d", shm->name(), slot);
|
|
_queue_sem = shm->sem(buf, owner);
|
|
}
|
|
|
|
void ShmQueue::takeOwnership()
|
|
{
|
|
_queue_sem->takeOwnership();
|
|
}
|
|
|
|
ShmQueue::~ShmQueue()
|
|
{
|
|
if (_owner) {
|
|
cleanup();
|
|
}
|
|
}
|
|
|
|
void ShmQueue::cleanup()
|
|
{
|
|
// does currently nothing
|
|
delete _queue_sem;
|
|
}
|
|
|
|
int ShmQueue::depth()
|
|
{
|
|
_shm->lock();
|
|
int d = _queue->count;
|
|
_shm->unlock();
|
|
return d;
|
|
}
|
|
|
|
void ShmQueue::enqueue(int cmd, const std::string &json_data)
|
|
{
|
|
size_t jd_size = json_data.size();
|
|
int str_place = _shm->alloc(jd_size + 1);
|
|
|
|
char *s;
|
|
ref(_shm, str_place, &s);
|
|
memcpy(s, json_data.data(), jd_size);
|
|
s[jd_size] = '\0';
|
|
|
|
int item_place = _shm->alloc(sizeof(ShmQueueItem));
|
|
ShmQueueItem *item;
|
|
ref(_shm, item_place, &item);
|
|
item->cmd = cmd;
|
|
item->data_place = str_place;
|
|
item->prev = _queue->last;
|
|
item->next = SHM_NULL;
|
|
if (_queue->last != SHM_NULL) {
|
|
ShmQueueItem *last_i;
|
|
ref(_shm, _queue->last, &last_i);
|
|
last_i->next = item_place;
|
|
}
|
|
|
|
_shm->lock();
|
|
_queue->last = item_place;
|
|
if (_queue->first == SHM_NULL) {
|
|
_queue->first = _queue->last;
|
|
}
|
|
_queue->count += 1;
|
|
_shm->unlock();
|
|
_queue_sem->post();
|
|
}
|
|
|
|
void ShmQueue::enqueue(int cmd)
|
|
{
|
|
std::string s;
|
|
enqueue(cmd, s);
|
|
}
|
|
|
|
|
|
bool ShmQueue::dequeue(int &cmd, std::string &json_data, int wait_ms)
|
|
{
|
|
bool something_came = _queue_sem->wait(wait_ms);
|
|
if (!something_came) {
|
|
return false;
|
|
}
|
|
|
|
_shm->lock();
|
|
|
|
ShmPlace item_place = _queue->first;
|
|
ShmQueueItem *item;
|
|
ref(_shm, item_place, &item);
|
|
_queue->first = item->next;
|
|
if (_queue->first == SHM_NULL) {
|
|
_queue->last = SHM_NULL;
|
|
}
|
|
_queue->count -= 1;
|
|
|
|
_shm->unlock();
|
|
|
|
ShmPlace data_place = item->data_place;
|
|
char *data;
|
|
ref(_shm, data_place, &data);
|
|
json_data = data;
|
|
|
|
_shm->free(data_place);
|
|
cmd = item->cmd;
|
|
|
|
_shm->free(item_place);
|
|
return true;
|
|
}
|