# Lock-free pooled queue

I am attempting to create a lock free pool of resources and I need the ability to access any one that is not already accessed and then return it back when I do not need it. This will happen very very often in many threads and I felt that a locked queue would create a lot of contention.

My solution is a relatively simple one (lots of atomics) but that simplicity and speed is pretty important to me.

I am not very familiar with atomics and I was wondering if I missed a potentially unsafe use. Although I am aware of the lack of protection against unsafe use of these functions (releasing an index that has not been acquired) but for my use case this will never happen. Just assume I am going to use the class correctly, I know its not good practice to write code that can easily break but the purpose is to be quick; therefor, I can not make useless checks.

That being said, I would like to know if there are any ways I can improve it: are there more optimal memory orders than sequential that I can use for some of them? Is the array of atomic ints safe? Can I get away with a simple volatile array? Is the do while loop necessary in the acquire method?

If you have any other thoughts please, let me know.

#pragma once
#include <atomic>

template <typename T>
class LFPoolQueue {
public:
const uint16_t END = 0xffff;

LFPoolQueue(uint16_t size)
{
m_size = size;
m_pool = new T[size];
m_ptrs = new std::atomic<uint16_t>[size];
uint16_t lindex = size - 1;
for (uint16_t i = 0; i < lindex; i++)
m_ptrs[i].store(i + 1);
m_ptrs[lindex].store(END);
m_next.store(0);
m_last.store(lindex);
}
~LFPoolQueue()
{
delete[] m_pool;
delete[] m_ptrs;
}

int dequeue()
{
bool wasMaxed = false;
struct VersionRef
{
uint16_t version;
uint16_t ref;
};
uint32_t newPtr = 0;
VersionRef vrOld = {};
VersionRef vrNew = {};
do
{
memcpy(&vrOld, &unsafePtr, sizeof(uint32_t));

vrNew.version = (vrOld.version + 1) & 0x7FFFU;

wasMaxed = vrOld.version & 0x8000U;
if (vrNew.ref == END)
{
if (wasMaxed)
return END;
else
{
vrNew.ref = vrOld.ref;
vrNew.version |= 0x8000U;
}
}

memcpy(&newPtr, &vrNew, sizeof(uint32_t));
} while (!m_next.compare_exchange_weak(unsafePtr, newPtr));

if (wasMaxed)
return dequeue();
else
return vrOld.ref;
}

void enqueue(uint16_t index)
{
m_ptrs[index].store(END);
m_ptrs[m_last.exchange(index)].store(index);
}

uint16_t size() { return m_size; }
T& operator[](const int index) { return m_pool[index]; }

private:
uint16_t m_size;
T* m_pool;
std::atomic<uint16_t>* m_ptrs;

std::atomic<uint32_t> m_next;
std::atomic<uint16_t> m_last;
};


main.cpp test app

#include <iostream>
#include "LFPoolQueue.h"

int main()
{
LFPoolQueue<int> q(3);
q[0] = 0;
q[1] = 1;
q[2] = 2;

for (int i = 0; i < 6; i++)
{
if (i == 3)
{
q.enqueue(1);
std::cout << "returned 1" << std::endl;
}
else
{
int index = q.dequeue();
std::cout << i << ": " << index << std::endl;
}
}
}


Output:

0: 0
1: 1
2: 2
returned 1
4: 1
5: 65535

• Have you tested it and is the code working? Could you provide a working example of how to use it? – pacmaninbw Jun 30 at 13:58
• Not really, that's why I posted this code review. I can not really run tests on it because its so fundamentally unpredictable. The system I will be using it in is so complicated that if this is the part that fails it will be very difficult to discover. Therefore I am hoping, with enough skilled eyes, someone can spot a deadly error. – Christopher Silvas Jul 1 at 0:39
• Have you done any testing of this code? Even minimal, by putting it into a small test app and exercising the basic functionality? – 1201ProgramAlarm Jul 1 at 0:48
• I'll try some single threaded basic logic functionality and see what happens. – Christopher Silvas Jul 1 at 0:59
• Look at the related section to the right of these comments, it might give you some ideas for testing. On Code Review we have guidelines that say we can only review working code codereview.stackexchange.com/help/dont-ask and codereview.stackexchange.com/help/how-to-ask. – pacmaninbw Jul 1 at 12:10