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 <mutex>
      28                 :            : #include <new>
      29                 :            : #include <stdexcept>
      30                 :            : 
      31                 :            : #include "./base/utils.h"
      32                 :            : 
      33                 :            : namespace vlink {
      34                 :            : 
      35                 :            : // MpmcQueueBase
      36                 :         56 : MpmcQueueBase::MpmcQueueBase(size_t capacity) : capacity_(capacity) {
      37         [ +  + ]:         56 :   if VUNLIKELY (capacity_ < 1U) {
      38                 :          1 :     throw_mpmc_invalid_capacity();
      39                 :            :   }
      40                 :         57 : }
      41                 :            : 
      42         [ +  - ]:          1 : void MpmcQueueBase::throw_mpmc_invalid_capacity() { throw std::invalid_argument("capacity < 1U"); }
      43                 :            : 
      44                 :          1 : void MpmcQueueBase::throw_mpmc_alignment_failure() { throw std::bad_alloc(); }
      45                 :            : 
      46                 :          3 : size_t MpmcQueueBase::capacity() const noexcept { return capacity_; }
      47                 :            : 
      48                 :         57 : size_t MpmcQueueBase::size(bool real) const noexcept {
      49                 :         57 :   static auto safe_diff = [](size_t h, size_t t) -> size_t {
      50         [ +  - ]:         57 :     return (h >= t) ? (h - t) : 0U;
      51                 :            :   };  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
      52                 :            : 
      53         [ +  + ]:         57 :   if (real) {
      54                 :            :     static constexpr size_t kMaxRetry = 50U;
      55                 :            : 
      56                 :         39 :     size_t retry_cnt = 0;
      57                 :            : 
      58                 :         39 :     size_t t = tail_.load(kMemoryOrderAcquire);
      59                 :         39 :     size_t h = head_.load(kMemoryOrderAcquire);
      60                 :            : 
      61         [ +  - ]:         39 :     while (retry_cnt < kMaxRetry) {
      62                 :         39 :       size_t t2 = tail_.load(kMemoryOrderAcquire);
      63                 :            : 
      64         [ +  - ]:         39 :       if (t == t2) {
      65                 :         39 :         return safe_diff(h, t);
      66                 :            :       }
      67                 :            : 
      68                 :            :       // LCOV_EXCL_START GCOVR_EXCL_START
      69                 :            :       t = t2;
      70                 :            :       h = head_.load(kMemoryOrderAcquire);
      71                 :            :       retry_cnt++;
      72                 :            : 
      73                 :            :       Utils::yield_cpu();
      74                 :            :     }
      75                 :            : 
      76                 :            :     return safe_diff(h, t);
      77                 :            :     // LCOV_EXCL_STOP GCOVR_EXCL_STOP
      78                 :            :   }
      79                 :            : 
      80                 :         18 :   auto h = head_.load(kMemoryOrderAcquire);
      81                 :         18 :   auto t = tail_.load(kMemoryOrderAcquire);
      82                 :            : 
      83                 :         18 :   return safe_diff(h, t);
      84                 :            : }
      85                 :            : 
      86                 :         33 : bool MpmcQueueBase::empty(bool real) const noexcept { return size(real) == 0; }
      87                 :            : 
      88                 :         20 : bool MpmcQueueBase::is_full(bool real) const noexcept { return size(real) >= capacity_; }
      89                 :            : 
      90                 :          7 : bool MpmcQueueBase::wait_not_empty(std::chrono::milliseconds timeout) noexcept {
      91         [ -  + ]:          7 :   if (!empty(true)) {
      92                 :          0 :     return true;
      93                 :            :   }
      94                 :            : 
      95                 :          7 :   std::unique_lock lock(cv_mtx_);
      96                 :            : 
      97                 :          7 :   bool ret = true;
      98                 :            : 
      99         [ +  + ]:          7 :   if (timeout == std::chrono::milliseconds(0)) {
     100   [ +  +  +  + ]:         18 :     cv_not_empty_.wait(lock, [this]() { return !empty(true) || quit_flag_.value.load(kMemoryOrderAcquire); });
     101                 :            :   } else {
     102                 :          1 :     ret = cv_not_empty_.wait_for(lock, timeout, [this]() {
     103   [ +  -  -  + ]:          2 :       return !empty(true) || quit_flag_.value.load(kMemoryOrderAcquire);
     104                 :            :     });  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     105                 :            :   }
     106                 :            : 
     107         [ +  + ]:          7 :   if VUNLIKELY (quit_flag_.value.load(kMemoryOrderAcquire)) {
     108                 :          5 :     return false;
     109                 :            :   }
     110                 :            : 
     111                 :          2 :   return ret;
     112                 :          7 : }
     113                 :            : 
     114                 :          4 : bool MpmcQueueBase::wait_not_full(std::chrono::milliseconds timeout) noexcept {
     115         [ +  + ]:          4 :   if (!is_full(true)) {
     116                 :          1 :     return true;
     117                 :            :   }
     118                 :            : 
     119                 :          3 :   std::unique_lock lock(cv_mtx_);
     120                 :            : 
     121                 :          3 :   bool ret = true;
     122                 :            : 
     123         [ +  + ]:          3 :   if (timeout == std::chrono::milliseconds(0)) {
     124   [ +  +  +  + ]:          6 :     cv_not_full_.wait(lock, [this]() { return !is_full(true) || quit_flag_.value.load(kMemoryOrderAcquire); });
     125                 :            :   } else {
     126                 :          1 :     ret = cv_not_full_.wait_for(lock, timeout, [this]() {
     127   [ +  -  -  + ]:          2 :       return !is_full(true) || quit_flag_.value.load(kMemoryOrderAcquire);
     128                 :            :     });  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     129                 :            :   }
     130                 :            : 
     131         [ +  + ]:          3 :   if VUNLIKELY (quit_flag_.value.load(kMemoryOrderAcquire)) {
     132                 :          1 :     return false;
     133                 :            :   }
     134                 :            : 
     135                 :          2 :   return ret;
     136                 :          3 : }
     137                 :            : 
     138                 :         16 : void MpmcQueueBase::notify_to_quit() noexcept {
     139                 :         16 :   std::lock_guard lock(cv_mtx_);
     140                 :            : 
     141                 :         16 :   quit_flag_.value.store(true, kMemoryOrderRelease);
     142                 :            : 
     143                 :         16 :   cv_not_empty_.notify_all();
     144                 :         16 :   cv_not_full_.notify_all();
     145                 :         16 : }
     146                 :            : 
     147                 :            : }  // namespace vlink