コンテンツにスキップ
wired guide

優先度と GOAWAY

帯域が足りないときは、音声を映像より先に送るべきです。 サーバーを止めるときは、接続を突然切るのではなく、セッションに移動先と猶予を与えるべきです。 このページでは、その両方を動かします。 優先度の異なる 2 つの購読が、それぞれ別の送出クラス（送る順番を決める値の組）のリレーストリームを受け取ります。 その後のドレイン（接続を保ったまま段階的に止める手順）では、ハブが GOAWAY（まもなく閉じることを伝えるメッセージ）を送り、新しいリクエストを拒否し、許可済みの購読を終わらせ、猶予が尽きたところでセッションを閉じます。

コード

main.c
#define WIRED_MAIN
#include "wired.h"


/* The hub calls this right after it opens a subscriber stream, with the
 * stream's send class: urgency 4 for every Object stream, then the
 * subscription's Subscriber Priority and the stream's Publisher Priority
 * packed into fine (lower is sent first), then its place inside the
 * subscription. Logging it makes the class visible; the scheduling itself
 * is wired_server_wt_stream_schedule's job. */
static int stream_sched(
    wired_wt_session* s, u64 stream_id, const wired_wt_sched* k) {
  wired_dprintf(
      2, "subscriber stream urgency=%u subscriber=%u publisher=%u\n",
      k->urgency, k->fine >> 8, k->fine & 0xff);
  return wired_server_wt_stream_schedule(s, stream_id, k);
}


static wired_moqt_hub g_hub;


/* Any datagram is the operator's "drain now" trigger in this demo (a real
 * server would call this on SIGTERM or before a deploy). Every session
 * gets GOAWAY pointing at the new URI, with 300 ms to leave (short so the
 * page's client stays well inside its own deadline; the hub still waits
 * WIRED_MOQTRUN_GOAWAY_GRACE_MS past that before it actually closes). */
static void on_datagram(void* ctx, wired_wt_session* s, wired_span data) {
  (void)s;
  (void)data;
  static const char uri[] = "https://relay2.example/moqt";
  wired_moqt_goaway((wired_moqt_hub*)ctx, wired_span_cstr(uri), 300);
  wired_log_str("GOAWAY sent\n");
}


/* The tick enforces the GOAWAY Timeout: PUBLISH_DONE once it passes, the
 * session close on the tick after. */
static void on_step(void* ctx, u64 now_ms) {
  wired_moqt_tick((wired_moqt_hub*)ctx, now_ms);
}


int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the other MoQT pages, plus QUIC
   * DATAGRAM support, which WebTransport clients ask for. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-pg");
  id.max_datagram_frame_size = 65535;


  /* stream_sched turns the subscription's priority into the stream's
   * send class; stream_reset and close_session are what the drain uses to
   * end subscriptions and sessions once the GOAWAY Timeout passes. GOAWAY
   * itself goes out on the hub's uni control stream ("moqt-22", 9.2). */
  wired_moqt_io io = wired_moqraw_io();
  io.stream_sched  = stream_sched;
  wired_moqt_init(&g_hub, io);


  wired_srvrun_opt opt = {
      .incoming_cpu         = -1,
      .wt_protocols         = "moqt-22",
      .on_step              = on_step,
      .on_step_ctx          = &g_hub,
      .wt_on_session        = wired_moqt_on_session,
      .wt_session_ctx       = &g_hub,
      .wt_on_stream_data    = wired_moqt_on_stream_data,
      .wt_stream_data_ctx   = &g_hub,
      .wt_on_stream_reset   = wired_moqt_on_stream_reset,
      .wt_stream_reset_ctx  = &g_hub,
      .wt_on_datagram       = on_datagram,
      .wt_datagram_ctx      = &g_hub,
      .wt_on_session_close  = wired_moqt_on_session_close,
      .wt_session_close_ctx = &g_hub,
  };


  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;
}

クライアント

client.go
package main


import (
  "bufio"
  "bytes"
  "context"
  "errors"
  "fmt"
  "io"
  "log"
  "os"
  "time"


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


  "wired-guide/moqtclient"
)


// publish announces track guide/name with Track Alias alias: PUBLISH
// (0x1D), answered REQUEST_OK (0x7) on its request stream.
func publish(ctx context.Context, s *moqtclient.Session, id byte, name string, alias byte) {
  _, r := s.Request(ctx, 0x1D, append(moqtclient.Track(id, name), alias, 0))
  if t, _ := moqtclient.ReadMsg(r); t != 0x7 {
    log.Fatalf("PUBLISH %s answered with type %#x", name, t)
  }
}


// subscribe sends SUBSCRIBE (0x3) for guide/name with one parameter,
// SUBSCRIBER_PRIORITY (0x20, a single byte: lower is more important), and
// returns the request stream, on which PUBLISH_DONE arrives later.
func subscribe(ctx context.Context, s *moqtclient.Session, id byte, name string, prio byte) *bufio.Reader {
  reply, r := s.Subscribe(ctx, id, name, 1, 0x20, prio)
  if !reply.Ok {
    log.Fatalf("SUBSCRIBE %s refused, error_code=%d", name, reply.ErrCode)
  }
  fmt.Printf("subscriber: SUBSCRIBE %s priority=%d -> SUBSCRIBE_OK\n", name, prio)
  return r
}


// sendGroup starts Group 0 of track alias on a uni stream and leaves the
// stream open, the way a live encoder keeps writing: SUBGROUP_HEADER
// (Type 0x70: FIRST_OBJECT, which the Original Publisher opening a
// subgroup MUST set (SS2.2), Subgroup 0, default Publisher Priority),
// Object ID delta 0, length, payload.
func sendGroup(ctx context.Context, s *moqtclient.Session, alias byte, payload string) {
  st, err := s.OpenUniStreamSync(ctx)
  moqtclient.Check(err)
  _, err = st.Write(append([]byte{0x70, alias, 0, 0, byte(len(payload))}, payload...))
  moqtclient.Check(err)
}


// receive accepts the subscriber's relay stream and prints its first
// Object; the stream stays open and is returned.
func receive(ctx context.Context, s *moqtclient.Session, name string) *bufio.Reader {
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  r := bufio.NewReader(uni)
  _, group, payload := moqtclient.ReadObject(r)
  fmt.Printf("subscriber: %s group %d %q\n", name, group, payload)
  return r
}


func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
  defer cancel()


  // The publisher announces two tracks.
  pub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer pub.Conn.CloseWithError(0, "")
  publish(ctx, pub, 0, "video", 1)
  publish(ctx, pub, 2, "audio", 2)


  // The subscriber's session. The hub's GOAWAY arrives later on the hub's
  // uni control stream, the one whose SETUP Dial already read.
  sub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer sub.Conn.CloseWithError(0, "")


  // Audio matters more than video here: priority 10 against 200. The
  // server log shows the urgency the hub gave each relay stream.
  videoReq := subscribe(ctx, sub, 0, "video", 200)
  audioReq := subscribe(ctx, sub, 2, "audio", 10)
  sendGroup(ctx, pub, 1, "keyframe")
  video := receive(ctx, sub, "video")
  sendGroup(ctx, pub, 2, "frame")
  audio := receive(ctx, sub, "audio")


  // Ask the server to drain (this demo's trigger is any datagram).
  moqtclient.Check(sub.SendDatagram([]byte("drain")))
  typ, body := moqtclient.ReadMsg(sub.Control)
  if typ != 0x10 {
    log.Fatalf("expected GOAWAY, got type %#x", typ)
  }
  br := bufio.NewReader(bytes.NewReader(body))
  uri := make([]byte, moqtclient.Varint(br))
  _, err := io.ReadFull(br, uri)
  moqtclient.Check(err)
  fmt.Printf("subscriber: GOAWAY new_uri=%s timeout=%dms\n", uri, moqtclient.Varint(br))


  // After GOAWAY every new request is refused: REQUEST_ERROR (0x5)
  // GOING_AWAY (0x6). A real client SHOULD NOT start one (SS9.2) and
  // would open its next request on new_uri; this sample sends one on
  // purpose, to show the refusal.
  if reply, _ := sub.Subscribe(ctx, 4, "video"); !reply.Ok {
    fmt.Printf("subscriber: new SUBSCRIBE -> REQUEST_ERROR error_code=%d\n", reply.ErrCode)
  } else {
    log.Fatal("new SUBSCRIBE was accepted after GOAWAY")
  }


  // The subscriptions already granted keep flowing until the timeout.
  // Then the hub resets their relay streams and ends each with
  // PUBLISH_DONE (0xB) GOING_AWAY (0x4) on its request stream.
  for _, st := range []struct {
    name      string
    relay, rq *bufio.Reader
  }{{"video", video, videoReq}, {"audio", audio, audioReq}} {
    _, err := st.relay.ReadByte()
    var se *webtransport.StreamError
    if !errors.As(err, &se) {
      log.Fatalf("%s relay stream: %v", st.name, err)
    }
    fmt.Printf("subscriber: %s relay stream reset error_code=%#x\n", st.name, se.ErrorCode)
    typ, body := moqtclient.ReadMsg(st.rq)
    if typ != 0xB {
      log.Fatalf("expected PUBLISH_DONE, got type %#x", typ)
    }
    fmt.Printf("subscriber: %s PUBLISH_DONE status=%d\n", st.name, moqtclient.Varint(bufio.NewReader(bytes.NewReader(body))))
  }


  // On the next tick the hub closes the session: GOAWAY_TIMEOUT (0x10),
  // since this client left its request streams open.
  _, err = sub.AcceptUniStream(ctx)
  var se *webtransport.SessionError
  if !errors.As(err, &se) {
    log.Fatal(err)
  }
  fmt.Printf("subscriber: session closed error_code=%#x\n", se.ErrorCode)
}

実行結果

Terminal window
$ ./moqt-priority-goaway --port 4433
subscriber stream urgency=4 subscriber=200 publisher=128
subscriber stream urgency=4 subscriber=10 publisher=128
GOAWAY sent
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
subscriber: SUBSCRIBE video priority=200 -> SUBSCRIBE_OK
subscriber: SUBSCRIBE audio priority=10 -> SUBSCRIBE_OK
subscriber: video group 0 "keyframe"
subscriber: audio group 0 "frame"
subscriber: GOAWAY new_uri=https://relay2.example/moqt timeout=300ms
subscriber: new SUBSCRIBE -> REQUEST_ERROR error_code=6
subscriber: video relay stream reset error_code=0x4
subscriber: video PUBLISH_DONE status=4
subscriber: audio relay stream reset error_code=0x4
subscriber: audio PUBLISH_DONE status=4
subscriber: session closed error_code=0x10

仕組み

Section titled “仕組み”
  • SUBSCRIBE は SUBSCRIBER_PRIORITY パラメータ（0〜255、小さいほど重要、省略時は 128。draft-ietf-moq-transport-22 9.20.7、5.1.1）を運びます。 Publisher Priority は、ストリーム自身の SUBGROUP_HEADER が運びます。 ハブは購読者のリレーストリームを開くとき、そのストリームに送出クラス（wired_wt_sched）を与えます。 RFC 9218 urgency は 4 で、2 番目のキーが subscriber << 8 | publisher です（5.1.2 は購読者の優先度、次にパブリッシャーの優先度で順序づけると定めています）。 さらに購読の中での順番として、グループ順と、同じグループなら fill ストリームをライブのストリームより先に送る規則が加わります。 制御ストリームとリクエストストリームはトランスポートの既定値 3 のままなので、どのリレーストリームよりも先に送られます（5.1.2 の推奨どおりです）。
  • 送出クラスは、ハブの io 操作 stream_sched を通じてトランスポートに届きます。 このサンプルでは wired_server_wt_stream_schedule をログ付きで包んでいます。 音声（優先度 10）と映像（優先度 200）はどちらも urgency 4 ですが、2 番目のキーは音声のほうが小さいため、帯域が足りないときサーバーは音声のバイトを映像より先に送ります。 ログが見せるのは送出クラスまでで、スケジューリング自体はトランスポートの仕事です。
  • wired_moqt_goaway(hub, new_uri, timeout_ms) は、すべてのセッションの制御ストリームに、New Session URI 付きの GOAWAY（9.2）を送ります。 このサンプルの moqt-22 セッションでは、それはサーバーが開いた単方向の制御ストリームです（移行の手順は 6.6.1）。 このデモはデータグラムの受信を引き金にしていますが、実際のサーバーなら SIGTERM やデプロイ前に呼びます。 URI は WIRED_MOQTRUN_GOAWAY_URI_MAX（8192）バイトまでです。
  • GOAWAY の後、そのセッションの新しいリクエストはすべて REQUEST_ERROR で拒否されます。 エラーコードは GOING_AWAY（0x6）で、クライアントには次のリクエストを new_uri へ送ることが期待されます。 GOAWAY を受けたクライアントは新しいリクエストを始めるべきではありません（SHOULD NOT、9.2）。このサンプルは GOING_AWAY の応答を見せるため、あえて SUBSCRIBE を 1 つ送ります。
  • 許可済みの購読は、タイムアウトまで流れ続けます。 期限を守らせるのは、on_step から呼ぶハブの定期処理であるティック（wired_moqt_tick）です。 timeout_ms が過ぎると、各購読のリレーストリームがリセットされ、購読はそのリクエストストリーム上の PUBLISH_DONE（9.9）ステータス GOING_AWAY（0x4）で終わります。
  • このクライアントは期限を過ぎてもリクエストストリームを開いたままにしているため、次のティックでハブがセッションを閉じます。 セッションコードは GOAWAY_TIMEOUT（0x10、WIRED_MOQTRUN_CLOSE_GOAWAY_TIMEOUT）です。 期限内に自分のストリームとセッションを閉じるクライアントは、このコードを受け取りません。
  • ピア（接続相手）の側から退出を求めることもできます。 WT_DRAIN_SESSION を wired_moqt_on_session_draining につなぐと、そのセッションだけが同じ手順でドレインされます（URI は空、猶予は 5000 ミリ秒）。