Skip to content

get_wq: round robin per thread instead of through a shared counter #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
byrnedj merged 1 commit into from
Oct 27, 2026
Merged

get_wq: round robin per thread instead of through a shared counter #40

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 22 additions & 4 deletions dto.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,6 @@ struct dto_wq {
struct dto_device {
struct dto_wq* wqs[MAX_WQS];
uint8_t num_wqs;
atomic_uchar next_wq;
};
#endif

Expand Down Expand Up @@ -127,7 +126,11 @@ static struct dto_wq wqs[MAX_WQS];
static struct dto_device* devices[MAX_NUMA_NODES];
#endif
static uint8_t num_wqs;
static atomic_uchar next_wq;
/* round-robin start position for the next thread that picks a WQ (get_wq) */
static atomic_uint next_thread_wq;
/* this thread's round-robin position */
static __thread unsigned int thr_wq_rr;
static __thread bool thr_wq_rr_set;
static atomic_uchar dto_initialized;
static atomic_uchar dto_initializing;
static uint8_t use_std_lib_calls;
Expand Down Expand Up @@ -1603,6 +1606,21 @@ static void cleanup_dto(void)
#endif
}

/* Next round-robin position for this thread. Each thread keeps its own, so
* picking a WQ writes no cache line shared with other threads; a counter
* incremented by every submit on every thread was measurable on small
* operations. Threads start at successive positions, so concurrent threads
* still spread across the WQs. */
static __always_inline unsigned int thread_wq_rr(void)
{
if (unlikely(!thr_wq_rr_set)) {
thr_wq_rr = atomic_fetch_add_explicit(&next_thread_wq, 1,
memory_order_relaxed);
thr_wq_rr_set = true;
}
return thr_wq_rr++;
}

static __always_inline struct dto_wq *get_wq(void* buf)
{
struct dto_wq* wq = NULL;
Expand All @@ -1617,14 +1635,14 @@ static __always_inline struct dto_wq *get_wq(void* buf)
struct dto_device* dev = devices[numa_node];
if (dev != NULL &&
dev->num_wqs > 0) {
wq = dev->wqs[dev->next_wq++ % dev->num_wqs];
wq = dev->wqs[thread_wq_rr() % dev->num_wqs];
}
}
}
#endif

if (wq == NULL) {
wq = &wqs[next_wq++ % num_wqs];
wq = &wqs[thread_wq_rr() % num_wqs];
}

return wq;
Expand Down
Loading