パイプ、tee（）、スプライス（）を使って複数のソケットにデータを送信

Question

私はsplice（）を使って複数のソケットに書き込むためにtee（）で "master"パイプを複製します。もちろん、これらのパイプは、宛先ソケットにどのくらいスプライスすることができるかによって、異なる速度で空になります。だから私は次に "マスター"パイプにデータを追加してから、それを再びtee（）しようとすると、64KBをパイプに書き込むことができますが、 "スレーブ"パイプの1つに4KBだけティーインします。私は、ソケットにすべての "マスター"パイプをスプライスすると、残りの60KBをそのスレーブパイプにtee（）することができなくなると思います。本当？私は、私は "未取得"データの開始点に設定し、それを超えてスプライスしない（0から始まる）tee_offsetを追跡することができると思います。この場合、私はtee_offsetを4096に設定し、それを他のパイプにすべてティーすることができるまでスプライスしません。私は正しい道にここにいますか？私のためのヒント/警告？

Answer 1

私が正しく理解していれば、複数のソケットに多重化したいリアルタイムのデータソースがあります。 1つの "ソース"パイプがデータを生成するものに接続されており、データを送信する各ソケットの "宛先"パイプがあります。あなたは、tee()を使用してソースパイプから各宛先パイプにデータをコピーし、splice()を宛先パイプからソケット自体にコピーします。

ここでヒットする基本的な問題は、ソケットの1つが追いつけない場合です。送信できるデータよりも速くデータを生成している場合は、問題が発生します。これはパイプの使用とは関係なく、単なる根本的な問題です。だから、あなたはこのケースで対処する戦略を選びたいと思うでしょう - これらのことがしばしばあなたを後で噛んでくるので、それが一般的ではないと思っても、これを処理することをお勧めします。あなたの基本的な選択は、違反しているソケットを閉じるか、出力バッファがクリアされるまでデータをスキップすることです。後者は、例えばオーディオ/ビデオストリーミングに適しています。

パイプの使用に関連するの問題はですが、Linuxではパイプのバッファーのサイズがやや柔軟ではありません。 Linux 2.6.11以降はデフォルトで64Kになります（tee()呼び出しは2.6.17で追加されました）。pipe manpageを参照してください。 2.6.35以来、この値は、F_SETPIPE_SZオプションを使用して（fcntl manpageを参照）に変更することができますが、バッファリングはユーザ空間の動的割り当て方式さあ。つまり、遅いソケットに対処するあなたの能力はいくらか限定されます - これが受け入れられるかどうかは、あなたがデータを受信し、送信できると予想される速度に依存します。

このバッファリング戦略が受け入れられるとすれば、各宛先パイプがソースから消費したデータの量を追跡する必要があり、すべての宛先パイプが消費したデータを破棄するのが安全であることを前提としています。これは、tee()にオフセットの概念がないという事実によってやや複雑です。パイプの先頭からのみコピーできます。その結果、最も遅いソケットの速度でコピーできるのは、tee()を使用してデータの一部がソースから消費されるまで宛先パイプにコピーすることができず、実行できないからです。このは、すべてのソケットにあなたが消費しようとしているデータがあるまで。

これをどのように処理するかは、データの重要性によって決まります。tee()とsplice()という速度が本当に必要な場合は、遅いソケットが非常にまれなイベントになると確信しています。このようなことができます（ノンブロッキングIOと1つのスレッドを使用していると仮定しています）が、同様の何かがまた、複数のスレッドで動作します）：

1. すべてのパイプは、（）の各ファイルディスクリプタノンブロッキングにするためにfcntl(d, F_SETFL, O_NONBLOCK)を使用し、非ブロッキングされていることを確認してください。
2. 宛先パイプごとに変数read_counterをゼロに初期化します。
3. epoll()のようなものを使用して、ソースパイプに何かがあるまで待ちます。
4. read_counterが0のすべての宛先パイプをループし、それぞれにデータを転送するためにtee()を呼び出します。フラグにSPLICE_F_NONBLOCKを渡してください。
5. tee()によって転送された量によって、宛先パイプごとにインクリメントread_counterが追加されます。最小の結果値を記録しておきます。
6. read_counterの最小結果値を見つける - これは非ゼロである場合、（たとえば、/dev/nullに開口宛先にsplice()コールを使用して）、ソース管からのデータの量を廃棄します。データを破棄した後、read_counterから廃棄された量をすべてパイプから差し引きます（これが最も低い値であったため、いずれも負になりません）。
7. から繰り返す。。

注：過去に私を倒してしまっただ一つのことがSPLICE_F_NONBLOCKがパイプ上tee()とsplice()操作は、ノンブロッキングであり、あなたがfnctl()で設定O_NONBLOCKが他のコールとの相互作用するかどうかに影響するかどうかに影響していることです（例えばread()およびwrite()）は非ブロッキングである。すべてを非ブロックにしたい場合は、両方を設定します。また、ソケットをノンブロッキングにしたり、データを転送するためにsplice()コールをブロックすることも忘れないでください（スレッドアプローチを使用している場合は、そうでない限り）。

この戦略には大きな問題があります.1つのソケットがブロックされるとすべてが停止し、そのソケットの宛先パイプがいっぱいになり、ソースパイプが停滞します。だから、もしtee()がEAGAINを返す段階に達するならば、あなたはそのソケットを閉じるか、少なくとも書き留めないように（つまりループから取り除く）その出力バッファが空になるまで何か他にどちらを選択するかは、データストリームのスキップをスキップしてリカバリできるかどうかによって異なります。

あなたはもっと優雅にネットワーク遅延に対処したいなら、あなたはより多くのバッファリングを行うために必要になるだろう、これはユーザ空間のバッファのいずれかが関与するつもりは（むしろtee()とsplice()の利点を否定している）か、おそらくされますディスクベースのバッファ。ディスクベースのバッファリングは、ユーザ空間のバッファリングよりもかなり確実に遅くなるため、最初はtee()とsplice()を選択したので、おそらく多くの速度が必要な場合は適切ではありませんが、完全性について言及します。あなたが任意の時点で、ユーザ空間からデータを挿入してしまう場合は注目に値します

ことの一つは、writev()コールと同様に、パイプにユーザ空間から「出力を収集し、」実行できるvmsplice()コールです。これは、複数の異なる割り当てバッファ間でデータを分割して十分なバッファリングを行っている場合（たとえば、プールアロケータアプローチを使用している場合など）に便利です。

最後に、tee()とsplice()を使用する「高速」スキームと、遅いユーザー空間バッファリングに移動し続けることができない場合、ソケットを交換することができます。これはあなたの実装を複雑にしますが、多数の接続を処理していて、それらのうちのごく少数しか接続が遅い場合は、やはり関与するユーザスペースへのコピーの量を減らしています。しかし、これは一時的なネットワーク問題に対処するための短期的な措置に過ぎません。私が言ったように、あなたのソケットがあなたのソースよりも遅い場合、根本的な問題があります。最終的にはバッファリングの限界に達し、データをスキップしたり、接続を閉じる必要があります。

全体的に、なぜ速度がtee()とsplice()である必要があり、使用のためにメモリまたはディスクのユーザー空間のバッファリングがより適切かどうかを慎重に検討します。しかし、速度が常に高くなることが確かで、制限されたバッファリングが許容されるならば、上で概説したアプローチが有効になります。

はまた、私が言及すべき一つのことは、これはあなたのコードは非常にLinux固有作るということです - 私は他のUnixの変種でサポートされてこれらの呼び出しを認識していませんよ。 sendfile()コールはsplice()よりも制限されていますが、移植性はかなり高くなる可能性があります。あなたが本当に物事を移植したい場合は、ユーザー空間のバッファリングに固執してください。

は、あなたが上で詳細をたいと思い、私が紹介してきたものがありますなら、私に教えてください。