RCU(Read Copy Update)をちゃんと知る（2）-１ RCUにおけるポインタの更新と参照

はじめに

前回(http://tkokamo.hateblo.jp/entry/2017/08/12/191142)で概要は終わりにし、 今回からはコードを見ていこうと思います。

大体以下のような流れでRCUの実装を見ていこうかと思います。

• rcu_dereference(), rcu_assign_pointer() ← 今回
• Classic RCU
• Tree RCU

rcu_read_lock()、rcu_read_unlock()はプリエンプションの無効化/有効化である、ということは前回述べたので触れないです。

RCUでのポインタの更新（と参照）

概要(http://tkokamo.hateblo.jp/entry/2017/05/31/213427)で述べたように、
RCUにおけるデータの更新は、更新後データの初期化を行った後、古いデータを指していたポインタを新しいデータに置き換えることで行われます。（下図参照）

更新後データは、値の初期化後にポインタによって指されるため、参照側に中途半端な状態(更新中のデータ)が見えることは無いはずですが、
現在のシステムでこれを保証するには少しだけ細工をしなければいけません。

データの初期化後にポインタで指す、という例として以下にコードを示します。

struct foo {
   int a;
   int b;
};

struct foo *foop = NULL;
struct foo new;

/*foop!=NULLで値を出力*/
void reader()
{
    int a, b;
    if (foop != NULL) {
        a = foop->a;
        b = foop->b;
        printk("Set. a = %d, b = %d\n", a, b);
    }
    return;
}

/*newを初期化しfoopにセット*/
void writer()
{
    new.a = 1;     // (1)
    new.b = 1;     // (2)
    foop = &new;   // (3)
    return;
}

reader()は、writer()によりnewがfoopにセットされていればその値を読み出し出力する、という関数ですが、
writer()の(1), (2), (3)の順序が変わってしまう場合には、Set. a = 1, b = 1が出力されるとは限りません。
そして、これらは以下の2つのメモリアクセス順序の変更によって実際に起きることがあります。

• コンパイラによるメモリアクセス順序の変更
  静的なリオーダリング
• CPUによるメモリアクセス順序の変更（目に見えるのはSMP）
  動的なリオーダリング
  

RCUにおけるポインタの更新と参照は、上記2種類のリオーダリングを適切に防ぐことで実装されています。 以下ではそれぞれのリオーダリングについて見たあとに、rcu_dereference()、rcu_assign_pointer()の実装を見てみます。

コンパイラによるメモリアクセス順序の変更

人間はしばしばおバカなコードを書いてしまい、メモリアクセスの観点から効率が悪いことがありますが、
コンパイラはその尻拭いをしてくれます。

例を下に示します。

// おばかなコード
#include <stdio.h> 
 
int main() 
{ 
        unsigned long arr[16]; 
        int i; 
 
        arr[0] = 1; 
        arr[8] = 1; 
        arr[1] = 1; 
        arr[9] = 1; 
        arr[2] = 1; 
        arr[10] = 1; 
        arr[3] = 1; 
        arr[11] = 1; 
        arr[4] = 1; 
        arr[12] = 1; 
        arr[5] = 1; 
        arr[13] = 1; 
        arr[6] = 1; 
        arr[14] = 1; 
        arr[7] = 1; 
        arr[15] = 1; 
 
        for (i = 0; i < 16; i++) 
                printf("%lu", arr[i]); 
 
        printf("\n"); 
        return 0; 
}

上のコードは二つのキャッシュラインを交互に見て1を代入しているおばかなコードです。（家の環境のi7-2600は確かキャッシュラインサイズが64B）
これをgccで最適化なしでコンパイルすると、、、

; 最適化なしの代入コード
        movq    $1, -144(%rbp)  ;arr[0] = 1
        movq    $1, -80(%rbp)   ;arr[8] = 1
        movq    $1, -136(%rbp)  ;arr[1] = 1
        movq    $1, -72(%rbp)   ;arr[9] = 1
...

馬鹿な人間に忠実なコードを吐いてくれますが、-O3をつけると

; -O3付き
        movdqa  %xmm0, (%rsp)
        movdqa  %xmm0, 16(%rsp)
        movdqa  %xmm0, 32(%rsp)
...

となります。movdqa命令はいわゆるSIMD命令です。
-O3を付けてコンパイルをすると、配列アクセスがプログラマが書いたようにではなく、ストリーム的にアクセスされるよう最適化されています(ストリームというにはしょぼいか)。
上記のようなコードであれば、特に意味も変わりませんが、代入の順番が大きな意味を持つ場合、最適化されてしまっては困ります。

gccではasm volatile("" ::: "memory");を入れることで、その前後のメモリアクセス順序がコンパイラによって変更されることを防ぐことができます。
今回の例の各代入文の間にasm volatile("" ::: "memory");を入れると、-O3を付けたとしても、メモリアクセス順序が人間の意図したままに維持されます。（興味があれば試してみてください）

CPUによるメモリアクセス順序の変更

こちらの方が少し理解に苦しむかも知れませんが、コンパイラ同様CPUもメモリアクセス順序を変更し、プログラム実行の最適化を図っています。
CPUによるリオーダリングが効いてくる例を見てみます。

// mem.c
#include <stdio.h>
#include <pthread.h>

struct arg {
        int a;
        int b;
        int c;
        int d;
};

void *t1_func(void *p)
{
    struct arg *arg = (struct arg *)p;

    arg->a = 1;
    if (arg->b == 0)
        arg->c = 1;

    return NULL;
}

void *t2_func(void *p)
{
    struct arg *arg = (struct arg *)p;

    arg->b = 1;
    if (arg->a == 0)
        arg->d = 1;
    
    return NULL;
}

int main()
{
    pthread_t t1, t2;
    struct arg arg;       

        /* 全部0で初期化 */
    arg.a = 0;
    arg.b = 0;
    arg.c = 0;
    arg.d = 0;

        /* スレッド生成 */
    pthread_create(&t1, NULL, t1_func, &arg);
    pthread_create(&t2, NULL, t2_func, &arg);

    /* スレッド終了待ち */
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);

        /* a,b,c,dを出力 */
    printf("%d%d%d%d\n", arg.a, arg.b, arg.c, arg.d);
    return 0;
}

上記コードは、a,b,c,dを0で初期化したのちに2つのスレッドを生成し、各スレッドは以下の処理を行います。

// スレッド1
    arg->a = 1;
    if (arg->b == 0)     // (1)
        arg->c = 1;
// スレッド2
    arg->b = 1;
    if (arg->a == 0)     // (2)
        arg->d = 1;

親スレッドはスレッド1、スレッド2の終了後、a,b,c,dの値を出力しますが、
このコードは出力が1110または1101になることを期待しています。
その理由は以下の通りです。

• (1)が成立するということは、スレッド1のarg->a = 1は実行されているが、スレッド2のarg->b = 1は実行されていない状態である。
  この時、スレッド2の(2)は成り立たず、4値の出力は1110となる。

• (2)が成立するということは、スレッド2のarg->b = 1は実行されているが、スレッド2のarg->a = 1は実行されていない状態である。
  この時、スレッド1の(1)は成り立たず、4値の出力は1101となる。

しかし、SMP環境では1111が出力されてしまうことがあります。
以下では、上記コードを10000回実行し、出力が1111となってしまう回数を調べています。

答えが先になってしまいますが、1111が出力される理由はCPUによってメモリアクセス順序が変更されているからです。
下では、上記コードに対し、特定箇所にコンパイラ/CPUによるリオーダリングを禁止する命令を追加した場合、
シングルコアで上記コードを実行した場合に1111がどれくらい出力されてしまうかも合わせて示しています。

◯ 上で示したコード（SMP環境）
# for i in `seq 1 10000`; do ./mem >> mem.log; done
# grep 1111 mem.log | wc
    470     470    2350  ★470回1111が出力された

◯ 上で示したコード（シングルコア）
# echo 0 > /sys/devices/system/cpu/cpu1/online
# echo 0 > /sys/devices/system/cpu/cpu2/online
# echo 0 > /sys/devices/system/cpu/cpu3/online
# echo 0 > /sys/devices/system/cpu/cpu4/online
# echo 0 > /sys/devices/system/cpu/cpu5/online
# echo 0 > /sys/devices/system/cpu/cpu6/online
# echo 0 > /sys/devices/system/cpu/cpu7/online
# lscpu | grep On-line
On-line CPU(s) list:   0
# for i in `seq 1 10000`; do ./mem >> mem_uni.log; done
# grep 1111 mem_uni.log | wc
      0       0       0  ★1111は出力されなかった

◯ 特定箇所のコンパイラによるメモリリオーダリングを禁止したコード
# for i in `seq 1 10000`; do ./mem_comp >> mem_comp.log; done
# grep 1111 mem_comp.log | wc
    487     487    2435  ★487回1111が出力された

◯ 特定箇所のCPUによるメモリリオーダリングを禁止したコード
# for i in `seq 1 10000`; do ./mem_cpu >> mem_cpu.log; done
# grep 1111 mem_cpu.log | wc
      0       0       0  ★1111は出力されなかった

出力結果が1111となってしまう、ということはスレッド1の(1)、スレッド2の(2)が成り立ってしまった、ということです。
そして、この原因はCPUによるメモリアクセス順序の変更にあります。

/*再掲*/
// スレッド1
    arg->a = 1;
    if (arg->b == 0)     // (1)
        arg->c = 1;
// スレッド2
    arg->b = 1;
    if (arg->a == 0)     // (2)
        arg->d = 1;

SMP環境において、CPUは依存関係の無い二つのメモリアドレスに対するアクセス順序を動的に変更し、プログラム実行を高速化しています。
メモリアクセスはloadかstoreですから、依存関係の無い二つのメモリアドレスに関してload -> load、load -> store、store -> load、store -> storeの4種類のどのアクセス順序を変更して良いか、というポリシーが各CPUで決められています。
x86は比較的メモリアクセス順序を守るほうなのですが、store -> loadに関してはその順序を入れ替えても良い（コア間で見え方が違っても良い）、というポリシーになっています。(そうでないx86のCPUもあります)
上記の例では、このリオーダリングが見事に現れました。
つまり、スレッド1とスレッド2が別コアで動いた時、それぞれのarg->a = 1、arg->b = 1は互いに反映される前に、loadが先行され(1)、(2)が成立してしまったわけです。
コンパイラ同様、CPUによるメモリアクセスのリオーダリングを防ぐ方法もあり、asm volatile("mfence" ::: "memory");がこれに該当します。
この1行を以下の箇所に入れることで、期待通り1111の出力を防ぐことができます。

// mem.cの改良
void *t1_func(void *p)
{
...
    arg->a = 1;
    asm volatile("mfence" ::: "memory");
    if (arg->b == 0)
...
}

void *t2_func(void *p)
{
...
    arg->b = 1;
    asm volatile("mfence" ::: "memory");
    if (arg->a == 0)
...
}

CPUのメモリアクセスのリオーダリングなんて無い（プログラムからは順番どおりに見える）ほうが良いじゃないか！という気持ちはわかりますが、
異なるコア間、ましてやNUMAなどの異なるチップ感でメモリの見え方を完全に一致させるには、同期を行うコストが大きくなり、性能に与える影響も小さくありません。
中国の「神威・太湖之光」やpezy computingのプロセッサでは、CPUによるリオーダリングを自由にし、プログラマが必要な箇所のみで同期を行うことで、データ並列性の高いアプリケーションでの実行速度を上げているようです。

でRCUではなにやってんの？

かなり寄り道してしまった感もありますが、再び最初に見たコードを思い出してみましょう。

struct foo {
   int a;
   int b;
};

struct foo *foop = NULL;
struct foo new;

/*foop!=NULLで値を出力*/
void reader()
{
    int a, b;
    if (foop != NULL) {
        a = foop->a;
        b = foop->b;
        printk("Set. a = %d, b = %d\n", a, b);
    }
    return;
}

/*newを初期化しfoopにセット*/
void writer()
{
    new.a = 1;
    new.b = 1;
    foop = &new;
    return;
}

上記コードでは、参照時にfoop->a、foop->bが1になっていない可能性がある、ということを冒頭で述べました。
x86のCPUを前提とした場合、どうすればこれを防げるか、というと、、、

struct foo {
   int a;
   int b;
};

struct foo *foop = NULL;
struct foo new;

/*foop!=NULLで値を出力*/
void reader()
{
    int a, b;
    if (foop != NULL) {
        a = foop->a;
        b = foop->b;
        printk("Set. a = %d, b = %d\n", a, b);
    }
    return;
}

/*newを初期化しfoopにセット*/
void writer()
{
    new.a = 1;   // (1)
    new.b = 1;   // (2)
    asm volatile("" ::: "memory");  //★
    foop = &new; // (3)
    return;
}

asm volatile("" ::: "memory");がfoop = &new;の前に入っただけです。
「え〜っ」って感じですが、x86に関してはこれだけで、上記コードは思った通りに動作します。
以下その理由です。
x86のCPUは、store -> storeのメモリアクセス順序は変更しません。 つまり、writer()の(1), (2), (3)はコンパイラによってメモリ順序の変更が起きない限り、意図通りに動くわけです。
そのため、asm volatile("" ::: "memory");を(1)、(2)と(3)の間に入れ、この間のコンパイラによるメモリ順序の変更を防いでいます。

そして、何を隠そうこれがRCUにおけるアトミックなポインタの更新と参照になるのです。
※ アーキテクチャが異なると実装も異なります。

/*注意！！！以下は正しい使い方ではありません！！！*/

void reader()
{
    struct foo *p;
    int a, b;

    p = rcu_dereference(foop);  //★
    if (p != NULL) {
        a = p->a;
        b = p->b;
        printk("Set. a = %d, b = %d\n", a, b);
    }
    return;
}

void writer()
{
    new.a = 1;
    new.b = 1;
    rcu_assign_pointer(foop, &new); //★
    return;
}

では、念の為嘘をついていないことを確認しましょう。
以下はv4.3のコードです。

rcu_assign_pointer

#define rcu_assign_pointer(p, v) smp_store_release(&p, RCU_INITIALIZER(v))

#define smp_store_release(p, v)                                         \
do {                                                                    \
        compiletime_assert_atomic_type(*p);                             \
        // コンパイラによるリオーダリングを禁止
        // #define barrier() __asm__ __volatile__("": : :"memory")
        barrier();                                                      \
        // 代入
        WRITE_ONCE(*p, v);                                              \
} while (0)

rcu_dereference

#define rcu_dereference(p) rcu_dereference_check(p, 0)

#define rcu_dereference_check(p, c) \
        __rcu_dereference_check((p), (c) || rcu_read_lock_held(), __rcu)

#define __rcu_dereference_check(p, c, space) \
({ \
        /* Dependency order vs. p above. */ \
        typeof(*p) *________p1 = (typeof(*p) *__force)lockless_dereference(p); \ //★
        RCU_LOCKDEP_WARN(!(c), "suspicious rcu_dereference_check() usage"); \
        rcu_dereference_sparse(p, space); \
        ((typeof(*p) __force __kernel *)(________p1)); \
})

#define lockless_dereference(p) \
({ \
        // 代入しているだけ
        typeof(p) _________p1 = READ_ONCE(p); \
        // x86の場合からっぽ
        smp_read_barrier_depends(); /* Dependency order vs. p above. */ \
        (_________p1); \
})

（嘘を見つけた方はコメント頂けると大変助かります）

おわりに

もともとここはさらっと行こうと思ったのですが、考えてみたら、いつもコード書くときにメモリオーダリングとか全く意識しないなぁ、という感じだったので、自分の勉強がてら調べてみました。
今回はRCUというよりはプリミティブな部分の話でしたが、次回からはRCUっぽい話ができるかと思います。