Simple concurrent freelist

Question

To become better acquainted with the Intel Parallel Studio XE profilers I'm currently messing around with a simple concurrent freelist. I'm fairly certain that it's thread-safe. It uses Acq/Rel semantics, Tagged Pointers to avoid the ABA problem and users of the free list can't actually touch the freelist header data. The worker functions have their own local state, no statics etc. However the Intel tools are insisting I have a data race and I can't decide if the tools are correct or the tools are presenting a false positive.

The tools report the data race on the first opening bracket of the main and/or runner function (which makes me think its a false positive). I can provide screen dumps of the data race tool analysis.

Freelist Code:

#pragma once

#include <atomic>
#include <array>
#include <memory>

/**< Implements a non data contiguous thread safe pool. */
template < typename T, size_t size = 100 >
class MemoryPool
{
public:
/**< Creates a memory pool. */
MemoryPool() _NOEXCEPT
{
    RebuildStack( );
}

/**< Used for global cleanup, only execute if we know there are no contending threads or items still allocated. */
void Reset() _NOEXCEPT
{
    RebuildStack();
}

T* try_pop() _NOEXCEPT
{
    // Read current head
    PoolHead currHead = head.load(std::memory_order_acquire);
    PoolHead newHead;

    // Keep on trying to pop an element off the stack
    while ( currHead.list )
    {
        // Construct the new head by grabbing next pointer in list and incrementing the atomic tag
        newHead.list = currHead.list->next.load(std::memory_order_relaxed);
        newHead.tag = currHead.tag + 1;

        // Try to replace the existing head
        if (head.compare_exchange_weak(currHead, newHead, std::memory_order_release, std::memory_order_acquire))
        {
            // Succesfully popped an item, return pointer to data
            return &currHead.list->data;
        }
    }

    // Failed to get any items
    return nullptr;
}

/**< Tries to push an item back onto the memory pool. */
bool try_push(T* p) _NOEXCEPT
{
    // Check that the given pointer points inside the pool and lies along an item
    auto ptr_delta = reinterpret_cast<PoolNode*>(p)-reinterpret_cast<PoolNode*>(&pool[0].data);
    if ( ptr_delta < 0 || ptr_delta >= size || &pool[ptr_delta].data != p )
        return false;

    // Read current head 
    PoolHead currhead = head.load(std::memory_order_relaxed);
    PoolHead newHead(&pool[ptr_delta]);

    // Try to replace the current head until we succeed
    do {
        // Increment atomic tag of head
        newHead.tag = currhead.tag + 1;

        // Make the head of the current list the new tail
        pool[ptr_delta].next.store(currhead.list, std::memory_order_relaxed);
    } while (!head.compare_exchange_weak(currhead, newHead, std::memory_order_release, std::memory_order_relaxed));

    // Added item back to pool
    return true;
}

protected:

struct PoolNode
{
    std::atomic<PoolNode*> next;

    T data;

    PoolNode() _NOEXCEPT : next(nullptr) {}
};

struct PoolHead
{
    PoolNode* list;

    size_t tag;

    explicit PoolHead(PoolNode* v = nullptr) _NOEXCEPT : list(v), tag(0) {}
};

std::atomic<PoolHead> head;

std::array<PoolNode, size> pool;

void RebuildStack() _NOEXCEPT
{
    // Point the head to the first item in the pool
    head.store(PoolHead(&pool[0]), std::memory_order_relaxed);

    // Clear the last node in the pool
    new (&pool[size-1]) PoolNode();

    // Rebuild the body of the stack
    for (size_t i = 0; i < size - 1; i++)
        pool[i].next.store(&pool[i + 1], std::memory_order_relaxed);

    // Issue memory barrier so we can immediately start work
    std::atomic_thread_fence(std::memory_order_release);
    }
};

Test Code:

// Freelist.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"
#include "MemoryPool.h"

#include <thread>
#include <vector>
#include <random>

template <typename T, size_t size, size_t n>
void runner (MemoryPool<T, size> *pool)
{
std::vector<T*> vec;
std::default_random_engine generator;
std::uniform_int_distribution<T> dist(0, 6);

for (size_t i = 0; i < n; i++)
{
    switch (dist(generator))
    {
    case 0:
    case 1:
    case 2:
    case 3:
        if (auto p = pool->try_pop())
            vec.push_back(p);
        break;
    case 4:
        if (!vec.empty())
            *vec[generator() % vec.size()] = rand();
        break;
    case 5:
        if (!vec.empty())
        {
            assert(pool->try_push(vec.back()));
            vec.pop_back();
        }
        break;
    }
}
};

int _tmain(int argc, _TCHAR* argv[])
{
static const size_t n = 200000;
static const size_t t = 8;
static const size_t size = 10000;

MemoryPool<int, size> pool;

std::array<std::thread, t> threads;
for (int i = 0; i < t; i++)
{
    threads[i] = std::thread(&runner<int, size, n>, &pool);
}

for (int i = 0; i < t; i++)
{
    threads[i].join();
}

return 0;
}

Answer 1

I won't be of great help for concurrency-related issues but there are some other small things that could be improved in your code:

• std::array has the method data which is the preferred way to acces its underlying memory instead of &arr[0].

• Instead of index-based for loops, you could use some nicer range-based for loops in order to make your code pretty:

  for (auto& thrd: threads)
  {
      thrd = std::thread(&runner<int, size, n>, &pool);
  }
  
  for (auto& thrd: threads)
  {
      thrd.join();
  }
  

• static const is a bit old-school. static constexpr is the new static const:

  static constexpr size_t n = 200000;
  static constexpr size_t t = 8;
  static constexpr size_t size = 10000;
  

  Note that the names n and t don't really help when reading the code. You could find something more meaningful.

• What's _NOEXCEPT? A macro corresponding to noexcept on supported architectures? Is it defined by the implementation? If not it uses a reserved name since it begins by an underscore followed by a capital letter.

• std::rand should be fully qualified with the std:: namespace. It's not mandatory, but it's good practice.

• By the way, why use the thread-unsafe std::rand in a threaded program when you obviously already know about the new <random> module?

• generator and dist don't look thread-safe either. Also, you don't need to create a new one per call to the function. Make them thread_local to solve both these problems at once:

  thread_local std::default_random_engine generator;
  thread_local std::uniform_int_distribution<T> dist(0, 6);
  

• Also, you might want to seed your random engine before using it. The obvious way to do so would be initialize it with an std::random_device:

  thread_local std::random_device rd;
  thread_local std::default_random_engine generator(rd());
  thread_local std::uniform_int_distribution<T> dist(0, 6);
  

  If std::random_device does not work properly with your compiler (I'm looking at you MinGW), you can still resort to the old std::time(nullptr) but I fear that it might seed each thread's random engine with the same time.

Answer 2

In RebuildStack your placement new is to a location a location that already has a initialized node, you should first call the destructor on it or assign a default constructed node.