執行緒安全的無鎖記憶體池：free函式在多執行緒中行為不正確-有解無憂

我有一個相同大小緩沖區的執行緒安全分配器的簡單實作。在內部，實作是一個非常簡單的互鎖單鏈表，它利用未分配緩沖區中未使用的空間來維護單鏈表。

還寫了一些測驗，以單執行緒模式測驗代碼 - 一切似乎都很好。設法將問題隔離到免費功能，但我似乎無法找到它。

我必須提到，我使用Microsoft 的 Interlocked Singly Linked Lists使用完全相同的代碼運行了一些測驗，顯然它有效，但我仍然想找出我的實作有什么問題。甚至嘗試反匯編代碼并應用類似的內在函式，但它沒有幫助（還要注意，我不需要跟蹤串列條目的數量，所以這就是為什么我不需要互鎖功能來交換雙暫存器-size 元素，例如用于 x64 的InterlockedCompareExchange128 ）

這是分配器的代碼：

#ifndef _POOLNOLOCK_HPP_
#define _POOLNOLOCK_HPP_

#include <windows.h>

template<size_t TSizeOfElem>
class PoolNoLock {
public:
    PoolNoLock(size_t N) :
        n(N),
        arr(new ELEMENT[n])
    {
        for (size_t i = 0; (n - 1) > i;   i)
        {
            arr[i].next = &arr[i   1];
        }
        arr[n - 1].next = nullptr;

        for (size_t i = 0; n > i;   i)
        {
            arr[i].allocated = false;
        }
    }

    ~PoolNoLock() { delete[] arr; }

    void *Alloc()
    {
        ELEMENT *allocBuff;

        do
        {
            allocBuff = ptrFree;
            if (!allocBuff)
            {
                return nullptr;
            }
        } while (allocBuff != InterlockedCompareExchangePointer(
            reinterpret_cast<void *volatile *>(&ptrFree),
            allocBuff->next,
            allocBuff
        ));

        if (allocBuff->allocated)
        {
            __debugbreak(); //will break here
        }

        allocBuff->allocated = true;

        return &allocBuff->buff;
    }

    void Free(void *Address)
    {
        ELEMENT *const freeBuff = reinterpret_cast<ELEMENT *>(Address);

        if (!freeBuff->allocated)
        {
            __debugbreak();
        }

        freeBuff->allocated = false;

        ELEMENT *cmpFree = ptrFree;
        do
        {
            freeBuff->next = cmpFree;

            ELEMENT *const xchgFree =
                reinterpret_cast<ELEMENT *>(InterlockedCompareExchangePointer(
                    reinterpret_cast<void *volatile *>(&ptrFree),
                    freeBuff,
                    cmpFree
                ));

            if (xchgFree == cmpFree)
            {
                break;
            }

            cmpFree = xchgFree;
        } while (true);
    }

private:
    typedef struct _ELEMENT {
        union {
            _ELEMENT *next;
            unsigned char buff[TSizeOfElem];
        };
        bool allocated; //debug info
    }ELEMENT;

    const size_t n;
    ELEMENT *const arr; //array of list elements

    ELEMENT *volatile ptrFree = &arr[0]; //head of "singly" linked list
};

#endif // _POOLNOLOCK_HPP_

這是我用來對課程進行壓力測驗的代碼：

64 是WaitForMultipleObjects可以等待的最大物件數
需要執行緒中的等待來幫助實作一個場景，其中盡可能多的執行緒正在訪問資源
產生的執行緒數正好等于分配器中的元素數，這就是 alloc-only 測驗有效的原因

#include "PoolNoLock.hpp"
#include <vector>
#include <map>
#include <iostream>

static constexpr size_t N_THREAD = 64;
static constexpr size_t N_TEST_RUN = 4;
static constexpr size_t N_ALLOC_FREE = 1024;

struct ThreadParam {
    PoolNoLock<sizeof(size_t)> *allocator;
    const HANDLE &hStartEvent;
    void *addressAlloc = nullptr;

    ThreadParam(PoolNoLock<sizeof(size_t)> *Allocator, const HANDLE &StartEvent) :
        allocator(Allocator),
        hStartEvent(StartEvent)
    {};
};

template<bool TAllocOnly>
class Test {
public:
    ~Test()
    {
        CloseHandle(hStartEvent);
    }

    bool RunSingle(PoolNoLock<sizeof(size_t)> *Allocator)
    {
        std::vector<ThreadParam> params(N_THREAD, ThreadParam(Allocator, hStartEvent));

        if (TRUE != ResetEvent(hStartEvent))
        {
            return false;
        }

        for (size_t i = 0; N_THREAD != i;   i)
        {
            handles[i] = CreateThread(nullptr,
                0,
                reinterpret_cast<PTHREAD_START_ROUTINE>(threadProc),
                &params[i],
                CREATE_SUSPENDED,
                &tids[i]);
            if (!handles[i])
            {
                return false;
            }
        }

        for (HANDLE handle : handles)
        {
            if (1 != ResumeThread(handle))
            {
                return false;
            }
        }

        if (TRUE != SetEvent(hStartEvent))
        {
            return false;
        }

        if ((WAIT_OBJECT_0   N_THREAD - 1) < WaitForMultipleObjects(N_THREAD, handles, TRUE, INFINITE))
        {
            return false;
        }

        for (size_t i = 0; N_THREAD != i;   i)
        {
            if (WAIT_OBJECT_0 != WaitForSingleObject(handles[i], 0))
            {
                return false;
            }

            DWORD exitCode;
            if (TRUE != GetExitCodeThread(handles[i], &exitCode))
            {
                return false;
            }

            if (0 != exitCode)
            {
                return false;
            }

            if (TRUE != CloseHandle(handles[i]))
            {
                return false;
            }
        }

        if (TAllocOnly)
        {
            std::map<void *, DWORD> threadAllocations;

            for (size_t i = 0; N_THREAD != i;   i)
            {
                if (!params[i].addressAlloc)
                {
                    return false;
                }

                if (threadAllocations.end() == threadAllocations.find(params[i].addressAlloc))
                {
                    return false;
                }

                std::pair<std::map<void *, DWORD>::iterator, bool> res =
                    threadAllocations.insert(std::make_pair(params[i].addressAlloc, tids[i]));

                if (!res.second)
                {
                    return false;
                }

                Allocator->Free(params[i].addressAlloc);
            }

            if (N_THREAD != threadAllocations.size())
            {
                return false;
            }
        }

        return false;
    }

    bool RunMultiple()
    {
        for (size_t i = 0; N_TEST_RUN != i;   i)
        {
            PoolNoLock<sizeof(size_t)> allocator(N_THREAD);
            RunSingle(&allocator);
        }

        return true;
    }

private:
    const HANDLE hStartEvent = CreateEventW(nullptr, TRUE, FALSE, nullptr);

    HANDLE handles[N_THREAD] = { nullptr };
    DWORD tids[N_THREAD] = { 0 };

    static DWORD WINAPI ThreadProcAllocOnly(_In_ ThreadParam *Param)
    {
        if (WAIT_OBJECT_0 != WaitForSingleObject(Param->hStartEvent, INFINITE))
        {
            return 2;
        }

        Param->addressAlloc = Param->allocator->Alloc();
        if (!Param->addressAlloc)
        {
            return 3;
        }

        return 0;
    }

    static DWORD WINAPI ThreadProcAllocFree(_In_ ThreadParam *Param)
    {
        if (WAIT_OBJECT_0 != WaitForSingleObject(Param->hStartEvent, INFINITE))
        {
            return 2;
        }

        for (size_t i = 0; N_ALLOC_FREE != i;   i)
        {
            void *ptrTest = Param->allocator->Alloc();
            if (!ptrTest)
            {
                return 3;
            }

            Param->allocator->Free(ptrTest);
        }

        return 0;
    }

    const LPTHREAD_START_ROUTINE threadProc =
        TAllocOnly
        ? reinterpret_cast<LPTHREAD_START_ROUTINE>(ThreadProcAllocOnly)
        : reinterpret_cast<LPTHREAD_START_ROUTINE>(ThreadProcAllocFree);
};

int main()
{
    Test<true> testAllocOnly0;
    Test<false> TestAllocFree0;

    if (!testAllocOnly0.RunMultiple()) //this test will succeed
    {
        std::cout << "Test failed" << std::endl;
        return 1;
    }
    std::cout << "Alloc-ONLY tests succeeded" << std::endl;

    if (!TestAllocFree0.RunMultiple()) //this test will fail
    {
        std::cout << "Test failed" << std::endl;
        return 1;
    }
    std::cout << "Alloc/free tests succeeded" << std::endl;

    std::cout << "All tests succeeded" << std::endl;

    return 0;
}

uj5u.com熱心網友回復：

你在Alloc()日常作業中的錯誤。更準確地符合

InterlockedCompareExchangePointer(
            reinterpret_cast<void *volatile *>(&ptrFree),
            allocBuff->next, // <-- !!!
            allocBuff)

這里有 2 個操作：首先 cpuallocBuff->next從allocBuff指標讀取，然后嘗試CAS，ptrFree但這 2 個操作不是原子的，可以在它們之間中斷。在您嘗試使用時allocBuff->next-allocBuff可能已經被另一個執行緒分配并next覆寫到垃圾箱（例如無效指標）。

所以讓我們存在 2 個執行緒：T#1 和 T#2

T#1 讀取allocBuff自ptrFree
T#2 讀取allocBuff自ptrFree
T#2 回傳allocBuff給用戶
T#2 覆寫allocBuff->next用戶資料的背景關系，讓 -1。
T#1 讀取next = allocBuff->next并獲取了一些用戶資料（ -1 ）
T#2 釋放/推allocBuff回ptrFree
T#1 在 CAS 中正常，因為ptrFree再次指向allocBuff. ptrFree現在指向-1
T#2 從 -1 讀取ptrFree
T#2 嘗試從 -1 讀取
碰撞

這里甚至只有堆疊中的單個元素 (a) 和 2 個執行緒足以進行演示。再舉一個例子：堆疊中的頭（F）和元素（a）。初始狀態：F -> a -> 0

T#1從F中讀取a
T#2從F中讀取a
T#2從a中讀取0
T#2 將0寫入F并將 a 回傳給用戶：F -> 0
T#2 將-1寫入a : a = -1;
T#1從a中讀取-1
T#2 釋放a : F -> a -> 0
T#1 將-1寫入F并將 a 回傳給用戶：F -> -1
T#2從F讀取-1
T#2 嘗試從-1讀取

可能和這里的另一場比賽

F - a - b - c

并且您想要 pop a并將F分配給b。但在你這樣做之前，另一個執行緒首先彈出一個，所以現在

F - b - c

然后彈出b：

F - c

并推/釋放a：

F - a - c。

因為F再次指向a CAS 就可以了，你可以制作鏈

F - b - 垃圾

因為b現在真的在使用。

無論如何，您的實施遠非最佳。您在這里不需要模板，TSizeOfElem并且n只需要在 init 程序中知道，不再需要。壓力測驗不需要很多執行緒，但關鍵點有延遲

void NotifyAllocated(PSINGLE_LIST_ENTRY allocBuff)
{
    allocBuff->Next = (PSINGLE_LIST_ENTRY)INVALID_HANDLE_VALUE; // allocated = true;
    
    WCHAR sz[16], txt[64];
    swprintf_s(sz, _countof(sz), L"%x", GetCurrentThreadId());
    swprintf_s(txt, _countof(txt), L"alocated = %p", allocBuff);
    
    MessageBoxW(0, txt, sz, MB_ICONINFORMATION); // simulate delay !
}

void NotifyCheck(PVOID buf, PCWSTR fmt)
{
    WCHAR sz[16], txt[64];
    swprintf_s(sz, _countof(sz), L"%x", GetCurrentThreadId());
    swprintf_s(txt, _countof(txt), fmt, buf);
    MessageBoxW(0, txt, sz, MB_ICONWARNING); // simulate delay !
}

class PoolNoLock 
{
    PVOID _arr = 0;                     //array of list elements
    PSINGLE_LIST_ENTRY _ptrFree = 0;    //head of "singly" linked list

public:
    bool Init(size_t N, size_t SizeOfElem)
    {
        if (N)
        {
            if (SizeOfElem < sizeof(SINGLE_LIST_ENTRY))
            {
                SizeOfElem = sizeof(SINGLE_LIST_ENTRY);
            }

            union {
                PUCHAR buf;
                PSINGLE_LIST_ENTRY Next;
            };

            if (buf = new UCHAR[N * SizeOfElem])
            {
                _arr = buf;

                PSINGLE_LIST_ENTRY ptrFree = 0;

                do 
                {
                    Next->Next = ptrFree;
                    ptrFree = Next;
                } while (buf  = SizeOfElem, --N);

                _ptrFree = ptrFree;
            }

            return true;
        }

        return false;
    }

    ~PoolNoLock() 
    { 
        if (PVOID buf = _arr)
        {
            delete[] buf; 
        }
    }

    void *Alloc()
    {
        PSINGLE_LIST_ENTRY allocBuff, ptrFree = _ptrFree;

        for (;;)
        {
            allocBuff = ptrFree;

            if (!allocBuff)
            {
                return 0;
            }

            NotifyCheck(allocBuff, L"try: %p");

            // access allocBuff->Next !!
            PSINGLE_LIST_ENTRY Next = allocBuff->Next;

            NotifyCheck(Next, L"next: %p");

            ptrFree = (PSINGLE_LIST_ENTRY)InterlockedCompareExchangePointer((void**)&_ptrFree, Next, allocBuff);
            //ptrFree = (PSINGLE_LIST_ENTRY)InterlockedCompareExchangePointer((void**)&_ptrFree, allocBuff->Next, allocBuff);

            if (ptrFree == allocBuff)
            {
                NotifyAllocated(allocBuff);
                return allocBuff;
            }
        }
    }

    void Free(void *Address)
    {
        PSINGLE_LIST_ENTRY ptrFree = _ptrFree, newFree;

        for ( ; ; ptrFree = newFree)
        {
            reinterpret_cast<PSINGLE_LIST_ENTRY>(Address)->Next = ptrFree;

            newFree = (PSINGLE_LIST_ENTRY)InterlockedCompareExchangePointer((void**)&_ptrFree, Address, ptrFree);

            if (newFree == ptrFree)
            {
                return ;
            }
        }
    }
};

ULONG ThreadTest(PoolNoLock* p)
{
    ULONG n = 2;
    do 
    {
        WCHAR sz[16], txt[32];
        swprintf_s(sz, _countof(sz), L"%x", GetCurrentThreadId());
        swprintf_s(txt, _countof(txt), L"loop %x", n);
        MessageBoxW(0, txt, sz, MB_OK);
        if (void* buf = p->Alloc())
        {
            p->Free(buf);
        }
    } while (--n);

    return 0;
}

void DemoTest()
{
    PoolNoLock p;
    if (p.Init(1, sizeof(PVOID)))
    {
        ULONG n = 2;
        do 
        {
            CloseHandle(CreateThread(0, 0, (PTHREAD_START_ROUTINE)ThreadTest, &p, 0, 0));
        } while (--n);
    }

    MessageBoxW(0, 0, L"Wait", MB_OK);
}

這與您的代碼相同，只是經過優化。相同的錯誤 - 在

ptrFree = (PSINGLE_LIST_ENTRY)InterlockedCompareExchangePointer(
        (void**)&_ptrFree, allocBuff->Next, allocBuff);

為了測驗和更好地理解 - 更好地寫成

PSINGLE_LIST_ENTRY Next = allocBuff->Next;

// delay !!

ptrFree = (PSINGLE_LIST_ENTRY)InterlockedCompareExchangePointer(
    (void**)&_ptrFree, Next, allocBuff);

用于解決這個問題SLIST_HEADER- 實際上這是指向堆疊頂部的指標操作計數的組合。正確的實作可以是下一個（如果不是直接使用SLIST_HEADER和它的api）

class PoolNoLock 
{
    PVOID _arr = 0;                     //array of list elements

    struct U  
    {
        PSINGLE_LIST_ENTRY ptr; //head of "singly" linked list
        ULONG_PTR allocCount;

        void operator = (U& v)
        {
            ptr = v.ptr;
            allocCount = v.allocCount;
        }

        U(U* v)
        {
            ptr = v->ptr->Next;
            allocCount = v->allocCount   1;
        }

        U(PSINGLE_LIST_ENTRY ptr, ULONG_PTR allocCount) : ptr(ptr), allocCount(allocCount)
        {
        }

        U() : ptr(0), allocCount(0)
        {
        }

    } u;

    //   debug
    LONG _allocMiss = 0;
    LONG _freeMiss = 0;
    //-- debug

public:
    bool Init(size_t N, size_t SizeOfElem)
    {
        if (N)
        {
            if (SizeOfElem < sizeof(SINGLE_LIST_ENTRY))
            {
                SizeOfElem = sizeof(SINGLE_LIST_ENTRY);
            }

            union {
                PUCHAR buf;
                PSINGLE_LIST_ENTRY Next;
            };

            if (buf = new UCHAR[N * SizeOfElem])
            {
                _arr = buf;

                PSINGLE_LIST_ENTRY ptrFree = 0;

                do 
                {
                    Next->Next = ptrFree;
                    ptrFree = Next;
                } while (buf  = SizeOfElem, --N);

                u.ptr = ptrFree;
                u.allocCount = 0;
            }

            return true;
        }

        return false;
    }

    ~PoolNoLock() 
    { 
        if (PVOID buf = _arr)
        {
            delete[] buf; 
        }
    }

    void *Alloc()
    {
        for (;;)
        {
            U allocBuff = u;

            if (!allocBuff.ptr)
            {
                return 0;
            }

            U Next(&allocBuff);

            if (InterlockedCompareExchange128((LONG64*)&u, Next.allocCount, (LONG64)Next.ptr, (LONG64*)&allocBuff))
            {
                // for debug only
                allocBuff.ptr->Next = (PSINGLE_LIST_ENTRY)INVALID_HANDLE_VALUE;

                return allocBuff.ptr;
            } 

            // for debug only
            InterlockedIncrementNoFence(&_allocMiss);
        }
    }

    void Free(void *Address)
    {
        for ( ; ; )
        {
            U ptrFree = u;
            U a(reinterpret_cast<PSINGLE_LIST_ENTRY>(Address), ptrFree.allocCount);
            
            reinterpret_cast<PSINGLE_LIST_ENTRY>(Address)->Next = ptrFree.ptr;

            if (InterlockedCompareExchange128((LONG64*)&u, a.allocCount, (LONG64)a.ptr, (LONG64*)&ptrFree))
            {
                return ;
            }

            // for debug only
            InterlockedIncrementNoFence(&_freeMiss);
        }
    }
};

ULONG ThreadTest(PoolNoLock* p)
{
    ULONG n = 0x10000;
    do 
    {
        if (void* buf = p->Alloc())
        {
            p->Free(buf);
        }
    } while (--n);

    return 0;
}

void DemoTest()
{
    PoolNoLock p;
    if (p.Init(16, sizeof(PVOID)))
    {
        ULONG n = 8;
        do 
        {
            CloseHandle(CreateThread(0, 0, (PTHREAD_START_ROUTINE)ThreadTest, &p, 0, 0));
        } while (--n);
    }

    MessageBoxW(0, 0, L"Wait", MB_OK);
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/483267.html

標籤：C 多线程温纳皮线程安全联锁

上一篇：安裝wfp內核驅動后devcon狀態回傳39

下一篇：FluttertoJson缺少分號