線程安全隊列是不是線程安全的，如果peek返回一個引用，即使從未使用引用（與彙編！）

Question

我有一個非常簡單的隊列實現包裝一個固定的數組。它包含窺視，入隊和出隊。如果peek返回一個引用，我發現它最終會返回衝突的結果（衝突的結果意味着它將返回2個不同的值，沒有任何干預的出隊或排隊）。顯然，如果這個引用被保留和修改，就會發生這種情況，但據我所知，事實並非如此。事實上，再次打電話給出了預期的結果。

下面是Windows線程和互斥鎖的代碼。我也嘗試過在Linux上使用pthreads，結果相同。我顯然不明白一些東西......我拋棄了可執行文件，發現返回引用或值之間的唯一區別是當內存位置被解除引用時。例如：

如果返回一個參考，偷看包含：
lea eax,[edx+ecx*4+8]
然後在消費者線程：
cmp dword ptr [eax],1

但是，如果返回值，偷看包含：
mov eax,dword ptr [edx+ecx*4+8]
然後在消費者線程中：
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); 
} 

Answer 1

Peek返回數組中間的引用，而其他線程正在主動修改該內存。從該引用讀取任何屬性將是未定義的行爲。您不能在裏面偷看，您的出列應刪除元素並返回副本。

Answer 2

您的線程可能輪流循環，這會讓您的客戶每次都碰巧使用正確的號碼。

相反，有一個消費者搶引用一個數字，它針對的副本一樣比較：

int& valRef(queue.peek()); 
int valCopy = valRef; 
while(valRef == valCopy){} 
printf("Aha! It IS unsafe!\n"); 

最終，製片方將覆蓋，而你正在做比較，你引用的內存之一。

Answer 3

也許它是這樣發揮的：

隊列已滿。由於調度生產者的一些反常的＃2連續跑了兩次，所以在隊列中的下兩個插槽包含這些值：2,2

1. Consumer線程＃1是在它的旋轉環和剛剛叫peek()。它返回對第一個插槽的引用。但是在lea指令和cmp指令之間的時間段內：
2. 消費者線程＃2調用dequeue()。這釋放了消費者＃1的peek()剛剛返回的插槽，允許生產者線程繼續。第二個值2現在在隊列的頭部。

3. 生產者線程＃1現在覆蓋值爲1的第一個插槽。因爲隊列是循環的，所以第一個插槽現在是隊列的尾部。

   這兩個插槽現在包含這些值：1，2

4. 早在Consumer線程＃1，cmp指令發生時，你看到的所需值，並退出while循環

5. 消費者＃1通話再次看到peek()，並看到錯誤的值。

當peek()返回一個副本你沒有看到這場比賽的情形，因爲消費者＃1持有互斥鎖時，它檢索值。當peek()返回一個引用時，您正在檢索該值而不保留互斥鎖，因此受到CPU指令調度程序和OS線程調度程序的支配。