From 9eedfb9baa420c7897d5f053cfba6e571ca73550 Mon Sep 17 00:00:00 2001
From: John Spray <jcs@vectorized.io>
Date: Tue, 31 Aug 2021 14:00:57 +0100
Subject: [PATCH] core: use 64 bit head+tail pointers in fair_queue

When uint32_t offsets were used here, the frequent
integer wrapping caused queues to get stuck.  The _pending
tail values would compare as greater than the head values
forever if there weren't always new requests coming through
to wrap the head/tail pointers again.

Signed-off-by: John Spray <jcs@vectorized.io>
---
 include/seastar/core/fair_queue.hh | 7 +++----
 src/core/fair_queue.cc             | 9 ++++++---
 2 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/include/seastar/core/fair_queue.hh b/include/seastar/core/fair_queue.hh
index 7c4155534cb..d3f46784dd7 100644
--- a/include/seastar/core/fair_queue.hh
+++ b/include/seastar/core/fair_queue.hh
@@ -93,11 +93,11 @@ public:
 };
 
 class fair_group_rover {
-    uint32_t _weight = 0;
-    uint32_t _size = 0;
+    uint64_t _weight = 0;
+    uint64_t _size = 0;
 
 public:
-    fair_group_rover(uint32_t weight, uint32_t size) noexcept;
+    fair_group_rover(uint64_t weight, uint64_t size) noexcept;
 
     /*
      * For both dimentions checks if the current rover is ahead of the
@@ -175,7 +175,6 @@ using priority_class_ptr = lw_shared_ptr<priority_class>;
 /// the given time frame exceeds the disk throughput.
 class fair_group {
     using fair_group_atomic_rover = std::atomic<fair_group_rover>;
-    static_assert(fair_group_atomic_rover::is_always_lock_free);
 
     fair_group_atomic_rover _capacity_tail;
     fair_group_atomic_rover _capacity_head;
diff --git a/src/core/fair_queue.cc b/src/core/fair_queue.cc
index 44c72dedbef..95160dc7d7d 100644
--- a/src/core/fair_queue.cc
+++ b/src/core/fair_queue.cc
@@ -81,14 +81,17 @@ std::ostream& operator<<(std::ostream& os, fair_queue_ticket t) {
     return os << t._weight << ":" << t._size;
 }
 
-fair_group_rover::fair_group_rover(uint32_t weight, uint32_t size) noexcept
+fair_group_rover::fair_group_rover(uint64_t weight, uint64_t size) noexcept
         : _weight(weight)
         , _size(size)
 {}
 
 fair_queue_ticket fair_group_rover::maybe_ahead_of(const fair_group_rover& other) const noexcept {
-    return fair_queue_ticket(std::max<int32_t>(_weight - other._weight, 0),
-            std::max<int32_t>(_size - other._size, 0));
+
+    uint64_t weight_d = std::clamp<int64_t>(_weight - other._weight, 0, std::numeric_limits<int32_t>::max());
+    uint64_t size_d = std::clamp<int64_t>(_size - other._size, 0, std::numeric_limits<int32_t>::max());
+
+    return fair_queue_ticket(weight_d, size_d);
 }
 
 fair_group_rover fair_group_rover::operator+(fair_queue_ticket t) const noexcept {