編譯器優化打破多線程代碼

Question

在瞭解了shared variables are currently not guarded by memory barriers的難題之後，我現在遇到了另一個問題。無論是我做錯了什麼，或者dmd中的現有編譯器優化都可以通過重新排序shared變量的讀取來打破多線程代碼。

作爲一個例子，當我編譯dmd -O（完全優化）可執行文件，編譯器愉快地優化遠局部變量o在此代碼（其中cas是從core.atomic的比較並交換功能）

shared uint cnt; 
void atomicInc () { uint o; do { o = cnt; } while (!cas(&cnt, o, o + 1));} 

到這樣的事情（見下文的拆卸）：

shared uint cnt; 
void atomicInc () { while (!cas(&cnt, cnt, cnt + 1)) { } } 

在「優化的」代碼cnt從存儲器中讀取兩次，從而冒着另一個線程修改cnt的風險。優化基本上破壞了比較和交換算法。

這是一個錯誤，還是有正確的方法來實現預期的結果？到目前爲止我發現的唯一解決方法是使用匯編程序來實現代碼。

全部測試代碼和其他詳細信息
爲了完整起見，這裏是一個完整的測試代碼，顯示這兩個問題（沒有記憶障礙，優化問題）。它產生於三個不同的Windows機器下面的輸出兩個DMD 2.049和DMD 2.050（假設德克爾的算法不死鎖，這可能發生）：

dmd -O -run optbug.d 
CAS : failed 
Dekker: failed 

而且裏面atomicInc循環被編譯到這個全優化：

; cnt is stored at 447C10h 
; while (!cas(&cnt, o, o + 1)) o = cnt; 
; 1) prepare call cas(&cnt, o, o + 1): &cnt and o go to stack, o+1 to eax 
402027: mov ecx,447C10h   ; ecx = &cnt 
40202C: mov eax,[447C10h]  ; eax = o1 = cnt 
402031: inc eax     ; eax = o1 + 1 (third parameter) 
402032: push ecx     ; push &cnt (first parameter) 
    ; next instruction pushes current value of cnt onto stack 
    ; as second parameter o instead of re-using o1 
402033: push [447C10h]  
402039: call 4020BC    ; 2) call cas  
40203E: xor al,1    ; 3) test success 
402040: jne 402027    ; no success try again 
; end of main loop 

下面是測試代碼：

import core.atomic; 
import core.thread; 
import std.stdio; 

enum loops = 0xFFFF; 
shared uint cnt; 

/* ***************************************************************************** 
Implement atomicOp!("+=")(cnt, 1U); with CAS. The code below doesn't work with 
the "-O" compiler flag because cnt is read twice while calling cas and another 
thread can modify cnt in between. 
*/ 
enum threads = 8; 

void atomicInc () { uint o; do { o = cnt; } while (!cas(&cnt, o, o + 1));} 
void threadFunc () { foreach (i; 0..loops) atomicInc; } 

void testCas () { 
    cnt = 0; 
    auto tgCas = new ThreadGroup; 
    foreach (i; 0..threads) tgCas.create(&threadFunc); 
    tgCas.joinAll; 
    writeln("CAS : ", cnt == loops * threads ? "passed" : "failed"); 
} 

/* ***************************************************************************** 
Dekker's algorithm. Fails on ia32 (other than atom) because ia32 can re-order 
read before write. Most likely fails on many other architectures. 
*/ 
shared bool flag1 = false; 
shared bool flag2 = false; 
shared bool turn2 = false; // avoids starvation by executing 1 and 2 in turns 

void dekkerInc () { 
    flag1 = true; 
    while (flag2) if (turn2) { 
     flag1 = false; while (turn2) { /* wait until my turn */ } 
     flag1 = true; 
    } 
    cnt++;     // shouldn't work without a cast 
    turn2 = true; flag1 = false; 
} 

void dekkerDec () { 
    flag2 = true; 
    while (flag1) if (!turn2) { 
     flag2 = false; while (!turn2) { /* wait until my turn */ } 
     flag2 = true; 
    } 
    cnt--;     // shouldn't work without a cast 
    turn2 = false; flag2 = false; 
} 

void threadDekkerInc () { foreach (i; 0..loops) dekkerInc; } 
void threadDekkerDec () { foreach (i; 0..loops) dekkerDec; } 

void testDekker () { 
    cnt = 0; 
    auto tgDekker = new ThreadGroup; 
    tgDekker.create(&threadDekkerInc); 
    tgDekker.create(&threadDekkerDec); 
    tgDekker.joinAll; 
    writeln("Dekker: ", cnt == 0 ? "passed" : "failed"); 
} 

/* ************************************************************************** */ 
void main() { 
    testCas; 
    testDekker; 
} 

Answer 1

雖然問題似乎仍然存在，core.atomic現在公開atomicLoad，它使相對簡單的解決方法。爲了使cas例如工作，它足以載入cnt原子：

void atomicInc () { 
    uint o; 
    do { 
     o = atomicLoad(cnt); 
    } while (!cas(&cnt, o, o + 1)); 
} 

同樣，使德克爾的算法工作：

// ... 
while (atomicLoad(flag2)) if (turn2) { 
// ... 
while (atomicLoad(flag1)) if (!turn2) { 
// ... 

對於除IA32其他架構（忽略字符串操作和SSE），可以也重新排序

• 讀取相對於讀取
• 或寫入RELA略去對寫
• 或寫入和讀取到相同的存儲器位置

額外內存屏障將是必需的。

Answer 2

是，在彙編代碼它。如果你跳過使用cas（）函數，並且只是在彙編中編寫整個atomicInt函數，那隻會是幾行代碼。在你這樣做之前，你可能會反對編譯器的優化。除此之外，您可以使用x86 LOCK INC指令而不是CAS，並且您應該能夠將函數減少爲僅一行或兩個程序集。