boost.taskを使用したサーバータスクのスケジューリング

最近、あるプロファイルリソースで、あるプログラマーが「ストリームを操作するためにMMOサーバーで何を使用するのですか？」という質問をしました。 プログラマーはIntel TBBに傾倒していましたが、基本的なプリミティブではなく、カスタムタスクスケジューリングに傾倒していました。 まあ、TBBのように-まあ、大丈夫。 そして少し後に、MMOサーバーのソースコードを別のプログラマーが見ました。彼は最近、アーキテクチャを改善するために最初からそれに対応し始めました。 また、boostなどのサードパーティコンポーネントを使用する代わりに、プログラマー自身が書いた自転車がたくさんありました（たとえば、pthreadのラッパーのクラス。これは2010年で、 boost.threadはほぼ標準です）。 タスクスケジューラを使用したスレッドプールのサポートも実装されました。 このトピックは私にとって非常に興味深いものであり、タスク計画（TBBなど）の既製のソリューションに関する情報を掘り始め、 boost.taskを見つけました。

定義

タスクは、論理的に統合されたアクションのセットです。 タスクスケジューラは、どのスレッドで誰をどの瞬間に実行するかを選択するための特定の戦略によって導かれるタスクを非同期的に実行します。
タスクを使用すると、通常のフローから抽象化し、より高いレベルで操作できます。

なぜタスクスケジューラが必要なのですか？

球状サーバーは真空でどのように機能しますか？ 非常にシンプル：

1. クライアントからリクエストが来ます
2. 彼は処理されています！
3. 返信を送信しました

さらに、サーバーでクライアントのリクエストなしに実行されるプロセスが発生する可能性があります。 たとえば、ユーザーデータベース全体に通知を送信したり、データベースから古いデータ（ブラケット）を削除したり、毎日の統計情報を処理したりします。
キャッチは、リクエストの処理方法そのものです。 それを処理する方法を理解する必要があります。
たとえば、memcachedのようなサーバーを考えてみましょう。hash_mapにはデータがあり、読み取り要求があり、書き込み要求はハッシュマップを簡単に検索してデータを返すか、ハッシュマップに書き込みます。 すべてが1つのスレッドで行われますが、システムのすべてのプロセッサを使用する必要がある場合はどうでしょうか。
コアと同じ数のスレッドを作成します。 各スレッドでは、接続を作成するときにラウンドロビンの原則に従って散らばっているユーザーを処理します。 コンテナにアクセスするとき、rwlock（boost :: shared_mutex）を使用します。 素晴らしい。 しかし、コンテナから要素を削除するのはどうですか？ N秒ごとに起動し、コンテナーをクリーンアップするストリームを作成します。
これは単純な例でしたが、現在はより複雑な例です。ユーザーのリクエストに応じて、データベースにリクエストを送信し、Webサイトにhttpリクエストを送信できるサービスです。 サーバーが以前のモデルに従って作成されるとどうなりますか（他のコンポーネントへのすべてのリクエストは同期的に実行されます）？ さて、データベースはサーバーと同じプラットフォーム上にあり、答えは数ミリ秒の通路にあります。 電子メールの送信も問題ではありません。同じマシンにsendmailを配置し、データを送信すると、彼自身が手紙の送信方法を見つけます。
素晴らしい。 本当にそうではありませんが。 httpリクエストをどうしますか？ 非常に長い時間がかかる場合があります-それはすべてどこか遠くにあるサイトに依存し、リクエストが処理される時間はわかりません。 この場合、キューには実行可能な要求が多数ありますが、スレッドは非アクティブになりますが、このスレッドが解放されるまで待機します。
このような要求は非同期で実行する必要があります。 次のように実装できます。

class LongRequestHandler
{
public :
void Handle()
{
// read client request parameters
// mysql request 1
// mysql request 2
HttpRequestExecutor::GetInstance()->Execute(
"example.com?x=1" ,
boost::bind( this , &LongRequestHandler::HandleStage2)
);
}
void HandleStage2( const std:: string & http_request_result)
{
// mysql request 3
// write response to client
}
};

* This source code was highlighted with Source Code Highlighter .

HttpRequestExecutorは、要求のURLと、要求の完了時に呼び出す必要があるコールバックを受け入れます（コールバックのタイプはboost :: functionです）。
そして、このアプローチは機能しますが、あまり美しくはありません。
C ++で非同期的に考えるブログでは、非同期タスクの興味深い実装が示されています。 結果は次のとおりです。

template <typename Handler> void async_echo(
tcp::socket& socket,
mutable_buffer working_buffer,
Handler handler,
// coroutine state:
coroutine coro = coroutine(),
error_code ec = error_code(),
size_t length = 0)
{
reenter (coro)
{
entry:
while (!ec)
{
yield socket.async_read_some(
buffer(working_buffer),
bind(&async_echo<handler>,
ref (socket), working_buffer,
box(handler), coro, _1, _2));
if (ec) break ;
yield async_write(socket,
buffer(working_buffer, length),
bind(&async_echo<handler>,
ref (socket), working_buffer,
box(handler), coro, _1, _2));
}
handler(ec);
}
}

* This source code was highlighted with Source Code Highlighter .

C ++のコルーチンとyieldは異常に見えます;）これはdefineに実装されており、ブログで著者がどうやって管理したかを読むことができます。
徐々に、ロジックが複雑になり、非同期的に処理する必要がある新しい要素が追加され、実装も複雑になります。 今後の課題

  mysqlリクエスト1
 mysqlリクエスト2
 httpリクエスト1
 mysqlリクエスト3
 httpリクエスト2
 mysqlリクエスト4
 mysqlリクエスト5

そして、httpリクエストで停止して順次実行すると、そのリクエストが表示されます

  mysqlリクエスト2
 httpリクエスト1

そして

  mysqlリクエスト3
 httpリクエスト2
 mysqlリクエスト4

並行して行うことができます。これを行うには、ロジックをさらに複雑にする必要があります。 たとえば、次のような簡単なコードを書きたいと思います。

  mysqlリクエスト1
 x =実行（func1）
 y =実行（func2）
待機（x、y）
 mysqlリクエスト5

 func1：
   mysqlリクエスト2
   httpリクエスト1

 func2：
   mysqlリクエスト3
   httpリクエスト2
   mysqlリクエスト4

これは、タスクスケジューラが役立つ場所です。

実装

ここで、新しい0x標準のタスクスケジューラサポートについて読むことができます 。

• just :: thread -father boostからのC ++ 0x標準スレッドライブラリの実装:: thread
• Parallel Patterns Library（PPL）-Microsoftからの実装
• 非同期エージェントライブラリ -およびマイクロソフトの別のライブラリ
• Intel Threading Building Blocksは、 Intelの非常に強力な並列プログラミングライブラリです。 タスクスケジューラが含まれています。
• boost :: task --Oliver Kowalkeからの実装、まだboostで採用されていない

boost.taskが一番好きでした。 さらに詳細な検討。

boost.taskの説明

boost.task- C ++ 0x標準の提案の実装。 タスク実行戦略の設定、サブタスクの作成、タスクの中断をサポートします。
ライブラリは以下に依存します。

• ブースト > = 1.41
• boost.atomic-ブーストのためのアトミックオブジェクトC ++ 0xの実装
• boost.move-まだ右辺値参照を持たないC ++ 03標準のMoveセマンティクスの実装
• boost.fiber- 軽量 スレッド用のboost.threadの類似物

boost.taskおよびboost.fiberコンパイル済みライブラリ（boost.atomicおよびboost.move-ヘッダーのみ）-したがって、それらを収集する必要があります。 実験をより便利にするために、すべての依存関係を1か所に収集し、cmakeで味付けして、プロジェクトをgithubにアップロードしました。 Linux上で動作し、Windowsでビルドします。cmakeにファイルを追加するには2〜3行かかります。

使用例

ライブラリAPIは非常にシンプルで、上記のリクエストハンドラの実装は難しくありません。 私は再びそれをもたらします：

  mysqlリクエスト1

   mysqlリクエスト2
   httpリクエスト1

   mysqlリクエスト3
   httpリクエスト2
   mysqlリクエスト4

 mysqlリクエスト5

mysqlへのクエリのエミュレーションとして、ランダムな時間の通常のスリープを使用します。

 boost::this_thread::sleep(boost::posix_time::milliseconds(rand()%100 + 10)); 
 

boost::this_thread::sleep(boost::posix_time::milliseconds(rand()%100 + 10));

boost :: asioからの非同期タイマーは、外部http要求として使用されます。
だから：
要求-要求クラス。

class Request
{
public :
Request( const std:: string & data);
const std:: string & Read() const ;
void Write( const std:: string & answer);
};

* This source code was highlighted with Source Code Highlighter .

また、RequestHandlerは要求ハンドラーのクラスです。

class RequestHandler
{
public :
RequestHandler(boost::asio::io_service & io_service, const RequestPtr & request);
void Process() const ;
};

* This source code was highlighted with Source Code Highlighter .

io_service-外部呼び出しを行うことができるように渡されます（boost :: asio :: deadline_timerタイマーを使用します）。 タスクを処理するスレッドのプールを定義します。

boost::tasks::static_pool< boost::tasks::unbounded_fifo > pool( boost::tasks::poolsize( 5) );

* This source code was highlighted with Source Code Highlighter .

boost.taskは、2つの主要なタスクスケジューリング戦略をサポートしています。

• 制限 （制限あり）-実行されたタスクの数にしきい値があり、それに達すると、新しいタスクを追加するとこのアクションを実行するスレッドがブロックされます。 主なタスクは、タスクの追加速度が実行速度を超える場合のリソースの枯渇を回避することです
• 無制限 -無限の数のタスクをキューに追加できます

キュー内のタスクを処理するための戦略を設定することもできます。

• fifo-最初に追加されたタスクが最初に実行されます
• 優先度 -タスクに優先度があり、最も優先度の高いタスクが実行のために選択されます
• smart-このタイプのキューは、パラメーターをテンプレートに渡すことで強力にカスタマイズできます。 デフォルトでは、任意のキーでタスクのインデックスを作成し、古いタスクが存在する場合は新しいタスクで置き換えることができます

したがって、記述されたコード行は、タイプfifoの無制限のキューを持つ5つのスレッドのプールを作成します。
次に、外部リクエストを処理するためにio_serviceと3つのスレッドのプールを作成する必要があります。

boost::asio::io_service io_service;

* This source code was highlighted with Source Code Highlighter .

io_service :: callにタスクがないときに呼び出した場合、メソッドはすぐに終了します。通常の操作では、作業スレッドが必要です。 通常、これはio_serviceがクライアントが接続するための受け入れポートを追加したという事実によって達成されます。この場合、タイマーの実行を待機するio_serviceを取ることができます：

boost::asio::deadline_timer dummy_timer(io_service);
dummy_timer.expires_from_now(boost::posix_time::seconds(10));
// void dummy_handler(const boost::system::error_code&) {}
dummy_timer.async_wait(&dummy_handler);

* This source code was highlighted with Source Code Highlighter .

その後、スレッドプールを作成できます。

boost::thread_group io_service_thread_pool;
for ( int i = 0; i < 3; ++i)
io_service_thread_pool.create_thread(
boost::bind(&boost::asio::io_service::run, &io_service)
);

* This source code was highlighted with Source Code Highlighter .

次に、リクエストを作成します。

RequestPtr request( new Request( "some data" ));
RequestHandlerPtr handler( new RequestHandler(io_service, request));

* This source code was highlighted with Source Code Highlighter .

すべての準備が整ったら、次のタスクを実行できます。

boost::tasks::handle< void > request_processing(
boost::tasks::async(
boost::tasks::make_task( &RequestHandler::Process, handler ),
pool));

* This source code was highlighted with Source Code Highlighter .

boost :: tasks :: make_task（＆RequestHandler :: Process、handler）-実行可能なハンドラーオブジェクトでProcessを呼び出すタスクを作成します。 boost ::タスク:: asyncは非同期タスクの実行を開始します。 boost ::タスク::タスクの完了ステータスを追跡できるオブジェクトを処理し、結果があればそれを取得します。
boost ::タスク:: asyncは4つのタスク実行アルゴリズムをサポートしています：

• own_thread-同じスレッドでの同期実行
• new_thread-スレッドは、実行されるタスク用に作成され、その後、スレッドは完了します
• as_sub_task-現在のタスクがプールで実行されている場合-新しいタスクを追加します。それ以外の場合は、new_threadなどの新しいスレッドを作成します。 これがデフォルトの動作です。
• static_pool-スレッドプールでタスクを実行する

次に、タスクが完了するまで待ちます。

request_processing.wait();

* This source code was highlighted with Source Code Highlighter .

そして、io_serviceを停止します。

io_service.stop();
io_service_thread_pool.join_all();

* This source code was highlighted with Source Code Highlighter .

プロセス関数は驚くほど簡単です。

void Subtask1() const
{
Request( "query2" );
ExternalRequest( "extquery1" );
}

void Subtask2() const
{
Request( "query3" );
ExternalRequest( "extquery2" );
Request( "query4" );
}

void Process() const
{
std:: string data = request_->Read();

Request( "query1" );

boost::tasks::handle< void > subtask1(
boost::tasks::async(
boost::tasks::make_task( &RequestHandler::Subtask1, this )));
boost::tasks::handle< void > subtask2(
boost::tasks::async(
boost::tasks::make_task( &RequestHandler::Subtask2, this )));

boost::tasks::waitfor_all( subtask1, subtask2);

Request( "query5" );

request_->Write( "some answer" );
}

* This source code was highlighted with Source Code Highlighter .

サブタスクは、実行するポリシーを指定せずにboost :: tasks :: asyncを使用して実行され、親タスクと同じスレッドプールでタスクを実行するas_sub_taskアルゴリズムが自動的に選択されます。 サブタスク関数の実装も簡単です。
RequestHandler :: Request-boost :: this_thread :: sleepを呼び出し、ExternalRequestでは少し複雑です：

void ExternalRequest( const std:: string & what) const
{
ExternalRequestHandler external_handler(io_service_);
boost::tasks::spin::auto_reset_event ev;
external_handler.PerformExternalReqeust(what, &ev);
ev.wait();
}

* This source code was highlighted with Source Code Highlighter .

ハンドラーと、自動リセットを伴うイベントが作成されます-ブースト::タスク::スピン:: auto_reset_event。 このイベントは外部リクエストハンドラに渡され、その完了時にev.set（）が呼び出され、ev.wait（）がタスクをブロックするまで呼び出されます。
通常のスレッドおよび同期プリミティブ（boost :: condition）とは対照的に、ev.wait（）はストリームをブロックしませんが、タスクをブロックします（ループでthis_task :: yield（）を呼び出します）。 これは、プロセッサリソースが他のタスクによって使用されることを意味します。
ファイル全体はこちらにあります 。

結論

boost.taskは、タスクをスケジュールするための非常に便利なライブラリです。 これにより、非同期コード実行のサポートが新しいC ++ 0x標準でどのように見えるかを確認でき、標準のリリースを待たずにすぐに使用できます。
boost.taskを使用するコードは、通常のスレッドの使用よりも小さくなり、理解しやすくなります。
確かに欠点があります。コードはまだ最適化されておらず、まれに問題を引き起こす可能性があります。 ライブラリは（依存関係と共に）ブーストでまだ受け入れられていません。

トピックについて何を読むべきですか？

• ネイティブの同時性
• 平行道路
• C ++で非同期的に考える
• 無料のランチは終了しました
• マルチスレッドゲームエンジンのアーキテクチャ