summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/buffer/common.h3
-rw-r--r--src/buffer/common_internal.h4
-rw-r--r--src/buffer/video.c5
-rw-r--r--src/codec/ffmpeg/decoder.c4
-rw-r--r--src/codec/ffmpeg/demuxer.c4
-rw-r--r--src/codec/ffmpeg/encoder.c1
-rw-r--r--src/fruits/cmc/ui/ui.c4
-rw-r--r--src/fruits/cmsrv/cmsrv.c4
-rw-r--r--src/liana/list.c97
-rw-r--r--src/liana/list.h2
-rw-r--r--src/liana/server.c4
-rw-r--r--src/libsink/sink.c38
-rw-r--r--src/mixer/mixer.c4
-rw-r--r--src/portal/src/search.c8
-rw-r--r--src/server/server.c1
-rw-r--r--src/util/blocking_ring_buffer.c193
-rw-r--r--src/util/blocking_ring_buffer.h42
-rw-r--r--src/util/color_palette.c11
-rw-r--r--src/util/color_palette.h2
-rw-r--r--src/util/meson.build2
-rw-r--r--src/util/queue.h80
21 files changed, 155 insertions, 358 deletions
diff --git a/src/buffer/common.h b/src/buffer/common.h
index 64168ff..24964ba 100644
--- a/src/buffer/common.h
+++ b/src/buffer/common.h
@@ -5,5 +5,6 @@ enum {
CAMU_BUFFER_CORK,
CAMU_BUFFER_UNCORK,
CAMU_BUFFER_PAUSED,
- CAMU_BUFFER_EOF
+ CAMU_BUFFER_EOF,
+ CAMU_BUFFER_ERRORED
};
diff --git a/src/buffer/common_internal.h b/src/buffer/common_internal.h
index 0dc1e5f..e4cdd33 100644
--- a/src/buffer/common_internal.h
+++ b/src/buffer/common_internal.h
@@ -6,5 +6,7 @@ enum {
// Got request to flush.
FLUSHED,
// Realized flush.
- SIGNALED
+ SIGNALED,
+ // Errored.
+ ERRORED
};
diff --git a/src/buffer/video.c b/src/buffer/video.c
index 2879082..94c452d 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -237,6 +237,11 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out)
}
u8 ret = buf->queue->read(buf->queue, base, out);
u8 flow = al_atomic_load(u8)(&buf->flow, AL_ATOMIC_ACQUIRE);
+ if (ret == CAMU_QUEUE_ERR) {
+ buf->callback(buf->userdata, CAMU_BUFFER_ERRORED);
+ al_atomic_store(u8)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE);
+ return false;
+ }
if (flow == FLUSHED && (ret == CAMU_QUEUE_EOF || (buf->single_frame && ret == CAMU_QUEUE_OK))) {
buf->callback(buf->userdata, CAMU_BUFFER_EOF);
al_atomic_store(u8)(&buf->flow, SIGNALED, AL_ATOMIC_RELEASE);
diff --git a/src/codec/ffmpeg/decoder.c b/src/codec/ffmpeg/decoder.c
index 47fcc0c..3fdb534 100644
--- a/src/codec/ffmpeg/decoder.c
+++ b/src/codec/ffmpeg/decoder.c
@@ -113,7 +113,7 @@ static void ff_decoder_flush(struct camu_decoder *dec)
static s32 receive_frames(struct camu_ff_decoder *av)
{
s32 ret;
- do {
+ for (;;) {
struct camu_codec_frame *frame = al_alloc_object(struct camu_codec_frame);
frame->mode = CAMU_FFMPEG_COMPAT;
frame->av.frame = av_frame_alloc();
@@ -136,7 +136,7 @@ static s32 receive_frames(struct camu_ff_decoder *av)
// / (av->codec_context->sample_rate * av->time_base.num / (f64)av->time_base.den);
//}
av->callback(av->userdata, frame);
- } while (1);
+ }
return ret;
}
diff --git a/src/codec/ffmpeg/demuxer.c b/src/codec/ffmpeg/demuxer.c
index f756168..0184d4d 100644
--- a/src/codec/ffmpeg/demuxer.c
+++ b/src/codec/ffmpeg/demuxer.c
@@ -128,7 +128,7 @@ static s32 ff_demuxer_get_packet(struct camu_demuxer *demux, struct camu_codec_p
struct camu_ff_demuxer *av = (struct camu_ff_demuxer *)demux;
packet->mode = CAMU_FFMPEG_COMPAT;
s32 ret = 0;
- do {
+ for (;;) {
ret = av_read_frame(av->format_context, packet->av.pkt);
if (ret == AVERROR_EOF) {
av->eof = true;
@@ -143,7 +143,7 @@ static s32 ff_demuxer_get_packet(struct camu_demuxer *demux, struct camu_codec_p
continue;
}
break;
- } while (1);
+ }
return ret;
}
diff --git a/src/codec/ffmpeg/encoder.c b/src/codec/ffmpeg/encoder.c
index 7c7b010..0577ae9 100644
--- a/src/codec/ffmpeg/encoder.c
+++ b/src/codec/ffmpeg/encoder.c
@@ -108,6 +108,7 @@ void camu_ff_encoder_push(struct camu_ff_encoder *enc, u8 **data, s32 sample_cou
return;
}
enc->callback(enc->userdata, &pkt);
+ av_packet_unref(&pkt);
enc->frame->pts += SAMPLES_PER_FRAME;
}
}
diff --git a/src/fruits/cmc/ui/ui.c b/src/fruits/cmc/ui/ui.c
index b4c196d..72e4df2 100644
--- a/src/fruits/cmc/ui/ui.c
+++ b/src/fruits/cmc/ui/ui.c
@@ -21,7 +21,7 @@ static void input_poll_callback(void *userdata, s32 revents)
struct cmc_ui *ui = (struct cmc_ui *)userdata;
(void)revents;
struct ncinput input;
- do {
+ for (;;) {
if (!read_input(ui, &input)) break;
bool do_render = false;
if (input.evtype == NCTYPE_PRESS || input.evtype == NCTYPE_UNKNOWN) {
@@ -41,7 +41,7 @@ static void input_poll_callback(void *userdata, s32 revents)
if (do_render) {
notcurses_render(ui->nc);
}
- } while (1);
+ }
}
bool cmc_ui_init(struct cmc_ui *ui, struct nn_event_loop *loop, struct cmc *c)
diff --git a/src/fruits/cmsrv/cmsrv.c b/src/fruits/cmsrv/cmsrv.c
index 53555d3..00f5373 100644
--- a/src/fruits/cmsrv/cmsrv.c
+++ b/src/fruits/cmsrv/cmsrv.c
@@ -86,7 +86,7 @@ static void input_poll_callback(void *userdata, s32 revents)
struct cmsrv *s = (struct cmsrv *)userdata;
(void)revents;
struct ncinput input;
- do {
+ for (;;) {
if (!cmsrv_ui_read_input(&s->ui, &input)) break;
if (input.evtype == NCTYPE_PRESS || input.evtype == NCTYPE_UNKNOWN) {
switch (input.id) {
@@ -104,7 +104,7 @@ static void input_poll_callback(void *userdata, s32 revents)
break;
}
}
- } while (1);
+ }
}
static void quit_signal_callback(void *userdata)
diff --git a/src/liana/list.c b/src/liana/list.c
index 7c4979b..2280eec 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -49,7 +49,6 @@ void lia_list_init(struct lia_list *list, str *name)
{
al_str_clone(&list->name, name);
list->current = -1;
- list->previous = -1;
list->queued = -1;
list->idle = true;
list->increment = 0;
@@ -103,6 +102,50 @@ static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_e
return false;
}
+static void set_queued(struct lia_list *list)
+{
+ if (list->queued < 0)
+ return;
+
+ struct lia_list_entry *queued = al_array_at(list->entries, list->queued);
+
+ bool error;
+ struct lia_list_entry *current = al_array_at(list->entries, list->current);
+ if (!entry_load_and_get_duration(list, current, list->current, &error))
+ return;
+
+ queued->start = current->start + (current->duration - current->offset);
+ u8 pause = queued->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_RESUME : LIANA_PAUSE_NONE;
+ struct lia_timing time = {
+ .at = queued->start,
+ .seek_pos = queued->offset,
+ .pause = pause
+ };
+
+ struct lia_list_sink *sink;
+ al_array_foreach(list->sinks, i, sink) {
+ if (sink->queued != list->queued) {
+ sink->queued = list->queued;
+ sink->callback(sink->userdata, LIANA_SINK_BUFFER_AND_QUEUE, queued, list->queued, &time);
+ }
+ }
+}
+
+static void evaluate_queued(struct lia_list *list)
+{
+ s32 size = (s32)list->entries.size;
+ s32 next = list->current + 1;
+ if (next >= size || next == list->queued)
+ return;
+
+ bool error;
+ struct lia_list_entry *queued = al_array_at(list->entries, next);
+ if (!entry_load_and_get_duration(list, queued, next, &error))
+ return;
+
+ list->queued = next;
+}
+
static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
{
struct lia_list_entry *current;
@@ -184,26 +227,7 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
u8 meta = LIANA_META_PLAYING;
list->callback(list->userdata, LIANA_LIST_META, entry, &meta);
} else {
- /*
- if (list->queued == -1) {
- // TODODODO: this is based on addeding entry to list->entries BEFORE this point.
- struct lia_list_entry *current = al_array_at(list->entries, list->current);
- list->queued = list->current + 1;
- entry->start = current->start + (current->duration - current->offset);
- struct lia_timing time = {
- .at = entry->start,
- .seek_pos = entry->offset,
- .pause = LIANA_PAUSE_RESUME
- };
- struct lia_list_sink *sink;
- al_array_foreach(list->sinks, i, sink) {
- sink->queued = list->queued;
- sink->callback(sink->userdata, LIANA_SINK_BUFFER_AND_QUEUE, entry, list->queued, &time);
- }
- } else {
- */
- entry->start = LIANA_TIMESTAMP_INVALID;
- //}
+ entry->start = LIANA_TIMESTAMP_INVALID;
u8 meta = LIANA_META_QUEUED;
list->callback(list->userdata, LIANA_LIST_META, entry, &meta);
}
@@ -218,8 +242,8 @@ static void adjust_current(struct lia_list *list, struct lia_list_entry *previou
struct lia_list_entry *entry;
al_array_foreach(list->entries, i, entry) {
if (entry->opaque == previous->opaque) {
- list->previous = -1;
- if (i == (u32)list->current) return;
+ if (i == (u32)list->current)
+ return;
cmd->op = SKIPTO;
cmd->sequence = i;
cmd->value.i = list->current;
@@ -234,7 +258,6 @@ static void adjust_current(struct lia_list *list, struct lia_list_entry *previou
static void unset_all(struct lia_list *list)
{
list->current = -1;
- list->previous = -1;
list->queued = -1;
struct lia_list_sink *sink;
al_array_foreach(list->sinks, i, sink) {
@@ -356,18 +379,16 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
current->held = true;
}
- al_log_debug("list", "skipto [%u-%u]: pause: %hhu, held: %s, current_ended: %s, target_ended: %s.",
+ al_log_debug("list", "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));
list->current = index;
- list->previous = sequence;
list->idle = false;
struct lia_timing time = {
.at = at,
.seek_pos = target->offset,
.pause = pause,
- .previous_ended = current_ended,
.ended = target_ended
};
@@ -429,7 +450,7 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
break;
}
- al_log_debug("list", "toggle_pause [%u]: pts: %f, pause: %hhu.", entry->id, pts, pause);
+ al_log_debug("list", "toggle_pause [#%u]: pts: %f, pause: %hhu.", entry->id, pts, pause);
struct lia_timing time = {
.at = at,
@@ -472,7 +493,7 @@ static void handle_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent
list->idle = false;
- al_log_debug("list", "seek [%u]: pos: %f.", entry->id, pos / 1000000.0);
+ al_log_debug("list", "seek [#%u]: pos: %f.", entry->id, pos / 1000000.0);
struct lia_timing time = {
.at = at,
@@ -501,6 +522,7 @@ static bool handle_end(struct lia_list *list, s32 id)
al_log_warn("list", "Got end() from an already ended resource, ignoring.");
return true;
}
+ al_log_debug("list", "end [#%u].", entry->id);
entry->ended = true;
entry->offset = entry->duration;
@@ -509,11 +531,11 @@ static bool handle_end(struct lia_list *list, s32 id)
s32 next = sequence + 1;
if (list->queued >= 0) {
list->current = list->queued;
- list->previous = sequence;
list->queued = -1;
struct lia_list_sink *sink;
al_array_foreach(list->sinks, i, sink) {
sink->queued = -1;
+ sink->set = list->current;
}
struct lia_list_entry *current = al_array_at(list->entries, list->current);
u8 meta = LIANA_META_PLAYING;
@@ -527,12 +549,6 @@ static bool handle_end(struct lia_list *list, s32 id)
return false;
} else {
list->idle = true;
- /*
- struct lia_list_sink *sink;
- al_array_foreach(list->sinks, i, sink) {
- sink->set = -1;
- }
- */
}
}
@@ -585,7 +601,7 @@ static void handle_clear(struct lia_list *list)
list->entries.size = 0;
}
-void pump_queue(struct lia_list *list)
+static void run_queue(struct lia_list *list)
{
if (!list->cmd) {
if (list->queue.size == 0) return;
@@ -651,6 +667,13 @@ void pump_queue(struct lia_list *list)
pump_queue(list);
}
+void pump_queue(struct lia_list *list)
+{
+ run_queue(list);
+ //evaluate_queued(list);
+ //set_queued(list);
+}
+
void lia_list_pump(struct lia_list *list)
{
pump_queue(list);
diff --git a/src/liana/list.h b/src/liana/list.h
index 8d47b72..4e5e343 100644
--- a/src/liana/list.h
+++ b/src/liana/list.h
@@ -60,7 +60,6 @@ struct lia_timing {
u64 at;
u64 seek_pos;
u8 pause;
- bool previous_ended;
bool ended;
};
@@ -97,7 +96,6 @@ struct lia_list_cmd {
struct lia_list {
str name;
s32 current;
- s32 previous;
s32 queued;
bool idle;
u32 increment;
diff --git a/src/liana/server.c b/src/liana/server.c
index 27c253b..f4a6efd 100644
--- a/src/liana/server.c
+++ b/src/liana/server.c
@@ -37,14 +37,14 @@ static u8 packet_pool_callback(void *userdata, struct nn_packet *packet)
static nn_thread_result NNWT_THREADCALL handler_thread(void *userdata)
{
struct lia_node_connection *conn = (struct lia_node_connection *)userdata;
- do {
+ for (;;) {
struct nn_packet *packet = nn_packet_pool_get(&conn->pool);
if (!packet) break;
conn->handler->step(conn->handler);
conn->handler->write_packet(conn->handler, packet);
nn_packet_pool_submit(&conn->pool, packet);
if (conn->handler->status != CAMU_OK) break;
- } while (1);
+ }
nn_packet_pool_flush(&conn->pool);
return 0;
}
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 984a41c..9b5c05e 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -348,11 +348,11 @@ static void queue_signal_callback(void *userdata)
struct camu_sink *sink = (struct camu_sink *)userdata;
u32 size;
struct camu_sink_cmd cmd;
- do {
+ for (;;) {
camu_queue_try_pop(sink->queue, size, cmd);
if (size == 0) break;
handle_sink_cmd(sink, &cmd);
- } while (1);
+ }
}
static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
@@ -436,11 +436,10 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
// If neither the entries audio or video buffer is ADDED, we don't care about adding it
// to previous (waiting for the next added entry to remove it).
#ifndef CAMU_SINK_NO_VIDEO
- if (!(previous->audio.state == BUFFER_ADDED || previous->video.state == BUFFER_ADDED))
+ if (!(previous->audio.state == BUFFER_ADDED || previous->video.state == BUFFER_ADDED) || target->ended) {
#else
- if (!(previous->audio.state == BUFFER_ADDED))
+ if (!(previous->audio.state == BUFFER_ADDED) || target->ended)
#endif
- {
remove_entry_buffers(sink, previous);
return;
}
@@ -586,11 +585,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
al_assert(current != target);
if (!current->ended) {
- if (target->ended) {
- remove_entry_buffers(sink, current);
- } else {
- maybe_add_to_previous(sink, current, target);
- }
+ maybe_add_to_previous(sink, current, target);
} else {
#ifndef CAMU_SINK_NO_VIDEO
if (ENTRY_IS_SINGLE_FRAME(current)) {
@@ -645,9 +640,11 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
switch_to(sink, sink->target);
sink->target = NULL;
return true;
- }/* else if (sink->queued) {
+ } else if (sink->queued) {
+ switch_to(sink, sink->queued);
+ sink->queued = NULL;
return true;
- }*/
+ }
return false;
}
@@ -698,6 +695,8 @@ static void audio_buffer_callback(void *userdata, u8 op)
nn_mutex_unlock(&sink->mutex);
break;
}
+ case CAMU_BUFFER_ERRORED:
+ break;
}
}
@@ -743,6 +742,8 @@ static void video_buffer_callback(void *userdata, u8 op)
}
break;
}
+ case CAMU_BUFFER_ERRORED:
+ break;
}
}
#endif
@@ -1031,9 +1032,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
// If current was never fully added we need to do this here.
maybe_remove_previous(sink);
- }/* else if (entry == sink->queued) {
+ } else if (entry == sink->queued) {
sink->queued = NULL;
- }*/
+ }
lia_client_free(&entry->client);
camu_audio_buffer_free(&entry->audio.buf);
@@ -1118,9 +1119,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
u64 at = nn_packet_read_u64(packet);
u64 seek_pos = nn_packet_read_u64(packet);
u8 pause = nn_packet_read_u8(packet);
- bool previous_ended = nn_packet_read_bool(packet);
bool ended = nn_packet_read_bool(packet);
- (void)previous_ended;
bool created = false;
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
@@ -1149,10 +1148,13 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
if (op == LIANA_SINK_BUFFER) {
goto out;
- }/* else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
+ } else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
+ if (pause == LIANA_PAUSE_RESUME) {
+ camu_clock_resume(&entry->clock, at);
+ }
sink->queued = entry;
goto out;
- }*/
+ }
#ifdef CAMU_SINK_LOCAL
(void)at;
diff --git a/src/mixer/mixer.c b/src/mixer/mixer.c
index 18d6b8a..5a21b19 100644
--- a/src/mixer/mixer.c
+++ b/src/mixer/mixer.c
@@ -38,7 +38,7 @@ static s32 data_callback(void *userdata, u8 *data, s32 frame_count, bool *silenc
camu_audio_buffer_read(buf, data, req);
}
}
- do {
+ for (;;) {
if (al_array_is_empty(mixer->buffers)) {
al_memset(data, 0, req);
} else {
@@ -57,7 +57,7 @@ static s32 data_callback(void *userdata, u8 *data, s32 frame_count, bool *silenc
}
}
break;
- } while (1);
+ }
}
return frame_count;
}
diff --git a/src/portal/src/search.c b/src/portal/src/search.c
index 88cf1f1..447e44c 100644
--- a/src/portal/src/search.c
+++ b/src/portal/src/search.c
@@ -63,7 +63,7 @@ static nn_thread_result NNWT_THREADCALL queue_thread(void *userdata)
nn_thread_setcanceltype(NNWT_THREAD_CANCEL_ASYNCHRONOUS);
bool have_python = false;
nn_mutex_lock(&bridge->mutex);
- do {
+ for (;;) {
nn_cond_wait(&bridge->cond, &bridge->mutex);
if (bridge->quit) break;
if (!have_python) {
@@ -127,7 +127,7 @@ static nn_thread_result NNWT_THREADCALL queue_thread(void *userdata)
}
bridge->queue.size = 0;
nn_signal_send(&bridge->results_signal);
- } while (1);
+ }
nn_mutex_unlock(&bridge->mutex);
if (have_python) camu_python_close();
return 0;
@@ -138,11 +138,11 @@ static void results_signal_callback(void *userdata)
struct camu_portal_bridge *bridge = (struct camu_portal_bridge *)userdata;
u32 size;
struct camu_portal_result result;
- do {
+ for (;;) {
camu_queue_try_pop(bridge->results, size, result);
if (size == 0) break;
result.callback(bridge->userdata, result.userdata, &result);
- } while (1);
+ }
}
void camu_portal_init(struct camu_portal_bridge *bridge, struct camu_post_cache *cache,
diff --git a/src/server/server.c b/src/server/server.c
index 7d0b2fe..b7276eb 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -140,7 +140,6 @@ static void list_sink_callback(void *userdata, u8 op, struct lia_list_entry *ent
al_assert(timing->seek_pos <= INT64_MAX);
nn_packet_write_u64(packet, timing->seek_pos);
nn_packet_write_u8(packet, timing->pause);
- nn_packet_write_bool(packet, timing->previous_ended);
nn_packet_write_bool(packet, timing->ended);
nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
diff --git a/src/util/blocking_ring_buffer.c b/src/util/blocking_ring_buffer.c
deleted file mode 100644
index ec6cbb8..0000000
--- a/src/util/blocking_ring_buffer.c
+++ /dev/null
@@ -1,193 +0,0 @@
-#include <al/lib.h>
-
-#include "blocking_ring_buffer.h"
-
-#define LOCK(n) \
- buffer->req_start = n; \
- nn_cond_wait(&buffer->cond, &buffer->mutex);
-
-#define SIGNAL() \
- nn_cond_signal(&buffer->cond);
-
-void camu_ring_buffer_init(struct camu_ring_buffer *buffer, s32 size)
-{
- buffer->size = size;
- buffer->data = (u8 *)al_malloc(buffer->size);
- nn_mutex_init(&buffer->mutex);
- nn_cond_init(&buffer->cond);
- camu_ring_buffer_reset(buffer);
- buffer->reset_unlock = false;
- buffer->enabled = true;
-}
-
-// TODO: does locking here actually prevent this from messing something up.
-void camu_ring_buffer_reset(struct camu_ring_buffer *buffer)
-{
- nn_mutex_lock(&buffer->mutex);
- buffer->valid_size = 0;
- buffer->start = 0;
- buffer->end = 0;
- // If get_chunk was waiting it doesn't matter where req_start
- // was, there should be room now.
- if (nn_cond_is_waiting(&buffer->cond)) {
- buffer->reset_unlock = true;
- SIGNAL();
- }
- buffer->wrap = false;
- nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_disable(struct camu_ring_buffer *buffer)
-{
- nn_mutex_lock(&buffer->mutex);
- buffer->enabled = false;
- // No state on the buffer will be edited by the time a get_chunk
- // returns because of this. So, the next call to get_chunk after
- // the buffer is re-enabled should be correct.
- if (nn_cond_is_waiting(&buffer->cond)) {
- buffer->reset_unlock = true;
- SIGNAL();
- }
- nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_enable(struct camu_ring_buffer *buffer)
-{
- nn_mutex_lock(&buffer->mutex);
- buffer->enabled = true;
- nn_mutex_unlock(&buffer->mutex);
-}
-
-s32 camu_ring_buffer_space(struct camu_ring_buffer *buffer)
-{
- s32 available = 0;
- nn_mutex_lock(&buffer->mutex);
- if (buffer->end < buffer->start) {
- available = buffer->start - buffer->end;
- } else {
- available = buffer->size - buffer->end;
- }
- nn_mutex_unlock(&buffer->mutex);
- return available;
-}
-
-s32 camu_ring_buffer_occupied(struct camu_ring_buffer *buffer)
-{
- s32 filled = 0;
- nn_mutex_lock(&buffer->mutex);
- if (buffer->end < buffer->start) {
- filled = (buffer->valid_size - buffer->start) + buffer->end;
- } else {
- // If the buffer is empty this will equate to 0.
- filled = buffer->end - buffer->start;
- }
- nn_mutex_unlock(&buffer->mutex);
- return filled;
-}
-
-u8 *camu_ring_buffer_get_chunk(struct camu_ring_buffer *buffer, s32 n)
-{
- nn_mutex_lock(&buffer->mutex);
-
- if (!buffer->enabled) return NULL;
-
- if (buffer->end < buffer->start) {
- if (buffer->end + n > buffer->size) {
- // Unsatisfiable, need to increase buffer size.
- al_assert(n < buffer->end);
- // Wait until read to the end of the buffer, wrap around, and read
- // past the amount of data requested in this call (n).
- buffer->wrap = true;
- LOCK(n);
- } else if (buffer->end + n >= buffer->start) {
- // Wait until we read past our current write position plus the amount
- // requested in this call (n).
- LOCK(buffer->end + n);
- }
- } else if (buffer->end + n > buffer->size) {
- // valid_size now marks the end of valid data in the buffer, this may be
- // less than the actual buffer size.
- buffer->valid_size = buffer->end;
- if (n >= buffer->start) {
- // Unsatisfiable, need to increase buffer size.
- al_assert(n < buffer->valid_size);
- // Wait until we read past at least the amount of data requested in this call (n).
- LOCK(n);
- }
- // Set end to zero after waiting because it's possible start is also 0 at this point
- // and we would end up with start == end while the buffer is full with data.
- buffer->end = 0;
- }
-
- // If a reset happened while waiting, the behavior for the structure is to assume
- // you didn't want watever data you were waiting to write.
- if (buffer->reset_unlock) {
- buffer->reset_unlock = false;
- return NULL;
- }
-
- // Return a pointer to the start of the writable data and advance end by
- // exactly the amount requested (n).
- u8 *data = &buffer->data[buffer->end];
- buffer->end += n;
-
- return data;
-}
-
-s32 camu_ring_buffer_read(struct camu_ring_buffer *buffer, u8 **data, s32 n)
-{
- nn_mutex_lock(&buffer->mutex);
-
- // Set data even if we are going to read 0 bytes to avoid possible issues
- // with data being NULL after a call to read.
- *data = &buffer->data[buffer->start];
-
- // If start == end, the buffer is empty.
- if (buffer->start == buffer->end) return 0;
-
- if (buffer->end < buffer->start) {
- // If we would read past the valid data in the buffer, instead
- // read to the end of the valid data.
- if (buffer->start + n > buffer->valid_size) {
- n = buffer->valid_size - buffer->start;
- }
- } else if (buffer->start + n > buffer->end) {
- // If we would read to or past end, instead read to the end.
- n = buffer->end - buffer->start;
- }
-
- buffer->start += n;
-
- // If we read to the end of the valid data, wrap around to the begining.
- //if (buffer->start == buffer->valid_size && buffer->start != buffer->end) {
- if (buffer->start == buffer->valid_size) {
- buffer->start = 0;
- buffer->valid_size = 0;
- // We wrapped and aren't specifically waiting for a position after the wrap, so unlock.
- if (nn_cond_is_waiting(&buffer->cond) && !buffer->wrap) {
- SIGNAL();
- }
- buffer->wrap = false;
- } else if (nn_cond_is_waiting(&buffer->cond) && (!buffer->wrap && buffer->start > buffer->req_start)) {
- // We read past the position we were waiting for, so unlock.
- SIGNAL();
- }
-
- // After SIGNAL(), LOCK() won't proceed until this thread unlocks.
-
- return n;
-}
-
-// Unlock the mutex, this needs to be called after a call to either get_chunk() or read() but
-// only after the pointer has been used.
-void camu_ring_buffer_unlock(struct camu_ring_buffer *buffer)
-{
- nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_free(struct camu_ring_buffer *buffer)
-{
- if (buffer->data) al_free(buffer->data);
- nn_mutex_destroy(&buffer->mutex);
- nn_cond_destroy(&buffer->cond);
-}
diff --git a/src/util/blocking_ring_buffer.h b/src/util/blocking_ring_buffer.h
deleted file mode 100644
index a5288d0..0000000
--- a/src/util/blocking_ring_buffer.h
+++ /dev/null
@@ -1,42 +0,0 @@
-#pragma once
-
-#include <al/types.h>
-#include <nnwt/thread.h>
-
-// Ring buffer with blocking write and non-blocking read.
-// - Write can block and always returns the exact amount requested.
-// - Read will return as many bytes as possible, which can be less than requested or 0.
-
-struct camu_ring_buffer {
- u8 *data;
- s32 size;
- // The position of the last valid byte at the end of the buffer.
- // This can be less than the total size of the buffer.
- s32 valid_size;
- s32 start;
- s32 end;
- // The last byte in the chunk that we are waiting to be fully available.
- s32 req_start;
- // True if we need to wait for the start to wrap around
- // to the begining before considering signaling.
- bool wrap;
- struct nn_mutex mutex;
- struct nn_cond cond;
- // Signal get_chunk to return null and not modifiy the buffer
- // if it was blocking and reset was called.
- bool reset_unlock;
- // If the buffer is disabled, all calls to get_chunk will return
- // early and not block.
- bool enabled;
-};
-
-void camu_ring_buffer_init(struct camu_ring_buffer *buffer, s32 size);
-void camu_ring_buffer_reset(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_disable(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_enable(struct camu_ring_buffer *buffer);
-s32 camu_ring_buffer_space(struct camu_ring_buffer *buffer);
-s32 camu_ring_buffer_occupied(struct camu_ring_buffer *buffer);
-u8 *camu_ring_buffer_get_chunk(struct camu_ring_buffer *buffer, s32 n);
-s32 camu_ring_buffer_read(struct camu_ring_buffer *buffer, u8 **data, s32 n);
-void camu_ring_buffer_unlock(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_free(struct camu_ring_buffer *buffer);
diff --git a/src/util/color_palette.c b/src/util/color_palette.c
index b67f1bb..bb4da03 100644
--- a/src/util/color_palette.c
+++ b/src/util/color_palette.c
@@ -1,5 +1,7 @@
#include "color_palette.h"
+struct camu_color_palette global_color_palette = { 0 };
+
#ifdef NAUNET_HAS_JSON
#include <nnwt/file.h>
#include <jansson.h>
@@ -27,13 +29,9 @@ static char *normal_colors[16] = {
"color14",
"color15"
};
-#endif
-
-struct camu_color_palette global_color_palette = { 0 };
bool camu_color_palette_init(str *path)
{
-#ifdef NAUNET_HAS_JSON
struct nn_file file;
if (!nn_file_open(&file, path, 0)) {
return false;
@@ -62,8 +60,5 @@ bool camu_color_palette_init(str *path)
global_color_palette.colors[i + 2] = (u32)al_str_to_long(al_str_w((char *)color, 1, 6), 16);
}
return true;
-#else
- (void)path;
- return false;
-#endif
}
+#endif
diff --git a/src/util/color_palette.h b/src/util/color_palette.h
index a71360c..e3e4ed5 100644
--- a/src/util/color_palette.h
+++ b/src/util/color_palette.h
@@ -32,4 +32,6 @@ struct camu_color_palette {
extern struct camu_color_palette global_color_palette;
+#ifdef NAUNET_HAS_JSON
bool camu_color_palette_init(str *path);
+#endif
diff --git a/src/util/meson.build b/src/util/meson.build
index dd67515..a83a66a 100644
--- a/src/util/meson.build
+++ b/src/util/meson.build
@@ -1,4 +1,4 @@
-util_src = ['color_palette.c', 'blocking_ring_buffer.c']
+util_src = ['color_palette.c']
blake3 = dependency('libblake3', required: false, allow_fallback: false)
if not blake3.found()
blake3_opts = cmake.subproject_options()
diff --git a/src/util/queue.h b/src/util/queue.h
index ac0dfec..1d788cb 100644
--- a/src/util/queue.h
+++ b/src/util/queue.h
@@ -6,48 +6,52 @@
#define queue(type) \
struct { \
array(type) a; \
- struct nn_mutex mutex; \
+ struct nn_mutex mutex; \
}
-#define camu_queue_size(q, r) \
- do { \
- nn_mutex_lock(&(q).mutex); \
- r = (q).a.size; \
- nn_mutex_unlock(&(q).mutex); \
- } while (0)
+#define camu_queue_size(q, r) \
+AL_MACRO_WRAP \
+({ \
+ nn_mutex_lock(&(q).mutex); \
+ r = (q).a.size; \
+ nn_mutex_unlock(&(q).mutex); \
+})
-#define camu_queue_init(q) \
- do { \
- al_array_init((q).a); \
- nn_mutex_init(&(q).mutex); \
- } while (0)
+#define camu_queue_init(q) \
+AL_MACRO_WRAP \
+({ \
+ al_array_init((q).a); \
+ nn_mutex_init(&(q).mutex); \
+})
-#define camu_queue_push(q, item) \
- do { \
- nn_mutex_lock(&(q).mutex); \
- al_array_push((q).a, item); \
- nn_mutex_unlock(&(q).mutex); \
- } while (0)
+#define camu_queue_push(q, item) \
+AL_MACRO_WRAP \
+({ \
+ nn_mutex_lock(&(q).mutex); \
+ al_array_push((q).a, item); \
+ nn_mutex_unlock(&(q).mutex); \
+})
-#define camu_queue_pop(q, r) \
- do { \
- nn_mutex_lock(&(q).mutex); \
- al_array_pop_at((q).a, 0, r); \
- nn_mutex_unlock(&(q).mutex); \
- } while (0)
+#define camu_queue_pop(q, r) \
+AL_MACRO_WRAP \
+({ \
+ nn_mutex_lock(&(q).mutex); \
+ al_array_pop_at((q).a, 0, r); \
+ nn_mutex_unlock(&(q).mutex); \
+})
-#define camu_queue_try_pop(q, s, r) \
- do { \
- nn_mutex_lock(&(q).mutex); \
- s = (q).a.size; \
- if (s > 0) { \
- al_array_pop_at((q).a, 0, r); \
- } \
- nn_mutex_unlock(&(q).mutex); \
- } while (0)
+#define camu_queue_try_pop(q, s, r) \
+AL_MACRO_WRAP \
+({ \
+ nn_mutex_lock(&(q).mutex); \
+ if ((s = (q).a.size) > 0) \
+ al_array_pop_at((q).a, 0, r); \
+ nn_mutex_unlock(&(q).mutex); \
+})
-#define camu_queue_free(q) \
- do { \
- nn_mutex_destroy(&(q).mutex); \
- al_array_free((q).a); \
- } while (0)
+#define camu_queue_free(q) \
+AL_MACRO_WRAP \
+({ \
+ nn_mutex_destroy(&(q).mutex); \
+ al_array_free((q).a); \
+})
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:36:21 +0000