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

ライブトラックの公開

このページで扱うのは、クロックに合わせて配信するトラックです。 wired_moqt_publish_live は、フラグメント化された MP4 の各フラグメントを、 連続する Group（MoQT の配信単位）に変換します。 Group は group_ms ごとに 1 つ進み、サーバーのクロックがその Group の区間に入った瞬間、 すべての購読者へ送られます。

コード

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


/* send_uni2 (moqtrun.h): a live Group's framing (head) plus the fragment
 * bytes (body, a view this file does not own) go out on their own fresh
 * uni stream. wired_server_wt_open_uni holds a payload past its staging
 * as a VIEW until ACKed (srvrun.h), so the signal+head+body bytes must
 * stay put until then -- a single reused buffer would risk splicing a
 * later Group into an earlier one still in flight (the exact bug
 * examples/moqt_chat's send_uni2 comment documents). This demo expects
 * one session and a handful of small Groups, so a 2-slot ring (reused
 * once wired_server_wt_stream_inflight reports the slot's stream done) is
 * enough; a multi-session server needs one ring per session (see
 * examples/moqt_chat/wired_server.c). */
#define LIVE_RING 2
#define LIVE_SLOT_CAP 128
typedef struct {
  u64 stream_id;
  int used;
  u8  buf[LIVE_SLOT_CAP];
} live_slot;
static wired_wt_session* g_live_session;
static live_slot         g_live_ring[LIVE_RING];


static int live_slot_free(const live_slot* slot) {
  if (!slot->used) return 1;
  return !wired_server_wt_stream_inflight(g_live_session, slot->stream_id);
}


static live_slot* live_slot_claim(void) {
  for (usz i = 0; i < LIVE_RING; i++)
    if (live_slot_free(&g_live_ring[i])) return &g_live_ring[i];
  return 0;
}


static i64 send_uni2(wired_wt_session* s, wired_span head, wired_span body) {
  live_slot* slot = live_slot_claim();
  usz        sig;
  if (!slot || head.n + body.n > sizeof slot->buf - 9) return -1;
  g_live_session = s;
  sig            = wired_wtwire_signal_put(
      slot->buf, sizeof slot->buf, 0, s->connect_stream_id);
  if (sig == 0) return -1;
  memcpy(slot->buf + sig, head.p, head.n);
  memcpy(slot->buf + sig + head.n, body.p, body.n);
  i64 sid = wired_server_wt_open_uni(
      s, wired_span_of(slot->buf, sig + head.n + body.n));
  if (sid >= 0) {
    slot->stream_id = (u64)sid;
    slot->used      = 1;
  }
  return sid;
}


static wired_moqt_hub g_hub;
static mp4frag_layout g_layout;


static void on_step(void* ctx, u64 now_ms) {
  wired_moqt_tick((wired_moqt_hub*)ctx, now_ms);
}


/* Group 0 starts not at server boot but at the first WebTransport session
 * -- otherwise Group 0's window includes the client's own connect/
 * handshake latency, which on a loaded machine can exceed group_ms and
 * make the first Group (and so the whole fragment-size sequence the page
 * prints) nondeterministic. Anchoring here instead means the only
 * latency left before the first SUBSCRIBE is one already-established
 * QUIC round trip (SETUP out, SUBSCRIBE back), far inside the margin. */
static int g_live_published = 0;


static void publish_live_once(void) {
  if (g_live_published) return;
  g_live_published = 1;
  wired_moqt_publish_live(
      &g_hub, wired_span_cstr("movie"), 2, g_layout.frags, g_layout.n_frags,
      300, clock_mono_ms());
}


static void on_session(
    void* ctx, wired_wt_session* s, wired_span path, wired_span protocol) {
  publish_live_once();
  wired_moqt_on_session(ctx, s, path, protocol);
}


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


  wired_moqt_io io = wired_moqraw_io();
  io.send_uni2     = send_uni2;
  wired_moqt_init(&g_hub, io);


  /* movie-live.mp4 (committed next to this file, see gen_movie.py) is a
   * tiny synthetic fragmented MP4: an ftyp+moov init segment, then 3
   * moof+mdat fragments -- real box structure, filler media bytes (this
   * demo is about MoQT Group delivery, not video decoding). */
  static u8 file[4096];
  ssz n = wired_fio_read("movie-live.mp4", wired_mspan_of(file, sizeof file));
  if (n <= 0 || !mp4frag_scan(wired_span_of(file, (usz)n), &g_layout)) {
    wired_log_str("cannot scan movie-live.mp4\n");
    return 1;
  }


  /* "movie/init" (Track Alias 1): the init segment, sent once per
   * subscriber on its own stream (wired_moqt_publish_blob). */
  static u8 init_wire[MOQDATA_BLOB_WIRE_CAP(64)];
  wired_moqt_publish_blob(
      &g_hub, wired_span_cstr("movie/init"), 1, g_layout.init,
      wired_mspan_of(init_wire, sizeof init_wire));


  /* "movie" (Track Alias 2): Group g carries fragment g mod n_frags,
   * advancing every 300ms (draft-ietf-moq-transport-22 2.3's Group
   * model) -- publish_live_once (above) defers the actual
   * wired_moqt_publish_live call to the first WebTransport session so
   * Group 0 starts there, not at this still-booting instant. "moqt-22"
   * selects draft-22, as on the hub page. */
  wired_srvrun_opt opt     = {0};
  opt.incoming_cpu         = -1;
  opt.wt_protocols         = "moqt-22";
  opt.on_step              = on_step;
  opt.on_step_ctx          = &g_hub;
  opt.wt_on_session        = on_session;
  opt.wt_session_ctx       = &g_hub;
  opt.wt_on_stream_data    = wired_moqt_on_stream_data;
  opt.wt_stream_data_ctx   = &g_hub;
  opt.wt_on_datagram       = wired_moqt_on_datagram;
  opt.wt_datagram_ctx      = &g_hub;
  opt.wt_on_session_close  = wired_moqt_on_session_close;
  opt.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"
  "context"
  "fmt"
  "os"
  "time"


  "wired-guide/moqtclient"
)


func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
  defer cancel()
  // "moqt-22" session: SETUP exchanged on the two uni control streams.
  s := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer s.Conn.CloseWithError(0, "")


  // movie/init first: its SUBSCRIBE_OK, then its one-shot blob stream,
  // both read before movie is subscribed, so the next uni stream is
  // movie's.
  if reply, _ := s.Subscribe(ctx, 0, "movie/init"); !reply.Ok {
    fmt.Println("movie/init REQUEST_ERROR", reply.ErrCode)
    return
  }
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  _, _, initPayload := moqtclient.ReadObject(bufio.NewReader(uni))
  fmt.Printf("movie/init bytes=%d\n", len(initPayload))


  // movie: SUBSCRIBE_OK, then the current Group at once, then one more
  // Group every ~300ms as the server's clock advances.
  if reply, _ := s.Subscribe(ctx, 2, "movie"); !reply.Ok {
    fmt.Println("movie REQUEST_ERROR", reply.ErrCode)
    return
  }
  const groups = 4
  for i := 0; i < groups; i++ {
    u, err := s.AcceptUniStream(ctx)
    moqtclient.Check(err)
    _, group, payload := moqtclient.ReadObject(bufio.NewReader(u))
    fmt.Printf("movie group=%d bytes=%d\n", group, len(payload))
  }
}

実行結果

Terminal window
$ ./moqt-live --port 4433
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
movie/init bytes=28
movie group=0 bytes=36
movie group=1 bytes=44
movie group=2 bytes=52
movie group=3 bytes=36

仕組み

Section titled “仕組み”
  • mp4frag_scan は、フラグメント化された MP4 のトップレベルボックス （ISO/IEC 14496-12）を走査し、init セグメント（ftyp+moov）と、 moof+mdat の各フラグメントへのビュー（コピーではなく、元のバイト列を指す参照）の一覧に分けます。 サンプルテーブルは解析せず、公開に必要な境界だけを読みます。 movie-live.mp4（main.c の隣に置かれており、gen_movie.py で再生成できます）は、 小さな合成ファイルです。 ボックス構造は本物ですが、メディアのバイト列はダミーです。 このページが示すのは MoQT の Group 配信であり、動画のデコードではありません。
  • init セグメントは、前のページと同じく wired_moqt_publish_blob で、 単独のトラック（movie/init）として公開します。
  • wired_moqt_publish_live(&hub, name, alias, frags, n_frags, group_ms, now_ms) は、 各フラグメントをトラック movie として公開します。 Group g は、フラグメント g mod n_frags を運びます。 SUBSCRIBE には SUBSCRIBE_OK とその時点の Group が即座に返ります。 wired_srvrun_opt.on_step に接続した wired_moqt_tick が、 クロックが新しい Group に入った瞬間、それ以降の各 Group を送ります。
  • now_ms（Group 0 の起点）は、サーバーの起動時ではなく、最初の WebTransport セッションが張られた瞬間に取得します。 publish_live_once（wt_on_session から呼ばれます）は、クライアントが実際に 接続したその瞬間、一度だけ実行されます。 起動時に取得してしまうと、クライアント自身の接続・ハンドシェイクにかかる時間が Group 0 の区間に混ざり込みます。 負荷の高いマシンではこれが group_ms を超え、Group 0（とそれ以降の全 Group） が運ぶフラグメントがずれてしまいます。
  • Group の配信には、もう一つの io エントリ send_uni2 が必要です。 ハブは、短いフレーミング（head）を組み立て、ビューとして持っているだけの フラグメント（body）の手前に置きます。 アダプタ側は、WebTransport の signal の後ろでこの両方を連結し、 自前の領域へ書き込みます。 この領域は、wired_server_wt_stream_inflight がそのストリームを送信中ではないと 報告するようになるまで、書き換えてはいけません。 このページがスロットの小さなリングを持っているのはこのためです。 examples/moqt_chat のサーバーが、より大きな規模で使っているのと同じ仕組みです。