线程安全队列不是线程安全的，如果peek返回一个引用，即使引用永远不会使用（与汇编！） [英] thread safe queue isn't thread safe if peek returns a reference even if reference is never used (with assembly!)

问题描述

我有一个非常简单的队列实现，封装一个固定的数组。它包含窥视，入队和出列。如果peek返回一个引用，我发现它将最终返回冲突的结果（冲突的结果意味着它将返回2个不同的值，没有任何中间的出队或入队）。显然，如果该引用被持有和修改，这可能发生，但是据我所知，它不是。事实上，再次调用peek会产生预期的结果。

I have a very simple queue implementation that wraps a fixed array. It contains peek, enqueue, and dequeue. If peek returns a reference, I've found that it will return conflicting results eventually (conflicting results meaning it will return 2 different values without any intervening dequeues or enqueues). Obviously, this could happen if that reference is held onto and modified, but as far as I can tell, it isn't. In fact, calling peek again gives the expected result.

下面是Windows线程和互斥体的代码。我也试过这个使用pthreads在Linux上有相同的结果。我显然不明白的东西...我抛弃了可执行文件，发现返回一个引用或值之间的唯一区别是当内存位置被解引用。例如：

Below is the code with Windows threading and mutexes. I've also tried this using pthreads on Linux with the same result. I obviously don't understand something... I've dumped the executable and find the only difference between returning a reference or a value is when the memory location is dereferenced. For example:

如果返回引用，则peek包含：

lea eax，[edx + ecx * 4 + 8]

然后在消费者主题中：

cmp dword ptr [eax]，1

If a reference is returned, peek contains:
lea eax,[edx+ecx*4+8]
And then in the consumer thread:
cmp dword ptr [eax],1

但是，如果返回值，则peek包含：

mov eax，dword ptr [edx + ecx * 4 + 8]

然后在消费者线程中：

cmp eax，1

But, if a value is returned, peek contains:
mov eax,dword ptr [edx+ecx*4+8]
And then in the consumer thread:
cmp eax,1

谢谢！

#include <iostream>
#include <windows.h>

typedef void *(thread_func_type)(void *);

void start_thread(HANDLE &thread, thread_func_type *thread_func, void *arg)
{
    DWORD id;
    thread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)thread_func, arg, 0, &id);
    if (thread == NULL) {
        std::cerr << "ERROR: failed to create thread\n";
        ::exit(1);
    }
}

void join_thread(HANDLE &thread)
{
    WaitForSingleObject(thread, INFINITE);
}

class ScopedMutex
{
    HANDLE &mutex;

public:

    ScopedMutex(HANDLE &mutex_) : mutex(mutex_)
    {
        WORD result = WaitForSingleObject(mutex, INFINITE);
        if (result != WAIT_OBJECT_0) {
            std::cerr << "ERROR: failed to lock mutex\n";
            ::exit(1);
        }
    };

    ~ScopedMutex()
    {
        ReleaseMutex(mutex);
    };
};

template <typename T, unsigned depth>
class Queue
{
    unsigned head, tail;
    bool full;
    T data[depth];
    HANDLE mutex;

public:

    Queue() : head(0), tail(0), full(false)
    {
        mutex = CreateMutex(NULL, 0, NULL);
        if (mutex == NULL) {
            std::cerr << "ERROR: could not create mutex.\n";
            ::exit(1);
        }
    };

    T &peek()
    {
        while (true) {
            {
                ScopedMutex local_lock(mutex);
                if (full || (head != tail))
                    return data[tail];
            }
            Sleep(0);
        }
    };

    void enqueue(const T &t)
    {
        while (true) {
            {
                ScopedMutex local_lock(mutex);
                if (!full) {
                    data[head++] = t;
                    head %= depth;
                    full = (head == tail);
                    return;
                }
            }
            Sleep(0);
        }
    };

    void dequeue()
    {
        while (true) {
            {
                ScopedMutex local_lock(mutex);
                if (full || (head != tail)) {
                    ++tail;
                    tail %= depth;
                    full = false;
                    return;
                }
            }
            Sleep(0);
        }
    };
};

template <unsigned num_vals, int val, unsigned depth>
void *
producer(void *arg)
{
    Queue<int, depth> &queue = *static_cast<Queue<int, depth> *>(arg);
    for (unsigned i = 0; i < num_vals; ++i) {
        queue.enqueue(val);
    }
    std::cerr << "producer " << val << " exiting.\n";
    return NULL;
}

template <unsigned num_vals, int val, unsigned depth>
void *
consumer(void *arg)
{
    Queue<int, depth> &queue = *static_cast<Queue<int, depth> *>(arg);
    for (unsigned i = 0; i < num_vals; ++i) {
        while (queue.peek() != val)
            Sleep(0);
        if (queue.peek() != val) {
            std::cerr << "ERROR: (" << val << ", " << queue.peek() << ")" << std::endl;
            std::cerr << "But peeking again gives the right value " << queue.peek() << std::endl;
            ::exit(1);
        }
        queue.dequeue();
    }
    return NULL;
}

int
main(int argc, char *argv[])
{
    const unsigned depth = 10;
    const unsigned num_vals = 100000;
    Queue<int, depth> queue;
    HANDLE p1, p2, c1, c2;
    start_thread(p1, producer<num_vals, 1, depth>, &queue);
    start_thread(p2, producer<num_vals, 2, depth>, &queue);
    start_thread(c1, consumer<num_vals, 1, depth>, &queue);
    start_thread(c2, consumer<num_vals, 2, depth>, &queue);
    join_thread(p1);
    join_thread(p2);
    join_thread(c1);
    join_thread(c2);
}

推荐答案

Peek中间的数组，而其他线程正在主动修改该内存。从该引用读取任何属性将是未定义的行为。你不能偷看里面，你的dequeue应该删除元素并返回一个副本。

Peek returns a reference into the middle of array, while other threads are actively modifying that memory. Reading any property from that reference will be undefined behavior. You can't peek inside, your dequeue should remove the element and return a copy.

这篇关于线程安全队列不是线程安全的，如果peek返回一个引用，即使引用永远不会使用（与汇编！）的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！