diff options
| -rw-r--r-- | flake.lock | 24 | ||||
| -rw-r--r-- | src/buffer/video_null.h | 1 | ||||
| -rw-r--r-- | src/codec/ffmpeg/decoder.h | 2 | ||||
| -rw-r--r-- | src/fruits/cmv/cmv.c | 2 | ||||
| -rw-r--r-- | src/fruits/common.h | 3 | ||||
| -rw-r--r-- | src/liana/client.c | 15 | ||||
| -rw-r--r-- | src/liana/client.h | 5 | ||||
| -rw-r--r-- | src/liana/list.c | 73 | ||||
| -rw-r--r-- | src/libsink/sink.c | 259 | ||||
| -rw-r--r-- | src/libsink/sink.h | 2 |
10 files changed, 185 insertions, 201 deletions
@@ -23,11 +23,11 @@ ] }, "locked": { - "lastModified": 1741955947, - "narHash": "sha256-2lbURKclgKqBNm7hVRtWh0A7NrdsibD0EaWhahUVhhY=", + "lastModified": 1742326330, + "narHash": "sha256-Tumt3tcMXJniSh7tw2gW+WAnVLeB3WWm+E+yYFnLBXo=", "owner": "nix-community", "repo": "home-manager", - "rev": "4e12151c9e014e2449e0beca2c0e9534b96a26b4", + "rev": "22a36aa709de7dd42b562a433b9cefecf104a6ee", "type": "github" }, "original": { @@ -39,11 +39,11 @@ }, "nixos-hardware": { "locked": { - "lastModified": 1741792691, - "narHash": "sha256-f0BVt1/cvA0DQ/q3rB+HY4g4tKksd03ZkzI4xehC2Ew=", + "lastModified": 1742376361, + "narHash": "sha256-VFMgJkp/COvkt5dnkZB4D2szVdmF6DGm5ZdVvTUy61c=", "owner": "NixOS", "repo": "nixos-hardware", - "rev": "e1f12151258b12c567f456d8248e4694e9390613", + "rev": "daaae13dff0ecc692509a1332ff9003d9952d7a9", "type": "github" }, "original": { @@ -59,11 +59,11 @@ "nixpkgs": "nixpkgs" }, "locked": { - "lastModified": 1741870048, - "narHash": "sha256-odXRdNZGdXg1LmwlAeWL85kgy/FVHsgKlDwrvbR2BsU=", + "lastModified": 1742209789, + "narHash": "sha256-D3GWoPC7dbF4LSJ5VfskuD6K77Ej4WyTsynJ06eh9kw=", "owner": "nix-community", "repo": "NixOS-WSL", - "rev": "5d76001e33ee19644a598ad80e7318ab0957b122", + "rev": "bc827c2924c46f2344d3168fd82c6711aaceb610", "type": "github" }, "original": { @@ -91,11 +91,11 @@ }, "nixpkgs_2": { "locked": { - "lastModified": 1742069588, - "narHash": "sha256-C7jVfohcGzdZRF6DO+ybyG/sqpo1h6bZi9T56sxLy+k=", + "lastModified": 1742288794, + "narHash": "sha256-Txwa5uO+qpQXrNG4eumPSD+hHzzYi/CdaM80M9XRLCo=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "c80f6a7e10b39afcc1894e02ef785b1ad0b0d7e5", + "rev": "b6eaf97c6960d97350c584de1b6dcff03c9daf42", "type": "github" }, "original": { diff --git a/src/buffer/video_null.h b/src/buffer/video_null.h index 8423b7c..60e73f1 100644 --- a/src/buffer/video_null.h +++ b/src/buffer/video_null.h @@ -64,6 +64,7 @@ static void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_co static void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, struct camu_codec_packet *packet) { + // This packet gets freed in the client. (void)buf; (void)packet; } diff --git a/src/codec/ffmpeg/decoder.h b/src/codec/ffmpeg/decoder.h index db7a6f8..c6fb5fc 100644 --- a/src/codec/ffmpeg/decoder.h +++ b/src/codec/ffmpeg/decoder.h @@ -6,7 +6,7 @@ #include "../codec.h" #ifndef CAMU_SINK_NO_VIDEO -#define CAMU_FF_DECODER_HWACCEL +//#define CAMU_FF_DECODER_HWACCEL #endif struct camu_ff_decoder { diff --git a/src/fruits/cmv/cmv.c b/src/fruits/cmv/cmv.c index e539f01..86b6f75 100644 --- a/src/fruits/cmv/cmv.c +++ b/src/fruits/cmv/cmv.c @@ -182,7 +182,7 @@ s32 window_system_main(s32 argc, str *argv) if (local) { for (s32 i = 1; i < argc; i++) { struct nn_packet *packet = nn_packet_create(); -#ifdef CAMU_HAVE_PORTAL +#if defined CAMU_HAVE_PORTAL if (al_str_at(&argv[i], 0) == ';' || camu_is_url(&argv[i], 0)) { nn_packet_write_u8(packet, CAMU_RESOURCE_SIMPLE_SEARCH); #elif defined NAUNET_HAS_CURL diff --git a/src/fruits/common.h b/src/fruits/common.h index fb9a127..2df5f73 100644 --- a/src/fruits/common.h +++ b/src/fruits/common.h @@ -13,5 +13,8 @@ static str CAMU_TEST_CONTROL_PATH = al_str_c("/tmp/camu_control_sock"); AL_UNUSED_VARIABLE_POP +//#define CAMU_TEST_TYPE NNWT_SOCKET_UNIX +//#define CAMU_TEST_ADDR CAMU_TEST_PATH + #define CAMU_TEST_TYPE NNWT_SOCKET_TCP #define CAMU_TEST_ADDR CAMU_TEST_IP diff --git a/src/liana/client.c b/src/liana/client.c index 726f8ff..0aec3e5 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -179,12 +179,15 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream) .pause = LIANA_PAUSE_NONE }; client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); - bool unconfigured = client->mask == 0; - if (unconfigured) { + struct lia_reconnect_info rec = { + .reconnect = true, + .unconfigured = client->mask == 0 + }; + if (rec.unconfigured) { al_assert(client->connection_id == 0); al_log_warn("liana", "Handling reconnect on unconfigured client."); } - client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &unconfigured); + client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, &rec); } else { al_assert(client->connection_id == 0); } @@ -212,8 +215,12 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream * } else { lia_vcr_close_all(&client->vcr); } + struct lia_reconnect_info rec = { + .reconnect = client->reconnect == RECONNECT_ON_CONNECTION_CLOSED, + .unconfigured = client->mask == 0 + }; // We need to account for REMOVE_BUFFERS possibly running the event loop to wait. - client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->reconnect); + client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &rec); if (client->reconnect == RECONNECT_ON_CONNECTION_CLOSED) { // If reconnect() errors, this will close the client on recursion. client->reconnect = RECONNECT_SIGNAL_CLIENT; diff --git a/src/liana/client.h b/src/liana/client.h index 4e2ff9e..eb84051 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -18,6 +18,11 @@ enum { LIANA_CLIENT_CLOSED }; +struct lia_reconnect_info { + bool reconnect; + bool unconfigured; +}; + struct lia_prefs { u8 enabled_mask; s8 audio_lang; diff --git a/src/liana/list.c b/src/liana/list.c index 3e2a350..d5eef48 100644 --- a/src/liana/list.c +++ b/src/liana/list.c @@ -43,22 +43,6 @@ void lia_list_init(struct lia_list *list, str *name) list->cmd = NULL; } -static bool assume_ended(struct lia_list_entry *entry, u64 at) -{ - if (entry->duration == LIANA_TIMESTAMP_INVALID) return false; - if (entry->ended || entry->duration == 0) return true; - if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) { - if (entry->offset > entry->duration) { - return true; - } - if (at < entry->start) return false; - if (at - entry->start >= entry->duration - entry->offset) { - return true; - } - } - return false; -} - #define META_OPAQUE(op) ((u8[]){ op }) static void signal_meta(struct lia_list *list, struct lia_list_entry *entry, u8 meta) @@ -169,11 +153,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) u8 pause; u64 at = LIANA_TIMESTAMP_INVALID; u64 seek_pos = current->offset; - bool ended = assume_ended(current, now); - if (ended && current->duration != LIANA_TIMESTAMP_INVALID) { - seek_pos = current->duration; - } - if (ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { + if (current->paused_at != LIANA_TIMESTAMP_INVALID) { pause = LIANA_PAUSE_NONE; } else { at = now + LIANA_BASE_DELAY; @@ -188,7 +168,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) .at = at, .seek_pos = seek_pos, .pause = pause, - .ended = ended + .ended = current->ended }; sink_set(sink, current, list->current, &time); } @@ -303,19 +283,13 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) // Target was never started before. target->start = at; } else { - // An ended entry may never have been paused, but a non-ended entry that wasn't set - // cannot be unpaused. Checking assume_ended(target) should be safe here as long - // as it can't go from true to false (consideration for seek?). - //al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID); + // Target is paused or static. + al_assert(target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID); } - // These are not equivalent to current/target->ended. - bool current_ended = assume_ended(current, now); - bool target_ended = assume_ended(target, now); - - // If current is ended or paused. - if (current_ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { - if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) { + // If current->duration = LIANA_TIMESTAMP_INVALID, handling of a static entry happens on the sink. + if (current->duration == 0 || current->paused_at != LIANA_TIMESTAMP_INVALID) { + if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) { // Swap entries and ignore their clocks. pause = LIANA_PAUSE_NONE; } else { @@ -324,7 +298,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) pause = LIANA_PAUSE_RESUME; } } else { // Current is playing. - if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) { + if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) { // Pause-swap current, don't touch target's clock. pause = LIANA_PAUSE_PAUSE; } else { @@ -334,13 +308,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) } } - // Pause current and set the held flag indicating it should be resumed - // if it becomes the target of a skip. - // - // @TODO: Why not hold an ended entry? The idea of ended entries being - // unpaused but never held seems like a over-complication. - // sink-side ended should only reference buffer state? i.e. pause the entries clock even if ended. - if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) { + // Pause current to be resumed if it becomes the target of a skip (hold). + if (current->duration != 0 && current->paused_at == LIANA_TIMESTAMP_INVALID) { current->paused_at = at; if (current->paused_at < current->start) { current->paused_at = current->start; @@ -350,14 +319,13 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) current->held = true; } - trace("skipto (#%u-#%u): pause: %hhu, held: %s, current_ended: %s, target_ended: %s.", - current->id, target->id, pause, BOOLSTR(current->held), BOOLSTR(current_ended), BOOLSTR(target_ended)); + trace("skipto(#%u-#%u): pause: %hhu, held: %s.", current->id, target->id, pause, BOOLSTR(current->held)); struct lia_timing time = { .at = at, .seek_pos = target->offset, .pause = pause, - .ended = target_ended + .ended = target->ended }; set_current(list, target, index, &time); @@ -382,13 +350,12 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) } struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); - al_assert(entry && !entry->held); + if (!entry || entry->ended || entry->duration == 0) return; + al_assert(!entry->held); u64 now = nn_get_timestamp(); u8 pause = (entry->paused_at == LIANA_TIMESTAMP_INVALID) ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME; u64 at; - bool ended = assume_ended(entry, now); - if (ended) return; switch (pause) { case LIANA_PAUSE_PAUSE: @@ -409,13 +376,13 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) break; } - trace("toggle_pause (#%u): pts: %f, pause: %hhu.", entry->id, pts, pause); + trace("toggle_pause(#%u): pts: %f, pause: %hhu.", entry->id, pts, pause); struct lia_timing time = { .at = at, .seek_pos = LIANA_TIMESTAMP_INVALID, .pause = pause, - .ended = false + .ended = entry->ended }; struct lia_list_sink *sink; @@ -436,7 +403,7 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos) struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); al_assert(entry); - if (entry->duration == LIANA_TIMESTAMP_INVALID || entry->duration == 0) { + if (entry->duration == 0 || entry->duration == LIANA_TIMESTAMP_INVALID) { warn("Skipping seek on entry with no duration."); return; } @@ -455,13 +422,13 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos) list->idle = false; - trace("seek (#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause); + trace("seek(#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause); struct lia_timing time = { .at = at, .seek_pos = pos, .pause = pause, - .ended = false + .ended = entry->ended }; struct lia_list_sink *sink; @@ -491,7 +458,7 @@ static bool handle_end(struct lia_list *list, u32 id, u32 reset_id) return true; } - trace("end (#%u).", entry->id); + trace("end(#%u).", entry->id); entry->ended = true; entry->offset = entry->duration; diff --git a/src/libsink/sink.c b/src/libsink/sink.c index d3647f1..5038566 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -57,7 +57,7 @@ enum { #define SINK_LRU_MAX UINT16_MAX // Number of entries to keep buffered at one time. -#define ENTRY_MAX_AGE 3 +#define ENTRY_MAX_AGE 4 // printf format for entries. #if defined AL_WE_64BIT @@ -205,9 +205,10 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry) } } -// It's possible for some of an entries buffers to be ENDED while others are still -// ADDED and playing. We handle that by making remove_entry_buffers() and -// add_audio/video_if_set_and_buffered() no-ops for ENDED buffers. +// It's possible for some of an entry's buffers to be ENDED while others are still ADDED. +// This means entry->ended and BUFFER_ENDED have two distinct considerations. +// entry->ended: Completely ignored, needs special consideration in CLIENT_REMOVE_BUFFERS. +// BUFFER_ENDED: No-op'd in remove_entry_buffers() and add_or_queue_entry() but otherwise unchanged. static void remove_entry_buffers(struct camu_sink_entry *entry) { trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); @@ -225,16 +226,17 @@ static void add_video_if_set_and_buffered(struct camu_sink_entry *entry); static void add_or_queue_entry(struct camu_sink_entry *entry) { + trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); + al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); + al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); if (AUDIO_STATE(entry) == BUFFER_INIT) { AUDIO_STATE(entry) = BUFFER_QUEUED; - } else { - al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); + } else if (AUDIO_STATE(entry) != BUFFER_ENDED) { add_audio_if_set_and_buffered(entry); } if (VIDEO_STATE(entry) == BUFFER_INIT) { VIDEO_STATE(entry) = BUFFER_QUEUED; - } else { - al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); + } else if (VIDEO_STATE(entry) != BUFFER_ENDED) { add_video_if_set_and_buffered(entry); } } @@ -490,16 +492,16 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->current = NULL; sink->queued = NULL; sink->target = NULL; - sink->reconnecting = NULL; + sink->detached = 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; - sink->video.renderer = renderer; + sink->audio.mixer = mixer; sink->video.state = SINK_PAUSED; + sink->video.renderer = renderer; return true; } @@ -557,31 +559,27 @@ static void maybe_remove_previous(struct camu_sink *sink) sink->previous.count = 0; } +// Due to the looseness of the previous queue, we may have to explicitly remove an +// entry if it becomes incorrect to attempt removing it's buffers. static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) { + bool removed = false; struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { if (previous == key) { maybe_remove_previous(sink); + removed = true; break; } } -} - -// Due to the looseness of the previous queue, we may have to explicitly remove an entry -// if it becomes incorrect to attempt removing it's buffers. -// An obvious example of this is at the point an entry gets freed. -static void maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry) -{ - trace("maybe_remove_from_previous("ENTRY_FMT").", ENTRY_ARG(entry)); - al_array_remove_all(sink->previous, entry); + trace("remove_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); } // Every call to maybe_add_to_previous() must map to a remove_entry_buffers(). static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target) { - trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(previous), ENTRY_ARG(target), AUDIO_STATE(previous), VIDEO_STATE(previous)); + trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT").", ENTRY_ARG(previous), ENTRY_ARG(target)); al_assert(previous != target); al_assert(!previous->ended); // If none of the entry's buffers are added, we don't care about adding it to previous. @@ -624,10 +622,8 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) al_assert(AUDIO_STATE(entry) != BUFFER_INIT); al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); + al_assert(AUDIO_STATE(entry) != BUFFER_ENDED); switch (AUDIO_STATE(entry)) { - case BUFFER_ENDED: - warn("Tried to add an ended audio buffer."); - return; case BUFFER_CONFIGURED: AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED; break; @@ -646,17 +642,18 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) al_assert(VIDEO_STATE(entry) != BUFFER_INIT); al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); - if (VIDEO_STATE(entry) == BUFFER_ENDED) { - warn("Tried to add an ended video buffer."); - return; - } else if (VIDEO_STATE(entry) == BUFFER_CONFIGURED) { + al_assert(VIDEO_STATE(entry) != BUFFER_ENDED); + switch (VIDEO_STATE(entry)) { + case BUFFER_CONFIGURED: VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED; - } else if (VIDEO_STATE(entry) == BUFFER_SET_OR_BUFFERED) { + break; + case BUFFER_SET_OR_BUFFERED: VIDEO_STATE(entry) = BUFFER_ADDED; if (VIDEO_IS_SINGLE_FRAME(entry)) add_entry_video_buffer(entry); if (AUDIO_ADDED_OR_EMPTY(entry)) { do_add_entry(entry); } + break; } } @@ -666,12 +663,16 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (sink->current) { struct camu_sink_entry *current = sink->current; + struct camu_sink_entry *detached = sink->detached; al_assert(current != target); - bool ensure_removed = sink->reconnecting || current->ended; - if (sink->reconnecting) { - al_assert(sink->reconnecting == current); - sink->reconnecting = NULL; - warn("Unset reconnecting as a substitute for remove."); + bool ensure_removed = detached || current->ended; + if (detached) { + al_assert(detached == current); + sink->detached = NULL; + warn("Unset detached as a substitute for remove."); + // This should only matter if detached was unconfigured. + if (AUDIO_EMPTY(detached)) AUDIO_STATE(detached) = BUFFER_INIT; + if (VIDEO_EMPTY(detached)) VIDEO_STATE(detached) = BUFFER_INIT; } else if (!current->ended) { maybe_add_to_previous(sink, current, target); } @@ -721,24 +722,26 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta struct camu_sink_entry *current = sink->current; trace("pause_and_swap_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(current)); al_assert(target != current); - if (!current || current->ended) { - switch_to(sink, target); - } else { - sink->target = target; + if (current) { + if (!current->ended) sink->target = target; current->audio.ignore_paused = true; camu_clock_pause(¤t->clock, at); } + if (!current || current->ended) { + switch_to(sink, target); + } } #endif static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry) { info("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry)); + remove_previous_if_contains(sink, entry); #ifdef CAMU_SINK_ONESHOT sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL); return false; #endif - maybe_remove_from_previous(sink, entry); + // This entry's buffers cannot be added again until after a reset. entry->ended = true; queue_cmd(sink, (struct camu_sink_cmd){ .op = END, @@ -750,6 +753,7 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink return true; #endif if (sink->target) { + al_assert(!sink->queued); switch_to(sink, sink->target); sink->target = NULL; return true; @@ -790,16 +794,14 @@ static void audio_buffer_callback(void *userdata, u8 op) nn_mutex_unlock(&sink->lock); break; case CAMU_BUFFER_EOF: { - // @TODO: Sync around BUFFER_EOF is not well tested. info("Audio EOF."); nn_mutex_lock(&sink->lock); - // Getting EOF on a buffer that isn't ADDED is very possible if the audio - // output is threaded. Even more if we had to wait on the lock above. + // This buffer's state could be ADDED, SET_OR_BUFFERED, or CONFIGURED. + // Having threaded outputs means anything could have happened while waiting on the lock. + // If we were waiting in CLIENT_REMOVE_BUFFERS, state could very well be CONFIGURED. if (AUDIO_STATE(entry) == BUFFER_ADDED) { remove_entry_audio_buffer(entry); } - // This assert likely doesn't matter due to the handling of the ENDED state. - al_assert(AUDIO_STATE(entry) == BUFFER_SET_OR_BUFFERED); AUDIO_STATE(entry) = BUFFER_ENDED; if (VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry)) { end_entry_and_advance_queue(sink, entry); @@ -840,6 +842,7 @@ static void video_buffer_callback(void *userdata, u8 op) bool swapped = false; bool single_frame = VIDEO_IS_SINGLE_FRAME(entry); nn_mutex_lock(&sink->lock); + // This buffer's state could be ADDED, SET_OR_BUFFERED, or CONFIGURED. if (!AUDIO_EMPTY(entry)) { camu_audio_buffer_set_no_video(&entry->audio.buf, true); } @@ -847,7 +850,6 @@ static void video_buffer_callback(void *userdata, u8 op) if (VIDEO_STATE(entry) == BUFFER_ADDED) { remove_entry_video_buffer(entry); } - al_assert(VIDEO_STATE(entry) == BUFFER_SET_OR_BUFFERED); VIDEO_STATE(entry) = BUFFER_ENDED; if (AUDIO_ENDED_OR_EMPTY(entry)) { swapped = end_entry_and_advance_queue(sink, entry); @@ -963,7 +965,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str return; } nn_mutex_lock(&sink->lock); - al_assert(AUDIO_STATE(entry) == BUFFER_QUEUED || AUDIO_STATE(entry) == BUFFER_INIT); + al_assert(AUDIO_EMPTY(entry)); if (AUDIO_STATE(entry) == BUFFER_QUEUED) { AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED; } else if (AUDIO_STATE(entry) == BUFFER_INIT) { @@ -979,7 +981,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str return; } nn_mutex_lock(&sink->lock); - al_assert(VIDEO_STATE(entry) == BUFFER_QUEUED || VIDEO_STATE(entry) == BUFFER_INIT); + al_assert(VIDEO_EMPTY(entry)); if (VIDEO_STATE(entry) == BUFFER_QUEUED) { VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED; } else if (VIDEO_STATE(entry) == BUFFER_INIT) { @@ -1034,70 +1036,77 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str break; } case LIANA_CLIENT_REMOVE_BUFFERS: { - bool reconnect = *(bool *)opaque; - + struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); + trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current)); - trace("remove_buffers("ENTRY_FMT", %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(reconnect), BOOLSTR(entry == sink->current)); + // This entry might be in previous if it was added to previous then, + // 1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. + // 2. it was seeked. + remove_previous_if_contains(sink, entry); - if (reconnect && entry == sink->current) { - sink->reconnecting = entry; + // If this entry is still current on CLIENT_RECONNECTED, re-add buffers. + if (rec->reconnect && entry == sink->current) { + sink->detached = entry; } - // Entry might be in previous here if it was added to previous then, - // - it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. - // - it was seeked. - remove_previous_if_contains(sink, entry); + // Ignore unconfigured entries. + if (rec->unconfigured) { + al_assert(!entry->ended); + al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry)); + nn_mutex_unlock(&sink->lock); + return; + } - if (AUDIO_STATE(entry) == BUFFER_ADDED) { + // An empty buffer is guaranteed to not be held. + if (!AUDIO_ENDED_OR_EMPTY(entry)) { + // Remove for re-add in CLIENT_RECONNECTED. + remove_entry_audio_buffer(entry); + // Remove again for BUFFER_BUFFERED. remove_entry_audio_buffer(entry); } - bool ignore_video = reconnect && VIDEO_IS_SINGLE_FRAME(entry); - if (!ignore_video && VIDEO_STATE(entry) == BUFFER_ADDED) { + // Slight optimization. A duplicate frame will still be sent but discarded in the video buffer. + bool ignore_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry); + if (!ignore_video && !VIDEO_ENDED_OR_EMPTY(entry)) { + remove_entry_video_buffer(entry); remove_entry_video_buffer(entry); } - // Resolve any queued REMOVE_BUFFER requests before blocking. - // This is why we are safe to block the event loop thread even if - // MIXER_THREADED_START_STOP is not set. + // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER happens from the sink's + // command queue. So, this is necessary for safely blocking the loop here. run_queue_by_opaque(sink, entry); + // Unlock to wait. nn_mutex_unlock(&sink->lock); while (entry_audio_buffer_held(entry) || (!ignore_video && entry_video_buffer_held(entry))) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } - // Reset possible ENDED state here in case the entry ended at - // some point after unlocking to block above. + // Reset possible ENDED state here in case the entry ended at some point after unlocking above. nn_mutex_lock(&sink->lock); - if (reconnect && entry->ended) { - entry->ended = false; - if (AUDIO_STATE(entry) == BUFFER_QUEUED) { - AUDIO_STATE(entry) = BUFFER_INIT; - } - if (VIDEO_STATE(entry) == BUFFER_QUEUED) { - VIDEO_STATE(entry) = BUFFER_INIT; - } - } - // SET_OR_BUFFERED, ADDED, or ENDED. - if (AUDIO_STATE(entry) > BUFFER_CONFIGURED) { + if (AUDIO_STATE(entry) == BUFFER_ENDED) { + al_assert(!AUDIO_EMPTY(entry)); AUDIO_STATE(entry) = BUFFER_CONFIGURED; + } else if (entry->ended && AUDIO_EMPTY(entry)) { + // If entry is ended, it may have forewent a remove_entry_buffers(). + AUDIO_STATE(entry) = BUFFER_INIT; } - if (!ignore_video && VIDEO_STATE(entry) > BUFFER_CONFIGURED) { + if (VIDEO_STATE(entry) == BUFFER_ENDED) { + al_assert(!VIDEO_EMPTY(entry)); VIDEO_STATE(entry) = BUFFER_CONFIGURED; + } else if (entry->ended && !ignore_video && VIDEO_EMPTY(entry)) { + VIDEO_STATE(entry) = BUFFER_INIT; } + entry->ended = false; nn_mutex_unlock(&sink->lock); - if (reconnect) { - if (!AUDIO_EMPTY(entry)) { - camu_audio_buffer_reset(&entry->audio.buf); - } - if (!VIDEO_EMPTY(entry) && !ignore_video) { - camu_video_buffer_reset(&entry->video.buf); - } + // This buffer won't be re-added until after a CLIENT_RECONNECTED event. + if (rec->reconnect) { + if (!AUDIO_EMPTY(entry)) camu_audio_buffer_reset(&entry->audio.buf); + if (!ignore_video && !VIDEO_EMPTY(entry)) camu_video_buffer_reset(&entry->video.buf); } break; @@ -1109,39 +1118,42 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str time->at = 0; #endif nn_mutex_lock(&sink->lock); -#if defined LIANA_LIST_SCUFFED_LOOP +#ifdef LIANA_LIST_SCUFFED_LOOP if (time->seek_pos == 0) { camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock)); } else { #endif camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at); -#if defined LIANA_LIST_SCUFFED_LOOP +#ifdef LIANA_LIST_SCUFFED_LOOP } #endif nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_RECONNECTED: { - bool unconfigured = *(bool *)opaque; + struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); - trace("reconnected("ENTRY_FMT"), reconnecting: "ENTRY_FMT".", ENTRY_ARG(entry), ENTRY_ARG(sink->reconnecting)); - if (entry == sink->reconnecting) { + trace("reconnected("ENTRY_FMT"), detached: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), ENTRY_ARG(sink->detached), AUDIO_STATE(entry), VIDEO_STATE(entry)); + if (entry == sink->detached) { al_assert(entry == sink->current); - if (AUDIO_STATE(entry) > BUFFER_QUEUED) { - al_assert(!unconfigured); - add_audio_if_set_and_buffered(entry); + al_assert(AUDIO_STATE(entry) != BUFFER_ENDED); + al_assert(VIDEO_STATE(entry) != BUFFER_ENDED); + if (rec->unconfigured) { + // The value of unconfigured is consistent from CLIENT_REMOVE_BUFFERS. + al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry)); } else { - AUDIO_STATE(entry) = BUFFER_QUEUED; - } - if (VIDEO_STATE(entry) > BUFFER_QUEUED) { - al_assert(!unconfigured); - if (!VIDEO_IS_SINGLE_FRAME(entry)) { + if (AUDIO_EMPTY(entry)) { + AUDIO_STATE(entry) = BUFFER_QUEUED; + } else { + add_audio_if_set_and_buffered(entry); + } + if (VIDEO_EMPTY(entry)) { + VIDEO_STATE(entry) = BUFFER_QUEUED; + } else if (!VIDEO_IS_SINGLE_FRAME(entry)) { add_video_if_set_and_buffered(entry); } - } else { - VIDEO_STATE(entry) = BUFFER_QUEUED; } - sink->reconnecting = NULL; + sink->detached = NULL; } nn_mutex_unlock(&sink->lock); break; @@ -1165,7 +1177,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str break; } case LIANA_CLIENT_CLOSED: { - // LIANA_CLIENT_REMOVE_BUFFERS has been called on this entry before we're here. + // CLIENT_REMOVE_BUFFERS has been called on this entry before we're here. nn_mutex_lock(&sink->lock); if (VIDEO_STATE(entry) == BUFFER_ADDED) { @@ -1183,9 +1195,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str sink->target = (struct camu_sink_entry *)0xb00b; warn("Attempting to handle a disconnected target."); } else if (entry == sink->current) { - if (sink->reconnecting) { - al_assert(sink->reconnecting == sink->current); - sink->reconnecting = NULL; + if (sink->detached) { + al_assert(sink->detached == sink->current); + sink->detached = NULL; } if (sink->target) { switch_to(sink, sink->target); @@ -1210,7 +1222,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str lia_client_free(&entry->client); camu_audio_buffer_free(&entry->audio.buf); camu_video_buffer_free(&entry->video.buf); - info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "error" : "cleanup"); + info("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup"); al_free(entry); break; @@ -1240,6 +1252,11 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id) camu_clock_init(&entry->clock, clock_callback, entry); entry->paused = false; + entry->client.callback = client_callback; + entry->client.userdata = entry; + entry->client.renderer = sink->video.renderer; + entry->client.prefs = sink->prefs; + AUDIO_STATE(entry) = BUFFER_INIT; entry->audio.ignore_paused = false; camu_audio_buffer_init(&entry->audio.buf, &entry->clock); @@ -1251,11 +1268,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u32 id) entry->video.buf.callback = video_buffer_callback; entry->video.buf.userdata = entry; - entry->client.callback = client_callback; - entry->client.userdata = entry; - entry->client.renderer = sink->video.renderer; - entry->client.prefs = sink->prefs; - al_array_push(sink->entries, entry); return entry; @@ -1304,44 +1316,30 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, u64 seek_pos = nn_packet_read_u64(packet); u8 pause = nn_packet_read_u8(packet); bool ended = nn_packet_read_bool(packet); + (void)ended; u32 reset_id = nn_packet_read_u32(packet); - struct camu_sink_entry *current = sink->current; struct camu_sink_entry *entry = get_entry_from_id(sink, id); bool create = !entry; if (create) entry = create_entry(sink, id); entry->sequence = sequence; - sink->lru = al_u16_add_wrap(sink->lru, SINK_LRU_MAX); + sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX); entry->lru = sink->lru; entry->reset_id = reset_id; - if (create) { + entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; camu_clock_set(&entry->clock, seek_pos / 1000000.0); - // ended here does not map to entry->ended, it means the server expects the entry to be ended. - if (!ended) { - entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; - } else { -#ifndef CAMU_SINK_LOCAL - camu_clock_resume(&entry->clock, 0); -#endif - } lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos); } if (op == LIANA_SINK_BUFFER) { goto out; } else if (op == LIANA_SINK_BUFFER_AND_QUEUE) { -#ifdef CAMU_SINK_LOCAL - camu_clock_resume(&entry->clock, 0); -#else - if (pause == LIANA_PAUSE_RESUME) { - camu_clock_resume(&entry->clock, at); - } -#endif sink->queued = entry; goto out; } + struct camu_sink_entry *current = sink->current; #ifdef CAMU_SINK_LOCAL (void)at; al_assert(entry != current); @@ -1629,8 +1627,11 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; u64 duration = 0; - if (current) duration = current->client.duration; - f64 pts = camu_clock_get_last_pts(¤t->clock); + f64 pts = 0.0; + if (current) { + duration = current->client.duration; + pts = camu_clock_get_last_pts(¤t->clock); + } nn_mutex_unlock(&sink->lock); if (!current || duration == 0) return; struct camu_sink_cmd cmd = { diff --git a/src/libsink/sink.h b/src/libsink/sink.h index d1bcd26..ba37241 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -88,7 +88,7 @@ struct camu_sink { struct camu_sink_entry *current; struct camu_sink_entry *queued; struct camu_sink_entry *target; - struct camu_sink_entry *reconnecting; + struct camu_sink_entry *detached; array(struct camu_sink_entry *) previous; array(struct camu_sink_entry *) entries; u16 lru; |