From 9026e7f49f11e920eac666bd4159ec459b979623 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Thu, 26 Dec 2024 11:25:14 -0500 Subject: Incomplete buffer errored and sink queued Signed-off-by: Andrew Opalach --- src/buffer/common.h | 3 +- src/buffer/common_internal.h | 4 +- src/buffer/video.c | 5 ++ src/codec/ffmpeg/decoder.c | 4 +- src/codec/ffmpeg/demuxer.c | 4 +- src/codec/ffmpeg/encoder.c | 1 + src/fruits/cmc/ui/ui.c | 4 +- src/fruits/cmsrv/cmsrv.c | 4 +- src/liana/list.c | 97 ++++++++++++-------- src/liana/list.h | 2 - src/liana/server.c | 4 +- src/libsink/sink.c | 38 ++++---- src/mixer/mixer.c | 4 +- src/portal/src/search.c | 8 +- src/server/server.c | 1 - src/util/blocking_ring_buffer.c | 193 ---------------------------------------- src/util/blocking_ring_buffer.h | 42 --------- src/util/color_palette.c | 11 +-- src/util/color_palette.h | 2 + src/util/meson.build | 2 +- src/util/queue.h | 88 +++++++++--------- 21 files changed, 159 insertions(+), 362 deletions(-) delete mode 100644 src/util/blocking_ring_buffer.c delete mode 100644 src/util/blocking_ring_buffer.h (limited to 'src') 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 - -#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 -#include - -// 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 #include @@ -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_init(q) \ - do { \ - al_array_init((q).a); \ - nn_mutex_init(&(q).mutex); \ - } while (0) - -#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_pop(q, r) \ - do { \ - nn_mutex_lock(&(q).mutex); \ +#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) \ +AL_MACRO_WRAP \ +({ \ + al_array_init((q).a); \ + nn_mutex_init(&(q).mutex); \ +}) + +#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) \ +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) \ +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); \ - } while (0) - -#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_free(q) \ - do { \ - nn_mutex_destroy(&(q).mutex); \ - al_array_free((q).a); \ - } while (0) + nn_mutex_unlock(&(q).mutex); \ +}) + +#define camu_queue_free(q) \ +AL_MACRO_WRAP \ +({ \ + nn_mutex_destroy(&(q).mutex); \ + al_array_free((q).a); \ +}) -- cgit v1.2.3-101-g0448