summaryrefslogtreecommitdiff
path: root/src/lib/ssm/rbuff.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib/ssm/rbuff.c')
-rw-r--r--src/lib/ssm/rbuff.c166
1 files changed, 118 insertions, 48 deletions
diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c
index e4558c31..04978d82 100644
--- a/src/lib/ssm/rbuff.c
+++ b/src/lib/ssm/rbuff.c
@@ -74,12 +74,13 @@ struct ssm_rbuff {
ssize_t * shm_base; /* start of shared memory */
size_t * head; /* start of ringbuffer */
size_t * tail;
- size_t * acl; /* access control */
+ size_t * flags; /* out-of-band flags (RB_*) */
pthread_mutex_t * mtx; /* lock for cond vars only */
pthread_cond_t * add; /* signal when new data */
pthread_cond_t * del; /* signal when data removed */
pid_t pid; /* pid of the owner */
int flow_id; /* flow_id of the flow */
+ size_t n_users; /* in-flight users */
};
#define MM_FLAGS (PROT_READ | PROT_WRITE)
@@ -113,12 +114,13 @@ static struct ssm_rbuff * rbuff_create(pid_t pid,
rb->shm_base = shm_base;
rb->head = (size_t *) (rb->shm_base + (SSM_RBUFF_SIZE));
rb->tail = (size_t *) (rb->head + 1);
- rb->acl = (size_t *) (rb->tail + 1);
- rb->mtx = (pthread_mutex_t *) (rb->acl + 1);
+ rb->flags = (size_t *) (rb->tail + 1);
+ rb->mtx = (pthread_mutex_t *) (rb->flags + 1);
rb->add = (pthread_cond_t *) (rb->mtx + 1);
rb->del = rb->add + 1;
rb->pid = pid;
rb->flow_id = flow_id;
+ rb->n_users = 0;
return rb;
@@ -179,7 +181,7 @@ struct ssm_rbuff * ssm_rbuff_create(pid_t pid,
if (pthread_cond_init(rb->del, &cattr))
goto fail_del;
- *rb->acl = ACL_RDWR;
+ *rb->flags = RB_RDWR;
*rb->head = 0;
*rb->tail = 0;
@@ -228,27 +230,47 @@ void ssm_rbuff_close(struct ssm_rbuff * rb)
{
assert(rb);
+ /*
+ * Caller must set RB_FLOWDOWN first; if a user becomes
+ * cancellable, push a cleanup that decrements n_users.
+ */
+ while (__atomic_load_n(&rb->n_users, __ATOMIC_SEQ_CST) > 0) {
+ struct timespec tic = { 0, 100000 };
+ nanosleep(&tic, NULL);
+ }
+
rbuff_destroy(rb);
}
+/* Cancel cleanup for a blocked reader: unlock mtx AND drop the n_users ref. */
+static void __cleanup_rbuff_reader(void * o)
+{
+ struct ssm_rbuff * rb = (struct ssm_rbuff *) o;
+
+ pthread_mutex_unlock(rb->mtx);
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+}
+
int ssm_rbuff_write(struct ssm_rbuff * rb,
- size_t idx)
+ size_t off)
{
- size_t acl;
+ size_t flags;
bool was_empty;
int ret = 0;
assert(rb != NULL);
- acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
- if (acl != ACL_RDWR) {
- if (acl & ACL_FLOWDOWN) {
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
+ flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
+ if (flags != RB_RDWR) {
+ if (flags & RB_FLOWDOWN) {
ret = -EFLOWDOWN;
- goto fail_acl;
+ goto fail_flags;
}
- if (acl & ACL_RDONLY) {
+ if (!(flags & RB_WR)) {
ret = -ENOTALLOC;
- goto fail_acl;
+ goto fail_flags;
}
}
@@ -261,7 +283,7 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
was_empty = IS_EMPTY(rb);
- HEAD(rb) = (ssize_t) idx;
+ HEAD(rb) = (ssize_t) off;
ADVANCE_HEAD(rb);
if (was_empty)
@@ -269,43 +291,47 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
pthread_mutex_unlock(rb->mtx);
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return 0;
fail_mutex:
pthread_mutex_unlock(rb->mtx);
- fail_acl:
+ fail_flags:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
int ssm_rbuff_write_b(struct ssm_rbuff * rb,
- size_t idx,
+ size_t off,
const struct timespec * abstime)
{
- size_t acl;
+ size_t flags;
int ret = 0;
bool was_empty;
assert(rb != NULL);
- acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
- if (acl != ACL_RDWR) {
- if (acl & ACL_FLOWDOWN) {
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
+ flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
+ if (flags != RB_RDWR) {
+ if (flags & RB_FLOWDOWN) {
ret = -EFLOWDOWN;
- goto fail_acl;
+ goto fail_flags;
}
- if (acl & ACL_RDONLY) {
+ if (!(flags & RB_WR)) {
ret = -ENOTALLOC;
- goto fail_acl;
+ goto fail_flags;
}
}
robust_mutex_lock(rb->mtx);
- pthread_cleanup_push(__cleanup_mutex_unlock, rb->mtx);
+ pthread_cleanup_push(__cleanup_rbuff_reader, rb);
while (IS_FULL(rb) && ret != -ETIMEDOUT) {
- acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
- if (acl & ACL_FLOWDOWN) {
+ flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
+ if (flags & RB_FLOWDOWN) {
ret = -EFLOWDOWN;
break;
}
@@ -316,7 +342,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
if (ret != -ETIMEDOUT && ret != -EFLOWDOWN) {
was_empty = IS_EMPTY(rb);
- HEAD(rb) = (ssize_t) idx;
+ HEAD(rb) = (ssize_t) off;
ADVANCE_HEAD(rb);
if (was_empty)
pthread_cond_broadcast(rb->add);
@@ -324,24 +350,28 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
pthread_mutex_unlock(rb->mtx);
- fail_acl:
+ fail_flags:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
-static int check_rb_acl(struct ssm_rbuff * rb)
+static int check_rb_flags(struct ssm_rbuff * rb)
{
- size_t acl;
+ size_t flags;
assert(rb != NULL);
- acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
+ flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
- if (acl & ACL_FLOWDOWN)
+ if (flags & RB_FLOWDOWN)
return -EFLOWDOWN;
- if (acl & ACL_FLOWPEER)
+ if (flags & RB_FLOWPEER)
return -EFLOWPEER;
+ if (!(flags & RB_RD))
+ return -ENOTALLOC;
+
return -EAGAIN;
}
@@ -351,11 +381,21 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb)
assert(rb != NULL);
- if (IS_EMPTY(rb))
- return check_rb_acl(rb);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
+ if (IS_EMPTY(rb)) {
+ ret = check_rb_flags(rb);
+ goto out;
+ }
robust_mutex_lock(rb->mtx);
+ if (IS_EMPTY(rb)) {
+ pthread_mutex_unlock(rb->mtx);
+ ret = check_rb_flags(rb);
+ goto out;
+ }
+
ret = TAIL(rb);
ADVANCE_TAIL(rb);
@@ -363,6 +403,8 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb)
pthread_mutex_unlock(rb->mtx);
+ out:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
@@ -370,21 +412,25 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb,
const struct timespec * abstime)
{
ssize_t idx = -1;
- size_t acl;
+ size_t flags;
assert(rb != NULL);
- acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
- if (IS_EMPTY(rb) && (acl & ACL_FLOWDOWN))
- return -EFLOWDOWN;
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
+ flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
+ if (IS_EMPTY(rb) && (flags & RB_FLOWDOWN)) {
+ idx = -EFLOWDOWN;
+ goto out;
+ }
robust_mutex_lock(rb->mtx);
- pthread_cleanup_push(__cleanup_mutex_unlock, rb->mtx);
+ pthread_cleanup_push(__cleanup_rbuff_reader, rb);
while (IS_EMPTY(rb) &&
idx != -ETIMEDOUT &&
- check_rb_acl(rb) == -EAGAIN) {
+ check_rb_flags(rb) == -EAGAIN) {
idx = -robust_wait(rb->add, rb->mtx, abstime);
}
@@ -395,43 +441,67 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb,
ADVANCE_TAIL(rb);
pthread_cond_broadcast(rb->del);
} else if (idx != -ETIMEDOUT) {
- idx = check_rb_acl(rb);
+ idx = check_rb_flags(rb);
}
pthread_mutex_unlock(rb->mtx);
assert(idx != -EAGAIN);
+ out:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return idx;
}
-void ssm_rbuff_set_acl(struct ssm_rbuff * rb,
- uint32_t flags)
+void ssm_rbuff_set_bits(struct ssm_rbuff * rb,
+ uint32_t bits)
{
assert(rb != NULL);
- __atomic_store_n(rb->acl, (size_t) flags, __ATOMIC_SEQ_CST);
+ robust_mutex_lock(rb->mtx);
+ __atomic_fetch_or(rb->flags, (size_t) bits, __ATOMIC_SEQ_CST);
+ pthread_cond_broadcast(rb->add);
+ pthread_cond_broadcast(rb->del);
+ pthread_mutex_unlock(rb->mtx);
+}
+
+void ssm_rbuff_clr_bits(struct ssm_rbuff * rb,
+ uint32_t bits)
+{
+ assert(rb != NULL);
+
+ robust_mutex_lock(rb->mtx);
+ __atomic_fetch_and(rb->flags, ~(size_t) bits, __ATOMIC_SEQ_CST);
+ pthread_cond_broadcast(rb->add);
+ pthread_cond_broadcast(rb->del);
+ pthread_mutex_unlock(rb->mtx);
}
-uint32_t ssm_rbuff_get_acl(struct ssm_rbuff * rb)
+uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb)
{
assert(rb != NULL);
- return (uint32_t) __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
+ return (uint32_t) __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
}
void ssm_rbuff_fini(struct ssm_rbuff * rb)
{
assert(rb != NULL);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
robust_mutex_lock(rb->mtx);
- pthread_cleanup_push(__cleanup_mutex_unlock, rb->mtx);
+ pthread_cleanup_push(__cleanup_rbuff_reader, rb);
while (!IS_EMPTY(rb))
robust_wait(rb->del, rb->mtx, NULL);
- pthread_cleanup_pop(true);
+ pthread_cleanup_pop(false);
+
+ pthread_mutex_unlock(rb->mtx);
+
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
}
size_t ssm_rbuff_queued(struct ssm_rbuff * rb)
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-05 05:09:23 +0000