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 "./impl/ack_manager.h"
      25                 :            : 
      26                 :            : #include <memory>
      27                 :            : #include <mutex>
      28                 :            : 
      29                 :            : namespace vlink {
      30                 :            : 
      31                 :            : // AckManager
      32                 :        148 : AckManager::AckManager() noexcept = default;
      33                 :            : 
      34                 :        148 : AckManager::~AckManager() noexcept = default;
      35                 :            : 
      36                 :        103 : AckManager::RequestPtr AckManager::create_request() noexcept {
      37                 :        103 :   std::unique_lock manager_lock(mtx_);
      38                 :            : 
      39                 :        103 :   auto request = std::make_shared<Request>();
      40                 :            : 
      41                 :        103 :   request->seq = request_seq_++;
      42                 :        103 :   request->generation = generation_;
      43                 :            : 
      44                 :        103 :   return request;
      45                 :        103 : }
      46                 :            : 
      47                 :         94 : bool AckManager::process(RequestPtr request, int ms, ProcessCallback&& process_callback) noexcept {
      48                 :            :   {
      49                 :         94 :     std::lock_guard manager_lock(mtx_);
      50                 :            : 
      51   [ +  +  -  +  :         94 :     if VUNLIKELY (is_interrupted_ || request->generation != generation_) {
                   +  + ]
      52                 :          1 :       return false;
      53                 :            :     }
      54                 :            : 
      55                 :         93 :     request_set_.emplace(request);
      56         [ +  + ]:         94 :   }
      57                 :            : 
      58         [ +  - ]:         93 :   if VLIKELY (process_callback) {
      59         [ +  + ]:         93 :     if VUNLIKELY (!process_callback()) {
      60                 :          3 :       remove(request);
      61                 :          3 :       return false;
      62                 :            :     }
      63                 :            :   } else {
      64                 :          0 :     remove(request);
      65                 :          0 :     return false;
      66                 :            :   }
      67                 :            : 
      68                 :         90 :   std::unique_lock lock(request->mtx);
      69                 :            : 
      70                 :        146 :   auto predicate = [&request]() -> bool { return request->status != Request::Status::kPending; };
      71                 :            : 
      72         [ +  + ]:         90 :   if VUNLIKELY (ms < 0) {
      73                 :          4 :     request->cv.wait(lock, predicate);
      74                 :            :   } else {
      75                 :         86 :     request->cv.wait_for(lock, std::chrono::milliseconds(ms), predicate);
      76                 :            :   }
      77                 :            : 
      78         [ +  + ]:         90 :   if (request->status == Request::Status::kPending) {
      79                 :         12 :     std::lock_guard manager_lock(mtx_);
      80                 :         12 :     request_set_.erase(request);
      81                 :         12 :     request->status = Request::Status::kCancelled;
      82                 :         12 :   }
      83                 :            : 
      84                 :         90 :   return request->status == Request::Status::kAcknowledged;
      85                 :         90 : }
      86                 :            : 
      87                 :         78 : bool AckManager::notify(RequestPtr request, NotifyCallback&& notify_callback) noexcept {
      88         [ +  + ]:         78 :   if VUNLIKELY (!request) {
      89                 :          1 :     return false;
      90                 :            :   }
      91                 :            : 
      92                 :         77 :   std::lock_guard lock(request->mtx);
      93                 :            : 
      94                 :            :   {
      95                 :         77 :     std::lock_guard manager_lock(mtx_);
      96                 :            : 
      97         [ +  + ]:         77 :     if VUNLIKELY (request_set_.erase(request) == 0) {
      98                 :          3 :       return false;
      99                 :            :     }
     100         [ +  + ]:         77 :   }
     101                 :            : 
     102         [ +  + ]:         74 :   if VLIKELY (notify_callback) {
     103                 :         64 :     notify_callback();
     104                 :            :   }
     105                 :            : 
     106                 :         74 :   request->status = Request::Status::kAcknowledged;
     107                 :         74 :   request->cv.notify_one();
     108                 :            : 
     109                 :         74 :   return true;
     110                 :         77 : }
     111                 :            : 
     112                 :          6 : bool AckManager::remove(RequestPtr request) noexcept {
     113         [ +  + ]:          6 :   if VUNLIKELY (!request) {
     114                 :          1 :     return false;
     115                 :            :   }
     116                 :            : 
     117                 :          5 :   std::lock_guard lock(request->mtx);
     118                 :            : 
     119                 :            :   {
     120                 :          5 :     std::lock_guard manager_lock(mtx_);
     121                 :            : 
     122         [ +  + ]:          5 :     if (request_set_.erase(request) == 0) {
     123                 :          2 :       return false;
     124                 :            :     }
     125         [ +  + ]:          5 :   }
     126                 :            : 
     127                 :          3 :   request->status = Request::Status::kCancelled;
     128                 :          3 :   request->cv.notify_one();
     129                 :            : 
     130                 :          3 :   return true;
     131                 :          5 : }
     132                 :            : 
     133                 :        122 : void AckManager::clear() noexcept {
     134                 :        122 :   decltype(request_set_) temp_set;
     135                 :            : 
     136                 :            :   {
     137                 :        122 :     std::lock_guard manager_lock(mtx_);
     138                 :            : 
     139                 :        122 :     is_interrupted_ = true;
     140                 :        122 :     ++generation_;
     141                 :            : 
     142                 :        122 :     temp_set.swap(request_set_);
     143                 :        122 :   }
     144                 :            : 
     145         [ +  + ]:        126 :   for (const auto& request : temp_set) {
     146                 :          4 :     std::lock_guard lock(request->mtx);
     147                 :          4 :     request->status = Request::Status::kCancelled;
     148                 :          4 :     request->cv.notify_all();
     149                 :          4 :   }
     150                 :        122 : }
     151                 :            : 
     152                 :         82 : void AckManager::reset_interrupted() noexcept {
     153                 :         82 :   std::lock_guard manager_lock(mtx_);
     154                 :            : 
     155                 :         82 :   is_interrupted_ = false;
     156                 :         82 : }
     157                 :            : 
     158                 :            : }  // namespace vlink