Cでの単純な協調ストリームの実装

こんにちは、Habr！



以前に翻訳されたRESTに関する投稿に注目していただきありがとうございます。今日は、システム設計のトピックを少し異なる角度から見て、Linuxの著名人であるStephen Brennanによる記事の翻訳を公開することを提案します。彼は、ユーザースペースでのマルチタスクの独自の実装とその有用性について説明しています。



マルチタスクは、オペレーティングシステムによって提供される他の多くの機能と同様に、当然のことと見なすだけでなく、通常のものとして認識します。私たちの強力なコンピューターとスマートフォンでは、コンピューターが何百ものプロセスを処理できないという考えは奇妙に思えます。これがコンピュータをとても便利にした可能性の1つだと思いますが、これが原因でコンピュータは非常に複雑になり、魔法のように見えることもあります。



マルチタスクを実装するコードに手を出すことは困難であり、どの場合に自分で実装する方がよいかが常に明確であるとは限らないため、オペレーティングシステム全体を作成する必要はありません。自分で気付くまでは、この現象を完全に理解することは不可能だと確信しています。そのため、簡単なスレッドの実装でどのように遊ぶことができるかを説明する記事を書くことにしました。この記事では、

（オペレーティングシステムではなく）通常のCプログラムに単純なストリームを実装します。

setjmpとlongjmpについての叙情的な逸脱

スケジューラーは、関数setjmp()とに大きく依存しますlongjmp()。それらは少し魔法のように見えるので、最初にそれらが何をするのかを説明し、次に少し時間をかけて正確な方法を説明します。



この関数をsetjmp()使用すると、プログラムが実行のどの段階にあったかに関する情報を記録できるため、このポイントに再びジャンプできます。型変数が渡されjmp_buf、この情報が格納されます。初めて戻ったとき、関数setjmp()は0を返します。



後で、関数longjmp(jmp_buf, value)を使用して、呼び出されたポイントから即座に実行を再開できますsetjmp()。あなたのプログラムでは、この状況はsetjmp()2度目に戻ったように見えます。今回は引数が返されますvalue合格したこと-2longjmp()番目のリターンと最初のリターンを区別する方が便利です。この点を説明する例を次に示します。

#include <stdio.h>
#include <setjmp.h>

jmp_buf saved_location;
int main(int argc, char **argv)
{
    if (setjmp(saved_location) == 0) {
        printf("We have successfully set up our jump buffer!\n");
    } else {
        printf("We jumped!\n");
        return 0;
    }

    printf("Getting ready to jump!\n");
    longjmp(saved_location, 1);
    printf("This will never execute...\n");
    return 0;
}

このプログラムをコンパイルして実行すると、次の出力が得られます。

We have successfully set up our jump buffer!
Getting ready to jump!
We jumped!

うわー！これはgotoステートメントと同じですが、この場合、関数の外にジャンプするために使用することもできます。またgoto、通常の関数呼び出しのように見えるため、読みにくいです。あなたのコードを使用している場合setjmp()と豊富でlongjmp()、（あなたを含む）、誰のためにそれを読み取ることが非常に困難になります。



の場合と同様にgoto、一般的にはとを避けることsetjmp()をお勧めしlongjmp()ます。しかしのようにgoto、上記の関数は、控えめかつ一貫して使用すると便利です。スケジューラーはコンテキストを切り替えることができる必要があるため、これらの機能を責任を持って使用します。最も重要なことは、プランナーユーザーがこの種の複雑さに対処する必要がないように、APIからこれらの関数を使用することです。



Setjmpとlongjmpはスタックを保存しません

True、関数setjmp()、longjmp()いかなる種類のホッピングもサポートすることを目的としていません。それらは非常に特殊な実用的なケースのために設計されました。HTTPリクエストの作成など、複雑な操作を実行していると想像してください。この場合、複雑な一連の関数呼び出しが含まれ、それらのいずれかが失敗した場合は、それぞれから特別なエラーコードを返す必要があります。このようなコードは、関数を呼び出す場所（おそらく数十回）で、次のリストのようになります。

int rv = do_function_call();
if (rv != SUCCESS) {
    return rv;
}

その意味setjmp()とlongjmp()はsetjmp()、本当に難しい仕事に着手する前に場所を賭けるのに役立つものです。次に、すべてのエラー処理を1か所に一元化できます。

int rv;
jmp_buf buf;
if ((rv = setjmp(buf)) != 0) {
    /*    */
    return;
}
do_complicated_task(buf, args...);

関係する機能のいずれかが失敗した場合do_complicated_task()、それはただ起こりlongjmp(buf, error_code)ます。これは、コンポジション内の各関数do_complicated_task()が任意の関数呼び出しが成功したと想定できることを意味します。つまり、各関数呼び出しのエラーを処理するためにこのコードを配置することはできません（実際には、これはほとんど実行されませんが、これは別の記事のトピックです） ..。



ここでの基本的な考え方はlongjmp()、深くネストされた関数からのみジャンプできるということです。以前にジャンプした、深くネストされた関数にジャンプすることはできません。これは、関数からジャンプしたときのスタックの外観です。アスタリスク（*）は、それが格納されているスタックポインタを意味しますsetjmp()。

  | Stack before longjmp    | Stack after longjmp
      +-------------------------+----------------------------
stack | main()              (*) | main()
grows | do_http_request()       |
down  | send_a_header()         |
 |    | write_bytes()           |
 v    | write()  - fails!       |

ご覧のとおり、スタックを後方に移動することしかできないため、データが破損する危険はありません。一方、タスク間をジャンプしたい場合はどうなるか想像してみてください。電話をかけsetjmp()てから戻ってきて、他のことをして、以前に行っていた作業を再開しようとすると、問題が発生します。

      | Stack at setjmp() | Stack later      | Stack after longjmp()
      +-------------------+------------------+----------------------
stack | main()            | main()           | main()
grows | do_task_one()     | do_task_two()    | do_stack_two()
down  | subtask()         | subtask()        | subtask()
 |    | foo()             |                  | ???
 v    | bar()         (*) |              (*) | ???               (*)

保存setjmp()されたスタックポインタは、存在しなくなったスタックフレームを指し、ある時点で他のデータによって上書きされた可能性があります。助けを借りlongjmp()て戻ってきた関数に戻ろうとすると、非常に奇妙なことが始まり、プログラム全体が崩壊する可能性があります。



道徳：これらのような複雑なタスクを使用setjmp()しlongjmp()てジャンプする場合は、各タスクに独自の個別のスタックがあることを確認する必要があります。この場合、longjmp()スタックポインタがリセットされると、プログラム自体がスタックを目的のスタックに置き換え、スタックの消去が発生しないため、問題は完全に解消されます。

スケジューラAPIを書いてみましょう

逸脱は少し長いですが、私たちが学んだことを武器に、ユーザースペースフローを実装できるようになりました。まず、スレッドの初期化、作成、開始のためにAPIを自分で設計することは非常に便利であることに注意してください。事前にこれを行うと、何を構築しようとしているのかをよりよく理解できるようになります。

void scheduler_init(void);
void scheduler_create_task(void (*func)(void*), void *arg);
void scheduler_run(void);

これらの関数は、スケジューラーを初期化し、タスクを追加し、最後にスケジューラーでタスクを開始するために使用されます。起動すると、scheduler_run()すべてのタスクが完了するまで実行されます。現在実行中のタスクには、次のAPIがあります。

void scheduler_exit_current_task(void);
void scheduler_relinquish(void);

最初の関数は、タスクを終了する役割を果たします。タスクを終了することもその機能から戻るときに可能であるため、この構造は便宜上存在するだけです。 2番目の関数は、別のタスクをしばらく実行する必要があることをスレッドがスケジューラーに通知する方法を説明します。タスクがを呼び出すとscheduler_relinquish()、他のタスクの実行中に一時的に中断できます。しかし、最終的には関数が戻り、最初のタスクを続行できます。



具体的な例として、APIの架空の使用例を考えてみましょう。これを使用してスケジューラをテストします。

#include <stdlib.h>
#include <stdio.h>

#include "scheduler.h"

struct tester_args {
    char *name;
    int iters;
};

void tester(void *arg)
{
    int i;
    struct tester_args *ta = (struct tester_args *)arg;
    for (i = 0; i < ta->iters; i++) {
        printf("task %s: %d\n", ta->name, i);
        scheduler_relinquish();
    }
    free(ta);
}

void create_test_task(char *name, int iters)
{
    struct tester_args *ta = malloc(sizeof(*ta));
    ta->name = name;
    ta->iters = iters;
    scheduler_create_task(tester, ta);
}

int main(int argc, char **argv)
{
    scheduler_init();
    create_test_task("first", 5);
    create_test_task("second", 2);
    scheduler_run();
    printf("Finished running all tasks!\n");
    return EXIT_SUCCESS;
}

この例では、同じ機能を実行するが異なる引数を取る2つのタスクを作成します。したがって、それらの実装は個別に追跡できます。各タスクは、設定された回数の反復を実行します。各反復で、メッセージを出力してから、別のタスクを実行します。プログラムの出力のようなものが表示されることを期待しています。

task first: 0
task second: 0
task first: 1
task second: 1
task first: 2
task first: 3
task first: 4
Finished running all tasks!

スケジューラAPIを実装しましょう

APIを実装するには、タスクのある種の内部表現が必要です。それでは、ビジネスに取り掛かりましょう。必要なフィールドを集めましょう：

struct task {
    enum {
        ST_CREATED,
        ST_RUNNING,
        ST_WAITING,
    } status;

    int id;

    jmp_buf buf;

    void (*func)(void*);
    void *arg;

    struct sc_list_head task_list;

    void *stack_bottom;
    void *stack_top;
    int stack_size;
};

これらの各フィールドについて個別に説明しましょう。作成されたすべてのタスクは、実行前に「作成済み」状態である必要があります。タスクの実行が開始されると、タスクは「実行中」状態になり、タスクが何らかの非同期操作を待機する必要がある場合は、「待機中」状態にすることができます。フィールドidは、単にタスクの一意の識別子です。これにbufは、longjmp()タスクをいつ実行して再開するかに関する情報が含まれています。funcおよびフィールドargはに渡されscheduler_create_task()、タスクを開始するために必要です。このフィールドはtask_list、すべてのタスクの二重にリンクされたリストを実装するために必要です。フィールドstack_bottom、stack_topおよびstack_sizeすべては、このタスク専用の個別のスタックに属しています。下はによって返されるアドレスですmalloc()ただし、「top」は、メモリ内の指定された領域のすぐ上のアドレスへのポインタです。x86スタックは下に向かって大きくなるため、スタックポインタをstack_topではなく値に設定する必要がありますstack_bottom。



このような状況では、次の関数を実装できますscheduler_create_task()。

void scheduler_create_task(void (*func)(void *), void *arg)
{
    static int id = 1;
    struct task *task = malloc(sizeof(*task));
    task->status = ST_CREATED;
    task->func = func;
    task->arg = arg;
    task->id = id++;
    task->stack_size = 16 * 1024;
    task->stack_bottom = malloc(task->stack_size);
    task->stack_top = task->stack_bottom + task->stack_size;
    sc_list_insert_end(&priv.task_list, &task->task_list);
}

を使用static intすると、関数が呼び出されるたびにidフィールドがインクリメントされ、そこに新しい番号が表示されることが保証されます。sc_list_insert_end()単にstruct taskグローバルリストに追加する機能を除いて、他のすべては説明なしで明確でなければなりません。グローバルリストは、スケジューラのすべてのプライベートデータを含む2番目の構造内に格納されます。以下は、構造自体とその初期化関数です。

struct scheduler_private {
    jmp_buf buf;
    struct task *current;
    struct sc_list_head task_list;
} priv;

void scheduler_init(void)
{
    priv.current = NULL;
    sc_list_init(&priv.task_list);
}

このフィールドはtask_list、タスクリストを参照するために使用されます（当然のことながら）。このフィールドにcurrentは、現在実行されているタスクが格納されます（またはnull、現時点でそのようなタスクがない場合）。最も重要なのは、フィールドbufがコードにジャンプするために使用されることですscheduler_run()。

enum {
    INIT=0,
    SCHEDULE,
    EXIT_TASK,
};

void scheduler_run(void)
{
    /*     ! */
    switch (setjmp(priv.buf)) {
    case EXIT_TASK:
        scheduler_free_current_task();
    case INIT:
    case SCHEDULE:
        schedule();
        /*       ,    */
        return;
    default:
        fprintf(stderr, "Uh oh, scheduler error\n");
        return;
    }
}

関数が呼び出されるとすぐに、いつでもその関数に戻ることができるようscheduler_run()にバッファーを設定setjmp()します。初回は0（INIT）が返され、すぐに呼び出しますschedule()。その後、SCHEDULEまたはEXIT_TASK定数をに渡すことができlongjmp()ます。これにより、さまざまな動作が引き起こされます。今のところ、EXIT_TASKのケースを無視して、実装に直接ジャンプしましょうschedule()。

static void schedule(void)
{
    struct task *next = scheduler_choose_task();

    if (!next) {
        return;
    }

    priv.current = next;
    if (next->status == ST_CREATED) {
        /*
         *     .   
         * ,        .
         */
        register void *top = next->stack_top;
        asm volatile(
            "mov %[rs], %%rsp \n"
            : [ rs ] "+r" (top) ::
        );

        /*
         *   
         */
        next->status = ST_RUNNING;
        next->func(next->arg);

        /*
         *     ,    .   – ,   
         *   
         */
        scheduler_exit_current_task();
    } else {
        longjmp(next->buf, 1);
    }
    /*   */
}

まず、内部関数を呼び出して、次に実行するタスクを選択します。このスケジューラーは通常のカルーセルのように機能するため、リストから新しいタスクを選択するだけです。この関数がNULLを返す場合、実行するタスクはこれ以上ないので、戻ります。それ以外の場合は、タスクの実行を開始するか（ST_CREATED状態の場合）、実行を再開する必要があります。



作成したタスクを実行するには、x86_64のアセンブリ命令を使用して、フィールドをstack_topレジスタrsp（スタックポインタ）に割り当てます。次に、タスクの状態を変更し、関数を実行して終了します。注：スタックポインターの保存と再配置のsetjmp()両方があるlongjmp()ため、ここでは、アセンブラーを使用してスタックポインターを変更するだけで済みます。



タスクがすでに開始されている場合、フィールドにはタスクを再開bufするlongjmp()ために必要なコンテキストが含まれている必要があるため、これを実行します。

次に、次に実行するタスクを選択するヘルパー関数を見てみましょう。これがスケジューラーの心臓部であり、前に述べたように、このスケジューラーはカルーセルのように機能します。

static struct task *scheduler_choose_task(void)
{
    struct task *task;

    sc_list_for_each_entry(task, &priv.task_list, task_list, struct task)
    {
        if (task->status == ST_RUNNING || task->status == ST_CREATED) {
            sc_list_remove(&task->task_list);
            sc_list_insert_end(&priv.task_list, &task->task_list);
            return task;
        }
    }

    return NULL;
}

リンクリストの実装（Linuxカーネルから取得）に精通していない場合は、大したことはありません。関数sc_list_for_each_entry()は、タスクリスト内のすべてのタスクを反復処理できるようにするマクロです。最初に選択可能なタスク（保留状態ではない）が現在の位置から削除され、タスクリストの最後に移動します。これにより、次にスケジューラを実行するときに、別のタスクを受け取ることが保証されます（タスクがある場合）。選択可能な最初のタスクを返します。タスクがまったくない場合はNULLを返します。



最後に、実装scheduler_relinquish()に移って、タスクがどのように自己排除できるかを見てみましょう。

void scheduler_relinquish(void)
{
    if (setjmp(priv.current->buf)) {
        return;
    } else {
        longjmp(priv.buf, SCHEDULE);
    }
}

これはsetjmp()、スケジューラーの関数のもう1つの使用例です。基本的に、このオプションは少し混乱しているように見えるかもしれません。タスクがこの関数を呼び出すとき、それを使用しsetjmp()て現在のコンテキスト（実際のスタックポインターを含む）を保存します。次に、それを使用longjmp()してスケジューラーに入り（ここでもscheduler_run()）、SCHEDULE関数を渡します。したがって、新しいタスクを割り当てるようにお願いします。



タスクが再開されると、関数setjmp()はゼロ以外の値で戻り、以前に実行していた可能性のあるタスクをすべて終了します。

最後に、タスクが終了するとどうなりますか（これは、明示的に、exit関数を呼び出すか、対応するタスク関数から戻ることによって行われます）。

void scheduler_exit_current_task(void)
{
    struct task *task = priv.current;
    sc_list_remove(&task->task_list);
    longjmp(priv.buf, EXIT_TASK);
    /*   */
}

static void scheduler_free_current_task(void)
{
    struct task *task = priv.current;
    priv.current = NULL;
    free(task->stack_bottom);
    free(task);
}

これは2つの部分からなるプロセスです。最初の関数は、タスク自体によって直接返されます。これに対応するエントリは割り当てられなくなるため、タスクのリストから削除します。次に、を使用longjmp()して、関数に戻りますscheduler_run()。今回はEXIT_TASKを使用します。これは、新しいタスクを割り当てる前に、スケジューラーに何を呼び出すべきかを指示する方法scheduler_free_current_task()です。説明に戻ると、scheduler_run()これがまさにその機能であることがわかりますscheduler_run()。



これは2つのステップで行いました。scheduler_exit_current_task()、タスク構造に含まれるスタックを積極的に使用します。スタックを解放して引き続き使用すると、解放したばかりのスタックメモリに関数がアクセスできる可能性があります。これが発生しないようにするには、ヘルプlongjmp()を使用して別のスタックを使用してスケジューラに戻す必要があります。その後、タスクに関連するデータを安全に解放できます。



そのため、スケジューラの実装全体を完全に分析しました。リンクリストの実装と上記のメインプログラムを追加してコンパイルしようとすると、完全に機能するスケジューラーが得られます。コピー＆ペーストに煩わされないように、この記事のすべてのコードが含まれているgithubのリポジトリに移動します。

説明されているアプローチの用途は何ですか？

これまで読んだことがあれば、その例が面白いと納得させる必要はないと思います。しかし、一見、あまり役に立たないように見えるかもしれません。結局のところ、Cでは、並行して実行でき、いずれかがを呼び出すまで相互に待機する必要がない「実際の」スレッドを使用できますscheduler_relinquish()。



ただし、これは、便利な機能の一連のエキサイティングな実装全体の開始点と見なしています。Pythonの新しい非同期ユーティリティが機能するのと同じように、重いI / Oタスク、シングルスレッドの非同期アプリケーションの実装について話すことができます。このようなシステムを使用して、ジェネレーターとコロチンを実装することもできます。最後に、ハードワークがあれば、このシステムを「実際の」オペレーティングシステムスレッドと組み合わせて、必要に応じて追加の同時実行性を提供することもできます。興味深いプロジェクトがこれらのアイデアのそれぞれの背後に隠されています。読者の皆様、自分でそれらの1つを完成させることをお勧めします。

安全です？

はいよりもいいえの可能性が高いと思います！スタックポインタに影響を与えるインラインアセンブリコードは安全とは見なされません。これらのものを本番環境で使用するリスクを冒さないでください。ただし、それらをいじって調査してください。



このようなシステムのより安全な実装は、「非コンテキスト」API（man getcontextを参照）を使用して構築できます。これにより、アセンブリコードを埋め込むことなく、これらのタイプのユーザースペース「ストリーム」を切り替えることができます。残念ながら、そのようなAPIは標準でカバーされていません（POSIX仕様から削除されています）。ただし、それでもglibcの一部として使用できます。

どうすればそのようなメカニズムを置き換えることができますか？

ここに示すように、このスケジューラーは、スレッドが明示的に制御をスケジューラーに戻す場合にのみ機能します。これは、一般的なプログラム、たとえばオペレーティングシステムには適していません。スレッドの作成が不十分だと、他のすべての実行がブロックされる可能性があるためです（ただし、MS-DOSでの協調マルチタスクの使用は妨げられませんでした）。これにより、協調マルチタスクが明らかに悪くなるとは思いません。それはすべてアプリケーションに依存します。



非標準の「コンテキスト外」APIを使用する場合、POSIXシグナルは、以前に実行されたコードのコンテキストを格納します。タイマーをビープ音に設定することにより、ユーザースペーススケジューラーは実際にプリエンプティブマルチタスクの動作バージョンを提供できます。これは別の記事に値する別のクールなプロジェクトです。