From 692785bc9da6904cf17e986fb034730ed3d78231 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Sat, 14 Dec 2024 11:13:38 -0500 Subject: Work on pause and seek Signed-off-by: Andrew Opalach --- src/libsink/sink.c | 311 ++++++++++++++++++++++++++++++----------------------- src/libsink/sink.h | 7 +- 2 files changed, 184 insertions(+), 134 deletions(-) (limited to 'src/libsink') diff --git a/src/libsink/sink.c b/src/libsink/sink.c index d6d54fd..26c144b 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -109,15 +109,6 @@ static inline bool entry_video_buffer_held(struct camu_sink_entry *entry) } #endif -static inline bool entry_buffers_held(struct camu_sink_entry *entry) -{ -#ifndef CAMU_SINK_NO_VIDEO - return entry_audio_buffer_held(entry) || entry_video_buffer_held(entry); -#else - return entry_audio_buffer_held(entry); -#endif -} - static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry) { al_assert(!entry->ended && entry->audio.state != BUFFER_ENDED); @@ -177,13 +168,14 @@ static void add_or_queue_entry(struct camu_sink_entry *entry) if (entry->audio.state == BUFFER_INIT) { entry->audio.state = BUFFER_QUEUED; } else { + al_assert(entry->audio.state != BUFFER_QUEUED); add_audio_if_set_and_buffered(entry); } - #ifndef CAMU_SINK_NO_VIDEO if (entry->video.state == BUFFER_INIT) { entry->video.state = BUFFER_QUEUED; } else { + al_assert(entry->video.state != BUFFER_QUEUED); add_video_if_set_and_buffered(entry); } #endif @@ -216,6 +208,17 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent } #endif +static inline s32 get_sequence_for_command(struct camu_sink *sink) +{ + s32 sequence = LIANA_SEQUENCE_ANY; + if (sink->target) { + sequence = sink->target->sequence; + } else if (sink->current) { + sequence = sink->current->sequence; + } + return sequence; +} + static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) { switch (cmd->op) { @@ -266,13 +269,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_SKIP); - s32 sequence = LIANA_SEQUENCE_ANY; - if (sink->target) { - sequence = sink->target->sequence; - } else if (sink->current) { - sequence = sink->current->sequence; - } - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_s32(packet, cmd->value.i); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -293,8 +290,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE); - s32 sequence = sink->current ? sink->current->sequence : LIANA_SEQUENCE_ANY; - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_f64(packet, cmd->value.f); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); #endif @@ -305,8 +301,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); aki_packet_write_str(packet, &sink->default_list); aki_packet_write_u8(packet, CAMU_LIST_SEEK); - s32 sequence = sink->current ? sink->current->sequence : LIANA_SEQUENCE_ANY; - aki_packet_write_s32(packet, sequence); + aki_packet_write_s32(packet, get_sequence_for_command(sink)); aki_packet_write_f64(packet, cmd->value.f); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -509,6 +504,10 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) #endif if (VIDEO_ADDED_OR_EMPTY(entry)) { entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = entry->audio.buffer_paused ? STOP : START, + .value.i = CAMU_SINK_AUDIO + }); #ifndef CAMU_SINK_NO_VIDEO bool stop_video = false; // Single frames are unconditionally added in add_video_if_set_and_buffered(). @@ -532,10 +531,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) } #endif } - queue_cmd(entry->sink, (struct camu_sink_cmd){ - .op = START, - .value.i = CAMU_SINK_AUDIO - }); + } } @@ -555,6 +551,10 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) if (AUDIO_ADDED_OR_EMPTY(entry)) { if (entry->audio.state == BUFFER_ADDED) { entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = entry->audio.buffer_paused ? STOP : START, + .value.i = CAMU_SINK_AUDIO + }); } entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); maybe_remove_previous(entry->sink); @@ -600,16 +600,19 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } sink->current = target; + sink->current->audio.ignore_paused = false; } #ifndef CAMU_SINK_LOCAL static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) { - if (!sink->current || sink->current->ended) { + struct camu_sink_entry *current = sink->current; + if (!current || current->ended) { switch_to(sink, target); } else { sink->target = target; - camu_clock_pause(&sink->current->clock, at); + current->audio.ignore_paused = true; + camu_clock_pause(¤t->clock, at); } } #endif @@ -647,7 +650,8 @@ static void audio_buffer_callback(void *userdata, u8 op) break; case CAMU_BUFFER_PAUSED: aki_mutex_lock(&sink->mutex); - if (!entry->audio.armed) { + al_log_info("sink", "Audio buffer paused."); + if (!entry->audio.ignore_paused) { queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_AUDIO @@ -808,10 +812,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (!camu_audio_buffer_configure(&entry->audio.buf, stream, sink->audio.mixer)) { al_log_warn("sink", "Audio buffer failed to configure."); lia_client_disconnect(&entry->client); + return; } aki_mutex_lock(&sink->mutex); - entry->audio.state = entry->audio.state == BUFFER_QUEUED ? - BUFFER_SET_OR_BUFFERED : BUFFER_CONFIGURED; + if (entry->audio.state == BUFFER_QUEUED) { + entry->audio.state = BUFFER_SET_OR_BUFFERED; + } else if (entry->audio.state == BUFFER_INIT) { + entry->audio.state = BUFFER_CONFIGURED; + } else { + assert(false); + } if (VIDEO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry); aki_mutex_unlock(&sink->mutex); break; @@ -821,10 +831,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (!camu_video_buffer_configure(&entry->video.buf, stream, sink->video.renderer)) { al_log_warn("sink", "Video buffer failed to configure."); lia_client_disconnect(&entry->client); + return; } aki_mutex_lock(&sink->mutex); - entry->video.state = entry->video.state == BUFFER_QUEUED ? - BUFFER_SET_OR_BUFFERED : BUFFER_CONFIGURED; + if (entry->video.state == BUFFER_QUEUED) { + entry->video.state = BUFFER_SET_OR_BUFFERED; + } else if (entry->video.state == BUFFER_INIT) { + entry->video.state = BUFFER_CONFIGURED; + } else { + assert(false); + } if (AUDIO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry); aki_mutex_unlock(&sink->mutex); break; @@ -889,37 +905,46 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str bool reconnect = *(bool *)opaque; aki_mutex_lock(&sink->mutex); - al_assert(!(reconnect && entry->ended)); - - // There are 2 reasons this entry might be in previous. - // 1. It's getting cleaned up before any entries set after it are buffered. - // 2. It got added to previous then seeked. + // We need to call this if entry was added to previous then, + // - it's being cleaned up after (ENTRY_MAX_AGE - 1) entries were added but none buffered. + // - it was seeked. maybe_remove_from_previous(sink, entry); - if (entry->audio.state == BUFFER_ENDED) { - entry->audio.state = BUFFER_SET_OR_BUFFERED; - } else if (BUFFER_NOT_EMPTY(&entry->audio)) { - remove_entry_audio_buffer(sink, entry); + if (reconnect) { + // This may not be correct if entry->ended can ever be set anywhere + // besides end_entry_and_advance_queue(). + entry->ended = false; } + bool skip_audio = sink->audio.state == SINK_PAUSED; + if (entry->audio.state == BUFFER_ADDED) { + remove_entry_audio_buffer(sink, entry); + } + // SET_OR_BUFFERED, ADDED, or ENDED. + if (entry->audio.state > BUFFER_CONFIGURED) { + entry->audio.state = BUFFER_CONFIGURED; + } #ifndef CAMU_SINK_NO_VIDEO bool ignore_video = BUFFER_EMPTY(&entry->video); - if (entry->video.state == BUFFER_ENDED) { - entry->video.state = BUFFER_SET_OR_BUFFERED; - } else if (BUFFER_NOT_EMPTY(&entry->video)) { + if (!ignore_video) { ignore_video = reconnect && ENTRY_IS_SINGLE_FRAME(entry); - if (!ignore_video) { + } + if (!ignore_video) { + if (entry->video.state == BUFFER_ADDED) { remove_entry_video_buffer(sink, entry); } + if (entry->video.state > BUFFER_CONFIGURED) { + entry->video.state = BUFFER_CONFIGURED; + } } #endif aki_mutex_unlock(&sink->mutex); while ( // Block until buffers are removed. #ifndef CAMU_SINK_NO_VIDEO - ignore_video ? entry_audio_buffer_held(entry) : entry_buffers_held(entry) + (!skip_audio && entry_audio_buffer_held(entry)) || (!ignore_video && entry_video_buffer_held(entry)) #else - entry_buffers_held(entry) + (!skip_audio && entry_audio_buffer_held(entry)) #endif ) { BLOCKING_SLEEP(AKI_TS_FROM_USEC(2500)); } @@ -938,8 +963,24 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } case LIANA_CLIENT_RESUME_AT: { struct lia_timing *time = (struct lia_timing *)opaque; - camu_clock_set(&entry->clock, time->seek_pos / 1000000.0); - camu_clock_resume(&entry->clock, time->at); + aki_mutex_lock(&sink->mutex); + camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at); + aki_mutex_unlock(&sink->mutex); + break; + } + case LIANA_CLIENT_RECONNECTED: { + aki_mutex_lock(&sink->mutex); + if (entry == sink->current) { + if (entry->audio.state > BUFFER_QUEUED) { + add_audio_if_set_and_buffered(entry); + } +#ifndef CAMU_SINK_NO_VIDEO + if (entry->video.state > BUFFER_QUEUED && !ENTRY_IS_SINGLE_FRAME(entry)) { + add_video_if_set_and_buffered(entry); + } +#endif + } + aki_mutex_unlock(&sink->mutex); break; } case LIANA_CLIENT_EOF: { @@ -1004,25 +1045,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } } -static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, bool *created) +static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id) { - struct camu_sink_entry *entry; - al_array_foreach(sink->entries, i, entry) { - if (entry->id == id) { - *created = false; - return entry; - } - } - - entry = al_alloc_object(struct camu_sink_entry); + struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry); + entry->sink = sink; entry->id = id; entry->ended = false; - entry->sink = sink; camu_clock_init(&entry->clock, clock_callback, entry); entry->audio.state = BUFFER_INIT; - entry->audio.armed = false; + entry->audio.ignore_paused = false; camu_audio_buffer_init(&entry->audio.buf, &entry->clock); entry->audio.buf.callback = audio_buffer_callback; entry->audio.buf.userdata = entry; @@ -1039,11 +1072,20 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u16 id, al_array_push(sink->entries, entry); - *created = true; - return entry; } +static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u32 id) +{ + struct camu_sink_entry *entry; + al_array_foreach(sink->entries, i, entry) { + if (entry->id == id) { + return entry; + } + } + return NULL; +} + static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { @@ -1059,10 +1101,14 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn goto out; } + // Liana node info. str addr; aki_packet_read_str(packet, &addr); u16 port = aki_packet_read_u16(packet); - u16 node_id = aki_packet_read_u16(packet); + u32 node_id = aki_packet_read_u32(packet); + + // List entry info. + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u64 seek_pos = aki_packet_read_u64(packet); @@ -1070,16 +1116,25 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn bool previous_ended = aki_packet_read_bool(packet); bool ended = aki_packet_read_bool(packet); (void)previous_ended; - (void)ended; - bool created; - struct camu_sink_entry *entry = get_entry_from_id(sink, node_id, &created); + bool created = false; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + entry = create_entry(sink, id); + created = true; + } entry->sequence = sequence; sink->lru = al_u16_inc_wrap(sink->lru); entry->lru = sink->lru; if (created) { camu_clock_set(&entry->clock, seek_pos / 1000000.0); + // ended here does not map to entry->ended, we use it as a hint. + if (!ended) { + entry->audio.buffer_paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; + } else { + camu_clock_resume(&entry->clock, 0); + } struct camu_renderer *renderer = NULL; #ifndef CAMU_SINK_NO_VIDEO renderer = sink->video.renderer; @@ -1108,81 +1163,66 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn #else switch (pause) { case LIANA_PAUSE_NONE: { - struct camu_sink_entry *target = sink->target; if (sink->target) { - al_log_warn("sink", "Ignoring target on NONE."); - if (!camu_clock_is_armed(&target->clock)) { // TMP - camu_clock_pause(&target->clock, at); - } sink->target = NULL; + } else { + al_assert(sink->current != entry); } if (entry != sink->current) { - if (camu_clock_is_armed(&entry->clock)) { // TMP - camu_clock_resume(&entry->clock, at); - } switch_to(sink, entry); } break; } case LIANA_PAUSE_RESUME: { - struct camu_sink_entry *target = sink->target; if (sink->target) { - al_log_warn("sink", "Ignoring target on RESUME."); - if (!camu_clock_is_armed(&target->clock)) { // TMP - camu_clock_pause(&target->clock, at); - } sink->target = NULL; + } else { + al_assert(sink->current != entry); } - camu_clock_resume(&entry->clock, at); if (entry != sink->current) { switch_to(sink, entry); } else { camu_audio_buffer_unpause(&entry->audio.buf); } + camu_clock_resume(&entry->clock, at); break; } case LIANA_PAUSE_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. + // 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 { + al_assert(prev_target != entry); + if (sink->current != entry) { sink->target = entry; + } else { + sink->target = NULL; } - // 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_armed(&prev_target->clock)) { // TMP - camu_clock_pause(&prev_target->clock, at); - } + camu_clock_pause(&prev_target->clock, at); } else { - if (camu_clock_is_armed(&entry->clock)) { // TMP - camu_clock_resume(&entry->clock, at); - } + al_assert(sink->current != entry); pause_and_swap_to(sink, entry, at); } break; } case LIANA_PAUSE_BOTH: { - 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 { + al_assert(prev_target != entry); + if (sink->current != entry) { sink->target = entry; + } else { + sink->target = NULL; camu_audio_buffer_unpause(&entry->audio.buf); } - if (!camu_clock_is_armed(&prev_target->clock)) { // TMP - camu_clock_pause(&prev_target->clock, at); - } + camu_clock_pause(&prev_target->clock, at); } else { + al_assert(sink->current != entry); pause_and_swap_to(sink, entry, at); } + camu_clock_resume(&entry->clock, at); break; } } @@ -1198,7 +1238,6 @@ out: return false; } -// TODO: pause and seek shouldn't depend on the entry being current. static bool pause_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { @@ -1206,33 +1245,38 @@ static bool pause_command_callback(void *userdata, struct aki_rpc_connection *co (void)conn; (void)rpacket; + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u8 pause = aki_packet_read_u8(packet); aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *current = sink->current; - if (!current) goto out; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + aki_mutex_unlock(&sink->mutex); + goto out; + } + al_assert(entry->sequence == sequence); #ifdef CAMU_SINK_LOCAL - (void)sequence; (void)at; (void)pause; - sink_local_pause(sink, current); + sink_local_pause(sink, entry); #else - if (current->sequence == sequence) { - if (pause == LIANA_PAUSE_PAUSE) { - camu_clock_pause(&sink->current->clock, at); - current->audio.armed = false; - } else { - camu_clock_resume(&sink->current->clock, at); - if (sink->audio.state == SINK_PAUSED) { - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); - sink->audio.state = SINK_PLAYING; - } else { - current->audio.armed = true; - } - } + switch (pause) { + case LIANA_PAUSE_PAUSE: + entry->audio.buffer_paused = true; + camu_clock_pause(&entry->clock, at); + break; + case LIANA_PAUSE_RESUME: + entry->audio.buffer_paused = false; + camu_clock_resume(&entry->clock, at); + camu_audio_buffer_unpause(&entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = START, + .value.i = CAMU_SINK_AUDIO + }); + break; } #endif out: @@ -1249,22 +1293,29 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con (void)conn; (void)rpacket; + u32 id = aki_packet_read_u32(packet); s32 sequence = aki_packet_read_s32(packet); u64 at = aki_packet_read_u64(packet); u64 pos = aki_packet_read_u64(packet); aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *current = sink->current; + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) { + aki_mutex_unlock(&sink->mutex); + goto out; + } + if (entry == sink->current && sink->target) { + switch_to(sink, sink->target); + sink->target = NULL; + } + al_assert(entry->sequence == sequence); aki_mutex_unlock(&sink->mutex); - if (!current) goto out; - if (current->sequence == sequence) { - current->ended = false; #ifdef CAMU_SINK_LOCAL - at = 0; + at = 0; #endif - lia_client_seek(¤t->client, pos, at); - } + // The rest of the seek is handled in LIANA_CLIENT_REMOVE_BUFFERS. + lia_client_seek(&entry->client, pos, at); out: aki_packet_free(packet); @@ -1364,14 +1415,10 @@ void camu_sink_toggle_pause(struct camu_sink *sink) aki_mutex_lock(&sink->mutex); struct camu_sink_entry *current = sink->current; aki_mutex_unlock(&sink->mutex); - f64 pts = -1.0; - if (!camu_clock_is_paused(¤t->clock)) { - pts = camu_clock_get_pts(¤t->clock, 0.0); - } if (current) { queue_cmd(sink, (struct camu_sink_cmd){ .op = TOGGLE_PAUSE, - .value.f = pts, + .value.f = camu_clock_get_pts(¤t->clock, 0.0), .opaque = current }); } diff --git a/src/libsink/sink.h b/src/libsink/sink.h index f544a36..674b684 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -37,7 +37,7 @@ enum { }; struct camu_sink_entry { - u16 id; + u32 id; s32 sequence; u16 lru; bool ended; @@ -45,7 +45,10 @@ struct camu_sink_entry { struct lia_client client; struct { u8 state; - bool armed; + // Don't start the audio output for this entry. + bool buffer_paused; + // Don't stop the audio output when this entry is paused. + bool ignore_paused; struct camu_audio_buffer buf; struct lia_vcr_track *track; } audio; -- cgit v1.2.3-101-g0448