[PATCH] lockfree ring_spsc: make counter type a template parameter

[Waldemar Kozaczuk]

This patch does not change the ring_spsc template in any significant way
but merely makes the counter type (of _begin and _end) a parameter. This
allows us to use smaller type - unsigned short (2 bytes) - in the unit test
tst-ring-spsc-wraparound.cc to make it execute more rapidly.

Before this change this unit test would run for almost 3 minutes on
aarch64 and almost 10 seconds on x64. Now it executes way under a second
and still verifies the same edge condition (see
1ba76eb03cba4431b557183d1001b16991cd1fa4).

Signed-off-by: Waldemar Kozaczuk <jwkoz...@gmail.com>
---
include/lockfree/ring.hh | 22 +++++++++++-----------
include/lockfree/unordered-queue-spsc.hh | 2 +-
include/lockfree/unordered_ring_mpsc.hh | 2 +-
include/osv/net_channel.hh | 2 +-
include/osv/percpu_xmit.hh | 2 +-
tests/misc-free-perf.cc | 2 +-
tests/misc-lfring.cc | 2 +-
tests/tst-nway-merger.cc | 2 +-
tests/tst-ring-spsc-wraparound.cc | 4 ++--
9 files changed, 20 insertions(+), 20 deletions(-)

diff --git a/include/lockfree/ring.hh b/include/lockfree/ring.hh
index c9fefccc..434a14b7 100644
--- a/include/lockfree/ring.hh
+++ b/include/lockfree/ring.hh
@@ -20,7 +20,7 @@
//
// spsc ring of fixed size
//
-template<class T, unsigned MaxSize, unsigned MaxSizeMask = MaxSize - 1>
+template<class T, typename COUNTER_TYPE, COUNTER_TYPE MaxSize, COUNTER_TYPE MaxSizeMask = MaxSize - 1>
class ring_spsc {
public:
ring_spsc(): _begin(0), _end(0)
@@ -31,7 +31,7 @@ public:
template<typename... Args>
inline bool emplace(Args&&... args)
{
- unsigned end = _end.load(std::memory_order_relaxed);
+ COUNTER_TYPE end = _end.load(std::memory_order_relaxed);

//
// It's ok to load _begin with relaxed ordering (in the size()) since
@@ -56,7 +56,7 @@ public:

bool pop(T& element)
{
- unsigned beg = _begin.load(std::memory_order_relaxed);
+ COUNTER_TYPE beg = _begin.load(std::memory_order_relaxed);

if (empty()) {
return false;
@@ -83,15 +83,15 @@ public:
* @return TRUE if there are no elements
*/
bool empty() const {
- unsigned beg = _begin.load(std::memory_order_relaxed);
- unsigned end = _end.load(std::memory_order_acquire);
+ COUNTER_TYPE beg = _begin.load(std::memory_order_relaxed);
+ COUNTER_TYPE end = _end.load(std::memory_order_acquire);
return beg == end;
}

const T& front() const {
DEBUG_ASSERT(!empty(), "calling front() on an empty queue!");

- unsigned beg = _begin.load(std::memory_order_relaxed);
+ COUNTER_TYPE beg = _begin.load(std::memory_order_relaxed);

return _ring[beg & MaxSizeMask];
}
@@ -102,16 +102,16 @@ public:
*
* @return the current number of the elements.
*/
- unsigned size() const {
- unsigned end = _end.load(std::memory_order_relaxed);
- unsigned beg = _begin.load(std::memory_order_relaxed);
+ COUNTER_TYPE size() const {
+ COUNTER_TYPE end = _end.load(std::memory_order_relaxed);
+ COUNTER_TYPE beg = _begin.load(std::memory_order_relaxed);

return (end - beg);
}

private:
- std::atomic<unsigned> _begin CACHELINE_ALIGNED;
- std::atomic<unsigned> _end CACHELINE_ALIGNED;
+ std::atomic<COUNTER_TYPE> _begin CACHELINE_ALIGNED;
+ std::atomic<COUNTER_TYPE> _end CACHELINE_ALIGNED;
T _ring[MaxSize];
};

diff --git a/include/lockfree/unordered-queue-spsc.hh b/include/lockfree/unordered-queue-spsc.hh
index 72f77790..702da681 100644
--- a/include/lockfree/unordered-queue-spsc.hh
+++ b/include/lockfree/unordered-queue-spsc.hh
@@ -26,7 +26,7 @@ namespace lockfree {
template <typename LT, unsigned RingSize>
class unordered_queue_spsc {
private:
- ring_spsc<LT*,RingSize> _ring;
+ ring_spsc<LT*,unsigned,RingSize> _ring;
unordered_queue_mpsc<LT> _queue;
public:

diff --git a/include/lockfree/unordered_ring_mpsc.hh b/include/lockfree/unordered_ring_mpsc.hh
index 72599bae..6c6d1165 100644
--- a/include/lockfree/unordered_ring_mpsc.hh
+++ b/include/lockfree/unordered_ring_mpsc.hh
@@ -26,7 +26,7 @@ template<class T, unsigned MaxSizePerCpu>
class unordered_ring_mpsc
{
private:
- std::vector<ring_spsc<T,MaxSizePerCpu>> rings;
+ std::vector<ring_spsc<T,unsigned,MaxSizePerCpu>> rings;
public:
using ring_mpsc_t = unordered_ring_mpsc<T,MaxSizePerCpu>;

diff --git a/include/osv/net_channel.hh b/include/osv/net_channel.hh
index 2784e9e7..11cc09cb 100644
--- a/include/osv/net_channel.hh
+++ b/include/osv/net_channel.hh
@@ -33,7 +33,7 @@ extern void* memory::alloc_page();
class net_channel {
private:
std::function<void (mbuf*)> _process_packet;
- ring_spsc<mbuf*, 256> _queue;
+ ring_spsc<mbuf*, unsigned, 256> _queue;
sched::thread_handle _waiting_thread CACHELINE_ALIGNED;
// extra list of threads to wake
osv::rcu_ptr<std::vector<pollreq*>> _pollers;
diff --git a/include/osv/percpu_xmit.hh b/include/osv/percpu_xmit.hh
index 7ec6f1be..4b44bf6a 100644
--- a/include/osv/percpu_xmit.hh
+++ b/include/osv/percpu_xmit.hh
@@ -151,7 +151,7 @@ public:

private:
lockfree::queue_mpsc<wait_record> _waitq;
- ring_spsc<value_type, CpuTxqSize> _r;
+ ring_spsc<value_type, unsigned, CpuTxqSize> _r;

//
// We don't want to wake the waiters when the Tx worker is going to sleep.
diff --git a/tests/misc-free-perf.cc b/tests/misc-free-perf.cc
index b9dc24b2..c7478adf 100644
--- a/tests/misc-free-perf.cc
+++ b/tests/misc-free-perf.cc
@@ -21,7 +21,7 @@ struct linked_object {
};

using _clock = std::chrono::high_resolution_clock;
-using queue_t = ring_spsc<void*,64*1024*1024>;
+using queue_t = ring_spsc<void*,unsigned,64*1024*1024>;

// Manages threads, allocates each thread on a different CPU
class thread_allocator
diff --git a/tests/misc-lfring.cc b/tests/misc-lfring.cc
index 83e3de84..ade201f1 100644
--- a/tests/misc-lfring.cc
+++ b/tests/misc-lfring.cc
@@ -64,7 +64,7 @@ public:

private:

- ring_spsc<int, 4096> _ring;
+ ring_spsc<int,unsigned,4096> _ring;

int _stats[2][max_random] = {};

diff --git a/tests/tst-nway-merger.cc b/tests/tst-nway-merger.cc
index 5f37a16a..83a0cc7d 100644
--- a/tests/tst-nway-merger.cc
+++ b/tests/tst-nway-merger.cc
@@ -69,7 +69,7 @@ public:
bool empty() const { return _r.empty(); }

private:
- ring_spsc<T, MaxSize> _r;
+ ring_spsc<T,unsigned,MaxSize> _r;
};

typedef my_spsc_ring<my_struct, 8> my_spsc_queue;
diff --git a/tests/tst-ring-spsc-wraparound.cc b/tests/tst-ring-spsc-wraparound.cc
index 2490755d..64953aed 100644
--- a/tests/tst-ring-spsc-wraparound.cc
+++ b/tests/tst-ring-spsc-wraparound.cc
@@ -17,8 +17,8 @@ using namespace std;

int main(int argc, char *argv[])
{
- ring_spsc<int, 256> test_ring;
- unsigned count;
+ ring_spsc<int, unsigned short, 256> test_ring;
+ unsigned short count;
int val;

for (count = 1; count != 0; count++) {
--
2.27.0

[Commit Bot]

From: Waldemar Kozaczuk <jwkoz...@gmail.com>
Committer: Waldemar Kozaczuk <jwkoz...@gmail.com>
Branch: master

lockfree ring_spsc: make counter type a template parameter

This patch does not change the ring_spsc template in any significant way
but merely makes the counter type (of _begin and _end) a parameter. This
allows us to use smaller type - unsigned short (2 bytes) - in the unit test
tst-ring-spsc-wraparound.cc to make it execute more rapidly.

Before this change this unit test would run for almost 3 minutes on
aarch64 and almost 10 seconds on x64. Now it executes way under a second
and still verifies the same edge condition (see
1ba76eb03cba4431b557183d1001b16991cd1fa4).

Signed-off-by: Waldemar Kozaczuk <jwkoz...@gmail.com>

---

diff --git a/include/lockfree/ring.hh b/include/lockfree/ring.hh

--- a/include/lockfree/unordered-queue-spsc.hh
+++ b/include/lockfree/unordered-queue-spsc.hh
@@ -26,7 +26,7 @@ namespace lockfree {
template <typename LT, unsigned RingSize>
class unordered_queue_spsc {
private:
- ring_spsc<LT*,RingSize> _ring;
+ ring_spsc<LT*,unsigned,RingSize> _ring;
unordered_queue_mpsc<LT> _queue;
public:

diff --git a/include/lockfree/unordered_ring_mpsc.hh b/include/lockfree/unordered_ring_mpsc.hh

--- a/include/lockfree/unordered_ring_mpsc.hh
+++ b/include/lockfree/unordered_ring_mpsc.hh
@@ -26,7 +26,7 @@ template<class T, unsigned MaxSizePerCpu>
class unordered_ring_mpsc
{
private:
- std::vector<ring_spsc<T,MaxSizePerCpu>> rings;
+ std::vector<ring_spsc<T,unsigned,MaxSizePerCpu>> rings;
public:
using ring_mpsc_t = unordered_ring_mpsc<T,MaxSizePerCpu>;

diff --git a/include/osv/net_channel.hh b/include/osv/net_channel.hh

--- a/include/osv/net_channel.hh
+++ b/include/osv/net_channel.hh
@@ -33,7 +33,7 @@ extern void* memory::alloc_page();
class net_channel {
private:
std::function<void (mbuf*)> _process_packet;
- ring_spsc<mbuf*, 256> _queue;
+ ring_spsc<mbuf*, unsigned, 256> _queue;
sched::thread_handle _waiting_thread CACHELINE_ALIGNED;
// extra list of threads to wake
osv::rcu_ptr<std::vector<pollreq*>> _pollers;
diff --git a/include/osv/percpu_xmit.hh b/include/osv/percpu_xmit.hh

--- a/include/osv/percpu_xmit.hh
+++ b/include/osv/percpu_xmit.hh
@@ -151,7 +151,7 @@ public:

private:
lockfree::queue_mpsc<wait_record> _waitq;
- ring_spsc<value_type, CpuTxqSize> _r;
+ ring_spsc<value_type, unsigned, CpuTxqSize> _r;

//
// We don't want to wake the waiters when the Tx worker is going to sleep.
diff --git a/tests/misc-free-perf.cc b/tests/misc-free-perf.cc

--- a/tests/misc-free-perf.cc
+++ b/tests/misc-free-perf.cc
@@ -21,7 +21,7 @@ struct linked_object {
};

using _clock = std::chrono::high_resolution_clock;
-using queue_t = ring_spsc<void*,64*1024*1024>;
+using queue_t = ring_spsc<void*,unsigned,64*1024*1024>;

// Manages threads, allocates each thread on a different CPU
class thread_allocator
diff --git a/tests/misc-lfring.cc b/tests/misc-lfring.cc

--- a/tests/misc-lfring.cc
+++ b/tests/misc-lfring.cc
@@ -64,7 +64,7 @@ class test_spsc_ring {

private:

- ring_spsc<int, 4096> _ring;
+ ring_spsc<int,unsigned,4096> _ring;

int _stats[2][max_random] = {};

diff --git a/tests/tst-nway-merger.cc b/tests/tst-nway-merger.cc

--- a/tests/tst-nway-merger.cc
+++ b/tests/tst-nway-merger.cc
@@ -69,7 +69,7 @@ class my_spsc_ring {

bool empty() const { return _r.empty(); }

private:
- ring_spsc<T, MaxSize> _r;
+ ring_spsc<T,unsigned,MaxSize> _r;
};

typedef my_spsc_ring<my_struct, 8> my_spsc_queue;
diff --git a/tests/tst-ring-spsc-wraparound.cc b/tests/tst-ring-spsc-wraparound.cc