The following implementation of a lock-free fixed-size allocator is a follow-up of:
Implementation of a lock-free fixed-sized allocator
§1 - Introduction
The purpose is self-learning of atomic types and synchronization. I would appreciate a review that does any of the following (from most important to least):
- Verification of assumption 2 in §3.1.
- Finds issues.
- Touches upon the efficiency of the code.
The previous version had many issues that were kindly pointed out by both answers.
§2 - Allocator implementation
This implementation now contains alignment for the data and alignment to make sure that highly contested variables aren't on the same cache line which provided a substantial performance improvement.
§2.1 - Function description
§2.1.1 - Verification function: try_decrease_available_blocks()
Checks that if there are any available blocks for allocate(). It's purpose is to atomically compare num_avail_blocks_ with 0 and decrease num_avail_blocks_ by 1 of it is not equal to 0.
This is required, since fetch/sub operations would cause it to underflow on their own.
§2.1.2 - Allocation function: allocate()
Checks that there are available blocks by calling try_decrease_available_blocks(). This condition assures that only capacity (maximum capacity specified in the constructor) calls actually go through to the next part.
Inside a CAS loop, the next_alloc_offset_ tag is increased by one to prevent the ABA issue. The pointer to be returned p_block is then obtained by the offset that was in next_alloc_offset_'s value data member.
The new allocation offset is then set to the value at the first sizeof(size_type) bytes of p_block. The CAS operation then takes place.
§2.1.3 - Deallocation function: deallocate(pointer)
Since allocate() returns unique pointers to every caller, then every pointer argument p received by deallocate() is unique. We calculate the deallocated pointer's offset starting from the initial memory location at p_memory_ as that will be the value of next_allocation_offset_.value.
In a CAS loop, the new allocation offset's tag is decreased to prevent the ABA problem and p's next offset is set to the current next_allocation_offset_.value value before next_allocation_offset_ is updated.
The number of available blocks is then incremented and allocate() can now be called one more time.
§2.2 - Implementation
#ifndef OAG_LOCK_FREE_MEMORY_CHUNK_H
#define OAG_LOCK_FREE_MEMORY_CHUNK_H
#pragma warning( disable : 4324 ) // ignore alignof(n) warning
#include <type_traits>
#include <atomic>
namespace oag
{
template <typename T, std::size_t cache_line_size = 128>
class lock_free_memory_chunk
{
private:
template <typename Tag, typename Value, std::size_t lock_free_max_size>
struct tagged_value // change
{
static_assert( sizeof( Tag ) + sizeof( Value ) <= lock_free_max_size,
"sizeof(tagged_value) > lock_free_max_size : cannot be lock-free" );
Tag tag;
Value value;
};
public:
using value_type = T;
using pointer = value_type*;
using size_type = std::size_t;
private:
// determines the type of memory allocated to preserve alignment of
// the biggest type between `value_type` and `size_type`;
// we only need the biggest of either to store its associated `next` offset
// at the first sizeof(size_type) bytes of each block.
using block_t = std::conditional_t<
sizeof( value_type ) < sizeof( size_type ), size_type, value_type>;
using byte = unsigned char;
using tagged_value_t = tagged_value<size_type, size_type, 8>;
public:
explicit lock_free_memory_chunk( size_type const );
~lock_free_memory_chunk();
// TODO: deal with copy & move operations
pointer allocate() noexcept;
void deallocate( pointer ) noexcept;
private:
bool try_decrease_available_blocks() noexcept;
private:
block_t* p_memory_;
alignas( cache_line_size ) std::atomic<size_type> num_avail_blocks_;
alignas( cache_line_size ) std::atomic<tagged_value_t> next_alloc_offset_;
};
}
namespace oag
{
template <typename T, std::size_t cache_line_size>
lock_free_memory_chunk<T, cache_line_size>::lock_free_memory_chunk( size_type const capacity ) :
p_memory_{ new block_t[ capacity ] },
num_avail_blocks_{ capacity },
next_alloc_offset_{ { 0, 0 } }
{
size_type i{ 0 };
for ( block_t* p{ p_memory_ }; i < capacity; ++p )
{
// assign offset to next block at first sizeof(size_type) bytes
// i.e.
// offset: [ 1 ][ 2 ][ 3 ]...[ N + 1 ]
// location: [ 0 ][ 1 ][ 2 ]...[ N ]
*reinterpret_cast<size_type*>( p ) = ++i;
}
}
template <typename T, std::size_t cache_line_size>
inline
lock_free_memory_chunk<T, cache_line_size>::~lock_free_memory_chunk()
{
delete[] p_memory_;
}
template <typename T, std::size_t cache_line_size>
inline typename lock_free_memory_chunk<T, cache_line_size>::pointer
lock_free_memory_chunk<T, cache_line_size>::allocate() noexcept
{
// can we decrement the number of available blocks down to 0 or greater
// IFF it is not already 0; in other words, assure we have available blocks
if ( !try_decrease_available_blocks() )
{
// no free blocks: cannot allocate -> return nullptr
return nullptr;
}
tagged_value_t alloc_offset = next_alloc_offset_.load();
pointer p_block{ nullptr };
tagged_value_t new_next_alloc_offset;
do
{
// increase the id tag to prevent the ABA problem
new_next_alloc_offset.tag = alloc_offset.tag + 1;
// cast to pointer to `T`
p_block = reinterpret_cast<pointer>( p_memory_ + alloc_offset.value );
// set the next alloc offset to the `size_type` value contained
// in the first `sizeof(size_type)` bytes of this block
new_next_alloc_offset.value = *reinterpret_cast<size_type*>( p_block );
}
while ( !next_alloc_offset_.compare_exchange_strong(
alloc_offset, new_next_alloc_offset ) );
return p_block;
}
template <typename T, std::size_t cache_line_size>
inline void
lock_free_memory_chunk<T, cache_line_size>::deallocate( pointer p ) noexcept
{
tagged_value_t alloc_offset_from_p = next_alloc_offset_.load();
tagged_value_t new_next_alloc_offset;
// calculate the next allocation offset based on argument `p`
// this will be a value between 0 and N, where N is total capacity
new_next_alloc_offset.value = static_cast<size_type>(
reinterpret_cast<block_t*>( p ) - p_memory_ );
do
{
// decrease the id tag to prevent the ABA problem
new_next_alloc_offset.tag = alloc_offset_from_p.tag - 1;
// set the next alloc offset of block `p` to the current alloc offset
*reinterpret_cast<size_type*>( p ) = alloc_offset_from_p.value;
}
while ( !next_alloc_offset_.compare_exchange_strong(
alloc_offset_from_p, new_next_alloc_offset ) );
num_avail_blocks_.fetch_add( 1 );
}
template <typename T, std::size_t cache_line_size>
inline bool
lock_free_memory_chunk<T, cache_line_size>::try_decrease_available_blocks() noexcept
{
auto n = num_avail_blocks_.load();
do
{
// if the number of available blocks is 0
if ( n == 0 )
{
// true: there are no available blocks -> return false
return false;
}
} // false: decrease the number by 1
while ( !num_avail_blocks_.compare_exchange_strong( n, n - 1 ) );
// false: there is at least one available block -> return true
return true;
}
}
#endif // !OAG_LOCK_FREE_MEMORY_CHUNK_H
§3 - Solving the ABA problem
The purpose of the tagged_value<Tag, Value> class is to prevent the ABA problem by associating a tag with a value (and vice-versa). It is a nested class inside the actual allocator (shown above).
It is to be used as the template argument of std::atomic<T>. It will be lock free as long as the combined size of its template arguments is <= N; where N is the maximum supported size for std::atomic<tagged_value<Tag, Value, N>> implementations to be lock free.
template <typename Tag, typename Value, std::size_t required_size_for_lock_free>
struct tagged_value // change
{
static_assert( sizeof( Tag ) + sizeof( Value ) <= required_size_for_lock_free,
"sizeof(tagged_value) > required_size_for_lock_free : cannot be lock-free" );
Tag tag;
Value value;
};
§3.1 - Assumptions
For the allocator, the tag and the value are both
size_type. The tag is increased/decreased atomically when there is an allocation/deallocation. Since the allocator's size is fixed and that the tag is of the same type, there is no possibility for the ABA problem to occur due to underflow/overflow of the tag.The only possibility for the
tagged_valueto be equal to anothertagged_valueis if forNallocations, there areNdeallocations done in the reverse order (of addresses).Ndeallocations done in a non-reversed order will simply cause thetagged_valueto differ on thetagand/orvaluemember, and the CAS will fail as required. In the case that atagged_valueis equal to another, we can treat it as if no allocations/deallocations were actually done while the original thread was paused and the CAS operation thus succeeds without issue.
§4 - Sample tests
The following test calculates a grand average over the number of executions given to the run_test(memory_chunk, n)'s n parameter. The memory chunk can have a cache line size specified as its second argument in order to make it customizable for different processors. It is set to 128 by default.
It also verifies that the allocate() function does not give out the same address twice by comparing every address it gives out to every thread together.
#include <unordered_set>
#include <future>
#include <chrono>
#include <iostream>
#include "lock_free_memory_chunk.h"
template <typename MemoryChunk>
using mc_pointer_set_t = std::unordered_set<typename MemoryChunk::pointer>;
auto num_threads = std::thread::hardware_concurrency() - 1;
std::size_t const num_blocks_per_thread = 10000;
template <typename MemoryChunk>
decltype( auto ) call_alloc_and_dealloc( MemoryChunk& mc, std::size_t n, long long& execution_time_ms )
{
mc_pointer_set_t<MemoryChunk> mcps;
mcps.reserve( n );
auto time_begin = std::chrono::high_resolution_clock::now();
for ( size_t i = 0; i < n; ++i )
mcps.insert( mc.allocate() );
for ( auto* p : mcps )
mc.deallocate( p );
auto time_end = std::chrono::high_resolution_clock::now();
mcps.clear();
execution_time_ms =
std::chrono::duration_cast<std::chrono::milliseconds>( time_end - time_begin ).count();
return mcps;
}
template <typename MemoryChunk>
void check_sets( mc_pointer_set_t<MemoryChunk> const& s1, mc_pointer_set_t<MemoryChunk> const& s2 )
{
auto end = std::cend( s2 );
for ( auto* p : s1 )
if ( s2.find( p ) != end )
throw std::exception( "two sets have the same address" );
}
template <typename MemoryChunk>
void run_test( MemoryChunk& mc, unsigned int K )
{
std::vector<long long> execution_times( num_threads );
long long avg = 0;
for ( decltype( num_threads ) k{ 0 }; k < K; ++k )
{
std::vector<std::future<mc_pointer_set_t<MemoryChunk>>> v;
for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
{
v.emplace_back( std::async(
std::launch::async,
call_alloc_and_dealloc<MemoryChunk>,
std::ref( mc ),
num_blocks_per_thread,
std::ref( execution_times[ i ] ) ) );
}
for ( auto& t : v )
{
t.wait();
}
for ( decltype( num_threads ) i = 0; i < num_threads; ++i )
{
//std::cout << "Execution time for thread " << i << ": " << execution_times[ i ] << " ms\n";
avg += execution_times[ i ];
}
std::vector<mc_pointer_set_t<MemoryChunk>> comparisons;
for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
{
comparisons.emplace_back( std::move( v[ i ].get() ) );
}
std::vector<std::future<void>> exceptions;
for ( decltype( num_threads ) i{ 0 }; i < num_threads; ++i )
{
for ( decltype( num_threads ) j{ i + 1 }; j < num_threads; ++j )
{
exceptions.emplace_back( std::async(
std::launch::async,
check_sets<MemoryChunk>,
std::ref( comparisons[ i ] ),
std::ref( comparisons[ j ] ) ) );
}
}
for ( auto& f : exceptions )
{
f.get();
}
}
std::cout << "Average execution time: " << avg / execution_times.size() << '\n';
}
int main()
{
oag::lock_free_memory_chunk<int, 128> mc128{ num_blocks_per_thread * num_threads };
std::cout << "BEGIN: mc128 tests\n";
for ( size_t i = 0; i < 5; i++ )
run_test( mc128, 100 );
std::cout << "END: mc128 tests\n\n";
oag::lock_free_memory_chunk<int, 32> mc32{ num_blocks_per_thread * num_threads };
std::cout << "BEGIN: mc32 tests\n";
for ( size_t i = 0; i < 5; i++ )
run_test( mc32, 100 );
std::cout << "END: mc32 tests\n";
}
