moqtrun.h

Go to the documentation of this file.

#ifndef WIRED_MOQTRUN_H
#define WIRED_MOQTRUN_H
 
#include "app/http3/server/srvrun/srvrun.h"
#include "app/moqt/cache/moqcache.h"
#include "app/moqt/ctl/moqctl.h"
#include "app/moqt/data/moqdata.h"
#include "app/moqt/fetch/moqfetch.h"
#include "app/moqt/qraw/moqraw.h"
#include "app/moqt/run/moqtrel.h"
#include "app/moqt/run/moqtss.h"
#include "app/moqt/sess/moqsess.h"
#include "common/bytes/span/span.h"
#include "common/platform/sys/syscall.h"


#define WIRED_MOQTRUN_MAX_SESSIONS 32

#define WIRED_MOQTRUN_MAX_SUBS (WIRED_MOQTRUN_MAX_SESSIONS - 1)
 
/* moqtrel.h's WIRED_MOQTREL_MAX_SUBS mirrors this value as its own
 * literal (it cannot include this header -- this header includes
 * moqtrel.h, moqtrel.h:50-53). This header includes moqtrel.h above, so
 * this is the one place both are visible together; keep it in sync here
 * if either changes. */
_Static_assert(
    WIRED_MOQTREL_MAX_SUBS == WIRED_MOQTRUN_MAX_SUBS,
    "the relay ring's subscriber cursor table must match the hub's cap");

typedef struct {
  i64 (*open_bidi_stream)(wired_wt_session* s, wired_span payload);
  int (*stream_send)(
      wired_wt_session* s, u64 stream_id, wired_span payload, int fin);
  i64 (*send_uni)(wired_wt_session* s, wired_span payload);
  i64 (*open_uni_stream)(wired_wt_session* s, wired_span payload);
  int (*stream_fin)(wired_wt_session* s, u64 stream_id);
  int (*stream_reset)(wired_wt_session* s, u64 stream_id, u32 error_code);
  i64 (*send_uni2)(wired_wt_session* s, wired_span head, wired_span body);
  int (*send_datagram)(wired_wt_session* s, wired_span payload);
  int (*stream_hold)(wired_wt_session* s, u64 stream_id, int hold);
  usz (*send_budget)(wired_wt_session* s);
  int (*close_session)(wired_wt_session* s, u32 error_code, wired_span reason);
  int (*stream_reply_open)(
      wired_wt_session* s, u64 stream_id, wired_span payload);
  int (*stream_priority)(wired_wt_session* s, u64 stream_id, u8 urgency);
  int (*stream_stop)(wired_wt_session* s, u64 stream_id, u32 error_code);
  u64 (*est_kbps)(wired_wt_session* s);
  int (*stream_sched)(
      wired_wt_session* s, u64 stream_id, const wired_wt_sched* k);
} wired_moqt_io;

#define WIRED_MOQTRUN_URGENCY(sub_prio, pub_prio) \
  ((u8)(4 + ((sub_prio) >> 7) * 2 + ((pub_prio) >> 7)))

#define WIRED_MOQTRUN_CLOSE_PROTOCOL_VIOLATION 0x3

#define WIRED_MOQTRUN_CLOSE_INTERNAL_ERROR 0x1

#define WIRED_MOQTRUN_CLOSE_NO_ERROR 0x0

#define WIRED_MOQTRUN_CLOSE_INVALID_REQUEST_ID 0x4

#define WIRED_MOQTRUN_CLOSE_GOAWAY_TIMEOUT 0x10

#define WIRED_MOQTRUN_CLOSE_INVALID_PATH 0x8

#define WIRED_MOQTRUN_CLOSE_INVALID_AUTHORITY 0x19

#define WIRED_MOQTRUN_CLOSE_MALFORMED_PATH 0x9

#define WIRED_MOQTRUN_CLOSE_MALFORMED_AUTHORITY 0x1A

#define WIRED_MOQTRUN_CLOSE_TOO_MANY_REQUEST_UPDATES 0x1B

#define WIRED_MOQTRUN_CLOSE_AUTH_TOKEN_CACHE_OVERFLOW 0x13

#define WIRED_MOQTRUN_CLOSE_UNKNOWN_AUTH_TOKEN_ALIAS 0x17

#define WIRED_MOQTRUN_MAX_FILTER_RANGES 4

#define WIRED_MOQTRUN_MAX_REQ_UPDATES 4

#define WIRED_MOQTRUN_IMPL_MAX 64

#define WIRED_MOQTRUN_HOLD_BUF 2048

#define WIRED_MOQTRUN_EARLY_BUF 2048

#define WIRED_MOQTRUN_GOAWAY_URI_MAX 8192

#define WIRED_MOQTRUN_GOAWAY_MSG_MAX ((usz)WIRED_MOQTRUN_GOAWAY_URI_MAX + 32)

#define WIRED_MOQTRUN_DRAIN_TIMEOUT_MS 5000

typedef struct {
  u64 ptype;     
  u64 prop_type; 
  u64 start;     
  u64 end;       
  u8 set_id;
  u8 has_prop;
  u8 has_end;
} wired_moqtrun_rngrow;

typedef struct {
  usz session_idx;
  u64 track_alias;
  int active;
  u64 request_id;
  u64 delivery_timeout;
  u64 subgroup_timeout;
  moqctl_loc start;
  u64 end_group;
  u64 end_object;
  u8 priority;
  u8 has_priority;
  u8 group_order;
  u8 forward_off;
  u8 has_delivery_timeout;
  u8 has_end_group;
  u8 has_end_object;
  moqctl_loc jl;
  u8 has_jl;
  moqctl_rangeloc filter;
  u8 has_filter;
  wired_moqtrun_rngrow rngf[WIRED_MOQTRUN_MAX_FILTER_RANGES];
  u8 rngf_n;
  u8 blob_sent;
  u8 no_props;
  u64 stream_count;
  u64 newest_group;
  u64 newest_sid;
  /* SSTS (moqtssts_run.c) */
  u64 ssts_base;
  u16 ssts_known;
  u16 ssts_fwd;
  u8 ssts_on;
  /* SWITCH_FROM (moqtswitch.c) */
  u8 sw_role;
  u8 sw_mode;
  u8 sw_done;
  u64 sw_g;
  u64 sw_peer_rid;
  u64 sw_deadline;
} wired_moqtrun_sub;

#define WIRED_MOQTRUN_TRACK_PROPS_MAX 32

#define WIRED_MOQTRUN_RELAY_HDR_MAX 40

#define WIRED_MOQTRUN_RELAY_FRAG_MAX 16384

#define WIRED_MOQTRUN_FRAG_POOL 8

typedef struct {
  int in_use;
  u64 pub_stream_id;
  u8 hdr[WIRED_MOQTRUN_RELAY_HDR_MAX];
  usz hdr_len;
  u8 default_pub_prio;
  i32 frag_idx;
  usz frag_len;
  u64 sub_stream_id[WIRED_MOQTRUN_MAX_SUBS];
  int sub_stream_set[WIRED_MOQTRUN_MAX_SUBS];
  u8 sub_busy_streak[WIRED_MOQTRUN_MAX_SUBS];
  i32 rel_idx;
  moqdata_objseq seq;
  u64 group_id;
  u64 frag_ms;
  u32 sub_expired;
  u32 sub_reframe;
  u64 subgroup_id;
  u8 pub_prio;
} wired_moqtrun_relay;

#define WIRED_MOQTRUN_RESET_AFTER_BUSY 8

#define WIRED_MOQTRUN_MAX_RELAYS 4

#define WIRED_MOQTRUN_MAX_NAME 64

#define WIRED_MOQTRUN_MAX_NS 128

#define WIRED_MOQTRUN_MAX_TRACKS_PER_PEER 4

#define WIRED_MOQTRUN_SUB_NAMES ((usz)WIRED_MOQTRUN_MAX_TRACKS_PER_PEER * 4)

#define WIRED_MOQTRUN_CTL_REPLY_MAX 64

#define WIRED_MOQTRUN_CTL_MSG_MAX 1024

#define WIRED_MOQTRUN_CTL_HDR_MAX 11

typedef struct {
  u8  buf[WIRED_MOQTRUN_CTL_MSG_MAX + WIRED_MOQTRUN_CTL_HDR_MAX];
  usz n;    
  usz at;   
  usz skip; 
} wired_moqtrun_ctl_asm;

#define WIRED_MOQTRUN_MAX_REQS 96

#define WIRED_MOQTRUN_MAX_REQS_PER_SESSION (WIRED_MOQTRUN_MAX_REQS / 4)

#define WIRED_MOQTRUN_CTL_REPLIES_MAX                               \
  ((usz)WIRED_MOQTRUN_CTL_REPLY_MAX * (usz)WIRED_MOQTRUN_MAX_SUBS * \
   (usz)WIRED_MOQTRUN_MAX_TRACKS_PER_PEER)

#define WIRED_MOQTRUN_CTL_SEND_BUF                              \
  (WIRED_MOQTRUN_CTL_REPLIES_MAX > WIRED_MOQTRUN_GOAWAY_MSG_MAX \
       ? WIRED_MOQTRUN_CTL_REPLIES_MAX                          \
       : WIRED_MOQTRUN_GOAWAY_MSG_MAX)

#define WIRED_MOQTRUN_REQ_SEND_BUF ((usz)WIRED_MOQTRUN_CTL_REPLY_MAX * 4)

typedef struct {
  int in_use;
  wired_wt_session* wt;
  u64 stream_id;
  u64 kind;
  u64 request_id;
  int opened;
  wired_moqtrun_ctl_asm in;
  u8 send_bufs[2][WIRED_MOQTRUN_REQ_SEND_BUF];
  usz send_lens[2];
  int armed_idx;
  u64 pending_updates;
  int goaway;
  int live;
  u8 rdv_held;
  int fin_in;
  int fin_out;
  int done_pending;
  u64 done_status;
  u64 done_count;
  u8 ns[WIRED_MOQTRUN_MAX_NS];
  usz ns_len;
  u64 ns_seen[(WIRED_MOQTRUN_MAX_REQS + 63) / 64];
  u64 attempted_tag
      [WIRED_MOQTRUN_MAX_SESSIONS * WIRED_MOQTRUN_MAX_TRACKS_PER_PEER];
  u64 pub_origin_rid;
  u64 pub_track_tag;
  u64 pub_alias;
  u8 has_forward;
  u8 forward;
  u8 group_order;
  wired_moqtrun_rngrow rngf[WIRED_MOQTRUN_MAX_FILTER_RANGES];
  u8 rngf_n;
  u8 pub_answered;
} wired_moqtrun_req;

#define WIRED_MOQTRUN_MAX_RDV 16

#define WIRED_MOQTRUN_RDV_PER_SESSION (WIRED_MOQTRUN_MAX_RDV / 4)

#define WIRED_MOQTRUN_RDV_MAX_MS 1500

#define WIRED_MOQTRUN_RDV_RETRY 1001

typedef struct {
  int in_use;
  wired_wt_session* wt;
  u64 stream_id;
  u64 deadline_ms;
  u8 ns[WIRED_MOQTRUN_MAX_NS];
  usz ns_len;
  u8 name[WIRED_MOQTRUN_MAX_NAME];
  usz name_len;
  u8 body[WIRED_MOQTRUN_CTL_MSG_MAX];
  usz body_len;
  u8 up_only;
  u8 repath;
  u64 done_status;
  u64 done_count;
  u64 tried;
} wired_moqtrun_rdv;

#define WIRED_MOQTRUN_MAX_UP WIRED_MOQTRUN_MAX_RDV

typedef struct {
  int in_use;
  wired_wt_session* wt;
  u64 stream_id;
  u64 request_id;
  u64 track_tag;
  u8 done;
  u8 ns[WIRED_MOQTRUN_MAX_NS];
  usz ns_len;
  u8 name[WIRED_MOQTRUN_MAX_NAME];
  usz name_len;
  wired_moqtrun_ctl_asm in;
} wired_moqtrun_up;

#define WIRED_MOQTRUN_MAX_FETCHES 8

#define WIRED_MOQTRUN_UPFETCH_BUDGET_MS 2000

#define WIRED_MOQTRUN_UPF_NONE 0
#define WIRED_MOQTRUN_UPF_SENT 1
#define WIRED_MOQTRUN_UPF_OK 2

#define WIRED_MOQTRUN_GOAWAY_GRACE_MS 1000

#define WIRED_MOQTRUN_PUBDONE_WAIT_MS 2000

typedef struct {
  wired_wt_session* wt;
  u64 request_id;
  u64 cache_tag;
  moqctl_loc cursor;
  moqctl_loc end;
  u64 stream_id;
  moqfetch_seq seq;
  u64 owner_rid;
  moqctl_loc lo;
  u64 last_ok_ms;
  wired_wt_session* up_wt;
  u64 up_req_sid;
  u64 up_rid;
  u64 up_data_sid;
  moqfetch_seq up_seq;
  moqctl_loc up_lo;
  moqctl_loc up_span_end;
  moqctl_loc up_last;
  u64 up_pl_left;
  u64 up_pl_at;
  moqctl_loc up_end;
  u64 up_deadline_ms;
  u64 up_budget_ms;
  u64 req_sid;
  int in_use;
  int opened;
  int is_fill;
  int failed;
  int descending;
  int up_state;
  int up_data_open;
  int up_data_end;
  int has_up_last;
  int up_eot;
  int up_tmo;
  int no_track;
  int answered;
} wired_moqtrun_fetch;

typedef struct {
  wired_moqtrun_ctl_asm in;
  u8 ns[WIRED_MOQTRUN_MAX_NS];
  usz ns_len;
  u8 name[WIRED_MOQTRUN_MAX_NAME];
  usz name_len;
} wired_moqtrun_upf_leg;

typedef struct {
  int in_use;
  u8 name[WIRED_MOQTRUN_MAX_NAME];
  usz name_len;
  u64 own_alias;
  wired_moqtrun_sub subs[WIRED_MOQTRUN_MAX_SUBS];
  wired_moqtrun_relay relays[WIRED_MOQTRUN_MAX_RELAYS];
  u8 ns[WIRED_MOQTRUN_MAX_NS];
  usz ns_len;
  moqctl_loc largest;
  int has_largest;
  u64 request_id;
  u64 subgroup_timeout_ms;
  u8 default_pub_prio;
  u64 cache_tag;
  u64 up_streams;
  u64 pubdone_status;
  u64 pubdone_streams;
  u64 pubdone_deadline;
  u8 pubdone_pending;
  moqctl_loc track_end;
  moqctl_loc group_end;
  u8 end_flags;
  u8 malformed;
  u8 props_len;
  u8 props[WIRED_MOQTRUN_TRACK_PROPS_MAX];
} wired_moqtrun_track;

typedef struct {
  /* --- 8-byte aligned: session, control streams, reassembly --- */
  wired_wt_session* wt;
  u64 control_stream_id;
  u64 peer_ctl_stream_id;
  u64 rx_sid;
  wired_moqtrun_ctl_asm* rx;
  wired_moqtrun_ctl_asm peer_ctl_asm;
  wired_moqtrun_ctl_asm ctl_asm;
  usz peer_impl_len;
  usz hold_len;
  usz early_n;
  u64 early_sid;
  int early_fin;
  u64 pre_sid;
  /* --- 8-byte aligned: Request IDs and session clocks --- */
  u64 request_id_next;
  u64 peer_rid_next;
  u64 join_seq;
  wired_moqtrun_req* req;
  u64 goaway_deadline;
  u64 goaway_flushed_at;
  /* --- 8-byte aligned: tracks, remembered subscriptions, replies --- */
  wired_moqtrun_track tracks[WIRED_MOQTRUN_MAX_TRACKS_PER_PEER];
  usz sub_name_lens[WIRED_MOQTRUN_SUB_NAMES];
  usz sub_ns_lens[WIRED_MOQTRUN_SUB_NAMES];
  wired_moqtrun_sub sub_state[WIRED_MOQTRUN_SUB_NAMES];
  usz send_lens[2];
  /* SSTS (moqtssts_run.c) */
  moqtss_sess ssts;
  /* --- 4-byte aligned --- */
  int in_use;
  int ver;
  int armed_idx;
  moqsess sess;
  /* --- 1-byte flags and counters --- */
  u8 legacy;
  u8 raw;
  u8 ctl_opened;
  u8 peer_ctl_set;
  u8 setup_recv;
  u8 peer_has_impl;
  u8 pre_n;
  u8 sub_names_n;  
  u8 sub_names_at; 
  u8 closing;
  u8 wt_drained;
  /* --- byte arrays --- */
  u8 pre[9];
  u8 peer_impl[WIRED_MOQTRUN_IMPL_MAX];
  u8 sub_names[WIRED_MOQTRUN_SUB_NAMES][WIRED_MOQTRUN_MAX_NAME];
  u8 hold[WIRED_MOQTRUN_HOLD_BUF];
  u8 early[WIRED_MOQTRUN_EARLY_BUF];
  u8 sub_ns[WIRED_MOQTRUN_SUB_NAMES][WIRED_MOQTRUN_MAX_NS];
  u8 send_bufs[2][WIRED_MOQTRUN_CTL_SEND_BUF];
} wired_moqtrun_peer;

typedef struct {
  wired_moqtrun_track track;    
  const wired_span*   frags;    
  usz                 n_frags;  
  u64                 t0_ms;    
  u64                 group_ms; 
  u64 last_now_ms;
  u64 sent_group[WIRED_MOQTRUN_MAX_SUBS]; 
  int sent_any[WIRED_MOQTRUN_MAX_SUBS];   
} wired_moqtrun_live;

typedef int (*wired_moqt_authorize_fn)(
    void* ctx, const moqctl_ftn* name, const moqctl_token* token);

typedef int (*wired_moqt_authorize_ns_fn)(
    void* ctx, u64 msg_type, const moqctl_ns* ns, const moqctl_token* token);

typedef struct {
  wired_moqtrun_peer peers[WIRED_MOQTRUN_MAX_SESSIONS];
  wired_moqt_io io;
  u64 join_seq_next;
  wired_moqt_authorize_fn authorize_subscribe;
  void* authorize_ctx;
  wired_moqt_authorize_fn authorize_publish;
  void* authorize_pub_ctx;
  wired_moqtrun_track blob_track;
  wired_span blob_wire;
  wired_moqtrun_live live;
  u64 stat_live_sent;
  u64 stat_live_drop;
  u8 relay_scratch[WIRED_MOQTRUN_RELAY_FRAG_MAX + WIRED_SRVLOOP_WT_BUF_CAP];
  u8 alias_scratch[WIRED_MOQTRUN_RELAY_FRAG_MAX + WIRED_SRVLOOP_WT_BUF_CAP + 8];
  u64 stat_frag_drop;
  u64 stat_relay_sent;
  u64 stat_relay_drop;
  u64 stat_open_drop;
  u64 stat_relay_reset;
  u64 stat_relay_full;
  u64 stat_dg_sent;
  u64 stat_dg_drop;
  u64 stat_dg_bad;
  u64 reliable_alias_limit;
  moqtrel_buf rel_pool[WIRED_MOQTREL_POOL];
  u64 stat_rel_stall;
  u64 stat_rel_overflow;
  u64 stat_rel_wait;
  u64 stat_rel_sent;
  u64 stat_rel_refused;
  u64 stat_rel_rings;
  u64 stat_rel_in_bytes;
  u64 stat_rel_fin_in;
  u64 stat_rel_fin_out;
  u64 stat_rel_hold;
  u64 stat_rel_early_return;
  wired_moqtrun_req reqs[WIRED_MOQTRUN_MAX_REQS];
  u8 frag_pool[WIRED_MOQTRUN_FRAG_POOL][WIRED_MOQTRUN_RELAY_FRAG_MAX];
  wired_moqtrun_relay* frag_owner[WIRED_MOQTRUN_FRAG_POOL];
  moqcache cache;
  u64 cache_tag_next;
  wired_moqtrun_fetch fetches[WIRED_MOQTRUN_MAX_FETCHES];
  wired_moqtrun_fetch fetch_waits[WIRED_MOQTRUN_MAX_FETCHES];
  wired_moqtrun_upf_leg upf_legs[WIRED_MOQTRUN_MAX_FETCHES];
  u8 upf_buf[WIRED_MOQTRUN_RELAY_FRAG_MAX + WIRED_SRVLOOP_WT_BUF_CAP];
  usz upf_len;
  wired_moqtrun_fetch* upf_owner;
  wired_moqtrun_rdv rdv[WIRED_MOQTRUN_MAX_RDV];
  wired_moqtrun_up ups[WIRED_MOQTRUN_MAX_UP];
  wired_moqt_authorize_ns_fn authorize_namespace;
  void* authorize_ns_ctx;
  u64 stat_timeout_reset;
  moqraw_policy raw_policy;
  /* SSTS (moqtssts_run.c) */
  const u64* ssts_algs;
  usz ssts_alg_n;
  u64 ssts_cap_kbps;
  /* SWITCH_FROM (moqtswitch.c) */
  u8 switch_track;
  u32 sw_live;
} wired_moqt_hub;

void wired_moqt_init(wired_moqt_hub* hub, wired_moqt_io io);

int wired_moqt_cache_attach(wired_moqt_hub* hub, u8* arena, usz size);

void wired_moqt_on_session(
    void* app_ctx, wired_wt_session* s, wired_span path, wired_span protocol);

void wired_moqt_on_session_raw(
    void* app_ctx, wired_wt_session* s, wired_span alpn);

void wired_moqt_on_stream_data(
    void*             app_ctx,
    wired_wt_session* s,
    u64               stream_id,
    wired_span        data,
    int               fin);

void wired_moqt_on_datagram(
    void* app_ctx, wired_wt_session* s, wired_span data);

void wired_moqt_on_session_close(void* app_ctx, wired_wt_session* s);

void wired_moqt_on_stream_reset(
    void*             app_ctx,
    wired_wt_session* s,
    u64               stream_id,
    int               mapped,
    u32               app_error_code);

int wired_moqt_goaway(wired_moqt_hub* hub, wired_span new_uri, u64 timeout_ms);

void wired_moqt_on_session_draining(void* app_ctx, wired_wt_session* s);

usz wired_moqt_publish_blob(
    wired_moqt_hub* hub,
    wired_span      name,
    u64             track_alias,
    wired_span      blob,
    wired_mspan     wire);

int wired_moqt_publish_live(
    wired_moqt_hub*   hub,
    wired_span        name,
    u64               track_alias,
    const wired_span* frags,
    usz               n_frags,
    u64               group_ms,
    u64               now_ms);

void wired_moqt_tick(wired_moqt_hub* hub, u64 now_ms);
 
#endif