summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/lib/dev.c5
-rw-r--r--src/lib/ssm/rbuff.c19
2 files changed, 20 insertions, 4 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c
index d0997273..c4c874cc 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -615,7 +615,12 @@ static int flow_rx_one(struct flow * flow,
continue;
}
+ pthread_cleanup_push(__cleanup_rwlock_unlock, &proc.lock);
+
idx = ssm_rbuff_read_b(rx_rb, &wait_abs);
+
+ pthread_cleanup_pop(false);
+
if (idx == -ETIMEDOUT) {
pthread_rwlock_unlock(&proc.lock);
if (deadline_passed(abs))
diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c
index 0121af89..7886b5c2 100644
--- a/src/lib/ssm/rbuff.c
+++ b/src/lib/ssm/rbuff.c
@@ -242,6 +242,15 @@ void ssm_rbuff_close(struct ssm_rbuff * rb)
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 off)
{
@@ -318,7 +327,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
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) {
flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
@@ -417,7 +426,7 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb,
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 &&
@@ -483,12 +492,14 @@ void ssm_rbuff_fini(struct ssm_rbuff * rb)
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);
}