バックプレッシャー: 遅い WebSocket クライアントのための送信キュー制限 | eSeGeCe Blog

バックプレッシャー: 遅い WebSocket クライアントのための送信キュー制限

· コンポーネント

高速なフィードをブロードキャストするサーバーは、最も遅いクライアントのペースではなく、自分自身のペースでメッセージを生み出します。マーケットデータ、テレメトリ、ライブのログ追尾。サーバーはプッシュし、接続しているすべてのクライアントはそれについてくることを期待されます。ほとんどのクライアントはついてきます。しかし遅かれ早かれ、そのうちの 1 つがついてこなくなります。

クライアントがソケットを読み出さなくなっても、送ったメッセージは消えてなくなりません。その接続の送信キューに積み上がり、キューはメモリです。列車のトンネルに入ったスマートフォン、スロットリングされたブラウザタブ、読み取りを呼ばなくなったクライアントプロセス。そのどれもが静かにサーバーのメモリを膨らませ、やがて何かが破綻します。sgcWebSockets 2026.7 では、そのキューに上限を設け、いっぱいになったときにどうするかを決められます。

遅いコンシューマー

問題はネットワークが遅いことではなく、ペースの食い違いです。フィードは毎秒 500 メッセージを発行します。回線状態の悪いクライアントは 50 しか消化できません。残りの 450 はどこかへ行かなければならず、その「どこか」とはサーバープロセス内の接続ごとのリストです。1 分後、その接続だけで数万件のメッセージを抱えています。10 分後には、あなたの RAM を抱えています。

被害は遅いクライアントだけにとどまりません。それが食いつぶすメモリは、健全なクライアントが得られなくなるメモリです。そしてプロセスがスワップを始めたり、アロケーターが苦しみ始めたりすれば、全員のレイテンシが道連れになります。トンネルの中のスマートフォン 1 台が、全員にとっての障害になるのです。

これまで sgcWebSockets の送信キューは無制限でした。すべてのクライアントがついてこられる間は問題ありませんが、1 つでもついてこられなくなった瞬間に厄介なことになります。解決策は、そこに上限を設けることです。

キューに上限を設ける

キューはコンポーネントの QueueOptions の下、既存の Text、Binary、Ping の各レベルのサブオプションの隣にあります。それに上限を設ける新しいプロパティが 2 つあります。MaxQueueSize は単一の接続が保持できる保留メッセージ数の上限を定め、OverflowPolicy はその上限に達した後に届いた次のメッセージをどうするかを決めます。

uses
  sgcWebSocket, sgcWebSocket_Classes, sgcWebSocket_Types;

begin
  // messages are queued per connection and flushed by the queue thread
  oServer.QueueOptions.Text.Level   := qmLevel1;
  oServer.QueueOptions.Binary.Level := qmLevel1;

  // at most 1000 pending messages per connection
  oServer.QueueOptions.MaxQueueSize := 1000;

  // when the queue is full, throw away the oldest pending message
  oServer.QueueOptions.OverflowPolicy := qopDropOldest;
end;

QueueOptions は TsgcWebSocketServer、TsgcWebSocketHTTPServer、TsgcWebSocketClient で published されており、コンポーネントは接続を受け入れるたびにその設定を各接続へコピーします。したがって制限はサーバー単位ではなく接続単位です。MaxQueueSize = 1000 で 1000 接続なら、最悪の場合は各接続に 1000 件の保留メッセージということになります。そしてその最悪ケースは、デプロイ前に紙の上で掛け算できる数字になりました。

オーバーフローポリシーを選ぶ

ここに普遍的な正解はありません。だからこそ 3 つ用意されています。どのポリシーが欲しいかは、メッセージが何を意味するかによって決まります。

qopDropOldest は、最新の値だけが意味を持つライブフィードに適しています。価格のティック、センサーの読み取り値、ダッシュボードのゲージ。4 秒前のメッセージは、次のメッセージが届いた瞬間に価値を失います。キューの先頭から捨てるということは、遅れて後から復帰したクライアントが、追いつくまで噛み砕かなければならない古いメッセージの山ではなく、現在の状況を受け取るということです。これが既定値です。

qopDropNewest は、意味を持つのが先に届いたメッセージのほうであり、新しい負荷を切り捨てたい場合に適しています。クライアントに一連のシーケンスを順に見せていて、その先頭部分こそが重要な意味を持つのなら、すでにキューにあるものを保持し、新参のメッセージを拒否します。

qopDisconnect は、ついてこられないクライアントなら要らない、という場合に適しています。これは正直な選択肢です。その接続に、穴だらけで壊れたストリーム像を黙って流し込む代わりに、接続を閉じるのです。クライアントの WatchDog が再接続し、再購読し、既知の状態からやり直します。切断されたことを知っているクライアントは復旧できます。メッセージを失ったことを知らないクライアントは復旧できません。

// A market data feed: only the newest tick is worth anything.
oServer.QueueOptions.MaxQueueSize   := 500;
oServer.QueueOptions.OverflowPolicy := qopDropOldest;

// A sequence where the head matters: shed the new load instead.
oServer.QueueOptions.OverflowPolicy := qopDropNewest;

// A client that falls this far behind is a client we do not want.
oServer.QueueOptions.OverflowPolicy := qopDisconnect;

どれを選ぶにせよ、意識して選んでください。メッセージを捨てることはトレードオフであり、この 3 つの値の意義は、本番で初めて思い知らされるのではなく、どのトレードオフを選ぶのかを自分で決められる点にあります。

捨てる前に知る: OnSendBufferFull

黙ってメッセージを捨てるキューは、あなたに嘘をつくキューです。OnSendBufferFull はその逃げ道です。何かが破棄される前に発火し、接続と現在のキューサイズを渡してくれます。そして var DropMessage パラメーターによって、コードから破棄を拒否できます。

procedure TForm1.WSServerSendBufferFull(Connection: TsgcWSConnection;
  const QueueSize: Integer; var DropMessage: Boolean);
begin
  // fires BEFORE anything is dropped: log it, count it, alert on it
  DoLog('slow consumer ' + Connection.IP + ' [' + Connection.Guid + '] queue=' +
    IntToStr(QueueSize));
  Inc(FSlowConsumerHits);

  // this connection is a paying feed we must not truncate: keep the message
  // and let the queue grow past the cap for this one send
  if IsPriorityConnection(Connection) then
    DropMessage := False;
end;

DropMessage := False を設定すると、そのメッセージについてはオーバーフローポリシーがスキップされ、上限を超えていてもキューに積まれます。これは控えめに使ってください。常に拒否権を行使するなら、制限がないのとまったく同じだからです。このイベントの日常的な使い方は地味なものであり、そしてそれこそが価値ある使い方です。つまり、それが起きたこと、どの接続で起きたこと、そしてどれだけ遅れていたかを知ることです。その数値は、プロセスがメモリを使い果たすはるか前に届く早期警告です。

OnSendBufferFull は TsgcWebSocketClient、TsgcWebSocketServer、TsgcWebSocketHTTPServer で published されています。

何ひとつ捨てられない場合にどうするか

ときには、1 件たりともメッセージを失ってはならない、というのが正直な答えであることもあります。そういう場合は、上の 3 つのポリシーはどれも求めるものではありませんし、キューを大きくすることも答えではありません。キューを大きくしても問題は解決せず、先送りするだけです。しかも、より悪い瞬間へと先送りします。バーストのさらに奥、メモリ圧迫のさらに深いところ、そしてまだ何か手を打てたはずの地点からさらに遠いところへ、です。

答えは別の設計です。未配信のメッセージをソケットのキューに抱えるのではなく、永続化して、クライアントが可能になったときに取りこぼしを取得できるようにするのです。それこそがメッセージ履歴と再接続リカバリーの役割です。サーバーはチャネルごとに上限つきの履歴をオフセット付きで保持し、遅れたり接続が切れたりしたクライアントは、再購読時に取りこぼしたメッセージを再生します。これを qopDisconnect と組み合わせれば、利用可能な最強の組み合わせが得られます。ついてこられないクライアントは素早く切り離され、戻ってきたときには、失っていないふりをするのではなく、取りこぼしたすべてを回復するのです。

既定は今も無制限

MaxQueueSize の既定は 0、つまり無制限であり、これはまさに sgcWebSockets のこれまでの挙動です。アップグレードしても何も変わりません。正の上限を設定しない限りオーバーフローの経路に入ることすらないので、既存のサーバーは、あなたが別の判断を下すまでこれまでどおりキューに積み続けます。

とはいえ、自分の管理下にないクライアントへブロードキャストしているなら、そしてインターネット上ではクライアントを管理下に置くことは決してできませんが、接続ごとの無制限キューとは、いつか接続してくる最悪のクライアントのために青天井のメモリを約束することにほかなりません。上限の設定にかかるコストは 1 行です。

提供状況

送信キュー制限、3 つのオーバーフローポリシー、そして OnSendBufferFull は sgcWebSockets 2026.7 で、Delphi 7 から 13 および C++Builder 向けに、Win32/Win64、Linux64、macOS、Android、iOS で提供されます。

有効なサブスクリプションをお持ちのお客様は、カスタマーエリアから新しいビルドをダウンロードできます。トライアルをご利用の方は esegece.com/products/websockets/download から更新版のインストーラーを入手できます。

ご質問、ご意見、あるいはご自身のフィードに適したポリシー選びのお手伝いが必要ですか。お問い合わせください。コードを書いた本人から返信が届きます。