summaryrefslogtreecommitdiff
path: root/src/lib/shm_rbuff_pthr.c
diff options
context:
space:
mode:
authorDimitri Staessens <dimitri.staessens@ugent.be>2019-04-06 13:34:17 +0200
committerSander Vrijders <sander@ouroboros.rocks>2019-04-06 13:40:52 +0200
commit3e446f29375c0908e8b68ad3a4fd0c5958538150 (patch)
tree323b8f4051567c4e552d622f55a06c224a78fd7a /src/lib/shm_rbuff_pthr.c
parent7039fd62a5863741adb757665696a67a8228db40 (diff)
downloadouroboros-3e446f29375c0908e8b68ad3a4fd0c5958538150.tar.gz
ouroboros-3e446f29375c0908e8b68ad3a4fd0c5958538150.zip
lib: Set independent size for rbuff
This allows setting the size of the rbuffs in a system independently of the main packet buffer using SHM_RBUFF_SIZE. The benefit of setting a smaller rbuff size is that a single process can't fully occupy the main packet buffer. Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be> Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>
Diffstat (limited to 'src/lib/shm_rbuff_pthr.c')
-rw-r--r--src/lib/shm_rbuff_pthr.c62
1 files changed, 59 insertions, 3 deletions
diff --git a/src/lib/shm_rbuff_pthr.c b/src/lib/shm_rbuff_pthr.c
index b4134bf6..5a58605b 100644
--- a/src/lib/shm_rbuff_pthr.c
+++ b/src/lib/shm_rbuff_pthr.c
@@ -72,7 +72,7 @@ int shm_rbuff_write(struct shm_rbuff * rb,
pthread_cond_broadcast(rb->add);
*head_el_ptr(rb) = (ssize_t) idx;
- *rb->head = (*rb->head + 1) & ((SHM_BUFFER_SIZE) -1);
+ *rb->head = (*rb->head + 1) & ((SHM_RBUFF_SIZE) - 1);
pthread_mutex_unlock(rb->lock);
@@ -82,6 +82,62 @@ int shm_rbuff_write(struct shm_rbuff * rb,
return ret;
}
+int shm_rbuff_write_b(struct shm_rbuff * rb,
+ size_t idx,
+ const struct timespec * abstime)
+{
+ int ret = 0;
+
+ assert(rb);
+ assert(idx < SHM_BUFFER_SIZE);
+
+#ifndef HAVE_ROBUST_MUTEX
+ pthread_mutex_lock(rb->lock);
+#else
+ if (pthread_mutex_lock(rb->lock) == EOWNERDEAD)
+ pthread_mutex_consistent(rb->lock);
+#endif
+
+ if (*rb->acl != ACL_RDWR) {
+ if (*rb->acl & ACL_FLOWDOWN)
+ ret = -EFLOWDOWN;
+ else if (*rb->acl & ACL_RDONLY)
+ ret = -ENOTALLOC;
+ goto err;
+ }
+
+ pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
+ (void *) rb->lock);
+
+ while (!shm_rbuff_free(rb) && ret != -ETIMEDOUT) {
+ if (abstime != NULL)
+ ret = -pthread_cond_timedwait(rb->add,
+ rb->lock,
+ abstime);
+ else
+ ret = -pthread_cond_wait(rb->add, rb->lock);
+#ifdef HAVE_ROBUST_MUTEX
+ if (ret == -EOWNERDEAD)
+ pthread_mutex_consistent(rb->lock);
+#endif
+ }
+
+ if (shm_rbuff_empty(rb))
+ pthread_cond_broadcast(rb->add);
+
+ if (ret != -ETIMEDOUT) {
+ *head_el_ptr(rb) = (ssize_t) idx;
+ *rb->head = (*rb->head + 1) & ((SHM_RBUFF_SIZE) - 1);
+ }
+
+ pthread_cleanup_pop(true);
+
+ return ret;
+ err:
+ pthread_mutex_unlock(rb->lock);
+ return ret;
+}
+
ssize_t shm_rbuff_read(struct shm_rbuff * rb)
{
ssize_t ret = 0;
@@ -102,7 +158,7 @@ ssize_t shm_rbuff_read(struct shm_rbuff * rb)
}
ret = *tail_el_ptr(rb);
- *rb->tail = (*rb->tail + 1) & ((SHM_BUFFER_SIZE) - 1);
+ *rb->tail = (*rb->tail + 1) & ((SHM_RBUFF_SIZE) - 1);
pthread_cond_broadcast(rb->del);
pthread_mutex_unlock(rb->lock);
@@ -147,7 +203,7 @@ ssize_t shm_rbuff_read_b(struct shm_rbuff * rb,
if (idx != -ETIMEDOUT) {
idx = *tail_el_ptr(rb);
- *rb->tail = (*rb->tail + 1) & ((SHM_BUFFER_SIZE) - 1);
+ *rb->tail = (*rb->tail + 1) & ((SHM_RBUFF_SIZE) - 1);
pthread_cond_broadcast(rb->del);
}