I've found a little task to create a block memory pool allocator. This allocator is required to allocate memory in single fixed-sized blocks from the pool in static memory. Sizes of block and pool are fixed at compile-time, but should be tweakable during the build. This allocator should work on various embedded multithreaded platforms (RTOS). It also should contain several tests.

This is the solution I've come up with, I'd really love to hear some feedback.

#include <array>
#include <bitset>
#include <cassert>
#include <condition_variable>
#include <cstddef>
#include <mutex>
#include <stdexcept>

namespace block_allocator {
    namespace detail {
        enum class ExhaustedBehaviour {
            throw_exception = 0,    // If the pool is exhausted (empty) throw the std::bad_alloc{}
            wait_for_release        // If the pool is exhausted wait for other thread to release an object

    template <std::size_t block_size = 8, std::size_t blocks = 32, detail::ExhaustedBehaviour behaviour = detail::ExhaustedBehaviour::throw_exception>
    class Pool final {
        using pointer = void *;

        // Thread-safe static singleton initialization ([stmt.dcl])
        // Singleton prevents the static initialization order fiasco
        static auto& get_instance() {
            static Pool<block_size, blocks> pool;
            return pool;

        // Thread-safe (not interrupt-safe!) allocator
        [[nodiscard]] pointer allocate() {
            auto lock = std::unique_lock(mutex);

            if constexpr (behaviour == detail::ExhaustedBehaviour::wait_for_release) {
                if (occupation_flags.all())

            // Find index of the vacant chunk to be returned
            auto vacant = FindVacant();
            occupation_flags[vacant] = 1;

            return &internal_pool[vacant];

        // Thread-safe deallocation and pool inclusion check
        void deallocate(pointer pointer) {
            auto lock = std::unique_lock(mutex);
            auto delta = GetPoolIndex(pointer);

            // Memory doesn't belong to the pool. 
            // In any other code I would've thrown an exception, but deallocations often occur in destructors. 
            if (delta >= blocks)

            const bool is_exhausted = occupation_flags.all();
            occupation_flags[delta] = 0;

            if constexpr (behaviour == detail::ExhaustedBehaviour::wait_for_release) {
                if (is_exhausted)

        // Type alias for an array of block_size bytes
        using Block = std::array<std::byte, block_size>;
        // Singleton helper
        Pool() = default;

        // Linear memory-efficient search time.
        // For big PoolSizes one could prefer a linked list for a O(1) search time on account of dynamic memory allocation.
        auto FindVacant() {
            for (std::size_t i = 0; i < blocks; ++i)
                if (occupation_flags[i] == 0)
                    return i;

            if constexpr (behaviour == detail::ExhaustedBehaviour::throw_exception)
                throw std::bad_alloc{};
                throw std::runtime_error("Unexpected end of pool (check the correct usage of the is_exhausted flag)");

        // O(1) chunk offset calculation
        auto GetPoolIndex(pointer pointer) {
            auto delta = static_cast<std::size_t>(reinterpret_cast<Block*>(pointer) -

            return delta;

        // Uninitialized storage for block_size * blocks bytes
        // Note the lack of alignment requirements. This may become a problem on ARM-alike architectures with placement new
        std::array<Block, blocks> internal_pool;
        // Occupation flags. Every bit represents the corresponding internal_pool element. 0 stands for vacant, 1 — occupied
        std::bitset<blocks> occupation_flags;
        // Condition variable for notifying other threads when the detail::ExhaustedBehaviour::wait_for_release is selected
        static inline std::condition_variable condition_variable;
        std::mutex mutex;

    template <std::size_t block_size = 8, std::size_t blocks = 32>
    class PoolTest {
        // Should be GTest, simplified for the sake of density
        static bool test() {
            auto dst = test_double_deallocation();
            dst &= test_block_size();
            dst &= test_exhaust();
            dst &= test_threads();
            return dst;

        // Simple (incomplete) RAII for internal test use. 
        // One should also delete the copy constructor and copy assignment operator
        template <typename T, typename pool>
        class AutoDeallocator {
            AutoDeallocator() {
                pointer = reinterpret_cast<T*>(pool::get_instance().allocate());

            ~AutoDeallocator() {

            operator T*() const {
                return pointer;

            T* pointer = nullptr;


        using current_pool = Pool <block_size, blocks>;
        using pool_wrapper = AutoDeallocator<std::byte, current_pool>;

        // Test the accidental double deallocation
        static bool test_double_deallocation() {
            try {
                auto& pool = current_pool::get_instance();
                auto ptr = pool.allocate();
            catch (...) {
                return false;
            return true;

        // Verify the block size via the pointer arithmetic
        static bool test_block_size() {
            auto first = pool_wrapper();
            auto second = pool_wrapper();
            auto delta = second - first;
            return delta == block_size;

        // Request all the blocks and catch the std::bad_alloc exception when requesting the extra one
        static bool test_exhaust() {
            std::array<pool_wrapper, blocks> arr{};
            try {
                pool_wrapper exhaust;
            catch (std::bad_alloc&) {
                return true;
            catch(...) {
                return false;
            return false;

        static bool test_threads() {
            auto func = [](std::exception_ptr& e) {
                try {
                    auto allocated = pool_wrapper();
                catch(...) {
                    e = std::current_exception();

            std::array<std::thread, blocks> array_of_threads;
            std::array<std::exception_ptr, blocks> array_of_exceptions;
            for(std::size_t i = 0; i < array_of_threads.size(); ++i) {
                array_of_threads[i] = std::thread([&]() {
            std::for_each(array_of_threads.begin(), array_of_threads.end(), [](auto&current_thread) { current_thread.join(); });
            return  std::all_of(array_of_exceptions.begin(), array_of_exceptions.end(), [](auto& val) {return val == nullptr; });

static bool pool_test = block_allocator::PoolTest<9, 4>::test();

int main() {

    return 0;
  • \$\begingroup\$ You don't need 90% of this code. It's as simple as allocating a static array of bytes and keep track of the used size. If you for some reason would need to free blocks, you should be using stack (RAII) allocation instead. \$\endgroup\$ – Lundin Aug 21 '19 at 14:06

from reading through your code I've collected following remarks:

  • Why do you only allow one instance of your allocator?
  • Note: the pointer typedef in std::allocator is deprecated in C++17.
  • In Pool::allocate, you are waiting using condition_variable.wait(). This function may return without being notified because of a spurious wakeup. The usual solution is to wait for the condition in a loop and check a boolean flag (here: occupation_flags.all()). C++ offers a second interface for wait() allowing you to pass a predicate that is checked in a loop:
   if (occupation_flags.all())
       condition_variable.wait(lock, []{ return occupation_flags.all(); });
  • In Pool::deallocate you are only notifying the condition if the pool is exhausted. What if two threads are waiting for a newly released block? You may leave the second thread waiting forever. There is no need to only notify if someone is listening, just send the signal.
  • In Pool::deallocate you are checking if the pointer belongs to the pool and silently throw this error away. C++ philosophy was always to not prevent the developer from shooting himself in the foot. You should fail as loud as possible, that is, make it an assert. Freeing memory that is not managed by the pool suggests there is a serious bug in the application and the user should know about it.
  • In Pool::FindVacant use auto and 0u:
for (auto i = 0u; i < blocks; ++i)
    if (occupation_flags[i] == 0)
        return i;

Some general comments:

  • I think offering two types of error handling for the allocator complicates the whole class. You may be better off implementing two different allocators (BlockingAllocator and ThrowingAllocator) than handling this inside of the class.
  • I would replace the block_size parameter by a type parameter. The allocator can then allocate elements of that type (the size can be derived from it using sizeof) and you don't need to reinterpret_cast your pointers in the class and outside of the class.


Just went over the test cases.

  1. static bool test_double_deallocation() usually double-free is an error. I would expect the application to either raise an error or walk into undefined behavior. With this test case you're documenting that your interface supports double-free, which is at least uncommon.
  2. Just a small one:
catch (std::bad_alloc&)

Better catch by const-ref.

  1. static bool test_threads() Whats the purpose of this test? You're not looking for exceptions, you're only allocating once per thread. IMO a test case should document a certain feature of the API. Here you're only checking that no exception was raised. You can achieve this easier and more readable by simply flipping an atomic flag in your catch.
  • \$\begingroup\$ Thanks for the time you've taken to review the code. Now about the remarks: 1. Only one instance is allowed because it was required by the task to store the internal pool in the static memory. 2. You are correct, however, this allocator is not compatible with the std::allocator and std::allocator_traits<> due to the lack of rebinding. 4. In Pool::deallocate it is only possible (by interface) to return one memory block to the allocator. Therefore, only one thread gets notified. The other thread will get notified when one more block will become available. \$\endgroup\$ – Michael K Aug 2 '19 at 12:28
  • \$\begingroup\$ 5. I don't quite like the error suppression in the Pool::deallocate myself. Yes, there should be an assert. 6. It's not correct on x86_64 systems (godbolt.org/z/uew9KQ). auto i = 0u resolves to the std::uint32_t, while on 64-bit systems sizes are stored in 64 bits. About the comments: 1. Splitting this allocator into two will lead to the massive code duplication. 2. Again, the task was to develop a memory block allocator. Of course, it'd be more convenient to make an object allocator and a placement new inside the construct function \$\endgroup\$ – Michael K Aug 2 '19 at 12:29
  • 1
    \$\begingroup\$ 4. Imagine the following situation: the pool is exhausted and two threads try to get a block. Both threads will wait until the condition variable is notified. Now two threads that already have allocated a block start deallocating it. The first thread will issue a notification, as the pool is exhausted. When the second thread releases it's block, the pool is not exhausted anymore, hence no notification takes place. In this situation you'll have only one of the two waiting threads awakened. \$\endgroup\$ – Cornholio Aug 2 '19 at 13:06
  • \$\begingroup\$ Regarding the splitting: I would argue that code duplication is not necessarly a bad thing ;-) Here you have a trade-off between data structure complexity and having a similar copy of the allocator class (but with fewer code). \$\endgroup\$ – Cornholio Aug 2 '19 at 13:16
  • \$\begingroup\$ 4. Now I get it, thanks. You're totally correct \$\endgroup\$ – Michael K Aug 2 '19 at 13:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.