如何让它运行同步？答案

【问题标题】：How to make it run sync?如何让它运行同步？
【发布时间】：2019-09-29 23:05:43
【问题描述】：

您好，我想同步两个线程，一个增加一个变量，另一个减少一个变量。我想要的结果如下：

Thread #0 j = 1

Thread #1 j = 0

Thread #0 j = 1

Thread #1 j = 0

等等.. 但我的代码有时会这样工作，在某些情况下它会打印出非常奇怪的值。我想我在某处有一些未定义的行为，但我无法弄清楚到底发生了什么。

我的代码包含在一个 HANDLE ghMutex 中，其中包含我的互斥锁的处理程序：

我的主要功能：

int main(void)
{
    HANDLE aThread[THREADCOUNT];

    ghMutex = CreateMutex(NULL, FALSE, NULL);             

    aThread[0] = (HANDLE)_beginthreadex(NULL, 0, &inc, NULL, CREATE_SUSPENDED, 0);
    aThread[1] = (HANDLE)_beginthreadex(NULL, 0, &dec, NULL, CREATE_SUSPENDED, 0);

    ResumeThread(aThread[0]);
    ResumeThread(aThread[1]);

    WaitForMultipleObjects(THREADCOUNT, aThread, TRUE, INFINITE);

    printf("j = %d\n", j);

    for (int i = 0; i < THREADCOUNT; i++)
        CloseHandle(aThread[i]);

    CloseHandle(ghMutex);

    return 0;
}

公司功能：

unsigned int __stdcall inc(LPVOID)
{
    for (volatile int i = 0; i < MAX; ++i)
    {
        WaitForSingleObject(
            ghMutex,    // handle to mutex
            INFINITE);  // no time-out interval

            j++;
            printf("Thread %d j = %d\n", GetCurrentThreadId(), j);
            ReleaseMutex(ghMutex);

    }
    _endthread();

    return TRUE;
}

十进制函数：

unsigned int __stdcall dec(void*)
{
    for (volatile int i = 0; i < MAX; ++i)
    {
        WaitForSingleObject(
            ghMutex,    // handle to mutex
            INFINITE);  // no time-out interval

        j--;
        printf("Thread %d j = %d\n", GetCurrentThreadId(), j);
        ReleaseMutex(ghMutex);
    }
    _endthread();

    return TRUE;
}

我需要一个标准 c++98 中的 win api 解决方案。

【问题讨论】：

为此您需要 2 个事件 (0,1)。每个线程 (i) 信号事件 (1-i) 并等待事件 (i)。说SignalObjectAndWait 这里很有用。在 win 8.1 之前是 object eventpair - 它非常适合此类任务，但由于未知原因，它已从 Windows 中删除
声明局部变量为volatile，只能从单线程访问——没有任何意义

标签： c++ windows multithreading winapi c++98

【解决方案1】：

互斥锁不是同步两个线程的正确工具，它用于保护资源。你确实有一个资源j，它受你的互斥锁保护，但是哪个线程获得锁的顺序是未定义的，所以你可能会遇到dec在inc有机会运行之前被多次调用的情况.

如果您想同步线程的顺序，您将不得不使用另一个同步原语，例如信号量。例如，您可以增加inc 中的信号量并减少dec 中的信号量。这将是经典的生产者 - 消费者关系，当信号量达到最大值时生产者将停止，消费者将等待项目消费。

抱歉，我没有提供 WinAPI C++98 解决方案，因为那会很愚蠢，但我希望我为您指明了正确的方向。

【讨论】：

抱歉，Mutex 和 BinarySemaphore 不一样吗？
@emanuelclur 不，这通常是未定义的行为，或者不允许互斥锁被另一个线程解锁，然后是锁定它的线程。二进制信号量正是为此而构建的。
我没有 WinAPI C++98 解决方案，因为那太愚蠢了我不同意这一点 - winapi（如果我们不关心可移植性） - 更多的功能和易于用于此类任务。（比较 c++ 库，这里非常差且效率不高）。但是当我们使用 winapi 解决方案时 - c++98 或 c++20 之间没有任何不同。
@RbMm 我认为现在坚持使用裸 WinAPI 或 C++98 的原因有很多，但这个答案的主要原因是问题与实现无关但是对线程同步原语的理解，我相信我已经在这个更高的层次上回答了这个问题。因此，我认为OP有能力编写他的实现，并且为WinAPI C ++ 98键入一些我不同意使用并且很长时间没有使用的代码不仅很愚蠢，而且是错误的签名给其他认为可以的读者。
我认为 c++ 库设施与 winapi 相比非常差和有限

【解决方案2】：

windows 互斥对象保证独占所有权，但不关心所有权顺序。这样同一个线程可以连续捕获多次而其他线程等待。

对于您的任务，您需要在任务完成后向另一个线程发出信号，然后等待来自另一个线程的信号。例如，此任务可以使用事件对。线程 (i) 信号事件 (1-i) 并等待事件 (i)。用于优化而不是 2 次调用 -

SetEvent(e[1-i]); WaitForSingleObject(e[i], INFINITE);

我们可以使用单呼SignalObjectAndWait

SignalObjectAndWait(e[1-i], e[i], INFINITE, FALSE)

当然，循环的开始和结束需要特别小心。为inc

    HANDLE hObjectToSignal  = _hEvent[1], hObjectToWaitOn  = _hEvent[0];

    for (;;)
    {
        _shared_value++;

        if (!--n)
        {
            SetEvent(hObjectToSignal);
            break;
        }

        SignalObjectAndWait(hObjectToSignal, hObjectToWaitOn, INFINITE, FALSE);
    }

对于dec

    HANDLE hObjectToSignal  = _hEvent[0], hObjectToWaitOn  = _hEvent[1];

    WaitForSingleObject(hObjectToWaitOn, INFINITE);
    for (;;)
    {
        --_shared_value;

        if (!--n)
        {
            break;
        }

        SignalObjectAndWait(hObjectToSignal, hObjectToWaitOn, INFINITE, FALSE);
    }

如果编写完整的测试，带有错误检查

struct Task 
{
    HANDLE _hEvent[4];
    ULONG _n;
    LONG _iTasks;
    LONG _shared_value;

    Task()
    {
        RtlZeroMemory(this, sizeof(*this));
    }

    ~Task()
    {
        ULONG n = RTL_NUMBER_OF(_hEvent);
        do 
        {
            if (HANDLE hEvent = _hEvent[--n]) CloseHandle(hEvent);
        } while (n);
    }

    ULONG WaitTaskEnd()
    {
        return WaitForSingleObject(_hEvent[2], INFINITE);
    }

    ULONG WaitTaskReady()
    {
        return WaitForSingleObject(_hEvent[3], INFINITE);
    }

    void SetTaskReady()
    {
        SetEvent(_hEvent[3]);
    }

    void End()
    {
        if (!InterlockedDecrement(&_iTasks)) SetEvent(_hEvent[2]);
    }

    void Begin()
    {
        InterlockedIncrementNoFence(&_iTasks);
    }

    static ULONG WINAPI IncThread(PVOID p)
    {
        return reinterpret_cast<Task*>(p)->Inc(), 0;
    }

    void Inc()
    {
        if (WaitTaskReady() == WAIT_OBJECT_0)
        {
            if (ULONG n = _n)
            {
                HANDLE hObjectToSignal  = _hEvent[1], hObjectToWaitOn  = _hEvent[0];

                for (;;)
                {
                    if (_shared_value) __debugbreak();

                    if (n < 17) DbgPrint("Inc(%u)\n", n);

                    _shared_value++;

                    if (!--n)
                    {
                        SetEvent(hObjectToSignal);
                        break;
                    }

                    if (SignalObjectAndWait(hObjectToSignal, hObjectToWaitOn, INFINITE, FALSE) != WAIT_OBJECT_0)
                    {
                        break;
                    }
                }
            }
        }

        End();
    }

    static ULONG WINAPI DecThread(PVOID p)
    {
        return reinterpret_cast<Task*>(p)->Dec(), 0;
    }

    void Dec()
    {
        if (WaitTaskReady() == WAIT_OBJECT_0)
        {
            if (ULONG n = _n)
            {
                HANDLE hObjectToSignal  = _hEvent[0], hObjectToWaitOn  = _hEvent[1];

                if (WaitForSingleObject(hObjectToWaitOn, INFINITE) == WAIT_OBJECT_0)
                {
                    for (;;)
                    {
                        --_shared_value;

                        if (_shared_value) __debugbreak();

                        if (n < 17) DbgPrint("Dec(%u)\n", n);

                        if (!--n)
                        {
                            break;
                        }

                        if (SignalObjectAndWait(hObjectToSignal, hObjectToWaitOn, INFINITE, FALSE) != WAIT_OBJECT_0)
                        {
                            break;
                        }
                    }
                }
            }
        }

        End();
    }

    ULONG Create()
    {
        ULONG n = RTL_NUMBER_OF(_hEvent);
        do 
        {
            if (HANDLE hEvent = CreateEventW(0, n > 2, 0, 0)) _hEvent[--n] = hEvent; 
            else return GetLastError();
        } while (n);

        return NOERROR;
    }

    ULONG Start()
    {
        static PTHREAD_START_ROUTINE aa[] = { IncThread, DecThread };

        ULONG n = RTL_NUMBER_OF(aa);

        do 
        {
            Begin();
            if (HANDLE hThread = CreateThread(0, 0, aa[--n], this, 0, 0))
            {
                CloseHandle(hThread);
            }
            else
            {
                n = GetLastError();
                End();
                return n;
            }
        } while (n);

        return NOERROR;
    }

    ULONG Start(ULONG n)
    {
        _iTasks = 1;

        ULONG dwError = Start();

        _n = dwError ? 0 : n;

        SetTaskReady();

        End();

        return dwError;
    }
};

void TaskTest(ULONG n)
{
    Task task;
    if (task.Create() == NOERROR)
    {
        task.Start(n);
        task.WaitTaskEnd();
    }
}

注意，没有任何意义将局部变量（只能从单线程访问，不会被任何中断等访问）声明为 volatile

在我们写代码的时候，比如：

// thread #1
write_shared_data();
SetEvent(hEvent);

// thread #2
WaitForSingleObject(hEvent, INFINITE);
read_shared_data();

在 SetEvent(hEvent) 中；是原子写入具有释放语义的事件状态（当然真的更强大）和内部等待事件功能 - 原子读取它的状态不仅仅是获取语义。结果是线程 #1 在 SetEvent 之前写入内存的所有内容 - 在等待事件之后线程 #2 将可见（如果由于从线程 #1 调用 Set 而等待完成）

【讨论】：