demux: remove weird tripple-buffering for the sh_stream list

The demuxer infrastructure was originally single-threaded. To make it
suitable for multithreading (specifically, demuxing and decoding on
separate threads), some sort of tripple-buffering was introduced. There
are separate "struct demuxer" allocations. The demuxer thread sets the
state on d_thread. If anything changes, the state is copied to d_buffer
(the copy is protected by a lock), and the decoder thread is notified.
Then the decoder thread copies the state from d_buffer to d_user (again
while holding a lock). This avoids the need for locking in the
demuxer/decoder code itself (only demux.c needs an internal, "invisible"
lock.)

Remove the streams/num_streams fields from this tripple-buffering
schema. Move them to the internal struct, and protect them with the
internal lock. Use accessors for read access outside of demux.c.

Other than replacing all field accesses with accessors, this separates
allocating and adding sh_streams. This is needed to avoid race
conditions. Before this change, this was awkwardly handled by first
initializing the sh_stream, and then sending a stream change event. Now
the stream is allocated, then initialized, and then declared as
immutable and added (at which point it becomes visible to the decoder
thread immediately).

This change is useful for PR #2626. And eventually, we should probably
get entirely of the tripple buffering, and this makes a nice first step.

13 files changed, 199 insertions, 146 deletions

diff --git a/demux/demux.c b/demux/demux.c
index 972bf31bae..8eb274647d 100644
--- a/demux/demux.c
+++ b/demux/demux.c
@@ -103,6 +103,11 @@ struct demux_internal {
     void (*wakeup_cb)(void *ctx);
     void *wakeup_cb_ctx;
 
+    struct sh_stream **streams;
+    int num_streams;
+
+    int events;
+
     bool warned_queue_overflow;
     bool last_eof;              // last actual global EOF status
     bool eof;                   // whether we're in EOF state (reset for retry)
@@ -198,42 +203,85 @@ void demux_set_ts_offset(struct demuxer *demuxer, double offset)
     pthread_mutex_unlock(&in->lock);
 }
 
-struct sh_stream *new_sh_stream(demuxer_t *demuxer, enum stream_type type)
+// Allocate a new sh_stream of the given type. It either has to be released
+// with talloc_free(), or added to a demuxer with demux_add_sh_stream(). You
+// cannot add or read packets from the stream before it has been added.
+struct sh_stream *demux_alloc_sh_stream(enum stream_type type)
 {
-    assert(demuxer == demuxer->in->d_thread);
-
-    if (demuxer->num_streams > MAX_SH_STREAMS) {
-        MP_WARN(demuxer, "Too many streams.\n");
-        return NULL;
-    }
-
-    int demuxer_id = 0;
-    for (int n = 0; n < demuxer->num_streams; n++) {
-        if (demuxer->streams[n]->type == type)
-            demuxer_id++;
-    }
-
-    struct sh_stream *sh = talloc_ptrtype(demuxer, sh);
+    struct sh_stream *sh = talloc_ptrtype(NULL, sh);
     *sh = (struct sh_stream) {
         .type = type,
-        .index = demuxer->num_streams,
-        .ff_index = demuxer->num_streams,
-        .demuxer_id = demuxer_id, // may be overwritten by demuxer
-        .ds = talloc(sh, struct demux_stream),
+        .index = -1,
+        .ff_index = -1,     // may be overwritten by demuxer
+        .demuxer_id = -1,   // ... same
     };
+    switch (sh->type) {
+    case STREAM_VIDEO: sh->video = talloc_zero(sh, struct sh_video); break;
+    case STREAM_AUDIO: sh->audio = talloc_zero(sh, struct sh_audio); break;
+    case STREAM_SUB:   sh->sub = talloc_zero(sh, struct sh_sub); break;
+    }
+
+    return sh;
+}
+
+// Add a new sh_stream to the demuxer. Note that as soon as the stream has been
+// added, it must be immutable, and must not be released (this will happen when
+// the demuxer is destroyed).
+void demux_add_sh_stream(struct demuxer *demuxer, struct sh_stream *sh)
+{
+    struct demux_internal *in = demuxer->in;
+    pthread_mutex_lock(&in->lock);
+
+    assert(!sh->ds); // must not be added yet
+
+    sh->ds = talloc(sh, struct demux_stream);
     *sh->ds = (struct demux_stream) {
-        .in = demuxer->in,
+        .in = in,
         .type = sh->type,
-        .selected = demuxer->in->autoselect,
+        .selected = in->autoselect,
     };
-    MP_TARRAY_APPEND(demuxer, demuxer->streams, demuxer->num_streams, sh);
-    switch (sh->type) {
-    case STREAM_VIDEO: sh->video = talloc_zero(demuxer, struct sh_video); break;
-    case STREAM_AUDIO: sh->audio = talloc_zero(demuxer, struct sh_audio); break;
-    case STREAM_SUB:   sh->sub = talloc_zero(demuxer, struct sh_sub); break;
+
+    sh->index = in->num_streams;
+    if (sh->ff_index < 0)
+        sh->ff_index = sh->index;
+    if (sh->demuxer_id < 0) {
+        sh->demuxer_id = 0;
+        for (int n = 0; n < in->num_streams; n++) {
+            if (in->streams[n]->type == sh->type)
+                sh->demuxer_id += 1;
+        }
     }
 
-    return sh;
+    MP_TARRAY_APPEND(in, in->streams, in->num_streams, sh);
+
+    in->events |= DEMUX_EVENT_STREAMS;
+    if (in->wakeup_cb)
+        in->wakeup_cb(in->wakeup_cb_ctx);
+    pthread_mutex_unlock(&in->lock);
+}
+
+// Return a stream with the given index. Since streams can only be added during
+// the lifetime of the demuxer, it is guaranteed that an index within the valid
+// range [0, demux_get_num_stream()) always returns a valid sh_stream pointer,
+// which will be valid until the demuxer is destroyed.
+struct sh_stream *demux_get_stream(struct demuxer *demuxer, int index)
+{
+    struct demux_internal *in = demuxer->in;
+    pthread_mutex_lock(&in->lock);
+    assert(index >= 0 && index < in->num_streams);
+    struct sh_stream *r = in->streams[index];
+    pthread_mutex_unlock(&in->lock);
+    return r;
+}
+
+// See demux_get_stream().
+int demux_get_num_stream(struct demuxer *demuxer)
+{
+    struct demux_internal *in = demuxer->in;
+    pthread_mutex_lock(&in->lock);
+    int r = in->num_streams;
+    pthread_mutex_unlock(&in->lock);
+    return r;
 }
 
 void free_demuxer(demuxer_t *demuxer)
@@ -247,8 +295,10 @@ void free_demuxer(demuxer_t *demuxer)
 
     if (demuxer->desc->close)
         demuxer->desc->close(in->d_thread);
-    for (int n = 0; n < demuxer->num_streams; n++)
-        ds_flush(demuxer->streams[n]->ds);
+    for (int n = in->num_streams - 1; n >= 0; n--) {
+        ds_flush(in->streams[n]->ds);
+        talloc_free(in->streams[n]);
+    }
     pthread_mutex_destroy(&in->lock);
     pthread_cond_destroy(&in->wakeup);
     talloc_free(demuxer);
@@ -388,8 +438,8 @@ static bool read_packet(struct demux_internal *in)
     // safe-guards against packet queue overflow.
     bool active = false, read_more = false;
     size_t packs = 0, bytes = 0;
-    for (int n = 0; n < in->d_buffer->num_streams; n++) {
-        struct demux_stream *ds = in->d_buffer->streams[n]->ds;
+    for (int n = 0; n < in->num_streams; n++) {
+        struct demux_stream *ds = in->streams[n]->ds;
         active |= ds->active;
         read_more |= ds->active && !ds->head;
         packs += ds->packs;
@@ -404,16 +454,16 @@ static bool read_packet(struct demux_internal *in)
         if (!in->warned_queue_overflow) {
             in->warned_queue_overflow = true;
             MP_ERR(in, "Too many packets in the demuxer packet queues:\n");
-            for (int n = 0; n < in->d_buffer->num_streams; n++) {
-                struct demux_stream *ds = in->d_buffer->streams[n]->ds;
+            for (int n = 0; n < in->num_streams; n++) {
+                struct demux_stream *ds = in->streams[n]->ds;
                 if (ds->selected) {
                     MP_ERR(in, "  %s/%d: %zd packets, %zd bytes\n",
                            stream_type_name(ds->type), n, ds->packs, ds->bytes);
                 }
             }
         }
-        for (int n = 0; n < in->d_buffer->num_streams; n++) {
-            struct demux_stream *ds = in->d_buffer->streams[n]->ds;
+        for (int n = 0; n < in->num_streams; n++) {
+            struct demux_stream *ds = in->streams[n]->ds;
             ds->eof |= !ds->head;
         }
         pthread_cond_signal(&in->wakeup);
@@ -433,8 +483,8 @@ static bool read_packet(struct demux_internal *in)
     pthread_mutex_lock(&in->lock);
 
     if (eof) {
-        for (int n = 0; n < in->d_buffer->num_streams; n++)
-            in->d_buffer->streams[n]->ds->eof = true;
+        for (int n = 0; n < in->num_streams; n++)
+            in->streams[n]->ds->eof = true;
         // If we had EOF previously, then don't wakeup (avoids wakeup loop)
         if (!in->last_eof) {
             if (in->wakeup_cb)
@@ -478,8 +528,8 @@ static void start_refreshing(struct demux_internal *in)
     in->start_refresh_seek = false;
 
     double start_ts = MP_NOPTS_VALUE;
-    for (int n = 0; n < demux->num_streams; n++) {
-        struct demux_stream *ds = demux->streams[n]->ds;
+    for (int n = 0; n < in->num_streams; n++) {
+        struct demux_stream *ds = in->streams[n]->ds;
         if (ds->type == STREAM_VIDEO || ds->type == STREAM_AUDIO)
             start_ts = MP_PTS_MIN(start_ts, ds->base_ts);
     }
@@ -488,8 +538,8 @@ static void start_refreshing(struct demux_internal *in)
         demux->partially_seekable || !demux->allow_refresh_seeks)
         return;
 
-    for (int n = 0; n < demux->num_streams; n++) {
-        struct demux_stream *ds = demux->streams[n]->ds;
+    for (int n = 0; n < in->num_streams; n++) {
+        struct demux_stream *ds = in->streams[n]->ds;
         // Streams which didn't read any packets yet can return all packets,
         // or they'd be stuck forever; affects newly selected streams too.
         if (ds->last_pos != -1)
@@ -697,21 +747,22 @@ bool demux_has_packet(struct sh_stream *sh)
 // Read and return any packet we find.
 struct demux_packet *demux_read_any_packet(struct demuxer *demuxer)
 {
-    assert(!demuxer->in->threading); // doesn't work with threading
+    struct demux_internal *in = demuxer->in;
+    assert(!in->threading); // doesn't work with threading
     bool read_more = true;
     while (read_more) {
-        for (int n = 0; n < demuxer->num_streams; n++) {
-            struct sh_stream *sh = demuxer->streams[n];
+        for (int n = 0; n < in->num_streams; n++) {
+            struct sh_stream *sh = in->streams[n];
             sh->ds->active = sh->ds->selected; // force read_packet() to read
             struct demux_packet *pkt = dequeue_packet(sh->ds);
             if (pkt)
                 return pkt;
         }
         // retry after calling this
-        pthread_mutex_lock(&demuxer->in->lock);
-        read_more = read_packet(demuxer->in);
-        read_more &= !demuxer->in->eof;
-        pthread_mutex_unlock(&demuxer->in->lock);
+        pthread_mutex_lock(&in->lock); // lock only because read_packet unlocks
+        read_more = read_packet(in);
+        read_more &= !in->eof;
+        pthread_mutex_unlock(&in->lock);
     }
     return NULL;
 }
@@ -795,13 +846,14 @@ static int decode_peak(demuxer_t *demuxer, const char *tag, float *out)
 
 static void apply_replaygain(demuxer_t *demuxer, struct replaygain_data *rg)
 {
-    for (int n = 0; n < demuxer->num_streams; n++) {
-        struct sh_stream *sh = demuxer->streams[n];
+    struct demux_internal *in = demuxer->in;
+    for (int n = 0; n < in->num_streams; n++) {
+        struct sh_stream *sh = in->streams[n];
         if (sh->audio && !sh->audio->replaygain_data) {
             MP_VERBOSE(demuxer, "Replaygain: Track=%f/%f Album=%f/%f\n",
                        rg->track_gain, rg->track_peak,
                        rg->album_gain, rg->album_peak);
-            sh->audio->replaygain_data = talloc_memdup(demuxer, rg, sizeof(*rg));
+            sh->audio->replaygain_data = talloc_memdup(in, rg, sizeof(*rg));
         }
     }
 }
@@ -853,11 +905,6 @@ static void demux_copy(struct demuxer *dst, struct demuxer *src)
         dst->start_time = src->start_time;
         dst->priv = src->priv;
     }
-    if (src->events & DEMUX_EVENT_STREAMS) {
-        // The stream structs themselves are immutable.
-        for (int n = dst->num_streams; n < src->num_streams; n++)
-            MP_TARRAY_APPEND(dst, dst->streams, dst->num_streams, src->streams[n]);
-    }
     if (src->events & DEMUX_EVENT_METADATA) {
         talloc_free(dst->metadata);
         dst->metadata = mp_tags_dup(dst, src->metadata);
@@ -905,6 +952,8 @@ void demux_update(demuxer_t *demuxer)
 
     pthread_mutex_lock(&in->lock);
     demux_copy(demuxer, in->d_buffer);
+    demuxer->events |= in->events;
+    in->events = 0;
     if (in->stream_metadata && (demuxer->events & DEMUX_EVENT_METADATA))
         mp_tags_merge(demuxer->metadata, in->stream_metadata);
     pthread_mutex_unlock(&in->lock);
@@ -1113,8 +1162,8 @@ struct demuxer *demux_open_url(const char *url,
 
 static void flush_locked(demuxer_t *demuxer)
 {
-    for (int n = 0; n < demuxer->num_streams; n++)
-        ds_flush(demuxer->streams[n]->ds);
+    for (int n = 0; n < demuxer->in->num_streams; n++)
+        ds_flush(demuxer->in->streams[n]->ds);
     demuxer->in->warned_queue_overflow = false;
     demuxer->in->eof = false;
     demuxer->in->last_eof = false;
@@ -1192,10 +1241,11 @@ void demux_set_enable_refresh_seeks(struct demuxer *demuxer, bool enabled)
 struct sh_stream *demuxer_stream_by_demuxer_id(struct demuxer *d,
                                                enum stream_type t, int id)
 {
-    for (int n = 0; n < d->num_streams; n++) {
-        struct sh_stream *s = d->streams[n];
+    int num = demux_get_num_stream(d);
+    for (int n = 0; n < num; n++) {
+        struct sh_stream *s = demux_get_stream(d, n);
         if (s->type == t && s->demuxer_id == id)
-            return d->streams[n];
+            return s;
     }
     return NULL;
 }
@@ -1205,8 +1255,9 @@ void demuxer_switch_track(struct demuxer *demuxer, enum stream_type type,
 {
     assert(!stream || stream->type == type);
 
-    for (int n = 0; n < demuxer->num_streams; n++) {
-        struct sh_stream *cur = demuxer->streams[n];
+    int num = demux_get_num_stream(demuxer);
+    for (int n = 0; n < num; n++) {
+        struct sh_stream *cur = demux_get_stream(demuxer, n);
         if (cur->type == type)
             demuxer_select_track(demuxer, cur, cur == stream);
     }
@@ -1408,8 +1459,8 @@ static int cached_demux_control(struct demux_internal *in, int cmd, void *arg)
         double *rates = arg;
         for (int n = 0; n < STREAM_TYPE_COUNT; n++)
             rates[n] = -1;
-        for (int n = 0; n < in->d_user->num_streams; n++) {
-            struct demux_stream *ds = in->d_user->streams[n]->ds;
+        for (int n = 0; n < in->num_streams; n++) {
+            struct demux_stream *ds = in->streams[n]->ds;
             if (ds->selected && ds->bitrate >= 0)
                 rates[ds->type] = MPMAX(0, rates[ds->type]) + ds->bitrate;
         }
@@ -1423,8 +1474,8 @@ static int cached_demux_control(struct demux_internal *in, int cmd, void *arg)
             .ts_duration = -1,
         };
         int num_packets = 0;
-        for (int n = 0; n < in->d_user->num_streams; n++) {
-            struct demux_stream *ds = in->d_user->streams[n]->ds;
+        for (int n = 0; n < in->num_streams; n++) {
+            struct demux_stream *ds = in->streams[n]->ds;
             if (ds->active) {
                 r->underrun |= !ds->head && !ds->eof;
                 r->ts_range[0] = MP_PTS_MAX(r->ts_range[0], ds->base_ts);
diff --git a/demux/demux.h b/demux/demux.h
index 96717d5eeb..db3b504c39 100644
--- a/demux/demux.h
+++ b/demux/demux.h
@@ -86,8 +86,6 @@ enum demux_event {
     DEMUX_EVENT_ALL = 0xFFFF,
 };
 
-#define MAX_SH_STREAMS 256
-
 struct demuxer;
 struct timeline;
 
@@ -199,9 +197,6 @@ typedef struct demuxer {
     // Bitmask of DEMUX_EVENT_*
     int events;
 
-    struct sh_stream **streams;
-    int num_streams;
-
     struct demux_edition *editions;
     int num_editions;
     int edition;
@@ -251,7 +246,11 @@ double demux_get_next_pts(struct sh_stream *sh);
 bool demux_has_packet(struct sh_stream *sh);
 struct demux_packet *demux_read_any_packet(struct demuxer *demuxer);
 
-struct sh_stream *new_sh_stream(struct demuxer *demuxer, enum stream_type type);
+struct sh_stream *demux_get_stream(struct demuxer *demuxer, int index);
+int demux_get_num_stream(struct demuxer *demuxer);
+
+struct sh_stream *demux_alloc_sh_stream(enum stream_type type);
+void demux_add_sh_stream(struct demuxer *demuxer, struct sh_stream *sh);
 
 struct demuxer *demux_open(struct stream *stream, struct demuxer_params *params,
                            struct mpv_global *global);
diff --git a/demux/demux_disc.c b/demux/demux_disc.c
index 1abae38d35..90c276b414 100644
--- a/demux/demux_disc.c
+++ b/demux/demux_disc.c
@@ -54,9 +54,10 @@ struct priv {
 static void reselect_streams(demuxer_t *demuxer)
 {
     struct priv *p = demuxer->priv;
-    for (int n = 0; n < MPMIN(p->slave->num_streams, p->num_streams); n++) {
+    int num_slave = demux_get_num_stream(p->slave);
+    for (int n = 0; n < MPMIN(num_slave, p->num_streams); n++) {
         if (p->streams[n]) {
-            demuxer_select_track(p->slave, p->slave->streams[n],
+            demuxer_select_track(p->slave, demux_get_stream(p->slave, n),
                 demux_stream_is_selected(p->streams[n]));
         }
     }
@@ -81,9 +82,7 @@ static void add_dvd_streams(demuxer_t *demuxer)
     struct stream_dvd_info_req info;
     if (stream_control(stream, STREAM_CTRL_GET_DVD_INFO, &info) > 0) {
         for (int n = 0; n < MPMIN(32, info.num_subs); n++) {
-            struct sh_stream *sh = new_sh_stream(demuxer, STREAM_SUB);
-            if (!sh)
-                break;
+            struct sh_stream *sh = demux_alloc_sh_stream(STREAM_SUB);
             sh->demuxer_id = n + 0x20;
             sh->codec = "dvd_subtitle";
             get_disc_lang(stream, sh);
@@ -114,6 +113,8 @@ static void add_dvd_streams(demuxer_t *demuxer)
 
             sh->extradata = s;
             sh->extradata_size = strlen(s);
+
+            demux_add_sh_stream(demuxer, sh);
         }
     }
 }
@@ -122,8 +123,8 @@ static void add_streams(demuxer_t *demuxer)
 {
     struct priv *p = demuxer->priv;
 
-    for (int n = p->num_streams; n < p->slave->num_streams; n++) {
-        struct sh_stream *src = p->slave->streams[n];
+    for (int n = p->num_streams; n < demux_get_num_stream(p->slave); n++) {
+        struct sh_stream *src = demux_get_stream(p->slave, n);
         if (src->sub) {
             struct sh_stream *sub = NULL;
             if (src->demuxer_id >= 0x20 && src->demuxer_id <= 0x3F)
@@ -134,9 +135,7 @@ static void add_streams(demuxer_t *demuxer)
                 continue;
             }
         }
-        struct sh_stream *sh = new_sh_stream(demuxer, src->type);
-        if (!sh)
-            break;
+        struct sh_stream *sh = demux_alloc_sh_stream(src->type);
         assert(p->num_streams == n); // directly mapped
         MP_TARRAY_APPEND(p, p->streams, p->num_streams, sh);
         // Copy all stream fields that might be relevant
@@ -159,6 +158,7 @@ static void add_streams(demuxer_t *demuxer)
         if (src->audio)
             sh->audio = src->audio;
         get_disc_lang(demuxer->stream, sh);
+        demux_add_sh_stream(demuxer, sh);
     }
     reselect_streams(demuxer);
 }
diff --git a/demux/demux_lavf.c b/demux/demux_lavf.c
index 7af41e6f09..5e7d8d8a9e 100644
--- a/demux/demux_lavf.c
+++ b/demux/demux_lavf.c
@@ -541,9 +541,7 @@ static void handle_new_stream(demuxer_t *demuxer, int i)
 
     switch (codec->codec_type) {
     case AVMEDIA_TYPE_AUDIO: {
-        sh = new_sh_stream(demuxer, STREAM_AUDIO);
-        if (!sh)
-            break;
+        sh = demux_alloc_sh_stream(STREAM_AUDIO);
         sh_audio_t *sh_audio = sh->audio;
 
         // probably unneeded
@@ -558,9 +556,7 @@ static void handle_new_stream(demuxer_t *demuxer, int i)
         break;
     }
     case AVMEDIA_TYPE_VIDEO: {
-        sh = new_sh_stream(demuxer, STREAM_VIDEO);
-        if (!sh)
-            break;
+        sh = demux_alloc_sh_stream(STREAM_VIDEO);
         sh_video_t *sh_video = sh->video;
 
         if (st->disposition & AV_DISPOSITION_ATTACHED_PIC) {
@@ -609,9 +605,7 @@ static void handle_new_stream(demuxer_t *demuxer, int i)
     }
     case AVMEDIA_TYPE_SUBTITLE: {
         sh_sub_t *sh_sub;
-        sh = new_sh_stream(demuxer, STREAM_SUB);
-        if (!sh)
-            break;
+        sh = demux_alloc_sh_stream(STREAM_SUB);
         sh_sub = sh->sub;
 
         if (codec->extradata_size) {
@@ -660,9 +654,8 @@ static void handle_new_stream(demuxer_t *demuxer, int i)
         sh->codec = mp_codec_from_av_codec_id(codec->codec_id);
         sh->codec_tag = codec->codec_tag;
         sh->lav_headers = avcodec_alloc_context3(NULL);
-        if (!sh->lav_headers)
-            return;
-        mp_copy_lav_codec_headers(sh->lav_headers, codec);
+        if (sh->lav_headers)
+            mp_copy_lav_codec_headers(sh->lav_headers, codec);
 
         if (st->disposition & AV_DISPOSITION_DEFAULT)
             sh->default_track = true;
@@ -680,10 +673,10 @@ static void handle_new_stream(demuxer_t *demuxer, int i)
         if (!sh->title && sh->hls_bitrate > 0)
             sh->title = talloc_asprintf(sh, "bitrate %d", sh->hls_bitrate);
         sh->missing_timestamps = !!(priv->avif_flags & AVFMT_NOTIMESTAMPS);
+        demux_add_sh_stream(demuxer, sh);
     }
 
     select_tracks(demuxer, i);
-    demux_changed(demuxer, DEMUX_EVENT_STREAMS);
 }
 
 // Add any new streams that might have been added
@@ -844,7 +837,7 @@ static int demux_open_lavf(demuxer_t *demuxer, enum demux_check check)
 
     // Often useful with OGG audio-only files, which have metadata in the audio
     // track metadata instead of the main metadata.
-    if (demuxer->num_streams == 1) {
+    if (demux_get_num_stream(demuxer) == 1) {
         priv->merge_track_metadata = true;
         for (int n = 0; n < priv->num_streams; n++) {
             if (priv->streams[n])
diff --git a/demux/demux_mf.c b/demux/demux_mf.c
index ab703c0ced..c1cc5d79f3 100644
--- a/demux/demux_mf.c
+++ b/demux/demux_mf.c
@@ -201,7 +201,7 @@ static int demux_mf_fill_buffer(demuxer_t *demuxer)
                 memcpy(dp->buffer, data.start, data.len);
                 dp->pts = mf->curr_frame / mf->sh->fps;
                 dp->keyframe = true;
-                demux_add_packet(demuxer->streams[0], dp);
+                demux_add_packet(mf->sh, dp);
             }
         }
         talloc_free(data.start);
@@ -316,7 +316,7 @@ static int demux_open_mf(demuxer_t *demuxer, enum demux_check check)
     mf->curr_frame = 0;
 
     // create a new video stream header
-    struct sh_stream *sh = new_sh_stream(demuxer, STREAM_VIDEO);
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_VIDEO);
     sh_video = sh->video;
 
     sh->codec = codec;
diff --git a/demux/demux_mkv.c b/demux/demux_mkv.c
index 562bddbfc6..af82dc8deb 100644
--- a/demux/demux_mkv.c
+++ b/demux/demux_mkv.c
@@ -1193,9 +1193,7 @@ static void add_coverart(struct demuxer *demuxer)
         const char *codec = mp_map_mimetype_to_video_codec(att->type);
         if (!codec)
             continue;
-        struct sh_stream *sh = new_sh_stream(demuxer, STREAM_VIDEO);
-        if (!sh)
-            break;
+        struct sh_stream *sh = demux_alloc_sh_stream(STREAM_VIDEO);
         sh->demuxer_id = -1 - sh->index; // don't clash with mkv IDs
         sh->codec = codec;
         sh->attached_picture = new_demux_packet_from(att->data, att->data_size);
@@ -1205,6 +1203,7 @@ static void add_coverart(struct demuxer *demuxer)
             sh->attached_picture->keyframe = true;
         }
         sh->title = att->name;
+        demux_add_sh_stream(demuxer, sh);
     }
 }
 
@@ -1266,9 +1265,7 @@ static int demux_mkv_open_video(demuxer_t *demuxer, mkv_track_t *track)
 {
     unsigned char *extradata = NULL;
     unsigned int extradata_size = 0;
-    struct sh_stream *sh = new_sh_stream(demuxer, STREAM_VIDEO);
-    if (!sh)
-        return 1;
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_VIDEO);
     init_track(demuxer, track, sh);
     sh_video_t *sh_v = sh->video;
 
@@ -1277,7 +1274,7 @@ static int demux_mkv_open_video(demuxer_t *demuxer, mkv_track_t *track)
     if (!strcmp(track->codec_id, "V_MS/VFW/FOURCC")) { /* AVI compatibility mode */
         // The private_data contains a BITMAPINFOHEADER struct
         if (track->private_data == NULL || track->private_size < 40)
-            return 1;
+            goto done;
 
         unsigned char *h = track->private_data;
         if (track->v_width == 0)
@@ -1355,7 +1352,7 @@ static int demux_mkv_open_video(demuxer_t *demuxer, mkv_track_t *track)
 
     if (extradata_size > 0x1000000) {
         MP_WARN(demuxer, "Invalid CodecPrivate\n");
-        return 1;
+        goto done;
     }
 
     sh->extradata = talloc_memdup(sh_v, extradata, extradata_size);
@@ -1378,6 +1375,9 @@ static int demux_mkv_open_video(demuxer_t *demuxer, mkv_track_t *track)
 
     sh_v->stereo_mode = track->stereo_mode;
 
+done:
+    demux_add_sh_stream(demuxer, sh);
+
     return 0;
 }
 
@@ -1469,9 +1469,7 @@ static const char *const mkv_audio_tags[][2] = {
 
 static int demux_mkv_open_audio(demuxer_t *demuxer, mkv_track_t *track)
 {
-    struct sh_stream *sh = new_sh_stream(demuxer, STREAM_AUDIO);
-    if (!sh)
-        return 1;
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_AUDIO);
     init_track(demuxer, track, sh);
     sh_audio_t *sh_a = sh->audio;
 
@@ -1680,12 +1678,15 @@ static int demux_mkv_open_audio(demuxer_t *demuxer, mkv_track_t *track)
     sh->extradata = extradata;
     sh->extradata_size = extradata_len;
 
+    demux_add_sh_stream(demuxer, sh);
+
     return 0;
 
  error:
     MP_WARN(demuxer, "Unknown/unsupported audio "
             "codec ID '%s' for track %u or missing/faulty\n"
             "private codec data.\n", track->codec_id, track->tnum);
+    demux_add_sh_stream(demuxer, sh); // add it anyway
     return 1;
 }
 
@@ -1717,9 +1718,7 @@ static int demux_mkv_open_sub(demuxer_t *demuxer, mkv_track_t *track)
     if (track->private_size > 0x10000000)
         return 1;
 
-    struct sh_stream *sh = new_sh_stream(demuxer, STREAM_SUB);
-    if (!sh)
-        return 1;
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_SUB);
     init_track(demuxer, track, sh);
 
     sh->codec = subtitle_type;
@@ -1734,6 +1733,8 @@ static int demux_mkv_open_sub(demuxer_t *demuxer, mkv_track_t *track)
     sh->extradata = track->private_data;
     sh->extradata_size = track->private_size;
 
+    demux_add_sh_stream(demuxer, sh);
+
     if (!subtitle_type)
         MP_ERR(demuxer, "Subtitle type '%s' is not supported.\n", track->codec_id);
 
diff --git a/demux/demux_mkv_timeline.c b/demux/demux_mkv_timeline.c
index a874f9c6a7..f784f2d010 100644
--- a/demux/demux_mkv_timeline.c
+++ b/demux/demux_mkv_timeline.c
@@ -458,8 +458,9 @@ static void check_track_compatibility(struct timeline *tl)
         if (p->source == mainsrc)
             continue;
 
-        for (int i = 0; i < p->source->num_streams; i++) {
-            struct sh_stream *s = p->source->streams[i];
+        int num_source_streams = demux_get_num_stream(p->source);
+        for (int i = 0; i < num_source_streams; i++) {
+            struct sh_stream *s = demux_get_stream(p->source, i);
             if (s->attached_picture)
                 continue;
 
@@ -473,8 +474,9 @@ static void check_track_compatibility(struct timeline *tl)
             }
         }
 
-        for (int i = 0; i < mainsrc->num_streams; i++) {
-            struct sh_stream *m = mainsrc->streams[i];
+        int num_main_streams = demux_get_num_stream(mainsrc);
+        for (int i = 0; i < num_main_streams; i++) {
+            struct sh_stream *m = demux_get_stream(mainsrc, i);
             if (m->attached_picture)
                 continue;
 
diff --git a/demux/demux_raw.c b/demux/demux_raw.c
index a67287792b..5e492334ae 100644
--- a/demux/demux_raw.c
+++ b/demux/demux_raw.c
@@ -117,6 +117,7 @@ const struct m_sub_options demux_rawvideo_conf = {
 };
 
 struct priv {
+    struct sh_stream *sh;
     int frame_size;
     int read_frames;
     double frame_rate;
@@ -125,14 +126,12 @@ struct priv {
 static int demux_rawaudio_open(demuxer_t *demuxer, enum demux_check check)
 {
     struct demux_rawaudio_opts *opts = demuxer->opts->demux_rawaudio;
-    struct sh_stream *sh;
-    sh_audio_t *sh_audio;
 
     if (check != DEMUX_CHECK_REQUEST && check != DEMUX_CHECK_FORCE)
         return -1;
 
-    sh = new_sh_stream(demuxer, STREAM_AUDIO);
-    sh_audio = sh->audio;
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_AUDIO);
+    struct sh_audio *sh_audio = sh->audio;
     sh_audio->channels = opts->channels;
     sh_audio->force_channels = true;
     sh_audio->samplerate = opts->samplerate;
@@ -142,9 +141,12 @@ static int demux_rawaudio_open(demuxer_t *demuxer, enum demux_check check)
     mp_set_pcm_codec(sh, f & 1, f & 2, f >> 3, f & 4);
     int samplesize = ((f >> 3) + 7) / 8;
 
+    demux_add_sh_stream(demuxer, sh);
+
     struct priv *p = talloc_ptrtype(demuxer, p);
     demuxer->priv = p;
     *p = (struct priv) {
+        .sh = sh,
         .frame_size = samplesize * sh_audio->channels.num,
         .frame_rate = sh_audio->samplerate,
         .read_frames = sh_audio->samplerate / 8,
@@ -156,8 +158,6 @@ static int demux_rawaudio_open(demuxer_t *demuxer, enum demux_check check)
 static int demux_rawvideo_open(demuxer_t *demuxer, enum demux_check check)
 {
     struct demux_rawvideo_opts *opts = demuxer->opts->demux_rawvideo;
-    struct sh_stream *sh;
-    sh_video_t *sh_video;
 
     if (check != DEMUX_CHECK_REQUEST && check != DEMUX_CHECK_FORCE)
         return -1;
@@ -219,17 +219,19 @@ static int demux_rawvideo_open(demuxer_t *demuxer, enum demux_check check)
         imgsize = width * height * bpp / 8;
     }
 
-    sh = new_sh_stream(demuxer, STREAM_VIDEO);
-    sh_video = sh->video;
+    struct sh_stream *sh = demux_alloc_sh_stream(STREAM_VIDEO);
+    struct sh_video *sh_video = sh->video;
     sh->codec = decoder;
     sh->codec_tag = imgfmt;
     sh_video->fps = opts->fps;
     sh_video->disp_w = width;
     sh_video->disp_h = height;
+    demux_add_sh_stream(demuxer, sh);
 
     struct priv *p = talloc_ptrtype(demuxer, p);
     demuxer->priv = p;
     *p = (struct priv) {
+        .sh = sh,
         .frame_size = imgsize,
         .frame_rate = sh_video->fps,
         .read_frames = 1,
@@ -256,7 +258,7 @@ static int raw_fill_buffer(demuxer_t *demuxer)
 
     int len = stream_read(demuxer->stream, dp->buffer, dp->len);
     demux_packet_shorten(dp, len);
-    demux_add_packet(demuxer->streams[0], dp);
+    demux_add_packet(p->sh, dp);
 
     return 1;
 }
diff --git a/demux/demux_tv.c b/demux/demux_tv.c
index ab584874cb..a2d06591e0 100644
--- a/demux/demux_tv.c
+++ b/demux/demux_tv.c
@@ -50,7 +50,7 @@ static int demux_open_tv(demuxer_t *demuxer, enum demux_check check)
     funcs = tvh->functions;
     demuxer->priv=tvh;
 
-    struct sh_stream *sh_v = new_sh_stream(demuxer, STREAM_VIDEO);
+    struct sh_stream *sh_v = demux_alloc_sh_stream(STREAM_VIDEO);
     sh_video = sh_v->video;
 
     /* get IMAGE FORMAT */
@@ -89,6 +89,8 @@ static int demux_open_tv(demuxer_t *demuxer, enum demux_check check)
     /* set height */
     funcs->control(tvh->priv, TVI_CONTROL_VID_GET_HEIGHT, &sh_video->disp_h);
 
+    demux_add_sh_stream(demuxer, sh_v);
+
     demuxer->seekable = 0;
 
     /* here comes audio init */
@@ -115,7 +117,7 @@ static int demux_open_tv(demuxer_t *demuxer, enum demux_check check)
                 goto no_audio;
         }
 
-        struct sh_stream *sh_a = new_sh_stream(demuxer, STREAM_AUDIO);
+        struct sh_stream *sh_a = demux_alloc_sh_stream(STREAM_AUDIO);
         sh_audio = sh_a->audio;
 
         funcs->control(tvh->priv, TVI_CONTROL_AUD_GET_SAMPLERATE,
@@ -128,6 +130,8 @@ static int demux_open_tv(demuxer_t *demuxer, enum demux_check check)
         // s16ne
         mp_set_pcm_codec(sh_a, true, false, 16, BYTE_ORDER == BIG_ENDIAN);
 
+        demux_add_sh_stream(demuxer, sh_a);
+
         MP_VERBOSE(tvh, "  TV audio: %d channels, %d bits, %d Hz\n",
                    nchannels, 16, sh_audio->samplerate);
     }
@@ -168,8 +172,9 @@ static int demux_tv_fil