パン屋のアルゴリズムを理解する

パン屋のアルゴリズムとは、並行プログラミングにおいて、安全に相互排除を行うためのアルゴリズムです。
特に、アトミック処理を前提としていない場合用いられます。。

並行プログラミング入門を読んでいて、詰まったので自分なりに整理します。

アトミック処理

アトミック処理とは、別スレッドからの干渉を受けずに実行することのできる処理です。

例えば、以下のようなcompare_and_swap関数を考えたときに、この処理はアトミック処理でしょうか？

うまくいかないコード

#include<stdint.h>
bool compare_and_swap(uint64_t *p, uint64_t val, uint64_t newval)  {
    if(*p != val) {
        return false;
    }
    *p = newval;
    return true;
}

//x86-64 gcc
compare_and_swap(unsigned long*, unsigned long, unsigned long):
        push    rbp
        mov     rbp, rsp
        mov     QWORD PTR [rbp-8], rdi
        mov     QWORD PTR [rbp-16], rsi
        mov     QWORD PTR [rbp-24], rdx
        mov     rax, QWORD PTR [rbp-8]
        mov     rax, QWORD PTR [rax]
        cmp     QWORD PTR [rbp-16], rax
        je      .L2
        mov     eax, 0
        jmp     .L3
.L2:
        mov     rax, QWORD PTR [rbp-8]
        mov     rdx, QWORD PTR [rbp-24]
        mov     QWORD PTR [rax], rdx
        mov     eax, 1
.L3:
        pop     rbp
        ret

一般的には、違うみたいです。
C言語で記した処理は、コンパイラ(gcc)でアセンブリコードにコンパイルすると、実際は複数の操作を組み合わせて実現されています。
従って、別スレッドからの干渉を受けずに実行できる処理ではない為、アトミック処理とは一般的には言えません。

うまくいくコード

しかし、最近のプログラミング言語ではアトミックに処理する機構が多く用意されています。
例えば、上のcompare_and_swapをアトミック処理する組み込み関数、__sync_bool_compare_and_swapが用意されています。
https://gcc.gnu.org/onlinedocs/gcc-4.1.0/gcc/Atomic-Builtins.html

#include<stdint.h>
bool compare_and_swap(uint64_t *p, uint64_t val, uint64_t newval)  {
    return __sync_bool_compare_and_swap(p, val, newval);
}

//x86-64 gcc
compare_and_swap(unsigned long*, unsigned long, unsigned long):
        push    rbp
        mov     rbp, rsp
        mov     QWORD PTR [rbp-8], rdi
        mov     QWORD PTR [rbp-16], rsi
        mov     QWORD PTR [rbp-24], rdx
        mov     rdx, QWORD PTR [rbp-8]
        mov     rax, QWORD PTR [rbp-16]
        mov     rcx, QWORD PTR [rbp-24]
        lock cmpxchg    QWORD PTR [rdx], rcx
        sete    al
        pop     rbp
        ret

lock cmpxchg

lock を用いて、cmpxchg命令を行うことで、アトミック処理を行います。
https://docs.microsoft.com/ja-jp/windows-hardware/drivers/debugger/x86-instructions

ミューテックス

ミューテックスは、並行プログラミングの際に同時に実行されているプログラム間で、リソースの排他制御、同期を行う仕組みの一つ。
クリティカルセクションを高々一つしか持たない。

うまくいかないコード

以下のコードだと、クリティカルセクションを同時に実行してしまう可能性があり、排他制御ができない。

bool lock = false;

void some_func() {
    retry:
    if(!lock) {
        lock = true;
    } else {
        goto retry;
    }
    lock = false;
}

//x86-64 gcc
lock:
        .zero   1
some_func():
        push    rbp
        mov     rbp, rsp
.L2:
        movzx   eax, BYTE PTR lock[rip]
        xor     eax, 1
        test    al, al
        je      .L2
        mov     BYTE PTR lock[rip], 1
        mov     BYTE PTR lock[rip], 0
        nop
        pop     rbp
        ret

うまくいくコード

以下のコードだと、クリティカルセクションを同時に実行してしまう可能性がなく、排他制御できる。

bool lock = false;

void some_func() {
    retry:
    if(!__sync_lock_test_and_set(&lock, 1)) {
        lock = true;
    } else {
        goto retry;
    }
    __sync_lock_release(&lock);
}

//x86-64 gcc
lock:
        .zero   1
some_func():
        push    rbp
        mov     rbp, rsp
.L2:
        mov     eax, 1
        xchg    al, BYTE PTR lock[rip]
        test    al, al
        sete    al
        test    al, al
        je      .L2
        mov     BYTE PTR lock[rip], 1
        mov     eax, 0
        mov     BYTE PTR lock[rip], al
        nop
        pop     rbp
        ret

セマフォ

ミューテックスは、並行プログラミングの際に同時に実行されているプログラム間で、リソースの排他制御、同期を行う仕組みの一つ。
クリティカルセクションをN個持つ。

Lamport's bakery algorithm

参考1
参考2
Lamport's bakery algorithm は、アトミック命令をサポートしない場合の同期処理手法です。
上に出てきた、アトミック処理をベースとしたミューテックス、セマフォ、RWロック等の手法が採用できない場合に用います。

実際にちゃんと動くソースコードは、著者様の参考1を確認して下さい。
今回は、メモリバリアを行わないで動かしてみた場合の結果を見てみようと思います。

unsafeなパン屋アルゴリズムと実行結果

use std::ptr::{read_volatile, write_volatile};
use std::thread;

const NUM_THREADS: usize = 4;
const NUM_LOOP: usize = 100000;

struct BakeryLock {
    entering: [bool; NUM_THREADS],
    tickets: [Option<u64>; NUM_THREADS],
}

impl BakeryLock {
    fn lock(&mut self, idx: usize) -> LockGuard {
        
        unsafe { write_volatile(&mut self.entering[idx], true)};
        

        let mut max = 0;
        for i in 0..NUM_THREADS {
            if let Some(t) = unsafe { read_volatile(&self.tickets[i])} {
                max = max.max(t);
            }
        }

        let ticket = max + 1;
        unsafe { write_volatile(&mut self.tickets[idx], Some(ticket))};

        
        unsafe { write_volatile(&mut self.entering[idx], false)};
        
        for i in 0..NUM_THREADS {
            if i == idx {
                continue;
            }

            while unsafe { read_volatile(&self.entering[i])} {}

            loop {
                match unsafe { read_volatile(&self.tickets[i])} {
                    Some(t) => {
                        if ticket < t || (ticket == t && idx < i) {
                            break;
                        }
                    }
                    None => {
                        break;
                    }
                }
            }
        }
        
        LockGuard { idx }
    }
}

struct LockGuard {
    idx: usize,
}

impl Drop for LockGuard {
    fn drop(&mut self) {
        
        unsafe { write_volatile(&mut LOCK.tickets[self.idx], None)};
    }
}

static mut LOCK: BakeryLock = BakeryLock {
    entering: [false; NUM_THREADS],
    tickets: [None; NUM_THREADS],
};

static mut COUNT: u64 = 0;

fn main() {
    let mut v = Vec::new();
    for i in 0..NUM_THREADS {
        let th = thread::spawn(move || {
            for _ in 0..NUM_LOOP {
                let _lock = unsafe { LOCK.lock(i) };
                unsafe {
                    let c = read_volatile(&COUNT);
                    write_volatile(&mut COUNT, c + 1);
                }
            }
        });
        v.push(th);
    }
    for th in v {
        th.join().unwrap();
    }

    println!(
        "COUNT = {} (expected = {})",
        unsafe { COUNT },
        NUM_LOOP * NUM_THREADS
    );
}

このコードでは、参考1と比べて、

メモリバリアをしない
read_volatile, write_volatileをマクロ化しない

ことをしています。

実際に、cargo run --release してみると、実行結果が不安定になります。

参考