1 /* 2 * Copyright (c) 2021-2022 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef PANDA_LIBPANDABASE_MEM_RING_BUF_LOCK_FREE_BUFFER 17 #define PANDA_LIBPANDABASE_MEM_RING_BUF_LOCK_FREE_BUFFER 18 19 #include <thread> 20 #include <atomic> 21 #include <cinttypes> 22 #include "libpandabase/utils/math_helpers.h" 23 24 namespace panda::mem { 25 /** 26 * Lock-free single-producer single-consumer ring-buffer. Push can take infinite amount of time if buffer is full. 27 */ 28 template <typename T, size_t RING_BUFFER_SIZE> 29 class LockFreeBuffer { 30 public: 31 static_assert(RING_BUFFER_SIZE > 0U, "0 is invalid size for ring buffer"); 32 static_assert(panda::helpers::math::IsPowerOfTwo(RING_BUFFER_SIZE)); 33 static constexpr size_t RING_BUFFER_SIZE_MASK = RING_BUFFER_SIZE - 1; 34 static_assert(RING_BUFFER_SIZE_MASK > 0); 35 36 // NOLINTNEXTLINE(cppcoreguidelines-pro-type-member-init) LockFreeBuffer()37 LockFreeBuffer() 38 { 39 // Atomic with release order reason: threads should see correct initialization 40 tail_index_.store(0, std::memory_order_release); 41 // Atomic with release order reason: threads should see correct initialization 42 head_index_.store(0, std::memory_order_release); 43 CheckInvariant(); 44 } 45 TryPush(const T val)46 bool TryPush(const T val) 47 { 48 CheckInvariant(); 49 // Atomic with acquire order reason: push should get the latest value 50 const auto current_tail = tail_index_.load(std::memory_order_acquire); 51 const auto next_tail = Increment(current_tail); 52 // Atomic with acquire order reason: push should get the latest value 53 auto local_head = head_index_.load(std::memory_order_acquire); 54 if (next_tail != local_head) { 55 ASSERT(current_tail < RING_BUFFER_SIZE); 56 buffer_[current_tail] = val; 57 // Atomic with release order reason: to allow pop to see the latest value 58 tail_index_.store(next_tail, std::memory_order_release); 59 return true; 60 } 61 return false; 62 } 63 Push(const T val)64 void Push(const T val) 65 { 66 // NOLINTNEXTLINE(readability-braces-around-statements) 67 while (!TryPush(val)) { 68 }; 69 } 70 IsEmpty()71 bool IsEmpty() 72 { 73 CheckInvariant(); 74 // Atomic with acquire order reason: get the latest value 75 auto local_head = head_index_.load(std::memory_order_acquire); 76 // Atomic with acquire order reason: get the latest value 77 auto local_tail = tail_index_.load(std::memory_order_acquire); 78 bool is_empty = (local_head == local_tail); 79 return is_empty; 80 } 81 TryPop(T * pval)82 bool TryPop(T *pval) 83 { 84 CheckInvariant(); 85 86 // Atomic with acquire order reason: get the latest value 87 auto currentHead = head_index_.load(std::memory_order_acquire); 88 // Atomic with acquire order reason: get the latest value 89 if (currentHead == tail_index_.load(std::memory_order_acquire)) { 90 return false; 91 } 92 93 *pval = buffer_[currentHead]; 94 size_t new_value = Increment(currentHead); 95 // Atomic with release order reason: let others threads to see the latest value 96 head_index_.store(new_value, std::memory_order_release); 97 return true; 98 } 99 Pop()100 T Pop() 101 { 102 T ret; 103 // NOLINTNEXTLINE(readability-braces-around-statements) 104 while (!TryPop(&ret)) { 105 } 106 return ret; 107 } 108 109 private: 110 std::atomic<size_t> tail_index_; 111 std::atomic<size_t> head_index_; 112 std::array<T, RING_BUFFER_SIZE> buffer_; 113 Increment(size_t n)114 size_t Increment(size_t n) 115 { 116 return (n + 1) & RING_BUFFER_SIZE_MASK; 117 } 118 CheckInvariant()119 void CheckInvariant() 120 { 121 // Atomic with acquire order reason: get the latest value 122 [[maybe_unused]] auto local_head = head_index_.load(std::memory_order_acquire); 123 ASSERT(local_head < RING_BUFFER_SIZE); 124 125 // Atomic with acquire order reason: get the latest value 126 [[maybe_unused]] auto local_tail = tail_index_.load(std::memory_order_acquire); 127 ASSERT(local_tail < RING_BUFFER_SIZE); 128 } 129 }; 130 } // namespace panda::mem 131 #endif 132