Implementation of a lock-free fixed-sized allocator

Question

This question now has a follow-up:

Implementation of a lock-free fixed-sized allocator - follow-up - with commented code

---

I've tried implementing a lock-free fixed-size allocator while trying to learn synchronization through atomic variables.

Here are the related classes:

template_utility.h

#ifndef OAG_TEMPLATE_UTLITY_H
#define OAG_TEMPLATE_UTLITY_H
namespace oag
{
    template <typename C>
    using Pointer_type = typename C::pointer;

    template <typename C>
    using Size_type = typename C::size_type;
}
#endif // !OAG_TEMPLATE_UTLITY_H

Implementation of the lock-free fixed-size allocator with default memory ordering for all atomic operations.

lock_free_memory_chunk.h

#ifndef OAG_LOCK_FREE_MEMORY_CHUNK_H
#define OAG_LOCK_FREE_MEMORY_CHUNK_H

#include <atomic>
#include "template_utility.h"

namespace oag
{
    template <typename T, typename SizeT = unsigned char>
    class lock_free_memory_chunk
    {
    public:
        using value_type = T;
        using pointer = value_type*;
        using size_type = SizeT;

    private:
        using byte = unsigned char;

    public:
        explicit lock_free_memory_chunk( size_type const );

        pointer allocate() noexcept;
        void deallocate( pointer ) noexcept;

    private:
        bool dec_avail_blocks();

    private:
        static auto constexpr block_sz = sizeof( value_type ) < sizeof( size_type ) ?
            sizeof( size_type ) : sizeof( value_type );

    private:
        byte* p_bytes_;
        std::atomic<size_type> next_alloc_idx_;
        std::atomic<size_type> num_avail_blocks_;
    };
}

namespace oag
{
    template <typename T, typename SizeT>
    lock_free_memory_chunk<T, SizeT>::lock_free_memory_chunk( size_type const capacity ) :
        p_bytes_{ new byte[ sizeof( value_type ) * capacity ] },
        next_alloc_idx_{ 0 },
        num_avail_blocks_{ capacity }
    {
        static_assert( sizeof( byte ) == 1, "sizeof(unsigned char) != 1" );

        size_type i{ 0 };
        for ( byte* p{ p_bytes_ }; i < capacity; p += block_sz )
        {
            *reinterpret_cast<size_type*>( p ) = ++i;
        }
    }

    template<typename T, typename SizeT>
    inline oag::lock_free_memory_chunk<T, SizeT>::~lock_free_memory_chunk()
    {
        delete[] p_bytes_;
    }

    template <typename T, typename SizeT>
    inline oag::Pointer_type<lock_free_memory_chunk<T, SizeT>>
    lock_free_memory_chunk<T, SizeT>::allocate() noexcept
    {
        if ( !dec_avail_blocks() )                                                  // 1A
            return nullptr;

        size_type alloc_idx{ next_alloc_idx_.load() };                              // 1B
        while ( !next_alloc_idx_.compare_exchange_weak(                             // 1C
            alloc_idx,
            *reinterpret_cast<size_type*>( p_bytes_ + alloc_idx * block_sz ) ) )
        {
        }

        return reinterpret_cast<pointer>( p_bytes_ + alloc_idx * block_sz );
    }

    template <typename T, typename SizeT>
    inline void
    lock_free_memory_chunk<T, SizeT>::deallocate( pointer p ) noexcept
    {
        auto next_alloc_from_p{ next_alloc_idx_.load() };                           // 2A
        auto new_next_alloc_idx                                                     // 2B
        {
            static_cast<size_type>(
            ( reinterpret_cast<byte*>( p ) - p_bytes_ ) / block_sz )
        };
        while ( !next_alloc_idx_.compare_exchange_weak(                             // 2C
            next_alloc_from_p,
            new_next_alloc_idx ) )
        {
        }
        *reinterpret_cast<size_type*>( p ) = next_alloc_from_p;                     // 2D

        num_avail_blocks_.fetch_add( 1 );
    }

    template<typename T, typename SizeT>
    inline bool
    oag::lock_free_memory_chunk<T, SizeT>::dec_avail_blocks()
    {
        auto curr_num_avail_blocks{ num_avail_blocks_.load() };                     // 3A
        auto dec_num_avail_blocks                                                   // 3B
        {
            curr_num_avail_blocks > 0 ? curr_num_avail_blocks - 1 : 0
        };
        while ( !num_avail_blocks_.compare_exchange_strong(                         // 3C
            curr_num_avail_blocks,
            dec_num_avail_blocks ) )
        {
            dec_num_avail_blocks = curr_num_avail_blocks > 0 ?                      // 3D
                curr_num_avail_blocks - 1 : 0;
        }

        return curr_num_avail_blocks > 0 ? true : false;
    }
}
#endif // !OAG_LOCK_FREE_MEMORY_CHUNK_H

General description

Memory is sizeof(value_type) or sizeof(size_type) multiplied by the desired capacity.

Every memory block stores an offset to the next block at its start.

Location    [0           ]|[1           ]|...|[N - 1       ]
Contents    [1, obj_bytes]|[2, obj_bytes]|...|[N, obj_bytes]

The offset is lost at allocation as the object takes its required space starting from the start of block.

Function description

allocate()

1A. Make sure that there are available blocks and decrease the number of available blocks by 1 before proceeding; true if blocks are available.

1B. Load the next allocation offset.

1C. Make sure that the allocation offset is unique for every thread in allocate().

deallocate(pointer)

2A. Load the offset of the next allocation.

2B. Calculate the offset from p_bytes_ to parameter p.

2C. Make sure that the value of the offset from p is unique.

2D. Set the offset of p to the unique value.

dec_avail_blocks()

3A. Load the current number of available blocks.

3B. Prevent underflow when decrementing the number of available blocks.

3C. Make sure that if the number of available blocks changes, that the new count is updated.

3D. Return whether there are available blocks or not to allocate().

Sample tests to make sure that no address is given out multiple times:

#include <unordered_set>
#include <future>
#include "lock_free_memory_chunk.h"

auto num_threads = std::thread::hardware_concurrency();
std::size_t const num_blocks_per_thread = 50000;
oag::lock_free_memory_chunk<int, std::size_t> mc{ num_blocks_per_thread * num_threads };

using mc_pointer_set = std::unordered_set<oag::lock_free_memory_chunk<int, std::size_t>::pointer>;
mc_pointer_set call_alloc( std::size_t n )
{
    mc_pointer_set mcps;
    for ( size_t i = 0; i < n; i++ )
        mcps.insert( mc.allocate() );

    for ( auto* p : mcps )
        mc.deallocate( p );

    for ( size_t i = 0; i < n; i++ )
        mcps.insert( mc.allocate() );

    return mcps;
}

void check_sets( mc_pointer_set const& s1, mc_pointer_set const& s2 )
{
    for ( auto* p : s1 )
    {
        if ( s2.find( p ) != std::cend( s2 ) )
        {
            throw std::runtime_error( "two sets contain the same address" );
        }
    }
}
int main()
{
    std::vector<std::future<mc_pointer_set>> v;
    for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
    {
        v.emplace_back( std::async(
            std::launch::async,
            call_alloc,
            num_blocks_per_thread ) );
    }
    for ( auto& t : v )
    {
        t.wait();
    }
    std::vector<mc_pointer_set> comparisons;
    for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
    {
        comparisons.emplace_back( std::move( v[ i ].get() ) );
    }
    for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
    {
        for ( decltype( num_threads ) j{ 1 }; j < num_threads; ++j )
        {
            std::async(
                std::launch::async,
                check_sets,
                std::ref( comparisons[ i ] ),
                std::ref( comparisons[ j ] ) );
        }
    }
}

Question

Are there any thread safety issues that can slip through the cracks? Am I overusing the synchronization constructs in any way (are any of them unnecessary)?

Assumptions made

§1 - Multiple threads in allocate()

§1.1

Any thread entering allocate() must get past the if(!dec_avail_blocks()) statement; there is no possibility for two threads to get past that block due to the indivisibility of the dec_avail_blocks() operation. Thus, threads only get past if the value of num_avail_blocks_ is greater than 0 before the decrement.

§1.2

Multiple threads can load() the same value from next_alloc_idx_ into alloc_idx, but the RMW operation in the while(...) loop assures that no two threads get past it while having the same alloc_idx value (§1.3).

§1.3

For any number of threads simultaneously looping in the while(...) statement, the indivisibility of an atomic CAS operation assures that if it succeeds, any other threads will fail on the comparison between next_alloc_idx_ and alloc_idx. In short, the threads will spin until they get a unique value (from the perspective of all involved threads) into alloc_idx.

§2 - Multiple threads in deallocate(pointer)

§2.1

It is not possible for any threads to share the same value of argument p, since allocate() assures that unique addresses are returned. However, it is possible for many threads to enter deallocate() and read the same value of next_alloc_idx_.

§2.2

The RMW operation in the while(...) loop assures that no two threads store the same value of next_alloc_idx_ to different addresses. This is key to preventing allocate() from returning the same address, since no two addresses can store the same offset. Once this value is determined to be unique (the loop ends), it is assigned as the offset of p, and the number of available blocks is incremented.

§3 - Multiple threads in dec_avail_blocks()

§3.1

Atomicity is ensured for the whole of this operation through the while(...) statement containing a RMW operation. The whole point of this is to properly synchronize a load from num_avail_blocks_ with a store as well, without decreasing past 0 to prevent underflow. The function then returns whether the current number of blocks is greater than 0.

Answer 1

ABA problem

It looks like your allocator is susceptible to the ABA problem. The allocate() function's compare and exchange is unsafe because the "next" value you are exchanging could be invalid by the time that the exchange happens.

ABA Example

Suppose your free list looks like this:

0 -> 1 -> 2 -> 3

1. Thread A enters allocate() and prepares to compare and exchange 0 with 1. But Thread A pauses for an extended period of time.

2. Thread B calls allocate() twice. The free list now looks like this:

   2 -> 3
   

3. Thread B calls deallocate() on item 0. The free list now looks like this:

   0 -> 2 -> 3
   

4. Thread A resumes. It exchanges 0 with 1. Notice item 1 is in use. The free list is now corrupted.

   1 -> ?
   

Deallocate problem

Also I noticed in the deallocate() function that you set the pointers in a bad order:

    while ( !next_alloc_idx_.compare_exchange_weak(                             // 2C
        next_alloc_from_p,
        new_next_alloc_idx ) )
    {
    }
    *reinterpret_cast<size_type*>( p ) = next_alloc_from_p;                     // 2D

If you succeed with the compare and exchange, but then pause before you set *p to next_alloc_from_p, then your free list is in a bad state. If another thread tries to allocate while this thread is paused before line 2D, it will load an unknown next pointer from *p. You need to assign the next pointer before exchanging:

    do {
        *reinterpret_cast<size_type*>( p ) = next_alloc_from_p;
    } while ( !next_alloc_idx_.compare_exchange_weak(
            next_alloc_from_p,
            new_next_alloc_idx ) );

Answer 2

Your code is of the type that has no obvious bugs, so I just started at the beginning and tried to understand the algorithm.

allocate()

1A. Make sure that there are available blocks and decrease the number of available blocks by 1 before proceeding; true if blocks are available.

template<typename T, typename SizeT>
inline bool
oag::lock_free_memory_chunk<T, SizeT>::dec_avail_blocks()
{
    auto curr_num_avail_blocks{ num_avail_blocks_.load() };                     // 3A
    auto dec_num_avail_blocks                                                   // 3B
    {
        curr_num_avail_blocks > 0 ? curr_num_avail_blocks - 1 : 0
    };
    while ( !num_avail_blocks_.compare_exchange_strong(                         // 3C
        curr_num_avail_blocks,
        dec_num_avail_blocks ) )
    {
        dec_num_avail_blocks = curr_num_avail_blocks > 0 ?                      // 3D
            curr_num_avail_blocks - 1 : 0;
    }

    return curr_num_avail_blocks > 0 ? true : false;
}

The word dec would be clearer if you spelled it try_decrement, and likewise avail should be available. The use of auto x{y}; syntax is much much harder to read than a simple = would have been. In the case that num_avail_blocks_ is 0, no CAS is necessary. Instead of repeating the assignment to dec_num_avail_blocks twice, refactor your loop. The expression x ? true : false is a verbose synonym for x.

Put it all together:

template<typename T, typename SizeT>
inline bool oag::lock_free_memory_chunk<T, SizeT>::try_decrement_available_blocks()
{
    auto n = num_avail_blocks_.load();
    do {
        if (n == 0) {
            return false;
        }
    } while (!num_avail_blocks_.compare_exchange_strong(n, n-1));
    return true;
}

1B. Load the next allocation offset.

1C. Make sure that the allocation offset is unique for every thread in allocate().

template <typename T, typename SizeT>
inline oag::Pointer_type<lock_free_memory_chunk<T, SizeT>>
lock_free_memory_chunk<T, SizeT>::allocate() noexcept
{
    if ( !dec_avail_blocks() )                                                  // 1A
        return nullptr;

    size_type alloc_idx{ next_alloc_idx_.load() };                              // 1B
    while ( !next_alloc_idx_.compare_exchange_weak(                             // 1C
        alloc_idx,
        *reinterpret_cast<size_type*>( p_bytes_ + alloc_idx * block_sz ) ) )

Stop! I'm pretty sure this is one of your bugs. You're saying, "Carefully load a value from next_alloc_idx_ into a register named alloc_idx. Then, without any kind of synchronization, compute p_bytes_ + alloc_idx*block_sz and fetch the size_type stored at that memory address." If two different threads do this at once... well, it's actually okay so far because both of them are reading, not writing. But one of them will succeed. That thread (call it Thread A) will update the value stored in next_alloc_idx_, and then return p_bytes_ + alloc_idx*block_sz reinterpret_casted to a pointer. Return it to the caller, who can do absolutely anything with that pointer, including write through it.

So now we have a data race: in Thread A, we've returned the pointer to our caller and our caller is writing to that memory, while simultaneously in Thread B, we're attempting to read from p_bytes_ + alloc_idx*block_sz without synchronization. Undefined behavior and boom.

---

When writing this kind of low-level concurrency code, never ever ever pack multiple memory operations into a single C++ expression, as in

*q = *p;

It's just asking for this kind of bug. Whenever you touch memory (via the * operator, mainly), that should be on a line by itself, with no other memory references on the same line.

The following is a cleaned-up version of your allocate() function — just changing the style to make it obvious where the bug is.

template <typename T, typename SizeT>
inline auto lock_free_memory_chunk<T, SizeT>::allocate() noexcept
    -> typename oag::lock_free_memory_chunk<T, SizeT>::pointer_type
{
    if (!try_decrement_available_blocks()) {
        return nullptr;
    }

    auto idx = next_alloc_idx_.load();
    void *p_block;
    do {
        p_block = &p_bytes_[idx * block_sz];
        size_type next_idx = *(size_type *)p_block;
    } while ( !next_alloc_idx_.compare_exchange_weak(idx, next_idx) );

    return p_block;
}

This makes it much more obvious that the load of next_idx (inside the loop) is completely unsynchronized, and therefore unsafe.

I'm sure there are more (non-obvious) bugs remaining to be found, but this one is sufficiently fatal that I don't mind stopping here.