再帰ロックのコード例にデータ競合が含まれる？

[kiripon]

４章４節の再帰ロックのコードサンプルにデータ競合がありそうです。
reentlock_acquire 関数
https://github.com/oreilly-japan/conc_ytakano/blob/main/chap4/4.4/ch4_4_reent_c/reent.c#L17

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 19 in 85a014d

if (lock->lock && lock->id == id) {

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 25 in 85a014d

spinlock_acquire(&lock->lock);

それぞれの行は lock->lock へのロードとストアをそれぞれ行っており、spinlock_acquire はアトミック操作で、もう片方は通常のロード操作です。

C11 の規格では、
https://port70.net/~nsz/c/c11/n1570.html#5.1.2.4p25

The execution of a program contains a data race if it contains two conflicting actions in different threads, at least one of which is not atomic, and neither happens before the other. Any such data race results in undefined behavior.

とあるのでこの lock->lock メンバへのアクセスはデータ競合を含んでいそうです。

また、この関数中の lock->id へのアクセスも同様です。

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 28 in 85a014d

lock->id = id;

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 19 in 85a014d

if (lock->lock && lock->id == id) {

lock->lock への操作と lock->id への操作はどちらも atomic なほうが良いのではないかと思いました。

[ytakano]

ありがとうございます！
ご指摘通りに思います。後ほど修正し、正誤表に追記いたします。

[ytakano]

確認しました。
こちらのコードはアトミック(TAS)と非アトミックなloadを利用していますが、これはTTASと同じ構造で問題無さそうです。
また、TASとReleaseはそれぞれ、acquireとreleaseのメモリオーダーなので、lock中のみ変数が更新されるので、こちらも問題無さそうです。

C11からは_Atomicが追加されて、メモリモデルが詳細に仕様化されて、これらを使うと良いのですが、本書はC99でも動作するコードを掲載しています。

いくつか未検討なパターンもあるので、引き続き検証してみます。

[kiripon]

確認ありがとうございます！

本書はC99でも動作するコードを掲載しています。

あっなるほど、であれば C11 のメモリモデルを持ち出すのは良くないですね…失礼しました。

これはTTASと同じ構造で問題無さそうです。

if 文の else 節の lock->lock 変数については TTAS と同じ構造であることを意図して書かれているのですね。 lock->lock は問題なさそうです。

一方で、lock->id についてはそのようになっていないような気がしています。
具体的には、 lock->id = id と、 lock->id == id の式が同時に実行された場合に問題が起きるのではないでしょうか？
この lock->id への代入と読み込みはアトミックでないため、代入中に読み込みを行うと代入しているものとは異なる値が観測されます。
スレッドID 0x11 のスレッドが lock->id への代入を行っている途中で別のスレッドが読み込みを行うと、途中状態として 0x10 が観測される可能性があります。仮に読み込んだスレッドがスレッドID 0x10 だったとすると、この if 文の条件は真になります。
こうなると、本来ロック保持者ではないスレッドID 0x10 のスレッドが勝手に lock->cnt をインクリメントしてしまいますし、スレッド ID 0x10 のスレッドはロック獲得時の処理に入ってしまうと思います。

[ytakano]

ありがとうございます！

C11の動作モデルについて

本書執筆時は、C11かC99かで迷ったのですが、C11で正確に記載するには、メモリバリアの説明が必要となります。しかし、メモリバリアはアドバンスなトピックであるため、中盤以降に持ってきたかったので、GCC拡張を利用しています。

C11の動作モデルで規定するundefinedですが、たしかに、x86では正しく動作するがAArch64では動作しないという挙動になるundefinedなコードは書けるとは思います。一方で、弱いメモリモデルにあわせたアルゴリズムなら、規格上はundefinedでも実用上は動作するのではと思います。実際C99はアトミック処理に関しては、どの動作もundefinedでコンパイラ実装とCPUの実装を把握して書くしかありませんでした。

ご指摘通り、C11で導入されたatomicを利用すればこれは綺麗に書けるのですが、本書ではその部分がRustに置き換わってしまっています。

lock->idについて

lock->idですが、アルゴリズム上はそのような挙動は起こりえないかと思います。

0x11スレッド

ロック獲得

以下の行を実行と仮定。

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 25 in 85a014d

spinlock_acquire(&lock->lock);

1. 初期状態：lock->id == 0
2. ロック獲得：lock->lock = true
3. lock->id = 0x11
4. lock->cnt++

ロック獲得

以下の行を実行と仮定。

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 37 in 85a014d

lock->cnt--;

1. lock->cnt--
2. lock->id = 0
3. lock->lock = false

0x10スレッド

上記のどのステップでlock->idを読み込んでも、0x10にはならない。ありあえるのは、0x00か0x11のみ。
自分でロック獲得後でないと、lock->lock == trueかつlock->id == 0x11にはならない。

モデル検査

本当にこのアルゴリズムで大丈夫かという疑問は残るので、モデル検査ツールのTLA+で検証してみました。
検証項目は、「デッドロックしない」と「クリティカルセクション実行のスレッドが高々1」という不変条件を検証しました。

結論から言うと、これらに対する反例は見つからず、現時点ではアルゴリズムの誤りは発見できませんでした。

以下が検証のコードとなります。
定数はMAX_REC = 3、PIDS = {1, 2, 3}としています。

また、以下の箇所がout-of-order実行されると想定し、メモリ読み込みをコードの逆順としています。

conc_ytakano/chap4/4.4/ch4_4_reent_c/reent.c

Line 19 in 85a014d

if (lock->lock && lock->id == id) {

------------------------------ MODULE reclock ------------------------------
EXTENDS Integers, TLC, Sequences, FiniteSets

CONSTANTS MAX_REC, PIDS

(*--algorithm lock

variables
    locking = {}, \* ロック中のプロセス集合。検証用の変数
    
    \* アルゴリズム用の変数
    lock = FALSE,
    id = 0,
    cnt = 0;

\* ロック獲得
procedure acquire()
variables
    regA, regB, regC, regD, regE;

begin
    Acquire:
        \* out-of-order実行を模擬
            regA := cnt;
        A0: regB := id;
        A1: regC := lock;
        A2: regD := regB = self;
        A3: regE := regC /\ regD;
        if regE then \* if lock /\ id = self
            A4: regA := regA + 1;
            A5: cnt := regA; \* cnt := cnt + 1
        else
            TAS:
                await lock = FALSE;
                lock := TRUE;
                locking := locking \union {self};
            A6: id := self;
            A7: cnt := cnt + 1;
        end if;
        A8: return
end procedure;

\* ロック開放
procedure release()
begin
    Release:
        cnt := cnt - 1;
        R0:
            if cnt = 0 then
                R1: id := 0;
                R2: lock := FALSE;
                    locking := locking \ {self};
            end if;
        R4: return;
end procedure;

\* 再帰ロック
procedure recursive_lock()
begin
    RecLock:
        either
            Rec:
                if cnt < MAX_REC then \* 最大MAX_REC回まで再帰ロック
                    RL0:
                        call lock_unlock() \* 再帰ロック
                end if;
        or
            NotRec:
                skip;
        end either;
        RL1: return;
end procedure;

procedure lock_unlock()
begin
    Lock:
        call acquire(); \* ロック獲得
        
        L0: call recursive_lock(); \* 再帰ロック
        
        L1: call release(); \* ロック開放
        L2: return;
end procedure;

fair process pid \in PIDS
begin
    T1:
        call lock_unlock()
end process;

end algorithm;*)
\* BEGIN TRANSLATION (chksum(pcal) = "b99e79b4" /\ chksum(tla) = "1fb7a28e")
CONSTANT defaultInitValue
VARIABLES locking, lock, id, cnt, pc, stack, regA, regB, regC, regD, regE

vars == << locking, lock, id, cnt, pc, stack, regA, regB, regC, regD, regE >>

ProcSet == (PIDS)

Init == (* Global variables *)
        /\ locking = {}
        /\ lock = FALSE
        /\ id = 0
        /\ cnt = 0
        (* Procedure acquire *)
        /\ regA = [ self \in ProcSet |-> defaultInitValue]
        /\ regB = [ self \in ProcSet |-> defaultInitValue]
        /\ regC = [ self \in ProcSet |-> defaultInitValue]
        /\ regD = [ self \in ProcSet |-> defaultInitValue]
        /\ regE = [ self \in ProcSet |-> defaultInitValue]
        /\ stack = [self \in ProcSet |-> << >>]
        /\ pc = [self \in ProcSet |-> "T1"]

Acquire(self) == /\ pc[self] = "Acquire"
                 /\ regA' = [regA EXCEPT ![self] = cnt]
                 /\ pc' = [pc EXCEPT ![self] = "A0"]
                 /\ UNCHANGED << locking, lock, id, cnt, stack, regB, regC, 
                                 regD, regE >>

A0(self) == /\ pc[self] = "A0"
            /\ regB' = [regB EXCEPT ![self] = id]
            /\ pc' = [pc EXCEPT ![self] = "A1"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regC, regD, 
                            regE >>

A1(self) == /\ pc[self] = "A1"
            /\ regC' = [regC EXCEPT ![self] = lock]
            /\ pc' = [pc EXCEPT ![self] = "A2"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, regD, 
                            regE >>

A2(self) == /\ pc[self] = "A2"
            /\ regD' = [regD EXCEPT ![self] = regB[self] = self]
            /\ pc' = [pc EXCEPT ![self] = "A3"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, regC, 
                            regE >>

A3(self) == /\ pc[self] = "A3"
            /\ regE' = [regE EXCEPT ![self] = regC[self] /\ regD[self]]
            /\ IF regE'[self]
                  THEN /\ pc' = [pc EXCEPT ![self] = "A4"]
                  ELSE /\ pc' = [pc EXCEPT ![self] = "TAS"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, regC, 
                            regD >>

A4(self) == /\ pc[self] = "A4"
            /\ regA' = [regA EXCEPT ![self] = regA[self] + 1]
            /\ pc' = [pc EXCEPT ![self] = "A5"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regB, regC, regD, 
                            regE >>

A5(self) == /\ pc[self] = "A5"
            /\ cnt' = regA[self]
            /\ pc' = [pc EXCEPT ![self] = "A8"]
            /\ UNCHANGED << locking, lock, id, stack, regA, regB, regC, regD, 
                            regE >>

TAS(self) == /\ pc[self] = "TAS"
             /\ lock = FALSE
             /\ lock' = TRUE
             /\ locking' = (locking \union {self})
             /\ pc' = [pc EXCEPT ![self] = "A6"]
             /\ UNCHANGED << id, cnt, stack, regA, regB, regC, regD, regE >>

A6(self) == /\ pc[self] = "A6"
            /\ id' = self
            /\ pc' = [pc EXCEPT ![self] = "A7"]
            /\ UNCHANGED << locking, lock, cnt, stack, regA, regB, regC, regD, 
                            regE >>

A7(self) == /\ pc[self] = "A7"
            /\ cnt' = cnt + 1
            /\ pc' = [pc EXCEPT ![self] = "A8"]
            /\ UNCHANGED << locking, lock, id, stack, regA, regB, regC, regD, 
                            regE >>

A8(self) == /\ pc[self] = "A8"
            /\ pc' = [pc EXCEPT ![self] = Head(stack[self]).pc]
            /\ regA' = [regA EXCEPT ![self] = Head(stack[self]).regA]
            /\ regB' = [regB EXCEPT ![self] = Head(stack[self]).regB]
            /\ regC' = [regC EXCEPT ![self] = Head(stack[self]).regC]
            /\ regD' = [regD EXCEPT ![self] = Head(stack[self]).regD]
            /\ regE' = [regE EXCEPT ![self] = Head(stack[self]).regE]
            /\ stack' = [stack EXCEPT ![self] = Tail(stack[self])]
            /\ UNCHANGED << locking, lock, id, cnt >>

acquire(self) == Acquire(self) \/ A0(self) \/ A1(self) \/ A2(self)
                    \/ A3(self) \/ A4(self) \/ A5(self) \/ TAS(self)
                    \/ A6(self) \/ A7(self) \/ A8(self)

Release(self) == /\ pc[self] = "Release"
                 /\ cnt' = cnt - 1
                 /\ pc' = [pc EXCEPT ![self] = "R0"]
                 /\ UNCHANGED << locking, lock, id, stack, regA, regB, regC, 
                                 regD, regE >>

R0(self) == /\ pc[self] = "R0"
            /\ IF cnt = 0
                  THEN /\ pc' = [pc EXCEPT ![self] = "R1"]
                  ELSE /\ pc' = [pc EXCEPT ![self] = "R4"]
            /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, regC, 
                            regD, regE >>

R1(self) == /\ pc[self] = "R1"
            /\ id' = 0
            /\ pc' = [pc EXCEPT ![self] = "R2"]
            /\ UNCHANGED << locking, lock, cnt, stack, regA, regB, regC, regD, 
                            regE >>

R2(self) == /\ pc[self] = "R2"
            /\ lock' = FALSE
            /\ locking' = locking \ {self}
            /\ pc' = [pc EXCEPT ![self] = "R4"]
            /\ UNCHANGED << id, cnt, stack, regA, regB, regC, regD, regE >>

R4(self) == /\ pc[self] = "R4"
            /\ pc' = [pc EXCEPT ![self] = Head(stack[self]).pc]
            /\ stack' = [stack EXCEPT ![self] = Tail(stack[self])]
            /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                            regE >>

release(self) == Release(self) \/ R0(self) \/ R1(self) \/ R2(self)
                    \/ R4(self)

RecLock(self) == /\ pc[self] = "RecLock"
                 /\ \/ /\ pc' = [pc EXCEPT ![self] = "Rec"]
                    \/ /\ pc' = [pc EXCEPT ![self] = "NotRec"]
                 /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, 
                                 regC, regD, regE >>

Rec(self) == /\ pc[self] = "Rec"
             /\ IF cnt < MAX_REC
                   THEN /\ pc' = [pc EXCEPT ![self] = "RL0"]
                   ELSE /\ pc' = [pc EXCEPT ![self] = "RL1"]
             /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, regC, 
                             regD, regE >>

RL0(self) == /\ pc[self] = "RL0"
             /\ stack' = [stack EXCEPT ![self] = << [ procedure |->  "lock_unlock",
                                                      pc        |->  "RL1" ] >>
                                                  \o stack[self]]
             /\ pc' = [pc EXCEPT ![self] = "Lock"]
             /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                             regE >>

NotRec(self) == /\ pc[self] = "NotRec"
                /\ TRUE
                /\ pc' = [pc EXCEPT ![self] = "RL1"]
                /\ UNCHANGED << locking, lock, id, cnt, stack, regA, regB, 
                                regC, regD, regE >>

RL1(self) == /\ pc[self] = "RL1"
             /\ pc' = [pc EXCEPT ![self] = Head(stack[self]).pc]
             /\ stack' = [stack EXCEPT ![self] = Tail(stack[self])]
             /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                             regE >>

recursive_lock(self) == RecLock(self) \/ Rec(self) \/ RL0(self)
                           \/ NotRec(self) \/ RL1(self)

Lock(self) == /\ pc[self] = "Lock"
              /\ stack' = [stack EXCEPT ![self] = << [ procedure |->  "acquire",
                                                       pc        |->  "L0",
                                                       regA      |->  regA[self],
                                                       regB      |->  regB[self],
                                                       regC      |->  regC[self],
                                                       regD      |->  regD[self],
                                                       regE      |->  regE[self] ] >>
                                                   \o stack[self]]
              /\ regA' = [regA EXCEPT ![self] = defaultInitValue]
              /\ regB' = [regB EXCEPT ![self] = defaultInitValue]
              /\ regC' = [regC EXCEPT ![self] = defaultInitValue]
              /\ regD' = [regD EXCEPT ![self] = defaultInitValue]
              /\ regE' = [regE EXCEPT ![self] = defaultInitValue]
              /\ pc' = [pc EXCEPT ![self] = "Acquire"]
              /\ UNCHANGED << locking, lock, id, cnt >>

L0(self) == /\ pc[self] = "L0"
            /\ stack' = [stack EXCEPT ![self] = << [ procedure |->  "recursive_lock",
                                                     pc        |->  "L1" ] >>
                                                 \o stack[self]]
            /\ pc' = [pc EXCEPT ![self] = "RecLock"]
            /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                            regE >>

L1(self) == /\ pc[self] = "L1"
            /\ stack' = [stack EXCEPT ![self] = << [ procedure |->  "release",
                                                     pc        |->  "L2" ] >>
                                                 \o stack[self]]
            /\ pc' = [pc EXCEPT ![self] = "Release"]
            /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                            regE >>

L2(self) == /\ pc[self] = "L2"
            /\ pc' = [pc EXCEPT ![self] = Head(stack[self]).pc]
            /\ stack' = [stack EXCEPT ![self] = Tail(stack[self])]
            /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                            regE >>

lock_unlock(self) == Lock(self) \/ L0(self) \/ L1(self) \/ L2(self)

T1(self) == /\ pc[self] = "T1"
            /\ stack' = [stack EXCEPT ![self] = << [ procedure |->  "lock_unlock",
                                                     pc        |->  "Done" ] >>
                                                 \o stack[self]]
            /\ pc' = [pc EXCEPT ![self] = "Lock"]
            /\ UNCHANGED << locking, lock, id, cnt, regA, regB, regC, regD, 
                            regE >>

pid(self) == T1(self)

(* Allow infinite stuttering to prevent deadlock on termination. *)
Terminating == /\ \A self \in ProcSet: pc[self] = "Done"
               /\ UNCHANGED vars

Next == (\E self \in ProcSet:  \/ acquire(self) \/ release(self)
                               \/ recursive_lock(self) \/ lock_unlock(self))
           \/ (\E self \in PIDS: pid(self))
           \/ Terminating

Spec == /\ Init /\ [][Next]_vars
        /\ \A self \in PIDS : /\ WF_vars(pid(self))
                              /\ WF_vars(lock_unlock(self))
                              /\ WF_vars(acquire(self))
                              /\ WF_vars(release(self))
                              /\ WF_vars(recursive_lock(self))

Termination == <>(\A self \in ProcSet: pc[self] = "Done")

\* END TRANSLATION 
TypeInvariant ==
    /\ MAX_REC \in Int
    /\ PIDS \subseteq Int
=============================================================================
\* Modification History
\* Last modified Mon Nov 22 09:29:41 JST 2021 by ytakano
\* Created Sun Nov 21 22:09:25 JST 2021 by ytakano

実行結果

[kiripon]

C11

本書執筆時は、C11かC99かで迷ったのですが、C11で正確に記載するには、メモリバリアの説明が必要となります。しかし、メモリバリアはアドバンスなトピックであるため、中盤以降に持ってきたかったので、GCC拡張を利用しています。

なるほど、そういう意図があったのですね。冒頭でメモリバリアの説明をするような怖い本だと自分は手に取れてなかった気がするので、そこのハードルを下げる構成になっていてすごくありがたいです！
__atomic_** 系の組み込み関数を使っていないのもこの時点でメモリオーダーの説明を避けるために考えた結果なんですね…

lock->id について

上記のどのステップでlock->idを読み込んでも、0x10にはならない。ありあえるのは、0x00か0x11のみ。

自分の言いたいことが伝わっていないような気がしています。
ちゃんと例を出して説明すべきでした。

lock->id == id にある、 lock->id の読み込みは atomic な操作ではありません。
読み込み操作が複数の操作に分割される可能性があることを意味します。
例は、 0x00 から 0x11 への変化の途中過程が検出されるような場合に 0x10 が見えることがあるということを言っています。

起こると考えている例

lock->id == id は、以下のような操作に分割される可能性があります。

1. lock->id の下位4bit を読み込む
2. lock->id の上位4bit を読み込む
3. lock->id の下位4bit と, id の下位4bit の比較
4. lock->id の上位4bit と, id の上位4bit の比較

自分が問題にしているのは、1 と 2 の間に行われる別のCPUからの代入です。
以下のような順序で操作が行われると、 lock->id から読み取った値が 0x10 になりえると考えています。

1. CPU A: lock->id の下位4bit を読み込む
2. CPU B: lock->id に 0x11 を書き込む (atomic)
3. CPU A: lock->id の上位4bit を読み込む
4. CPU A: lock->id の下位4bit と, id の下位4bit の比較
5. CPU A: lock->id の上位4bit と, id の上位4bit の比較

lock->id から読み取った値を保持する変数を tmp とすると、それぞれの変数の状態の変化は

0. 初期状態
   • tmp = 0x00
   • lock->id = 0x00
   • id = 0x10
1. lock->id の下位4bit を読み込む
   • tmp = 0x00
   • lock->id = 0x00
   • id = 0x10
2. 別のCPUが lock->id に 0x11 を書き込む (atomic)
   • tmp = 0x00
   • lock->id = 0x11
   • id = 0x10
3. lock->id の上位4bit を読み込む
   • tmp = 0x10
   • lock->id = 0x11
   • id = 0x10
4. lock->id の下位4bit と, id の下位4bit の比較
   • tmp == id なので true
5. lock->id の上位4bit と, id の上位4bit の比較
   • tmp == id なので true

となります。

TLA+ の検証に引っかからなかった理由

自分はTLA+については初見なのであまり自信はないのですが、lock->id の読み込みが atomic とみなされているのではないでしょうか？ （対応するのは A0: regB := id; の行？）
もしこの代入の競合までチェックされるようになっていたとしたら自分の指摘が誤りです！
時間を取らせて申し訳ありません！

たとえば id を上位ビットと下位ビットを表す2つの変数に分割すると競合を模倣できるのではないかと思います。
（本来こういうことは自分の手元で動かしてからいうべきなのですが、 TLA+ の環境構築に時間がかかっていてまだ検証器を回せていないです…）

修正案

C11 の場合

https://en.cppreference.com/w/c/atomic/atomic_load を用いて、
atomic_load_explicit(&lock->id, memory_order_relaxed) == id
とする。
ただ、C11 の使用の説明ができないのでこの案は使えなさそうです。

コンパイラ組み込み関数を使う場合

https://gcc.gnu.org/onlinedocs/gcc/_005f_005fatomic-Builtins.html
__atomic_load_n(&lock->id,__ATOMIC_RELAXED) == id とする？

（この場合は __ATOMIC_RELAXED というメモリオーダー指定が入ってしまうので、メモリバリアの説明前に入れるには良くないですが…）

参考にした文献

• https://yohhoy.hatenadiary.jp/entry/20121016/p1

  • volatile変数ではダメだという反例を挙げよ
    volatile変数への読み書きは不可分(atomic)操作であるとは保証されない。仮にレジスタサイズ2byte長のマシンを想定した場合、同環境では4byteサイズのvolatile変数アクセスはおそらく2回の2byteメモリアクセスに分割される。このとき異なるスレッドからvolatile変数へ読み／書きを同時に行うと、4byte領域のうち半分だけ更新された中間の値を読み込む可能性がある。つまり、プログラムの動作上は “どこにも存在しない値” を観測してしまうリスクがある。

  • volatile についての保証ですが、通常の変数についても言える説明かと思います。

[ytakano]

確かに、上位4ビットと下位4ビットで操作が分割されるようなCPUアーキテクチャの場合は、おこりえそうです。

__atomic_load_n(&lock->id, __ATOMIC_RELAXED) == id
が良さそうなので、次の増刷があれば、こちらに修正したいと思います。詳しい解析ありがとうございます！

https://yohhoy.hatenadiary.jp/entry/20121016/p1
こちらのブログで指摘されている状況は、intとx86_64かAArch64を用いている限りそのようなことは起きないとは思います。

AArch64だと、たとえば、0x1234_5678_9abc_def1のような定数値のメモリ書き込みはアトミックではなく複数回のメモリ書き込みになりますが、その他のメモリ読み込みとレジスタからの書き込みはアトミックになります。x86_64の場合はCISCなのでこれは起きなさそうに思います。

アトミック処理を含む現代的なCPUだと問題なさそうな気もしますが、__atomic_load_n()を使うのが無難に思えますね。ありがとうございます。

[kiripon]

https://yohhoy.hatenadiary.jp/entry/20121016/p1
こちらのブログで指摘されている状況は、intとx86_64かAArch64を用いている限りそのようなことは起きないとは思います。

はい。自分も現実的な環境では発生しないように思います。
（言語仕様を持ち出したのは、 aarch64 と x86_64 では問題がないが移植性はないコードになっているということをいうためでした。わかりにくくてすみません… 自分がここを気にしているのは、これら２つの命令セットを題材に正しい並列プログラミングの手法を伝えるというコンセプトの本だと思ったためです。）

その他のメモリ読み込みとレジスタからの書き込みはアトミックになります。

自分もデータがアラインメントされている限りは読み込みは atomic な振る舞いになると思っているのですが、その保証がまだ見つけられていません。
一応、LD64B のような 64bit atomic ロード専用命令もあるようなのでひょっとしたら LDR 命令には保証がないのかもしれないと疑っています。 こちらの命令は 64bit ではなく64byte でした。すみません。
https://developer.arm.com/documentation/ddi0596/2021-09/Base-Instructions/LD64B--Single-copy-Atomic-64-byte-Load-

命令レベルの問題のほかに気にしていたのはコンパイラの最適化です。
最適化によって lock->lock へのロードが失われてしまう危険性もあるかと思います。
（コンパイラの組み込み関数や、atomic なプリミティブを使う場合は最適化が抑制されるので問題ないです。）

lock->lock 変数についてもデータ競合は起きているので、 atomic ではないロードは良くないといえばよくないのですが

• スレッドID さえ atomic に見ていれば lock->lock が true かどうかのチェックはなくても正しく動作する
• 所詮 1bit の状態を保持しているだけなので(最適化を考慮しなければ) atomic になるはず（?）

という理由で問題はないと思います。
この後の Rust のスピンロック実装では Relaxed なメモリオーダーでのアトミック操作になっているので、こちらも atomic だと対応関係がとれるとは思います。ですが、test and test-and-set イディオムの最初の test だと考えると問題ないような気もします。(非 atomic に対する test として真面目にやるなら最適化に対する配慮も必要かもですが）
この辺に関しての問題は、TTAS で調べるよりもほぼ同じ問題の Double Checked Locking Problem とかで調べるとかなり情報が見つかると思います。

[ytakano]

ありがとうございます！

以下のようにコードを修正したいと思います。また、増刷があれば書籍にも反映したいと思います。

ytakano@c1c6bb3#diff-fb9cce8092ac6c6fa7da708205bc019ce70ac2ad0bec59426a0d9d8aff9ff437

並行プログラミングは難しいトピックで、間違いもまだまだあるとは思いますが、これに懲りずご指摘頂ければと思います。

[kiripon]

コードの修正を確認しました。問題ないと思います。
メモリバリアについても lock->lock 変数のアクセスを介して適切な順序付けがされているので問題ないようです。
（強いて言うならばロック獲得済みかどうかは lock->id == 0 で確認できるので lock->lock 変数はもはや不要なのですが、そこまで修正すると本文まで書き換えることになるのできついですね）

自分の指摘に長々付き合って頂いてありがとうございます。
また何か気になる点を見つけたら報告するようにします。

知り合いの間で本読みをしていますが、並行プログラミングの世界の概要が一冊で把握できる貴重な本だと感じています。
面白い本を書いていただいてありがとうございました！

[ytakano]

はい。lock->idのみでも大丈夫そうですね。機会があれば修正したいと思います。

こちらこそ、大変詳しく読んで頂きありがとうございます。
そう言って頂けると、励みになります！