Direct threaded code

2009-01-24
#c

Direct threadingってよく聞くんだけどどういうものか知らなかったので調べた。

Threaded code

  • threaded codeは、VMの実装をするときに命令のフェッチ→オペコードによって処理の分岐というのを、プログラム側に各オペコードの処理のアドレスを埋め込んで高速化する技のこと
  • threaded codeで実装するとコンパクト、だけど遅い。 でも小さいから場合によってはキャッシュミスが減って十分速いかも。
  • threaded codeはVMのバイトコードから実際の関数だかサブルーチンのアドレスを直接埋め込みをコンパイル時やプログラムロード時に行って、実行時は直接そのアドレスに飛ぶようにしたもの
    • しかしバイトコードよりは使う空間が大きくなる
  • direct threaded codeは、次の命令への分岐を先頭の一ヶ所で行わず、各オペコードの処理の最後で先頭に戻る代わりに次の命令に直接ジャンプする
  • YARV Maniacs 【第 3 回】 命令ディスパッチの高速化
    • threaded codeだと分岐予測がほぼはずれる、のでdirect threadedはその分有利

テストのプログラムを書いた（GCC拡張のgotoのラベルのアドレスを取る機能を使って）

/// VMのオペコード
typedef enum {
    HALT,
    LDA,
    DEC,
    JNZ,
} Opcode;

/// コード
typedef struct {
    union {
        Opcode op;
        const void* adr;
    } u;
    int operand;
} Code;

/// VM
typedef struct {
    struct {
        int a;
    } reg;
} Vm;

#ifndef DIRECT_THREADED

#define OP_SWITCH_LOOP      for (;;) { switch((pc++)->u.op) {
#define CASE(op)            case op:
#define BREAK               break;
#define OP_SWITCH_LOOP_END  } }

#else

#define OP_SWITCH_LOOP      goto *((pc++)->u.adr);
#define CASE(op)            op:
#define BREAK               goto *((pc++)->u.adr);
#define OP_SWITCH_LOOP_END  /* nothing */

/// 命令をラベルに置き換える
void replace_label(Code* code, const void* optbl[]) {
    Opcode op;
    do {
        op = code->u.op;
        code->u.adr = optbl[op];
    } while (++code, op != HALT);
}

#endif


void run(Vm* vm, Code* pc) {
#ifdef DIRECT_THREADED
    static const void* optbl[] = {
        &&HALT,
        &&LDA,
        &&DEC,
        &&JNZ,
    };
    replace_label(pc, optbl);
#endif

    OP_SWITCH_LOOP
    CASE(HALT)
        return;
    CASE(LDA)
        vm->reg.a = (pc-1)->operand;
        BREAK
    CASE(DEC)
        vm->reg.a -= 1;
        BREAK
    CASE(JNZ)
        if (vm->reg.a != 0) {
            pc += (pc-1)->operand;
        }
        BREAK
    OP_SWITCH_LOOP_END
}

int main() {
    Code code[] = {
        { LDA,  100000000 },    // １億
        { DEC },
        { JNZ,  -2 },
        { HALT },
    };
    Vm vm;
    run(&vm, code);

    return 0;
}

cygwin上でgcc -O2 でコンパイルした実行結果：

$ time ./thread.exe

real    0m2.795s
user    0m2.753s
sys     0m0.040s

$ time ./direct.exe

real    0m5.557s
user    0m5.367s
sys     0m0.050s

だめじゃん。 gcc -Sでアセンブリコードを見てみるとDIRECT_THREADEDでもディスパッチ部がひとつにまとめられてdirect threadedになってない。

; ディスパッチ部
L17:
    movl    (%ebx), %eax
    addl    $8, %ebx
    jmp     *%eax
    .p2align 4,,7
; JNZ
L10:
    movl    (%esi), %eax
    testl   %eax, %eax
    je      L17
    movl    -4(%ebx), %eax
    leal    (%ebx,%eax,8), %ebx
    jmp     L17    ; ディスパッチ部に戻る
    ...

gcc -S で出力された.sを直接書き換えてdirect threadingにして実行してみる：

$ time ./real-direct.exe

real    0m1.310s
user    0m1.261s
sys     0m0.060s

おお、確かに速くなった。 ジャンプ命令ひとつでこれだけ変わるというのは致命的かも。

  • .p2align 4,,7は、最大７バイト挿入して１６バイトにアライメントする、ということらしい。 ページにあわせると命令キャッシュに乗って？有利とか何とか。 勝手にやってくれるんか。
  • Threaded Code

Direct threadingでフィボナッチ計算

上のVMにいくつかてきとーに命令を足してフィボナッチ計算(30)をさせてみた：

/// VMのオペコード
typedef enum {
    HALT,
    NOP,
    LDI,
    MOV,
    DEC,
    ADD,
    CMPI,
    PUSH,
    POP,
    JMP,
    JNZ,
    JLT,
    CALL,
    RET,
    CFUNC,
} Opcode;

/// コード
typedef struct {
    union {
        Opcode op;
        const void* adr;
    } u;
    union {
        int l;
        short w[2];
    } operand;
} Code;

#define REG_NUM  (8)
/// VM
typedef struct {
    int* stack;
    int reg[REG_NUM];
    int sp;
    int opr1, opr2;
} Vm;

#define WORD(_0, _1)  ((_0) | ((_1)<<16))

//#define DIRECT_THREADED

#ifndef DIRECT_THREADED

#define OP_SWITCH_LOOP      for (;;) { switch((pc++)->u.op) {
#define CASE(op)            case op:
#define BREAK               break;
#define OP_SWITCH_LOOP_END  } }

#else

#define OP_SWITCH_LOOP      goto *((pc++)->u.adr);
#define CASE(op)            op:
#define BREAK               goto *((pc++)->u.adr);
#define OP_SWITCH_LOOP_END  /* nothing */

/// 命令をラベルに置き換える
void replace_label(Code* code, const void* optbl[]) {
    Opcode op;
    do {
        op = code->u.op;
        code->u.adr = optbl[op];
    } while (++code, op != HALT);
}

#endif

void run(Vm* vm, Code* pc) {
#ifdef DIRECT_THREADED
    static const void* optbl[] = {
        &&HALT,
        &&NOP,
        &&LDI,
        &&MOV,
        &&DEC,
        &&ADD,
        &&CMPI,
        &&PUSH,
        &&POP,
        &&JMP,
        &&JNZ,
        &&JLT,
        &&CALL,
        &&RET,
        &&CFUNC,
    };
    replace_label(pc, optbl);
#endif

#define JUMP_EXEC(pc)   (pc += (pc-1)->operand.l)
#define COND_JUMP(op)   do { if (vm->opr1 op vm->opr2) JUMP_EXEC(pc); } while (0)
#define PUSH_EXEC(v)    (vm->stack[vm->sp++] = (int)(v))
#define POP_EXEC()       (vm->stack[--vm->sp])

    vm->sp = 0;

    OP_SWITCH_LOOP
    CASE(HALT)
        return;
    CASE(NOP)
        BREAK
    CASE(LDI)
        vm->reg[(pc-1)->operand.w[0]] = (pc-1)->operand.w[1];
        BREAK
    CASE(MOV)
        vm->reg[(pc-1)->operand.w[0]] = vm->reg[(pc-1)->operand.w[1]];
        BREAK
    CASE(DEC)
        vm->reg[(pc-1)->operand.l] -= 1;
        BREAK
    CASE(ADD)
        vm->reg[(pc-1)->operand.w[0]] += vm->reg[(pc-1)->operand.w[1]];
        BREAK
    CASE(CMPI)
        vm->opr1 = vm->reg[(pc-1)->operand.w[0]];
        vm->opr2 = (pc-1)->operand.w[1];
        BREAK
    CASE(PUSH)
        PUSH_EXEC(vm->reg[(pc-1)->operand.l]);
        BREAK
    CASE(POP)
        vm->reg[(pc-1)->operand.l] = POP_EXEC();
        BREAK
    CASE(JMP)
        JUMP_EXEC(pc);
        BREAK
    CASE(JNZ)
        COND_JUMP(!=);
        BREAK
    CASE(JLT)
        COND_JUMP(<);
        BREAK
    CASE(CALL)
        PUSH_EXEC(pc);
        JUMP_EXEC(pc);
        BREAK
    CASE(RET)
        if (vm->sp == 0) {
            return;
        }
        pc = (Code*)POP_EXEC();
        BREAK
    CASE(CFUNC)
        {
            void (*f)() = (void (*)())(pc-1)->operand.l;
            f(vm);
        }
        BREAK
    OP_SWITCH_LOOP_END
}

#include <stdio.h>
#define STACK_SIZE  (256)

int main(int argc, char* argv[]) {
    int stack[STACK_SIZE];
    Code code[] = {
        { JMP,    13 },          // goto main

        // fib
        { CMPI,   WORD(0, 2) },  // if a < 2    return a;
        { JLT,    10 },
        { DEC,    0 },           // a -= 1
        { PUSH,   0 },           // push(a)
        { CALL,   -5 },          // a = fib(a)
        { POP,    1 },           // b = pop()
        { PUSH,   0 },           // push(a)
        { MOV,    WORD(0, 1) },  // a = b
        { DEC,    0 },           // a -= 1
        { CALL,   -10 },         // a = fib(a)
        { POP,    1 },           // b = pop()
        { ADD,    WORD(0, 1) },  // a += b
        { RET },

        // main
        { CALL,    -14 },
        { HALT },
    };
    int n = argc >= 2 ? atoi(argv[1]) : 10;
    Vm vm;
    vm.stack = stack;
    vm.reg[0] = n;
    run(&vm, code);

    printf("%d\n", vm.reg[0]);

    return 0;
}

threaded code の実行結果

$ time ./fib-threaded 30
832040

real    0m0.817s
user    0m0.771s
sys     0m0.050s

direct threaded code の実行結果

$ time ./fib-direct 30
832040

real    0m0.439s
user    0m0.400s
sys     0m0.030s

Cとか、他のスクリプト言語での実行結果

Cネイティブと、ゲームにおけるスクリプト言語の現状に出てきたスクリプト言語、あとはSchemeで速度検証。 Celeron 650MHz。

言語 real user sys
gcc 0m0.136s (1.00) 0m0.100s 0m0.040s
threaded 0m0.817s (6.01) 0m0.771s 0m0.050s
direct-threaded 0m0.439s (3.23) 0m0.400s 0m0.030s
Lua 0m2.060s (15.15) 0m2.032s 0m0.040s
Squirrel 0m12.117s (89.10) 0m12.117s 0m0.020s
xtal 0m2.874s (21.13) 0m0.020s 0m0.020s
Gauche 0m1.661s (12.21) 0m0.020s 0m0.030s
Ypsilon 0m1.593s (11.71) 0m0.010s 0m0.040s

YpsilonだけはWin32バイナリ版で他はCygwin上でソースをコンパイルしたもの。 純粋に計算だけだから、それほど問題ないよね。

Squirrelの遅さは異常。 これはリファレンスカウンタ方式によるものなのか？コンパイルオプションを見てもちゃんと-O2でやってるし。 もしも文法とか機能が飛びぬけてよかったとしても、遅い言語には興味ありません！ Xtalが結構速いのが驚き。

CRIScriptはとにかくデカすぎる、boostとか入ってるもんだから。 組み込むのはちょっと辛い。 そしてVisualStudioのプロジェクトしかついてなくて、Cygwinから簡単にコンパイルできない。 今使ってる旧マシンで動かそうとしたけど、Visual C++ 2008 Express Editionのインストールがうまくいかず断念 Ypsilonは以前試したときはGaucheの２倍くらいの速度が出たんだけど、今回はシングルコアで動かしたせいかな？

ということで、これから使うべき組み込みスクリプト言語はSchemeという結論が出たようだな！

関連記事

新着記事