ストリームを止める・中断する

このページでは、クライアント側から開いたストリームに背圧（バックプレッシャー）をかけます。アプリの処理が追いつく速さを超えて読み進めないよう、受信クレジット（ピアにあとどれだけ送らせるかの枠）を止めてから、あとで戻します。これとは別に、アプリがもう必要としないサーバー送信ストリームを中断する方法と、それがまだ送信中かどうかを調べる方法も示します。

コード

#define WIRED_MAIN
#include "wired.h"

/* A server-opened stream, kept open, whose inflight state the client's
 * second round below resets. */
static i64 g_uni  = -1;
static int g_held = 0;

static void on_session(
    void* ctx, wired_wt_session* s, wired_span path, wired_span protocol) {
  (void)ctx;
  (void)path;
  (void)protocol;
  u8  buf[16];
  usz n = wired_wtwire_signal_put(buf, sizeof buf, 0, s->connect_stream_id);
  memcpy(buf + n, "ready", 5);
  g_uni = wired_server_wt_open_uni_stream(s, wired_span_of(buf, n + 5));
}

/* An upload split across two writes: hold receive credit on the first
 * (pausing the peer's send window -- bytes it already has credit for
 * still arrive, nothing new is granted), release once it ends with FIN.
 * The final round's word then exercises reset + inflight on g_uni. */
static void on_stream_data(
    void* ctx, wired_wt_session* s, u64 stream_id, wired_span data, int fin) {
  (void)ctx;
  if (!fin) {
    if (!g_held) {
      wired_server_wt_stream_hold(s, stream_id, 1);
      wired_log_str("held\n");
      g_held = 1;
    }
    return;
  }
  if (g_held) {
    wired_server_wt_stream_hold(s, stream_id, 0);
    wired_log_str("released\n");
    g_held = 0;
  }
  if (wired_span_eq_cstr(data, "abort")) {
    wired_server_wt_stream_reset(s, (u64)g_uni, 1);
    wired_log_str("reset\n");
  }
  wired_dprintf(
      2, "inflight %d\n", wired_server_wt_stream_inflight(s, (u64)g_uni) != 0);
  wired_server_wt_stream_reply(s, stream_id, wired_span_cstr("ok"));
}

int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the HTTP/3 hello page. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-wt");
  id.max_datagram_frame_size = 65535; /* WebTransport requires DATAGRAM */

  wired_srvrun_opt opt  = {0};
  opt.incoming_cpu      = -1;
  opt.wt_on_session     = on_session;
  opt.wt_on_stream_data = on_stream_data;

  u16 port                 = (u16)wired_cliargs_int(argc, argv, "--port", 4433);
  wired_srvrun_handler h   = {0};
  wired_srvrun_obs     obs = {0};
  return wired_server_run_opt(port, &id, h, obs, &opt) ? 0 : 1;
}

クライアント

package main

import (
  "context"
  "crypto/tls"
  "fmt"
  "io"
  "net/http"
  "os"
  "time"

  "github.com/quic-go/quic-go"
  "github.com/quic-go/webtransport-go"
)

// sendRound opens a fresh bidi stream, writes parts as separate writes (so
// the server sees each as its own reassembly delta), closes it, and
// returns the server's reply.
func sendRound(ctx context.Context, s *webtransport.Session, parts ...string) string {
  str, err := s.OpenStreamSync(ctx)
  if err != nil {
    panic(err)
  }
  for i, p := range parts {
    str.Write([]byte(p))
    if i < len(parts)-1 {
      time.Sleep(100 * time.Millisecond) // let the hold take effect first
    }
  }
  str.Close()
  ack, _ := io.ReadAll(str)
  return string(ack)
}

func main() {
  ctx := context.Background()
  var conn *quic.Conn // kept so main can close it before exiting
  d := webtransport.Transport{
    TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
    QUICConfig:      &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true},
    DialAddr: func(ctx context.Context, addr string, t *tls.Config, q *quic.Config) (*quic.Conn, error) {
      c, err := quic.DialAddrEarly(ctx, addr, t, q)
      conn = c
      return c, err
    },
  }
  _, s, err := d.Dial(ctx, os.Args[1]+"/upload", http.Header{})
  if err != nil {
    panic(err)
  }
  defer conn.CloseWithError(0, "")

  fmt.Println("ack1", sendRound(ctx, s, "part1", "part2"))
  fmt.Println("ack2", sendRound(ctx, s, "abort"))
}

実行結果

$ ./wt-backpressure --port 4433
held
released
inflight 1
reset
inflight 0
conn closed by peer (CONNECTION_CLOSE)

$ go run . https://127.0.0.1:4433
ack1 ok
ack2 ok

仕組み

クライアントのアップロードの最初のデルタ（fin=0）で wired_server_wt_stream_hold(s, stream_id, 1) を呼びます。ループは、ピアに知らせる受信可能量 MAX_STREAM_DATA を上げるのを止めます。すでに許可した分のバイトはそれでも届きますが、新しいクレジットは渡されません。
最後のデルタ（fin=1）で hold(..., 0) を呼んで再開し、 on_session で開いた別のサーバー送信ストリームに対して wired_server_wt_stream_inflight(s, g_uni) を調べます。まだ開いているか、ACK（受信確認）が済んでいない間は 1 です。
代わりに "abort" を送ると wired_server_wt_stream_reset(s, g_uni, 1) を呼び、そのストリームの送信スロットを即座に解放します。直後に inflight を読むと 0 になります。
クライアントは、各アップロードを閉じる前に 2 回に分けて書き込みます。そのため、サーバーは 1 回ではなく 2 回の再構成デルタを見ます。 hold が働くのは、最後のチャンクより前にもう 1 つチャンクがあるときだけです。