diff --git a/app/meson.build b/app/meson.build index 3a5cb12a..d6fac580 100644 --- a/app/meson.build +++ b/app/meson.build @@ -25,6 +25,7 @@ src = [ 'src/server.c', 'src/stream.c', 'src/video_buffer.c', + 'src/util/acksync.c', 'src/util/file.c', 'src/util/intr.c', 'src/util/log.c', diff --git a/app/src/util/acksync.c b/app/src/util/acksync.c new file mode 100644 index 00000000..2899cdcb --- /dev/null +++ b/app/src/util/acksync.c @@ -0,0 +1,76 @@ +#include "acksync.h" + +#include +#include "util/log.h" + +bool +sc_acksync_init(struct sc_acksync *as) { + bool ok = sc_mutex_init(&as->mutex); + if (!ok) { + return false; + } + + ok = sc_cond_init(&as->cond); + if (!ok) { + sc_mutex_destroy(&as->mutex); + return false; + } + + as->stopped = false; + as->ack = SC_SEQUENCE_INVALID; + + return true; +} + +void +sc_acksync_destroy(struct sc_acksync *as) { + sc_cond_destroy(&as->cond); + sc_mutex_destroy(&as->mutex); +} + +void +sc_acksync_ack(struct sc_acksync *as, uint64_t sequence) { + sc_mutex_lock(&as->mutex); + + // Acknowledgements must be monotonic + assert(sequence >= as->ack); + + as->ack = sequence; + sc_cond_signal(&as->cond); + + sc_mutex_unlock(&as->mutex); +} + +enum sc_acksync_wait_result +sc_acksync_wait(struct sc_acksync *as, uint64_t ack, sc_tick deadline) { + sc_mutex_lock(&as->mutex); + + bool timed_out = false; + while (!as->stopped && as->ack < ack && !timed_out) { + timed_out = !sc_cond_timedwait(&as->cond, &as->mutex, deadline); + } + + enum sc_acksync_wait_result ret; + if (as->stopped) { + ret = SC_ACKSYNC_WAIT_INTR; + } else if (as->ack >= ack) { + ret = SC_ACKSYNC_WAIT_OK; + } else { + assert(timed_out); + ret = SC_ACKSYNC_WAIT_TIMEOUT; + } + sc_mutex_unlock(&as->mutex); + + return ret; +} + +/** + * Interrupt any `sc_acksync_wait()` + */ +void +sc_acksync_interrupt(struct sc_acksync *as) { + sc_mutex_lock(&as->mutex); + as->stopped = true; + sc_cond_signal(&as->cond); + sc_mutex_unlock(&as->mutex); +} diff --git a/app/src/util/acksync.h b/app/src/util/acksync.h new file mode 100644 index 00000000..1fd34444 --- /dev/null +++ b/app/src/util/acksync.h @@ -0,0 +1,61 @@ +#ifndef SC_ACK_SYNC_H +#define SC_ACK_SYNC_H + +#include "common.h" + +#include "thread.h" + +#define SC_SEQUENCE_INVALID 0 + +/** + * Helper to wait for acknowledgments + */ +struct sc_acksync { + sc_mutex mutex; + sc_cond cond; + + bool stopped; + + // Last acked value, initially SC_SEQUENCE_INVALID + uint64_t ack; +}; + +enum sc_acksync_wait_result { + // Acknowledgment received + SC_ACKSYNC_WAIT_OK, + + // Timeout expired + SC_ACKSYNC_WAIT_TIMEOUT, + + // Interrupted from another thread by sc_acksync_interrupt() + SC_ACKSYNC_WAIT_INTR, +}; + +bool +sc_acksync_init(struct sc_acksync *as); + +void +sc_acksync_destroy(struct sc_acksync *as); + +/** + * Acknowledge `sequence` + * + * The `sequence` must be greater than (or equal to) any previous acknowledged + * sequence. + */ +void +sc_acksync_ack(struct sc_acksync *as, uint64_t sequence); + +/** + * Wait for acknowledgment of sequence `ack` (or higher) + */ +enum sc_acksync_wait_result +sc_acksync_wait(struct sc_acksync *as, uint64_t ack, sc_tick deadline); + +/** + * Interrupt any `sc_acksync_wait()` + */ +void +sc_acksync_interrupt(struct sc_acksync *as); + +#endif