1. 程式人生 > >執行緒基礎之“其他一些例子”

執行緒基礎之“其他一些例子”

原文連結 , 譯文連結,  譯者: dabaosod011 ,校對:樑海艦

其他一些例子

我們對資料競爭的定義相當嚴格:必須以確切方式去執行原始、未經轉換的程式中並行的衝突操作。引入有害的資料競爭將給編譯器帶來無法“打斷”程式的負擔。

舉例來說,考慮如下程式,x和y是兩個普通變數,初始化值均為0:

執行緒 1

執行緒2
if (x != 0)
  y = 1;

y = 2;

這個例子並不會導致資料競爭的出現:x變數永遠都是0,因此執行緒1的 y = 1語句永遠都不會執行。

對於下面這個例子也同樣如此:

執行緒 1

執行緒2

if (x != 0)
  y = 1;

if (y != 0)
  x = 1;

跟上面例子一樣,這兩個執行緒同樣不存在資料競爭的問題。

(這兩個例子如果按照POSIX標準來看並不那麼清晰,是否存在資料競爭會有一些爭議。我們堅持認為不存在的理由就是考慮到,資料競爭只有在順序一致性的執行過程中才會發生)

(譯者注:POSIX強調的是,如果上述程式的執行結果依賴於兩個執行緒的交叉執行順序,則存在資料競爭)

然而,如果我們稍微改變一下第一個例子,就會導致資料競爭的出現了:

執行緒 1

執行緒2

y = ((x != 0)? 1 : y)

y = 2;

儘管在單執行緒執行時,上述兩個例子的執行緒1 是等價的。但在多執行緒執行時就不一樣了,在新版本的例子中y的值需要依情況而定。在特定情況下,y的值有可能線上程2執行的同時被改寫。由於執行緒1的程式碼往往會被分割為幾步執行,這種資料競爭隱含著很容易出現的問題,如果上述程式碼按照如下的順序執行:

tmp = ((x != 0)? 1 : y);   // tmp = 0
y = 2;
y = tmp;   // y = 0

這將導致非預期性的結果,而這在最初的例子中是不會出現的。因此編譯器不能以這種方式去生成程式碼,儘管硬體的條件轉移指令會讓這種程式碼執行速度更快,或者可以使迴圈語句被向量化(譯者注:https://wiki.engr.illinois.edu/download/attachments/114688007/9-Vectorization.pdf)。如果需要這種快速的版本,要麼明確地按照程式的執行順序生成程式碼,要麼需要以某種形式讓編譯器能夠決定哪些變數不會被共享訪問。

一些其他常見的編譯器優化技術也會潛在的引入資料競爭。例如考慮下面這個迴圈語句,用來統計list中負數的數量,其中count是一個全域性或者靜態類成員:

 void count_neg(T *q)
 {
    for (T*p = q; p != 0; p = p -> next;)
    {
        if (p -> data < 0) ++count;
    }
 }

這段程式碼很明顯會引入資料競爭,如果當前執行緒正在更新count值的同時,有另一個執行緒也剛好訪問到這個變數。

編譯器有可能會將count存放在某個臨時變數裡,然後將程式碼轉換為以下形式:

 void count_neg(T *q)
 {
     int local_count = count;

     for (p = q; p != 0; p = p -> next;)
     {
        if (p -> data < 0) ++local_count;
     }
     count = local_count;
 }

但這也會引入資料競爭。假設pos_list指向某個list,它包含兩個data值:1和2, 然後宣告count為全域性變數並初始化為0。接下來有兩個執行緒分別執行:

執行緒 1

執行緒2

count_neg(pos_list);

count++;

先前沒有使用臨時變數版本的程式不會存在資料競爭,因為pos_list並不包含負數,因此執行緒1不會讀也不會改寫全域性變數count。

(有些人可能會覺得關於資料競爭的這種定義,沒有必要苛刻要求按照特定的執行順序,因此這就需要考慮傳遞給函式的一些特定的引數。尤其是在實際程式中需要傳遞指標引數時,這是唯一講得通的定義,否則任何通過指標來寫資料的函式都會引入資料競爭了。)

上面這個被編譯器優化後的程式(引入臨時變數local_count),執行緒1不管何時都存在對count的寫操作,因此也就引入了資料競爭。

(儘管在Java和C++0x中都將這種轉換清楚地列為非法,然而在POSIX中卻沒有明確規定,而且在當前(2008年)許多C編譯器中都存在這種轉換。)

需要注意的是,至少在沒有異常出現的情況下,上述程式碼可允許被按如下方式轉換:

 void count_neg(T *q)
 {
    int local_count = 0;

    for (p = q; p != 0; p = p -> next;)
    {
        if (p -> data < 0) ++local_count;
    }
    if (local_count != 0) count += local_count;
 }