Eine nicht blockierende, thread-sichere Speicher-Pool-Implementierung

Question

Ich brauchte einen einfachen, nicht blockierenden Speicher-Pool mit statischer Blockgröße. Ich habe keinen solchen im Web gefunden. Also an alle, die eine solche Lösung brauchen. Diese hier ist kostenlos... funktioniert nur unter Win32.

Mit freundlichen Grüßen,

Friedrich

#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

Und der 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

Das Problem bei diesem Ansatz ist, dass es eine Wettlaufsituation in malloc :

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

Betrachten Sie die folgende Abfolge von Vorgängen:

1. Ursprünglich Root = A, A->next = B, ...
2. Ein Thread lautet r = Root also r = A und (in einem Register) heißt es ecx = r->Next = B
3. Der erste Thread wird vorzeitig beendet (oder, auf einer anderen CPU) eine Reihe von malloc y free auftreten, so dass A wird eine Zeit lang verwendet und zuletzt freigegeben.
4. Der neue Listenstatus lautet Root = A, A->next = ZZZ, ...
5. Original-Thread wacht auf und macht cmpxchg und ist erfolgreich, weil Root == r == A und setzt damit Root = ecx = B
6. Jetzt ist Ihre Liste beschädigt.

Sie können dieses Problem lösen, wenn Sie ein Doppelwort haben cmpxchg , wie zum Beispiel cmpxchg8b . Sie fügen einfach eine Seriennummer neben dem Listenkopf ein, so dass der Vergleich fehlschlägt, wenn Sie wie in (3) oben unterbrochen werden. Die free Seite kann die schmale Version verwendet werden, solange jede malloc beide tauschen den Zeiger y inkrementiert die Seriennummer.

Answer 2

Ich danke Ihnen für Ihre Kommentare. Dieses Programm kann mit WinXP und neueren Versionen verwendet werden. Die zuvor erwähnte Implementierung kann auch mit einer PowerPC-Architektur verwendet werden (wenn Sie eine geeignete Implementierung von CompareAndSwap haben, siehe "http://publib.boulder.ibm.com/infocenter/aix/v6r1/topic/com.ibm.aix.aixassem/doc/alangref/stwcx.htm").

Mit freundlichen Grüßen,

Friedrich

/// @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);
    }
};