Branch data     Line data    Source code

       1                 :            : /*
       2                 :            :  * Copyright (C) 2026 by Thun Lu. All rights reserved.
       3                 :            :  * Author: Thun Lu <thun.lu@zohomail.cn>
       4                 :            :  * Repo:   https://github.com/thun-res/vlink
       5                 :            :  *  _    __   __      _           __
       6                 :            :  * | |  / /  / /     (_) ____    / /__
       7                 :            :  * | | / /  / /     / / / __ \  / //_/
       8                 :            :  * | |/ /  / /___  / / / / / / / ,<
       9                 :            :  * |___/  /_____/ /_/ /_/ /_/ /_/|_|
      10                 :            :  *
      11                 :            :  * Licensed under the Apache License, Version 2.0 (the "License");
      12                 :            :  * you may not use this file except in compliance with the License.
      13                 :            :  * You may obtain a copy of the License at
      14                 :            :  *
      15                 :            :  *     http://www.apache.org/licenses/LICENSE-2.0
      16                 :            :  *
      17                 :            :  * Unless required by applicable law or agreed to in writing, software
      18                 :            :  * distributed under the License is distributed on an "AS IS" BASIS,
      19                 :            :  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      20                 :            :  * See the License for the specific language governing permissions and
      21                 :            :  * limitations under the License.
      22                 :            :  */
      23                 :            : 
      24                 :            : #include "./base/mpmc_queue.h"
      25                 :            : 
      26                 :            : #include <chrono>
      27                 :            : #include <limits>
      28                 :            : #include <mutex>
      29                 :            : #include <new>
      30                 :            : #include <stdexcept>
      31                 :            : 
      32                 :            : #include "./base/utils.h"
      33                 :            : 
      34                 :            : namespace vlink {
      35                 :            : 
      36                 :            : // MpmcQueueBase
      37                 :          3 : size_t MpmcQueueBase::capacity() const noexcept { return capacity_; }
      38                 :            : 
      39                 :         63 : size_t MpmcQueueBase::size(bool real) const noexcept {
      40                 :         63 :   static auto safe_diff = [](size_t h, size_t t) -> size_t {
      41         [ +  - ]:         63 :     return (h >= t) ? (h - t) : 0U;
      42                 :            :   };  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
      43                 :            : 
      44         [ +  + ]:         63 :   if (real) {
      45                 :            :     static constexpr size_t kMaxRetry = 50U;
      46                 :            : 
      47                 :         45 :     size_t retry_cnt = 0;
      48                 :            : 
      49                 :         45 :     size_t t = tail_.load(kMemoryOrderAcquire);
      50                 :         45 :     size_t h = head_.load(kMemoryOrderAcquire);
      51                 :            : 
      52         [ +  - ]:         45 :     while (retry_cnt < kMaxRetry) {
      53                 :         45 :       size_t t2 = tail_.load(kMemoryOrderAcquire);
      54                 :            : 
      55         [ +  - ]:         45 :       if (t == t2) {
      56                 :         45 :         return safe_diff(h, t);
      57                 :            :       }
      58                 :            : 
      59                 :            :       // LCOV_EXCL_START GCOVR_EXCL_START
      60                 :            :       t = t2;
      61                 :            :       h = head_.load(kMemoryOrderAcquire);
      62                 :            :       retry_cnt++;
      63                 :            : 
      64                 :            :       Utils::yield_cpu();
      65                 :            :     }
      66                 :            : 
      67                 :            :     return safe_diff(h, t);
      68                 :            :     // LCOV_EXCL_STOP GCOVR_EXCL_STOP
      69                 :            :   }
      70                 :            : 
      71                 :         18 :   auto h = head_.load(kMemoryOrderAcquire);
      72                 :         18 :   auto t = tail_.load(kMemoryOrderAcquire);
      73                 :            : 
      74                 :         18 :   return safe_diff(h, t);
      75                 :            : }
      76                 :            : 
      77                 :         36 : bool MpmcQueueBase::empty(bool real) const noexcept { return size(real) == 0; }
      78                 :            : 
      79                 :         23 : bool MpmcQueueBase::is_full(bool real) const noexcept { return size(real) >= capacity_; }
      80                 :            : 
      81                 :         10 : bool MpmcQueueBase::wait_not_empty(std::chrono::milliseconds timeout) noexcept {
      82         [ +  + ]:         10 :   if (!empty(true)) {
      83                 :          3 :     return !quit_flag_.value.load(kMemoryOrderAcquire);
      84                 :            :   }
      85                 :            : 
      86                 :          7 :   std::unique_lock lock(cv_mtx_);
      87                 :            : 
      88                 :          7 :   bool ret = true;
      89                 :            : 
      90         [ +  + ]:          7 :   if (timeout == std::chrono::milliseconds(0)) {
      91   [ +  +  +  + ]:         18 :     cv_not_empty_.wait(lock, [this]() { return !empty(true) || quit_flag_.value.load(kMemoryOrderAcquire); });
      92                 :            :   } else {
      93                 :          1 :     ret = cv_not_empty_.wait_for(lock, timeout, [this]() {
      94   [ +  -  -  + ]:          2 :       return !empty(true) || quit_flag_.value.load(kMemoryOrderAcquire);
      95                 :            :     });  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
      96                 :            :   }
      97                 :            : 
      98         [ +  + ]:          7 :   if VUNLIKELY (quit_flag_.value.load(kMemoryOrderAcquire)) {
      99                 :          5 :     return false;
     100                 :            :   }
     101                 :            : 
     102                 :          2 :   return ret;
     103                 :          7 : }
     104                 :            : 
     105                 :          7 : bool MpmcQueueBase::wait_not_full(std::chrono::milliseconds timeout) noexcept {
     106         [ +  + ]:          7 :   if (!is_full(true)) {
     107                 :          4 :     return !quit_flag_.value.load(kMemoryOrderAcquire);
     108                 :            :   }
     109                 :            : 
     110                 :          3 :   std::unique_lock lock(cv_mtx_);
     111                 :            : 
     112                 :          3 :   bool ret = true;
     113                 :            : 
     114         [ +  + ]:          3 :   if (timeout == std::chrono::milliseconds(0)) {
     115   [ +  +  +  + ]:          6 :     cv_not_full_.wait(lock, [this]() { return !is_full(true) || quit_flag_.value.load(kMemoryOrderAcquire); });
     116                 :            :   } else {
     117                 :          1 :     ret = cv_not_full_.wait_for(lock, timeout, [this]() {
     118   [ +  -  -  + ]:          2 :       return !is_full(true) || quit_flag_.value.load(kMemoryOrderAcquire);
     119                 :            :     });  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     120                 :            :   }
     121                 :            : 
     122         [ +  + ]:          3 :   if VUNLIKELY (quit_flag_.value.load(kMemoryOrderAcquire)) {
     123                 :          1 :     return false;
     124                 :            :   }
     125                 :            : 
     126                 :          2 :   return ret;
     127                 :          3 : }
     128                 :            : 
     129                 :         43 : void MpmcQueueBase::notify_to_quit() noexcept {
     130                 :         43 :   std::lock_guard lock(cv_mtx_);
     131                 :            : 
     132                 :         43 :   quit_flag_.value.store(true, kMemoryOrderRelease);
     133                 :            : 
     134                 :         43 :   cv_not_empty_.notify_all();
     135                 :         43 :   cv_not_full_.notify_all();
     136                 :         43 : }
     137                 :            : 
     138                 :         69 : MpmcQueueBase::MpmcQueueBase(size_t capacity) : capacity_(capacity) {
     139   [ +  +  +  +  :         69 :   if VUNLIKELY (capacity_ < 1U || capacity_ == std::numeric_limits<size_t>::max()) {
                   +  + ]
     140                 :          2 :     throw_mpmc_invalid_capacity();
     141                 :            :   }
     142                 :         71 : }
     143                 :            : 
     144         [ +  - ]:          2 : void MpmcQueueBase::throw_mpmc_invalid_capacity() { throw std::invalid_argument("capacity must be in [1, SIZE_MAX)"); }
     145                 :            : 
     146                 :          1 : void MpmcQueueBase::throw_mpmc_alignment_failure() { throw std::bad_alloc(); }
     147                 :            : 
     148                 :            : }  // namespace vlink