一个非阻塞线程安全的内存池实现

Question

我需要一个简单的非阻塞静态块大小的内存池。我在网上找不到这样的东西。所以每个人，谁需要这样的解决方案。这个是免费的..。仅适用于Win32。

诚挚的问候,

弗里德里希

#ifndef MEMPOOL_HPP_INCLUDED
#define MEMPOOL_HPP_INCLUDED

#include "atomic.hpp"
#include "static_assert.hpp"

#pragma warning( push )
#pragma warning( disable : 4311 ) // warning C4311: 'Typumwandlung'

/// @brief Block-free memory-pool implemenation
/// @tparam T Object-type to be saved within the memory-pool.
/// @tparam S Capacy of the memory-pool.
template <typename T, int S>
class MemoryPool
{
private:
    STATIC_ASSERT(sizeof(int) == sizeof(void*), "Well, ...");

public:
    /// @brief Object-type saved within the pool.
    typedef T TYPE;
    enum
    {
        /// @brief Capacy of the memory-pool.
        SIZE = S
    };

private:
    /// @brief Chunks, that holds the memory
    struct Chunk
    {
        /// @brief Single-linked list.
        Chunk* Next;
        /// @brief The value
        /// We do not call the default constructor this way.
        char Value[sizeof(TYPE)];
    };

    /// @brief The root object.
    Chunk* Root;

    /// @brief The pool
    Chunk Pool[SIZE];

private:
    // do not allow copying
    MemoryPool(const MemoryPool&);
    MemoryPool& operator=(const MemoryPool&);

    void free(Chunk* c)
    {
        c->Next = Root;
        while(!CompareAndSwap((int*)&Root, (int)c->Next, (int)c))
        {
            c->Next = Root;
        }
    }

public:
    /// Default constructor
    /// Creates an empty memory-pool.
    /// Invalidates all the memory.
    MemoryPool()
    :   Root(0)
    {
        for(int i = 0; i < SIZE; i++)
        {
            MemoryPool::free(&Pool[i]);
        }
    }

    /// @brief Frees a chunk of memory, that was allocated by MemoryPool::malloc
    /// @param _Chunk A chunk of memory, that was allocated by MemoryPool::malloc
    /// This function will not call the destructor.
    /// Thread-safe, non-blocking
    void free(T* _Chunk)
    {
        if(!_Chunk)
            return;

        Chunk* c = (Chunk*)((int)(_Chunk) - sizeof(Chunk*));

        if(c < &Pool[0] || c > &Pool[SIZE - 1])
            return;

        MemoryPool::free(c);
    }

    /// @brief Returns a pointer to a chunk of memory
    /// @return 0 on a memory shortage
    /// @return A pointer to a chunk of memory
    /// This function will not call the constructor.
    /// Thread-safe, non-blocking
    T* malloc()
    {
        Chunk* r = Root;
        if(!r)
            return 0;

        while(!CompareAndSwap((int*)&Root, (int)r, (int)r->Next))
        {
            r = Root;
            if(!r)
                return 0;
        }

        return &(r->Value);
    }
};

#pragma warning( pop )

#endif // MEMPOOL_HPP_INCLUDED

和CompareAndSwap

/// @brief Atomic compare and set
/// Atomically compare the value stored at *p with cmpval and if the
/// two values are equal, update the value of *p with newval. Returns
/// zero if the compare failed, nonzero otherwise.
/// @param p Pointer to the target
/// @param cmpval Value as we excpect it
/// @param newval New value
static inline int CompareAndSwap(volatile int *_ptr, int _old, int _new)
{
    __asm {
        mov eax, [_old]                // place the value of _old to EAX
        mov ecx, [_new]                // place the value of _new to ECX
        mov edx, [_ptr]                // place the pointer of _ptr to EDX
        lock cmpxchg [edx], ecx        // cmpxchg old (EAX) and *ptr ([EDX])
    }
    return 1;
}

Answer 1

这种方法的问题在于malloc中存在竞争条件

while(!CompareAndSwap((int*)&Root, (int)r, (int)r->Next))

请考虑以下操作序列：

新列表最初由Root = A, A->next = B, ...

• One线程读取，因此ecx = r->Next = B

• Initial线程被抢占(或，在另一个

• 上)发生一系列r = Root和r = A。由于ecx = r->Next = B

• Initial线程唤醒了Root = A, A->next = ZZZ, ...

• Original线程，因此D20>成功。< A >H225>G226

如果你有一个双字cmpxchg，比如cmpxchg8b，你可以解决这个问题。您只需在列表标题旁边添加一个序列号，以便在比较失败时，如果您被中断，如上面的(3)所示。只要每个malloc都交换指针并递增序列号，free端就可以使用窄版本。

Answer 2

感谢您的评论。这个可以与WinXP和更新版本一起使用。前面提到的实现仍然可以与PowerPC架构一起使用(如果您有正确的CompareAndSwap实现，请参阅“http://publib.boulder.ibm.com/infocenter/aix/v6r1/topic/com.ibm.aix.aixassem/doc/alangref/stwcx.htm"”)。

诚挚的问候,

弗里德里希

/// @brief Lock-free memory-pool implementation
/// @tparam T Type stored within the memory-pool
/// @tparam S Number of elements stored in the memory-pool.
template <typename T, int S>
class MemoryPool
{
public:
    /// @brief Type stored within the memory-pool.
    typedef T TYPE;
    enum
    {
        /// @brief Number of enrties in the memory-pool.
        SIZE = S
    };

private:

// we need to align the memory-pool-chunks.
#pragma pack(push, MEMORY_ALLOCATION_ALIGNMENT)

    /// @brief The memory-chunk used by the memory-pool.
    template <typename TYPE>
    struct MemoryChunk
    {
        /// @brief Next entry in the single-linked list.
        SLIST_ENTRY Next;
        /// @brief The value stored within the memory-pool.
        /// Do not call the constructor
        char Value[sizeof(TYPE)];
    };
    typedef MemoryChunk<TYPE> CHUNK_TYPE;

#pragma pack(pop, MEMORY_ALLOCATION_ALIGNMENT)

    /// @brief Head of the single-linked list.
    SLIST_HEADER Head;

    /// @brief The pool itself
    CHUNK_TYPE Pool[SIZE];

    // no copying is supported
    MemoryPool& operator=(const MemoryPool&);
    MemoryPool(const MemoryPool&);

public:
    /// @brief Constructs the memory-pool.
    MemoryPool()
    {
        InitializeSListHead(&Head);
        for(int i = 0; i < SIZE; i++)
        {
            InterlockedPushEntrySList(&Head, &Pool[i].Next);
        }
    }

    /// @brief Free the memory-pool.
    ~MemoryPool()
    {
        InterlockedFlushSList(&Head);
    }

    /// @brief Allocates a memory chunk
    /// @return 0 if none is free
    /// @return Pointer to a free memory chunk (the constructor is not called!)
    TYPE* Allocate()
    {
        CHUNK_TYPE* c = reinterpret_cast<CHUNK_TYPE*>(InterlockedPopEntrySList(&Head));
        if(c)
            return reinterpret_cast<TYPE*>(&c->Value[0]);
        else
            return 0;
    }

    /// @brief Deallocates a memory chunk (the destructor is not called)
    /// @param c Point to the memory-chunk allocated by us.
    void Deallocate(void* c)
    {
        if(c < static_cast<void*>(&Pool[0]) || c > static_cast<void*>(&Pool[SIZE]))
            return; // was not allocated by us
        char* p = static_cast<char*>(c);
        p -= sizeof(SLIST_ENTRY);
        CHUNK_TYPE* t = reinterpret_cast<CHUNK_TYPE*>(p);
        InterlockedPushEntrySList(&Head, &t->Next);
    }
};