I've been working on a lockless c++11 fifo buffer. And I've almost got it. However one small detail has gotten the better of me. The buffer has a head pointed to by:
std::shared_ptr<node<T>> m_head;
Of the type:
struct node
{
node(const T data)
:
data(new T(data)),
next(nullptr)
{}
std::shared_ptr<T> data;
std::shared_ptr<node<T>> next;
};
And then there are the produce:
void produce(const T &&data)
{
//bool indicating whether a notification should be sent after adding
bool l_notifyUponAdding;
//the new node to be added at the end of the array
std::shared_ptr<node<T>> l_newNode(new node<T>(std::forward<const T&&>(data)));
//pointer to the last node
std::shared_ptr<node<T>> l_lastNode(std::atomic_load(&m_head));
//value to compare the next of the last node with
std::shared_ptr<node<T>> l_expectedNullPointer;
//notify if this isn't the only node
l_notifyUponAdding = !l_lastNode;
if (!l_lastNode)//if there are no nodes, add this as the only node
if (std::atomic_compare_exchange_strong(&m_head, &l_expectedNullPointer, l_newNode))
return;
do
{
l_expectedNullPointer.reset();
while (l_lastNode->next)
{
l_lastNode = std::atomic_load(&l_lastNode)->next;
}
} while (!std::atomic_compare_exchange_weak(&l_lastNode->next, &l_expectedNullPointer, l_newNode));
//adding failed since another thread already did this.
l_lastNode = l_expectedNullPointer;
if (l_notifyUponAdding)
m_newDataWaiter.notify_one();
}
And consume:
std::shared_ptr<T> consume(bool blockingCall = false)
{
//Check if the head is null if it is:
if (!std::atomic_load(&m_head))
{
if (blockingCall)//And this is a blocking call,
{
do
{
m_newDataWaiter.wait(m_newDataWaiterLock, [this]{return std::atomic_load(&(this->m_head)) == nullptr; });//we block until
} while (!std::atomic_load(&m_head));// the load yields a head that is not null(to avoid unnecessary calls on spurious wake ups)
}
else//And this is not a blocking call we
{
return nullptr;
}
}
//If we've found a valid head we will now try to make the node pointed to by head the new head.
std::shared_ptr<node<T>> l_poppee = atomic_load(&m_head);
std::shared_ptr<node<T>> l_newHead = atomic_load(&m_head);
//note that l_poppee gets updated if the compare exchange fails
while (l_poppee && !std::atomic_compare_exchange_weak(&m_head, &l_poppee, l_poppee->next))
{
}
if (l_poppee)
return l_poppee->data;
else
return std::shared_ptr<T>();
}
Functions.
All seems to work well. However I reckon there is one flaw. If all nodes are consumed whilst executing a produce. The data will be added to the last element. Even though the element already has been deleted.
To be more precise, if this line has been executed:
if (std::atomic_compare_exchange_strong(&m_head, &l_expectedNullPointer, l_newNode))
And the loaded node wasn't zero. The next element of the last node will be changed. Regardless of whether the nodes are being deleted in the meantime or not. The nodes will not be physically deleted as long as the produce function is being excuted, because of the shared pointers.
However, the main pointer will be set to NULL. And therefore the new node will be deleted as soon as the produce function is exited.
Would anybody happen to know a solution for this problem:)?