summaryrefslogtreecommitdiff
path: root/src/libsink
diff options
context:
space:
mode:
Diffstat (limited to 'src/libsink')
-rw-r--r--src/libsink/input_simulator.c3
-rw-r--r--src/libsink/sink.c100
-rw-r--r--src/libsink/sink.h1
3 files changed, 61 insertions, 43 deletions
diff --git a/src/libsink/input_simulator.c b/src/libsink/input_simulator.c
index 80fde02..871f0ce 100644
--- a/src/libsink/input_simulator.c
+++ b/src/libsink/input_simulator.c
@@ -21,8 +21,9 @@ enum {
static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
+ nn_thread_set_name("input_simulator");
while (!quit) {
- nn_thread_sleep(NNWT_TS_FROM_USEC(30000));
+ nn_thread_sleep(NNWT_TS_FROM_USEC(60000));
switch (al_rand() % MARK) {
case SKIP:
camu_sink_skip(sink, (al_rand() % 5));
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 29af2f5..94ee3c6 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -204,7 +204,7 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry)
// 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 and needs special consideration in CLIENT_REMOVE_BUFFERS.
+// entry->ended: Completely ignored and needs special handling 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)
{
@@ -279,19 +279,15 @@ 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)
+static inline s32 get_sequence_for_command(struct camu_sink_entry *entry)
{
// SEQUENCE_ANY resolves order on the server.
s32 sequence = LIANA_SEQUENCE_ANY;
// If we're local this could only lead to feeling like your inputs were eaten.
#ifndef CAMU_SINK_LOCAL
- if (sink->target && sink->target != (struct camu_sink_entry *)0xb00b) {
- sequence = sink->target->sequence;
- } else if (sink->current) {
- sequence = sink->current->sequence;
- }
+ if (entry) sequence = entry->sequence;
#else
- (void)sink;
+ (void)entry;
#endif
return sequence;
}
@@ -393,7 +389,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
nn_packet_write_str(packet, &sink->default_list);
nn_packet_write_u8(packet, CAMU_LIST_SKIP);
- nn_packet_write_s32(packet, get_sequence_for_command(sink));
+ nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque));
nn_packet_write_s32(packet, (s32)cmd->value.i);
nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
@@ -406,7 +402,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
nn_packet_write_str(packet, &sink->default_list);
nn_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE);
- nn_packet_write_s32(packet, get_sequence_for_command(sink));
+ nn_packet_write_s32(packet, get_sequence_for_command((struct camu_sink_entry *)cmd->opaque));
nn_packet_write_f64(packet, cmd->value.f);
nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
#endif
@@ -674,7 +670,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
if (detached) {
al_assert(detached == current);
sink->detached = NULL;
- log_warn("Unset detached as a substitute for remove.");
+ log_debug("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;
@@ -690,7 +686,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
remove_previous_if_contains(sink, target);
add_or_queue_entry(target);
} else {
- if (VIDEO_IS_SINGLE_FRAME(target)) {
+ if (!VIDEO_EMPTY(target) && VIDEO_IS_SINGLE_FRAME(target)) {
add_video_if_set_and_buffered(target);
}
stop_video = true;
@@ -698,7 +694,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
}
if (ensure_removed) {
- if (VIDEO_IS_SINGLE_FRAME(current)) {
+ if (!VIDEO_EMPTY(current) && VIDEO_IS_SINGLE_FRAME(current)) {
remove_entry_video_buffer(current);
}
al_assert(AUDIO_NOT_ADDED(current));
@@ -849,24 +845,25 @@ static void video_buffer_callback(void *userdata, u8 op)
break;
case CAMU_BUFFER_EOF:
log_debug("Video EOF.");
- bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
- bool swapped = false;
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);
}
- if (!single_frame) {
- if (VIDEO_STATE(entry) == BUFFER_ADDED) {
- remove_entry_video_buffer(entry);
- }
- VIDEO_STATE(entry) = BUFFER_ENDED;
- if (AUDIO_ENDED_OR_EMPTY(entry)) {
- swapped = end_entry_and_advance_queue(sink, entry);
- }
+ if (VIDEO_IS_SINGLE_FRAME(entry)) {
+ nn_mutex_unlock(&sink->lock);
+ return;
+ }
+ if (VIDEO_STATE(entry) == BUFFER_ADDED) {
+ remove_entry_video_buffer(entry);
+ }
+ VIDEO_STATE(entry) = BUFFER_ENDED;
+ bool swapped = false;
+ if (AUDIO_ENDED_OR_EMPTY(entry)) {
+ swapped = end_entry_and_advance_queue(sink, entry);
}
nn_mutex_unlock(&sink->lock);
- if (!swapped && !single_frame) {
+ if (!swapped) {
queue_cmd(sink, (struct camu_sink_cmd){
.op = STOP,
.value.i = CAMU_SINK_VIDEO
@@ -907,8 +904,8 @@ static void clock_callback(void *userdata, u8 op)
static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry)
{
-#ifdef CAMU_SINK_LOCAL
bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
+#ifdef CAMU_SINK_LOCAL
if (!AUDIO_EMPTY(entry) && !ignore_video) {
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
s32 frames = audio / entry->video.buf.avg_frame_duration;
@@ -923,13 +920,12 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s
// To sync clients with differing audio latencies our only option is to factor the mixer
// latency directly into the audio buffer.
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
- if (!VIDEO_EMPTY(entry)) {
+ if (!ignore_video) {
s32 frames = audio / entry->video.buf.avg_frame_duration;
struct camu_renderer *renderer = sink->video.renderer;
if (renderer) frames += renderer->get_latency(renderer);
camu_video_buffer_set_latency(&entry->video.buf, frames);
}
- bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
camu_audio_buffer_set_latency(&entry->audio.buf, audio);
camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
#endif
@@ -1051,6 +1047,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
nn_mutex_lock(&sink->lock);
log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current));
+ // Immediately switch to a potential target to avoid excessive delay/catchup
+ // that would be caused by this entry being re-added with an in-between clock state.
+ if (sink->target) {
+ switch_to(sink, sink->target);
+ sink->target = NULL;
+ }
+
// 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.
@@ -1077,8 +1080,8 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
remove_entry_audio_buffer(entry);
}
// 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)) {
+ bool skip_video = rec->reconnect && VIDEO_IS_SINGLE_FRAME(entry);
+ if (!skip_video && !VIDEO_ENDED_OR_EMPTY(entry)) {
remove_entry_video_buffer(entry);
remove_entry_video_buffer(entry);
}
@@ -1090,7 +1093,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
// Unlock to wait.
nn_mutex_unlock(&sink->lock);
- while (entry_audio_buffer_held(entry) || (!ignore_video && entry_video_buffer_held(entry))) {
+ while (entry_audio_buffer_held(entry) || (!skip_video && entry_video_buffer_held(entry))) {
BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000));
}
@@ -1107,7 +1110,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
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)) {
+ } else if (entry->ended && VIDEO_EMPTY(entry)) {
VIDEO_STATE(entry) = BUFFER_INIT;
}
entry->ended = false;
@@ -1118,23 +1121,26 @@ 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;
- log_trace("resume_at("ENTRY_FMT"), paused_at: %f.", ENTRY_ARG(entry), entry->clock.paused_at);
+ log_trace("resume_at("ENTRY_FMT"), seek_pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->seek_pos / 1000000.0, entry->clock.paused_at);
// These buffers won't be re-added until after a CLIENT_RECONNECTED event.
+ bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
if (!AUDIO_EMPTY(entry)) {
camu_audio_buffer_reset(&entry->audio.buf);
+ // no_video is set in video BUFFER_EOF as a fail-safe. Reset it here.
+ camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
}
- if (!VIDEO_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
+ if (!ignore_video) {
camu_video_buffer_reset(&entry->video.buf, time->seek_pos);
}
-#ifdef CAMU_SINK_LOCAL
- time->at = 0;
-#endif
nn_mutex_lock(&sink->lock);
#ifdef LIANA_LIST_SCUFFED_LOOP
if (time->seek_pos == 0) {
camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock));
} else {
#endif
+#ifdef CAMU_SINK_LOCAL
+ time->at = 0;
+#endif
camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
#ifdef LIANA_LIST_SCUFFED_LOOP
}
@@ -1198,7 +1204,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); }
}
- // @TODO: Mark for removal here instead.
bool removed = al_array_remove(sink->entries, entry);
remove_from_queue_by_opaque(sink, entry);
@@ -1525,7 +1530,6 @@ static struct nn_rpc_command commands[] = {
{ .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL }
};
-// @TODO: Send sink id from list.
static void identify_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
@@ -1607,18 +1611,30 @@ void camu_sink_return_current(struct camu_sink *sink)
nn_mutex_unlock(&sink->lock);
}
+static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink)
+{
+ if (sink->target && sink->target != (struct camu_sink_entry *)0xb00b) {
+ return sink->target;
+ }
+ return sink->current;
+}
+
void camu_sink_skip(struct camu_sink *sink, s32 n)
{
+ nn_mutex_lock(&sink->lock);
+ struct camu_sink_entry *current = get_entry_for_command(sink);
+ nn_mutex_unlock(&sink->lock);
queue_cmd(sink, (struct camu_sink_cmd){
.op = SKIP,
- .value.i = n
+ .value.i = n,
+ .opaque = current
});
}
void camu_sink_toggle_pause(struct camu_sink *sink)
{
nn_mutex_lock(&sink->lock);
- struct camu_sink_entry *current = sink->current;
+ struct camu_sink_entry *current = get_entry_for_command(sink);
nn_mutex_unlock(&sink->lock);
if (!current) return;
queue_cmd(sink, (struct camu_sink_cmd){
@@ -1631,7 +1647,7 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
{
nn_mutex_lock(&sink->lock);
- struct camu_sink_entry *current = sink->current;
+ struct camu_sink_entry *current = get_entry_for_command(sink);
u64 duration = 0;
f64 pts = 0.0;
if (current) {
@@ -1713,7 +1729,7 @@ void camu_sink_close(struct camu_sink *sink)
void camu_sink_free(struct camu_sink *sink)
{
- al_assert(sink->entries.count == 0);
+ al_assert(!sink->entries.count);
al_array_free(sink->entries);
nn_rpc_free(&sink->client);
camu_queue_free(sink->queue);
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index c2ea6a1..40bd0a4 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -13,6 +13,7 @@
#ifndef CAMU_SINK_NO_VIDEO
#include "../buffer/video.h"
#else
+// Stub video_buffer usage within libsink only.
#include "../buffer/video_null.h"
#endif
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:35:58 +0000