summaryrefslogtreecommitdiff
path: root/src/libsink
diff options
context:
space:
mode:
Diffstat (limited to 'src/libsink')
-rw-r--r--src/libsink/desktop.c3
-rw-r--r--src/libsink/sink.c127
-rw-r--r--src/libsink/sink.h3
3 files changed, 72 insertions, 61 deletions
diff --git a/src/libsink/desktop.c b/src/libsink/desktop.c
index 780818c..8582cd4 100644
--- a/src/libsink/desktop.c
+++ b/src/libsink/desktop.c
@@ -2,6 +2,9 @@
#include <al/log.h>
#ifdef CAMU_NO_MINIAUDIO_BACKENDS
+// @TODO: audio_null is incomplete in that it never actually reads from the buffers.
+// This is an issue in CAMU_DIRECT_MODE because eventually the server-side packet_pool
+// will be starved by pending audio packets.
#define DESKTOP_NULL_AUDIO
#endif
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 7b173a9..e9b4df4 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -71,7 +71,8 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX);
#define CONNECTION_NUMBER(id) ((id >> 48) & 0xffff)
// Only sink->target can ever be 0xb00b.
-#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b)
+#define OxbOOb ((struct camu_sink_entry *)0xb00b)
+#define ENTRY_IS_VALID(entry) ((entry) && (entry) != OxbOOb)
// printf format for entries.
#ifdef AL_DEBUG
@@ -601,7 +602,7 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
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.
- bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
+ bool dangling_target = target == OxbOOb;
if (dangling_target || target->ended || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) {
remove_entry_buffers(previous);
return;
@@ -713,8 +714,10 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
// @TODO: Cleanup stop_video? and log_trace lengths.
bool stop_video = false;
- bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
- if (!dangling_target) {
+ bool dangling_target = target == OxbOOb;
+ if (dangling_target) {
+ log_trace("Ignored dangling target.");
+ } else {
if (!target->ended) {
remove_previous_if_contains(sink, target);
add_or_queue_entry(target);
@@ -725,8 +728,6 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
}
stop_video = true;
}
- } else {
- log_trace("Ignored dangling target.");
}
if (ensure_removed) {
@@ -737,15 +738,15 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
al_assert(VIDEO_STATE(current) != BUFFER_ADDED);
}
- if (!dangling_target) {
+ if (dangling_target) {
+ stop_video = true;
+ sink->current = NULL;
+ } else {
target->audio.ignore_paused = false;
if (!target->paused) {
camu_audio_buffer_resync(&target->audio.buf);
}
sink->current = target;
- } else {
- stop_video = true;
- sink->current = NULL;
}
if (stop_video) {
@@ -775,14 +776,17 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at)
{
struct camu_sink_entry *current = sink->current;
- log_trace("pause_and_swap_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(current));
+ log_trace("pause_and_swap_to("ENTRY_FMT", %.2f), current: "ENTRY_FMT".",
+ ENTRY_ARG(target), at / 1000000.0, ENTRY_ARG(current));
al_assert(target != current);
al_assert(!sink->target);
// This is extra verbose because the order is important.
// 1. sink->target has to be set before calling clock_pause().
// 2. current must still be paused even if it's ended.
// 3. In the immediate case, switch_to() has to come last.
- if (current && !current->ended) sink->target = target;
+ if (current && !current->ended) {
+ sink->target = target;
+ }
if (current) {
current->audio.ignore_paused = true;
camu_clock_pause(&current->clock, at);
@@ -803,7 +807,7 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
entry->ended = true;
queue_cmd(sink, (struct camu_sink_cmd){
.op = END,
- .value.u = entry->reset_id,
+ .value.u = entry->reset_token,
.opaque = entry
});
#ifdef LIANA_LIST_SCUFFED_LOOP
@@ -811,14 +815,9 @@ 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;
- } else if (sink->queued) {
- switch_to(sink, sink->queued);
- sink->queued = NULL;
- return true;
}
return false;
}
@@ -950,20 +949,23 @@ static void clock_callback(void *userdata, u8 op)
static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry)
{
- // @TODO: Video can't be ENDED here right?
bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
f64 avg_frame_duration = entry->video.buf.avg_frame_duration;
#ifdef CAMU_SINK_LOCAL
if (!AUDIO_EMPTY(entry) && !ignore_video) {
+ // Delay either the audio or video so we can start the clock as
+ // soon as possible while keeping A/V sync.
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
struct camu_renderer *renderer = sink->video.renderer;
f64 video = renderer->get_latency(renderer) * avg_frame_duration;
- // Start either the audio or video early so we can start the clock as
- // soon as possible while keeping A/V sync.
+ // If the audio buffer is delayed by less than the audio latency, data will be skipped.
+ f64 base = -audio;
if (audio > video) {
- camu_video_buffer_set_latency(&entry->video.buf, video - audio);
+ camu_audio_buffer_set_latency(&entry->audio.buf, base);
+ camu_video_buffer_set_latency(&entry->video.buf, base - (audio - video));
} else if (video > audio) {
- camu_audio_buffer_set_latency(&entry->audio.buf, audio - video);
+ camu_audio_buffer_set_latency(&entry->audio.buf, base);
+ camu_video_buffer_set_latency(&entry->video.buf, base + (video - audio));
}
}
// If we're local we don't have to worry about syncing audio-only entries.
@@ -971,7 +973,7 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s
camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
#else
// When trying to sync clients with different audio/video latencies (common case),
- // our only option is to factor the latency directly into the buffers.
+ // our only option is to shift each buffer forward directly by their latency.
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
if (!ignore_video) {
struct camu_renderer *renderer = sink->video.renderer;
@@ -1064,6 +1066,8 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
case LIANA_CLIENT_CONFIGURE_COMPLETE: {
// All present buffers were configured.
nn_mutex_lock(&sink->lock);
+ // This is mainly to assert DETACHED handling.
+ al_assert(!VIDEO_ENDED(entry) && !AUDIO_ENDED(entry));
evaluate_and_set_buffer_params(sink, entry);
nn_mutex_unlock(&sink->lock);
break;
@@ -1215,7 +1219,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
// List has no concept of a local sink, ignore it's request.
time->at = 0;
#endif
- log_trace("resume_at("ENTRY_FMT"), seek_pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->seek_pos / 1000000.0, entry->clock.paused_at);
+ log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->pos / 1000000.0, entry->clock.paused_at);
nn_mutex_lock(&sink->lock);
bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
if (!AUDIO_EMPTY(entry)) {
@@ -1224,14 +1228,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
}
if (!ignore_video) {
- camu_video_buffer_reset(&entry->video.buf, time->seek_pos);
+ camu_video_buffer_reset(&entry->video.buf, time->pos);
}
#ifdef LIANA_LIST_SCUFFED_LOOP
- if (time->seek_pos == 0) {
+ if (time->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);
+ camu_clock_seek(&entry->clock, time->pos / 1000000.0, time->at);
#ifdef LIANA_LIST_SCUFFED_LOOP
}
#endif
@@ -1305,7 +1309,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
remove_from_queue_by_opaque(sink, entry);
if (entry == sink->target) {
- sink->target = (struct camu_sink_entry *)0xb00b;
+ sink->target = OxbOOb;
log_warn("Attempting to handle a disconnected target.");
} else if (entry == sink->current) {
if (sink->suspended) {
@@ -1326,8 +1330,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
// If current was never fully added we need to call this here.
maybe_remove_previous(sink);
- } else if (entry == sink->queued) {
- sink->queued = NULL;
}
nn_mutex_unlock(&sink->lock);
@@ -1415,11 +1417,9 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
s32 sequence = nn_packet_read_s32(packet);
u64 at = nn_packet_read_u64(packet);
- u64 seek_pos = nn_packet_read_u64(packet);
+ u64 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);
+ u32 reset_token = nn_packet_read_u32(packet);
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
bool create = !entry;
@@ -1427,24 +1427,26 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
entry->sequence = sequence;
entry->lru = sink->lru;
sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
- entry->reset_id = reset_id;
+ entry->reset_token = reset_token;
if (create) {
entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
- camu_clock_set(&entry->clock, seek_pos / 1000000.0);
- lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, seek_pos);
+ camu_clock_set(&entry->clock, pos / 1000000.0);
+ lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos);
}
// Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect().
nn_mutex_lock(&sink->lock);
+ struct camu_sink_entry *current = sink->current;
+
if (op == LIANA_SINK_BUFFER) {
+ log_trace("buffered("ENTRY_FMT"), created: %s.", ENTRY_ARG(entry), BOOLSTR(create));
goto out;
} else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
- sink->queued = entry;
+ log_trace("queued("ENTRY_FMT"), %s, created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create));
goto out;
}
- struct camu_sink_entry *current = sink->current;
#ifdef CAMU_SINK_LOCAL
(void)at;
al_assert(entry != current);
@@ -1466,7 +1468,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
#else
// For target to be set that must mean current is set, armed to pause, and not ended.
struct camu_sink_entry *prev_target = sink->target;
- log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
+ log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry),
+ lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
switch (pause) {
case LIANA_PAUSE_NONE:
if (prev_target) {
@@ -1481,13 +1484,15 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
case LIANA_PAUSE_RESUME:
if (prev_target) {
sink->target = NULL;
+ if (entry == current) {
+ // Switched back to current before pause_and_swap_to() completed.
+ camu_audio_buffer_resync(&entry->audio.buf);
+ }
} else {
al_assert(entry != current);
}
if (entry != current) {
switch_to(sink, entry);
- } else {
- camu_audio_buffer_resync(&entry->audio.buf);
}
camu_clock_resume(&entry->clock, at);
break;
@@ -1496,16 +1501,17 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
al_assert(current);
al_assert(!current->ended);
al_assert(prev_target != entry);
- if (current != entry) {
- sink->target = entry;
- } else {
+ if (entry == current) {
+ // pause_and_swap_to() negated.
sink->target = NULL;
+ } else {
+ sink->target = entry;
}
- if (prev_target != (struct camu_sink_entry *)0xb00b) {
+ if (prev_target != OxbOOb) {
camu_clock_pause(&prev_target->clock, at);
}
} else {
- al_assert(current != entry);
+ al_assert(entry != current);
pause_and_swap_to(sink, entry, at);
}
break;
@@ -1514,17 +1520,17 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
al_assert(current);
al_assert(!current->ended);
al_assert(prev_target != entry);
- if (current != entry) {
- sink->target = entry;
- } else {
+ if (entry == current) {
sink->target = NULL;
camu_audio_buffer_resync(&entry->audio.buf);
+ } else {
+ sink->target = entry;
}
- if (prev_target != (struct camu_sink_entry *)0xb00b) {
+ if (prev_target != OxbOOb) {
camu_clock_pause(&prev_target->clock, at);
}
} else {
- al_assert(current != entry);
+ al_assert(entry != current);
pause_and_swap_to(sink, entry, at);
}
camu_clock_resume(&entry->clock, at);
@@ -1564,14 +1570,15 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
if (!entry->held) local_pause(sink, entry);
#else
switch (pause) {
- case LIANA_PAUSE_PAUSE:
+ case LIANA_PAUSE_PAUSE: {
entry->paused = true;
camu_clock_pause(&entry->clock, at);
log_info("Clock paused.");
// Audio will be stopped in a BUFFER_PAUSED callback.
// Video will be stopped in a CLOCK_PAUSED callback.
break;
- case LIANA_PAUSE_RESUME:
+ }
+ case LIANA_PAUSE_RESUME: {
entry->paused = false;
camu_clock_resume(&entry->clock, at);
log_info("Clock resumed.");
@@ -1590,6 +1597,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
}
break;
}
+ }
#endif
nn_mutex_unlock(&sink->lock);
@@ -1610,12 +1618,14 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn
(void)sequence;
u64 at = nn_packet_read_u64(packet);
u64 pos = nn_packet_read_u64(packet);
- u32 reset_id = nn_packet_read_u32(packet);
+ u32 reset_token = nn_packet_read_u32(packet);
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
if (!entry) goto out;
- log_trace("seek("ENTRY_FMT"), reset_id: %u.", ENTRY_ARG(entry), reset_id);
- entry->reset_id = reset_id;
+ nn_mutex_lock(&sink->lock);
+ log_trace("seek("ENTRY_FMT"), reset_token: %u.", ENTRY_ARG(entry), reset_token);
+ entry->reset_token = reset_token;
+ nn_mutex_unlock(&sink->lock);
// The rest of the seek is handled in CLIENT_REMOVE_BUFFERS/RESUME_AT/RECONNECTED.
lia_client_seek(&entry->client, pos, at);
@@ -1705,7 +1715,6 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
nn_signal_start(&sink->queue_signal);
camu_queue_init(sink->queue);
sink->current = NULL;
- sink->queued = NULL;
sink->target = NULL;
sink->suspended = NULL;
al_array_init(sink->previous);
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 2c96313..5958829 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -50,7 +50,7 @@ struct camu_sink_entry {
u16 lru;
bool disconnected;
bool ended;
- u32 reset_id;
+ u32 reset_token;
struct camu_clock clock;
bool held; // Only used in SINK_LOCAL.
bool paused;
@@ -89,7 +89,6 @@ struct camu_sink {
struct nn_signal queue_signal;
queue(struct camu_sink_cmd) queue;
struct camu_sink_entry *current;
- struct camu_sink_entry *queued;
struct camu_sink_entry *target;
// If an entry that was current was removed for a reconnect, it's "suspended".
struct camu_sink_entry *suspended;
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:47:17 +0000