michaelh/zephyr
1
0
Fork 0
Code Releases Activity

rtio: Use an atomic completion counter

Browse source 
Rather than looking at the pool of completions for spinning use an
atomic counter of total completions ever done. The relative number of
completions being waited on by rtio_submit may then always be correctly
done.

Prior to this a race was possible, and understood, as
rtio_cqe_consumable was a likely but not guaranteed count of completions.
Sure enough on an SMP system the likely count was ahead of the actual
available completions and a race was caught by the simple test case.

Signed-off-by: Tom Burdick <thomas.burdick@intel.com>
This commit is contained in:
Tom Burdick 2023-06-23 13:23:46 -05:00 • committed by Fabio Baltieri
commit 49a97540dc
2 changed files with 16 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

22
include/zephyr/rtio/rtio.h
View file

@ -354,6 +354,9 @@ struct rtio {
struct k_sem *consume_sem;
#endif
/* Total number of completions */
atomic_t cq_count;
/* Number of completions that were unable to be submitted with results
* due to the cq spsc being full
*/
@ -761,6 +764,7 @@ static inline void rtio_block_pool_free(struct rtio_block_pool *pool, void *buf,
IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM, (.submit_sem = &_submit_sem_##name,)) \
IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM, (.submit_count = 0,)) \
IF_ENABLED(CONFIG_RTIO_CONSUME_SEM, (.consume_sem = &_consume_sem_##name,)) \
.cq_count = ATOMIC_INIT(0), \
.xcqcnt = ATOMIC_INIT(0), \
.sqe_pool = _sqe_pool, \
.cqe_pool = _cqe_pool, \
@ -813,18 +817,6 @@ static inline uint32_t rtio_sqe_acquirable(struct rtio *r)
return r->sqe_pool->pool_free;
}
/**
* @brief Count of likely, but not gauranteed, consumable completion queue events
*
* @param r RTIO context
*
* @return Likely count of consumable completion queue events
*/
static inline uint32_t rtio_cqe_consumable(struct rtio *r)
{
return (r->cqe_pool->pool_size - r->cqe_pool->pool_free);
}
/**
* @brief Get the next sqe in the transaction
*
@ -1148,6 +1140,8 @@ static inline void rtio_cqe_submit(struct rtio *r, int result, void *userdata, u
cqe->flags = flags;
rtio_cqe_produce(r, cqe);
}
atomic_inc(&r->cq_count);
#ifdef CONFIG_RTIO_SUBMIT_SEM
if (r->submit_count > 0) {
r->submit_count--;
@ -1425,6 +1419,8 @@ static inline int z_impl_rtio_submit(struct rtio *r, uint32_t wait_count)
k_sem_reset(r->submit_sem);
r->submit_count = wait_count;
}
#else
uintptr_t cq_count = atomic_get(&r->cq_count) + wait_count;
#endif
/* Submit the queue to the executor which consumes submissions
@ -1444,7 +1440,7 @@ static inline int z_impl_rtio_submit(struct rtio *r, uint32_t wait_count)
"semaphore was reset or timed out while waiting on completions!");
}
#else
while (rtio_cqe_consumable(r) < wait_count) {
while (atomic_get(&r->cq_count) < cq_count) {
Z_SPIN_DELAY(10);
k_yield();
}

10
tests/subsys/rtio/rtio_api/src/test_rtio_api.c
View file

@ -117,6 +117,7 @@ void test_rtio_chain_(struct rtio *r)
uint32_t userdata[4] = {0, 1, 2, 3};
struct rtio_sqe *sqe;
struct rtio_cqe *cqe;
uintptr_t cq_count = atomic_get(&r->cq_count);
for (int i = 0; i < 4; i++) {
sqe = rtio_sqe_acquire(r);
@ -131,10 +132,11 @@ void test_rtio_chain_(struct rtio *r)
sqe->flags = 0;
TC_PRINT("submitting\n");
res = rtio_submit(r, 4);
TC_PRINT("checking cq\n");
zassert_ok(res, "Should return ok from rtio_execute");
zassert_equal(rtio_cqe_consumable(r), 4, "Should have 4 pending completions");
zassert_equal(atomic_get(&r->cq_count) - cq_count, 4, "Should have 4 pending completions");
for (int i = 0; i < 4; i++) {
cqe = rtio_cqe_consume(r);
@ -527,6 +529,7 @@ void test_rtio_transaction_(struct rtio *r)
struct rtio_sqe *sqe;
struct rtio_cqe *cqe;
bool seen[2] = { 0 };
uintptr_t cq_count = atomic_get(&r->cq_count);
sqe = rtio_sqe_acquire(r);
zassert_not_null(sqe, "Expected a valid sqe");
@ -551,9 +554,10 @@ void test_rtio_transaction_(struct rtio *r)
TC_PRINT("submitting userdata 0 %p, userdata 1 %p\n", &userdata[0], &userdata[1]);
res = rtio_submit(r, 4);
TC_PRINT("checking cq, completions available %u\n", rtio_cqe_consumable(r));
TC_PRINT("checking cq, completions available, count at start %lu, current count %lu\n",
cq_count, atomic_get(&r->cq_count));
zassert_ok(res, "Should return ok from rtio_execute");
zassert_equal(rtio_cqe_consumable(r), 4, "Should have 4 pending completions");
zassert_equal(atomic_get(&r->cq_count) - cq_count, 4, "Should have 4 pending completions");
for (int i = 0; i < 4; i++) {
TC_PRINT("consume %d\n", i);