diff options
Diffstat (limited to 'src/libsink')
| -rw-r--r-- | src/libsink/sink.c | 338 | ||||
| -rw-r--r-- | src/libsink/sink.h | 1 |
2 files changed, 192 insertions, 147 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 80414fd..ede3889 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -10,7 +10,9 @@ #include "sink.h" #include "common.h" +#ifndef CAMU_SINK_NO_VIDEO #include "../render/renderer_libplacebo.h" +#endif enum { SINK_EMPTY = 0, @@ -95,6 +97,7 @@ static bool entry_buffers_held(struct camu_sink_entry *entry) static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry) { + al_assert(!entry->ended); if (entry->audio.state == BUFFER_ADDED) { sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); entry->audio.state = BUFFER_SET_OR_BUFFERED; @@ -122,6 +125,7 @@ static void remove_entry_video_buffer(struct camu_sink *sink, struct camu_sink_e static void remove_entry_buffers(struct camu_sink *sink, struct camu_sink_entry *entry) { + al_assert(!entry->ended); #ifndef CAMU_SINK_NO_VIDEO remove_entry_video_buffer(sink, entry); #endif @@ -133,8 +137,9 @@ static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry); static void add_video_if_set_and_buffered(struct camu_sink_entry *entry); #endif -static void set_or_queue_entry(struct camu_sink_entry *entry) +static void add_or_queue_entry(struct camu_sink_entry *entry) { + al_assert(!entry->ended); if (entry->audio.state == BUFFER_INIT) { entry->audio.state = BUFFER_QUEUED; } else { @@ -315,6 +320,46 @@ static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd) aki_signal_send(&sink->queue_signal); } +static void mixer_callback(void *userdata, u8 op) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + if (op == CAMU_MIXER_EMPTY) { + al_log_debug("sink", "Mixer empty."); + queue_cmd(sink, (struct camu_sink_cmd){ + .op = STOP, + .value.i = CAMU_SINK_AUDIO + }); + } +} + +bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop, + struct camu_mixer *mixer +#ifndef CAMU_SINK_NO_VIDEO + , struct camu_renderer *renderer +#endif + ) +{ + sink->loop = loop; + aki_mutex_init(&sink->mutex); + aki_signal_init(&sink->queue_signal, queue_signal_callback, sink); + aki_signal_start(&sink->queue_signal, sink->loop); + camu_queue_init(sink->queue); + sink->queued = NULL; + sink->current = NULL; + al_array_init(sink->previous); + al_array_init(sink->entries); + sink->lru = 0; + mixer->callback = mixer_callback; + mixer->userdata = sink; + sink->audio.mixer = mixer; + sink->audio.state = SINK_PAUSED; +#ifndef CAMU_SINK_NO_VIDEO + sink->video.renderer = renderer; + sink->video.state = SINK_PAUSED; +#endif + return true; +} + static void maybe_remove_previous(struct camu_sink *sink) { struct camu_sink_entry *previous; @@ -324,28 +369,74 @@ static void maybe_remove_previous(struct camu_sink *sink) sink->previous.size = 0; } -static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *entry, struct camu_sink_entry *current) +static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target) { - al_assert(entry != current); + al_assert(previous != target); + // If our target is ended, remove previous immediately. + if (target->ended) { + remove_entry_buffers(sink, previous); + return; + } struct camu_sink_entry *rentry; al_array_foreach_rev(sink->previous, i, rentry) { - if (rentry == current) { - // If the entry we are about to add is in previous, - // remove it immediately. - remove_entry_buffers(sink, current); - al_array_remove_at(sink->previous, i); + // If the entry we are about to add is in previous, run the queue now. + if (rentry == target) { + maybe_remove_previous(sink); + return; } } // Don't accept duplicates. al_array_foreach_rev(sink->previous, i, rentry) { - if (rentry == entry) return; + if (rentry == previous) return; + } + al_array_push(sink->previous, previous); +} + +static s32 lru_compare(const void *a, const void *b) +{ + struct camu_sink_entry *aa = *((struct camu_sink_entry **)a); + struct camu_sink_entry *bb = *((struct camu_sink_entry **)b); + if (aa->lru > bb->lru) return -1; + else if (aa->lru < bb->lru) return 1; + return 0; +} + +static void maybe_cleanup_old_entries(struct camu_sink *sink) +{ + // We check size <= MAX_AGE in the loop because sink->lru + // is not indicative of the amount of entries we have loaded. + // There are various reasons for this but the most obvious is + // that it's incremented for buffer and queue operations. + // + // Handle sink->lru wrapping. + // 0 65532 65533 65534 65535 + // 0 1 65533 65534 65535 + // 0 1 2 65534 65535 + // 0 1 2 3 65535 + // 0 1 2 3 4 + al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare); + struct camu_sink_entry *entry; + al_array_foreach_rev(sink->entries, i, entry) { + if (entry->lru > sink->lru && (UINT16_MAX - (entry->lru - 1)) + sink->lru >= ENTRY_MAX_AGE) { + al_array_remove_at(sink->entries, i); + lia_client_disconnect(&entry->client); + } + if (sink->entries.size <= ENTRY_MAX_AGE) return; + } + if (sink->lru >= ENTRY_MAX_AGE) { + al_array_foreach_rev(sink->entries, i, entry) { + if (sink->lru - entry->lru >= ENTRY_MAX_AGE) { + al_array_remove_at(sink->entries, i); + lia_client_disconnect(&entry->client); + } + if (sink->entries.size <= ENTRY_MAX_AGE) return; + } } - al_array_push(sink->previous, entry); } void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) { - al_assert(entry->audio.state != BUFFER_ADDED); + al_assert(!entry->ended && entry->audio.state != BUFFER_ADDED); if (entry->audio.state == BUFFER_CONFIGURED) { entry->audio.state = BUFFER_SET_OR_BUFFERED; } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) { @@ -370,7 +461,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) #ifndef CAMU_SINK_NO_VIDEO void add_video_if_set_and_buffered(struct camu_sink_entry *entry) { - al_assert(entry->video.state != BUFFER_ADDED); + al_assert(!entry->ended && entry->video.state != BUFFER_ADDED); if (entry->video.state == BUFFER_CONFIGURED) { entry->video.state = BUFFER_SET_OR_BUFFERED; } else if (entry->video.state == BUFFER_SET_OR_BUFFERED) { @@ -391,16 +482,22 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry) { al_log_info("sink", "Entry ended."); - camu_clock_end(&entry->clock); + entry->ended = true; + // TODO: Can it make sense for this entry to be in previous? + maybe_remove_previous(sink); if (sink->target) { - set_or_queue_entry(sink->target); + if (!sink->target->ended) { + add_or_queue_entry(sink->target); + } sink->current = sink->target; sink->target = NULL; return true; } else if (sink->queued) { // TODO: What is the right behavior if sink->queued // and sink->target are both set. - set_or_queue_entry(sink->queued); + if (!sink->queued->ended) { + add_or_queue_entry(sink->queued); + } sink->current = sink->queued; sink->queued = NULL; return true; @@ -509,17 +606,51 @@ static void video_buffer_callback(void *userdata, u8 op) } #endif -static void switch_to(struct camu_sink *sink, struct camu_sink_entry *entry) +static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) { if (sink->current) { - if (camu_clock_is_ended(&entry->clock)) { - remove_entry_buffers(sink, sink->current); + struct camu_sink_entry *current = sink->current; + if (current->ended) { + bool single_frame = camu_video_buffer_is_single_frame(¤t->video.buf); + if (single_frame) { + remove_entry_video_buffer(sink, current); + } + al_assert(current->audio.state != BUFFER_ADDED && current->video.state != BUFFER_ADDED); } else { - maybe_add_to_previous(sink, sink->current, entry); + maybe_add_to_previous(sink, current, target); + } + } + if (!target->ended) { + add_or_queue_entry(target); + } + sink->current = target; +} + +static void set_target_and_pause(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) +{ + if (!sink->current || sink->current->ended) { + if (!target->ended) { + add_or_queue_entry(target); + } + sink->current = target; + } else { + sink->target = target; + camu_clock_pause(&sink->current->clock, at); + } +} + +static void clock_callback(void *userdata, u8 op) +{ + struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + struct camu_sink *sink = entry->sink; + if (op == CAMU_CLOCK_PAUSED) { + aki_mutex_lock(&sink->mutex); + if (sink->target) { + switch_to(sink, sink->target); + sink->target = NULL; } + aki_mutex_unlock(&sink->mutex); } - set_or_queue_entry(entry); - sink->current = entry; } static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *entry) @@ -633,10 +764,12 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str case LIANA_CLIENT_SUBTITLE: { switch (stream->type) { case CAMU_STREAM_SUBTITLE: { +#ifndef CAMU_SINK_NO_VIDEO #ifdef CAMU_HAVE_FFMPEG AVPacket *pkt = (AVPacket *)opaque; camu_video_buffer_push_subtitle(&entry->video.buf, pkt); #endif +#endif break; } } @@ -699,11 +832,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str aki_mutex_lock(&sink->mutex); if (entry == sink->target) { if (sink->current) { - remove_entry_buffers(sink, sink->current); + if (!sink->current->ended) { + remove_entry_buffers(sink, sink->current); + } sink->current = NULL; } sink->target = NULL; } else if (entry == sink->current) { + // current's buffers will already be removed. sink->current = NULL; if (sink->target) { switch_to(sink, sink->target); @@ -737,103 +873,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } } -static void mixer_callback(void *userdata, u8 op) -{ - struct camu_sink *sink = (struct camu_sink *)userdata; - if (op == CAMU_MIXER_EMPTY) { - al_log_debug("sink", "Mixer empty."); - queue_cmd(sink, (struct camu_sink_cmd){ - .op = STOP, - .value.i = CAMU_SINK_AUDIO - }); - } -} - -bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop, - struct camu_mixer *mixer -#ifndef CAMU_SINK_NO_VIDEO - , struct camu_renderer *renderer -#endif - ) -{ - sink->loop = loop; - aki_mutex_init(&sink->mutex); - aki_signal_init(&sink->queue_signal, queue_signal_callback, sink); - aki_signal_start(&sink->queue_signal, sink->loop); - camu_queue_init(sink->queue); - sink->queued = NULL; - sink->current = NULL; - al_array_init(sink->previous); - al_array_init(sink->entries); - sink->lru = 0; - mixer->callback = mixer_callback; - mixer->userdata = sink; - sink->audio.mixer = mixer; - sink->audio.state = SINK_PAUSED; -#ifndef CAMU_SINK_NO_VIDEO - sink->video.renderer = renderer; - sink->video.state = SINK_PAUSED; -#endif - return true; -} - -static s32 lru_compare(const void *a, const void *b) -{ - struct camu_sink_entry *aa = *((struct camu_sink_entry **)a); - struct camu_sink_entry *bb = *((struct camu_sink_entry **)b); - if (aa->lru > bb->lru) return -1; - else if (aa->lru < bb->lru) return 1; - return 0; -} - -static void maybe_cleanup_old_entries(struct camu_sink *sink) -{ - // We check size <= MAX_AGE in the loop because sink->lru - // is not indicative of the amount of entries we have loaded. - // There are various reasons for this but the most obvious is - // that it's incremented for buffer and queue operations. - // - // Handle sink->lru wrapping. - // 0 65532 65533 65534 65535 - // 0 1 65533 65534 65535 - // 0 1 2 65534 65535 - // 0 1 2 3 65535 - // 0 1 2 3 4 - al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare); - struct camu_sink_entry *entry; - al_array_foreach_rev(sink->entries, i, entry) { - if (entry->lru > sink->lru && (UINT16_MAX - (entry->lru - 1)) + sink->lru >= ENTRY_MAX_AGE) { - al_array_remove_at(sink->entries, i); - lia_client_disconnect(&entry->client); - } - if (sink->entries.size <= ENTRY_MAX_AGE) return; - } - if (sink->lru >= ENTRY_MAX_AGE) { - al_array_foreach_rev(sink->entries, i, entry) { - if (sink->lru - entry->lru >= ENTRY_MAX_AGE) { - al_array_remove_at(sink->entries, i); - lia_client_disconnect(&entry->client); - } - if (sink->entries.size <= ENTRY_MAX_AGE) return; - } - } -} - - -static void clock_callback(void *userdata, u8 op) -{ - struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; - struct camu_sink *sink = entry->sink; - if (op == CAMU_CLOCK_PAUSED) { - aki_mutex_lock(&sink->mutex); - if (sink->target) { - switch_to(sink, sink->target); - sink->target = NULL; - } - aki_mutex_unlock(&sink->mutex); - } -} - static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, bool *created) { struct camu_sink_entry *entry; @@ -846,6 +885,7 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, entry = al_alloc_object(struct camu_sink_entry); entry->id = id; + entry->ended = false; entry->sink = sink; camu_clock_init(&entry->clock, clock_callback, entry); @@ -884,9 +924,12 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn u8 op = aki_packet_read_u8(packet); #ifdef CAMU_SINK_LOCAL + // TODO: look over this. if (op == LIANA_SINK_UNSET) { if (sink->current) { - remove_entry_buffers(sink, sink->current); + if (!sink->current->ended) { + remove_entry_buffers(sink, sink->current); + } sink->current = NULL; } goto out; @@ -931,17 +974,17 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn #ifdef CAMU_SINK_LOCAL (void)at; (void)pause; - if (sink->current) { - if (!camu_clock_is_paused(&sink->current->clock)) { - camu_clock_pause(&sink->current->clock, 0); - } - maybe_add_to_previous(sink, sink->current, entry); + if (sink->current && !sink->current->ended && !camu_clock_is_paused(&sink->current->clock)) { + camu_clock_pause(&sink->current->clock, 0); } - set_or_queue_entry(entry); - sink->current = entry; + switch_to(sink, entry); // This will resume a user paused stream. camu_clock_resume(&entry->clock, 0); #else + // PAUSE_NONE and PAUSE_PAUSE mean the server expects the entry + // being set to be ended. Not acting accordingly here is the + // only place where local entry->ended and the servers expectation + // being mismatched can cause issues. switch (pause) { case LIANA_PAUSE_NONE: if (sink->target) { @@ -949,6 +992,9 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn sink->target = NULL; } if (entry != sink->current) { + if (camu_clock_is_paused(&entry->clock)) { // TMP + camu_clock_resume(&entry->clock, at); + } switch_to(sink, entry); } break; @@ -965,22 +1011,27 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn } break; case LIANA_PAUSE_PAUSE: { - // For target to be set that must mean that current is set and armed to pause. struct camu_sink_entry *prev_target = sink->target; if (prev_target) { + // For target to be set that must mean that current is set, + // armed to pause, and not ended. + al_assert(sink->current && !sink->current->ended); if (prev_target == entry) { sink->target = NULL; } else { sink->target = entry; } - camu_clock_pause(&prev_target->clock, at); + // The targets clock is only guaranteed to be resumed + // if it was set in the PAUSE_BOTH case. I feel like + // this needs to be simpler. + if (!camu_clock_is_paused(&prev_target->clock)) { // TMP + camu_clock_pause(&prev_target->clock, at); + } } else { - if (!sink->current || camu_clock_is_ended(&sink->current->clock)) { - switch_to(sink, entry); - } else { - sink->target = entry; - camu_clock_pause(&sink->current->clock, at); + if (camu_clock_is_paused(&entry->clock)) { // TMP + camu_clock_resume(&entry->clock, at); } + set_target_and_pause(sink, entry, at); } break; } @@ -988,20 +1039,18 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn camu_clock_resume(&entry->clock, at); struct camu_sink_entry *prev_target = sink->target; if (prev_target) { + al_assert(sink->current && !sink->current->ended); if (prev_target == entry) { sink->target = NULL; } else { sink->target = entry; camu_audio_buffer_unpause(&entry->audio.buf); } - camu_clock_pause(&prev_target->clock, at); - } else { - if (!sink->current || camu_clock_is_ended(&sink->current->clock)) { - switch_to(sink, entry); - } else { - sink->target = entry; - camu_clock_pause(&sink->current->clock, at); + if (!camu_clock_is_paused(&prev_target->clock)) { // TMP + camu_clock_pause(&prev_target->clock, at); } + } else { + set_target_and_pause(sink, entry, at); } break; } @@ -1078,6 +1127,7 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con if (!current) goto out; if (current->sequence == sequence) { + current->ended = false; #ifdef CAMU_SINK_LOCAL (void)at; lia_client_seek(¤t->client, pos, 0); @@ -1229,12 +1279,6 @@ void camu_sink_offset_volume(struct camu_sink *sink, f64 amount) void camu_sink_stop(struct camu_sink *sink) { - aki_mutex_lock(&sink->mutex); - if (sink->current) { - remove_entry_buffers(sink, sink->current); - sink->current = NULL; - } - aki_mutex_unlock(&sink->mutex); queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_AUDIO diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 18ef134..a9a6da0 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -40,6 +40,7 @@ struct camu_sink_entry { u16 id; s32 sequence; u16 lru; + bool ended; struct camu_clock clock; struct lia_client client; struct { |