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/schedule.h"
      25                 :            : 
      26                 :            : #include <chrono>
      27                 :            : #include <cstdint>
      28                 :            : #include <memory>
      29                 :            : #include <mutex>
      30                 :            : #include <optional>
      31                 :            : #include <utility>
      32                 :            : #include <vector>
      33                 :            : 
      34                 :            : #include "./base/logger.h"
      35                 :            : #include "./base/memory_resource.h"
      36                 :            : 
      37                 :            : namespace vlink {
      38                 :            : 
      39                 :            : // Schedule::Config
      40                 :         25 : Schedule::Config::Config() = default;
      41                 :            : 
      42                 :         19 : Schedule::Config::Config(uint32_t _delay_ms, uint16_t _priority, uint32_t _schedule_timeout_ms,
      43                 :         19 :                          uint32_t _execution_timeout_ms)
      44                 :         19 :     : delay_ms(_delay_ms),
      45                 :         19 :       priority(_priority),
      46                 :         19 :       schedule_timeout_ms(_schedule_timeout_ms),
      47                 :         19 :       execution_timeout_ms(_execution_timeout_ms) {}
      48                 :            : 
      49                 :            : // Schedule::Status
      50                 :         45 : Schedule::Status::Status() : impl_(MemoryResource::make_shared<Schedule::Status::Impl>()) {
      51                 :         45 :   impl_->is_valid.store(true, std::memory_order_relaxed);
      52                 :         45 : }
      53                 :            : 
      54                 :         75 : Schedule::Status::~Status() { commit(); }
      55                 :            : 
      56                 :         30 : Schedule::Status::Status(Status&& status) noexcept {
      57         [ -  + ]:         30 :   if VUNLIKELY (this == &status) {
      58                 :            :     return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
      59                 :            :   }
      60                 :            : 
      61                 :         30 :   impl_ = std::move(status.impl_);
      62                 :         30 :   launcher_ = std::move(status.launcher_);
      63                 :         30 :   committed_ = status.committed_;
      64                 :            : }
      65                 :            : 
      66                 :          2 : Schedule::Status& Schedule::Status::operator=(Status&& status) noexcept {
      67         [ -  + ]:          2 :   if VUNLIKELY (this == &status) {
      68                 :            :     return *this;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
      69                 :            :   }
      70                 :            : 
      71                 :          2 :   commit();
      72                 :            : 
      73                 :          2 :   impl_ = std::move(status.impl_);
      74                 :          2 :   launcher_ = std::move(status.launcher_);
      75                 :          2 :   committed_ = status.committed_;
      76                 :            : 
      77                 :          2 :   return *this;
      78                 :            : }
      79                 :            : 
      80                 :         86 : void Schedule::Status::commit() noexcept {
      81   [ +  +  +  +  :         86 :   if (!impl_ || committed_) {
                   +  + ]
      82                 :         41 :     return;
      83                 :            :   }
      84                 :            : 
      85                 :         45 :   committed_ = true;
      86                 :            : 
      87         [ +  + ]:         45 :   if VLIKELY (launcher_) {
      88                 :         29 :     launcher_();
      89                 :            :   }
      90                 :            : }
      91                 :            : 
      92                 :          4 : void Schedule::Status::set_valid(bool valid) {
      93         [ +  - ]:          4 :   if VLIKELY (impl_) {
      94                 :          4 :     impl_->is_valid.store(valid, std::memory_order_relaxed);
      95                 :            :   }
      96                 :          4 : }
      97                 :            : 
      98   [ +  -  +  + ]:         48 : bool Schedule::Status::is_valid() const { return impl_ && impl_->is_valid.load(std::memory_order_relaxed); }
      99                 :            : 
     100                 :          9 : bool Schedule::Status::dispatch() {
     101                 :          9 :   commit();
     102                 :            : 
     103                 :          9 :   return is_valid();
     104                 :            : }
     105                 :            : 
     106                 :          3 : Schedule::Status& Schedule::Status::on_execution_timeout(Callback&& callback) {
     107         [ +  - ]:          3 :   if VLIKELY (is_valid()) {
     108         [ +  - ]:          3 :     std::lock_guard lock(impl_->mtx);
     109                 :            : 
     110         [ -  + ]:          3 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     111                 :            :       VLOG_E("Schedule: on_execution_timeout registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     112                 :            :       return *this;                                                                  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     113                 :            :     }
     114                 :            : 
     115         [ -  + ]:          3 :     if VUNLIKELY (impl_->execution_timeout_callback) {
     116                 :            :       VLOG_E("Schedule: Execution timeout callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     117                 :            :       return *this;                                                    // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     118                 :            :     }
     119                 :            : 
     120                 :          3 :     impl_->execution_timeout_callback = std::move(callback);
     121         [ +  - ]:          3 :   }
     122                 :            : 
     123                 :          3 :   return *this;
     124                 :            : }
     125                 :            : 
     126                 :          4 : Schedule::Status& Schedule::Status::on_schedule_timeout(Callback&& callback) {
     127         [ +  - ]:          4 :   if VLIKELY (is_valid()) {
     128         [ +  - ]:          4 :     std::lock_guard lock(impl_->mtx);
     129                 :            : 
     130         [ -  + ]:          4 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     131                 :            :       VLOG_E("Schedule: on_schedule_timeout registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     132                 :            :       return *this;                                                                 // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     133                 :            :     }
     134                 :            : 
     135         [ -  + ]:          4 :     if VUNLIKELY (impl_->schedule_timeout_callback) {
     136                 :            :       VLOG_E("Schedule: Schedule timeout callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     137                 :            :       return *this;                                                   // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     138                 :            :     }
     139                 :            : 
     140                 :          4 :     impl_->schedule_timeout_callback = std::move(callback);
     141         [ +  - ]:          4 :   }
     142                 :            : 
     143                 :          4 :   return *this;
     144                 :            : }
     145                 :            : 
     146                 :          5 : Schedule::Status& Schedule::Status::on_catch(CatchCallback&& callback) {
     147         [ +  - ]:          5 :   if VLIKELY (is_valid()) {
     148         [ +  - ]:          5 :     std::lock_guard lock(impl_->mtx);
     149                 :            : 
     150         [ -  + ]:          5 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     151                 :            :       VLOG_E("Schedule: on_catch registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     152                 :            :       return *this;                                                      // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     153                 :            :     }
     154                 :            : 
     155         [ -  + ]:          5 :     if VUNLIKELY (impl_->catch_callback) {
     156                 :            :       VLOG_E("Schedule: Catch callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     157                 :            :       return *this;                                        // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     158                 :            :     }
     159                 :            : 
     160                 :          5 :     impl_->catch_callback = std::move(callback);
     161         [ +  - ]:          5 :   }
     162                 :            : 
     163                 :          5 :   return *this;
     164                 :            : }
     165                 :            : 
     166                 :            : // Schedule::RetStatus
     167                 :          6 : Schedule::Status& Schedule::RetStatus::on_else(Callback&& callback) {
     168         [ +  - ]:          6 :   if VLIKELY (is_valid()) {
     169         [ +  - ]:          6 :     std::lock_guard lock(impl_->mtx);
     170                 :            : 
     171         [ -  + ]:          6 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     172                 :            :       VLOG_E("Schedule: on_else registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     173                 :            :       return *this;                                                     // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     174                 :            :     }
     175                 :            : 
     176         [ -  + ]:          6 :     if VUNLIKELY (impl_->else_callback) {
     177                 :            :       VLOG_E("Schedule: Else callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     178                 :            :       return *this;                                       // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     179                 :            :     }
     180                 :            : 
     181                 :          6 :     impl_->else_callback = std::move(callback);
     182         [ +  - ]:          6 :   }
     183                 :            : 
     184                 :          6 :   return *this;
     185                 :            : }
     186                 :            : 
     187                 :         15 : Schedule::RetStatus& Schedule::RetStatus::on_then(RetCallback&& callback) {
     188         [ +  - ]:         15 :   if VLIKELY (is_valid()) {
     189         [ +  - ]:         15 :     std::lock_guard lock(impl_->mtx);
     190                 :            : 
     191         [ +  + ]:         15 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     192   [ +  -  +  - ]:          2 :       VLOG_E("Schedule: on_then registered after dispatch; ignored.");
     193                 :          1 :       return *this;
     194                 :            :     }
     195                 :            : 
     196         [ +  - ]:         14 :     impl_->then_callback_list.emplace_back(std::move(callback));
     197         [ +  + ]:         15 :   }
     198                 :            : 
     199                 :         14 :   return *this;
     200                 :            : }
     201                 :            : 
     202                 :            : // Schedule
     203                 :         29 : Schedule::Status Schedule::process(const Config& config, Callback&& callback, Callback& wrapper_callback) {
     204                 :         29 :   RetCallback adapted_callback = [callback = std::move(callback)]() mutable -> bool {
     205         [ +  - ]:         25 :     if VLIKELY (callback) {
     206                 :         25 :       callback();
     207                 :            :     }
     208                 :            : 
     209                 :         24 :     return true;
     210         [ +  - ]:         29 :   };
     211                 :            : 
     212         [ +  - ]:         58 :   return internal_process_with_ret(config, std::move(adapted_callback), wrapper_callback);
     213                 :         29 : }
     214                 :            : 
     215                 :         11 : Schedule::RetStatus Schedule::process_with_ret(const Config& config, RetCallback&& callback,
     216                 :            :                                                Callback& wrapper_callback) {
     217                 :         11 :   return internal_process_with_ret(config, std::move(callback), wrapper_callback);
     218                 :            : }
     219                 :            : 
     220                 :         40 : Schedule::RetStatus Schedule::internal_process_with_ret(const Config& config, RetCallback&& callback,
     221                 :            :                                                         Callback& wrapper_callback) {
     222                 :         40 :   Schedule::RetStatus status;
     223                 :            : 
     224                 :         40 :   wrapper_callback = [callback = std::move(callback), config, impl = status.impl_]() mutable {
     225                 :         36 :     Schedule::Callback schedule_timeout_cb;
     226                 :         36 :     Schedule::Callback execution_timeout_cb;
     227                 :         36 :     Schedule::CatchCallback catch_cb;
     228                 :         36 :     Schedule::Callback else_cb;
     229                 :         36 :     std::chrono::steady_clock::time_point submit_time;
     230                 :         36 :     std::vector<Schedule::RetCallback> then_callbacks;
     231                 :            : 
     232                 :            :     {
     233         [ +  - ]:         36 :       std::lock_guard lock(impl->mtx);
     234                 :            : 
     235         [ -  + ]:         36 :       if VUNLIKELY (!impl->is_valid.load(std::memory_order_relaxed)) {
     236                 :            :         return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     237                 :            :       }
     238                 :            : 
     239                 :         36 :       schedule_timeout_cb = std::move(impl->schedule_timeout_callback);
     240                 :         36 :       execution_timeout_cb = std::move(impl->execution_timeout_callback);
     241                 :         36 :       catch_cb = std::move(impl->catch_callback);
     242                 :         36 :       else_cb = std::move(impl->else_callback);
     243                 :         36 :       submit_time = impl->submit_time;
     244                 :         36 :       then_callbacks = std::move(impl->then_callback_list);
     245                 :            : 
     246                 :         36 :       impl->dispatched.store(true, std::memory_order_release);
     247         [ +  - ]:         36 :     }
     248                 :            : 
     249                 :         36 :     auto now = std::chrono::steady_clock::now();
     250                 :            : 
     251         [ +  + ]:         36 :     if (config.schedule_timeout_ms > 0) {
     252   [ +  -  +  - ]:          3 :       auto wait_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - submit_time).count();
     253                 :          3 :       auto timeout_ms = static_cast<uint64_t>(config.schedule_timeout_ms) + config.delay_ms;
     254                 :            : 
     255         [ +  + ]:          3 :       if (static_cast<uint64_t>(wait_ms) > timeout_ms) {
     256         [ +  - ]:          1 :         if (schedule_timeout_cb) {
     257         [ +  - ]:          1 :           schedule_timeout_cb();
     258                 :            :         }
     259                 :            : 
     260                 :          1 :         return;
     261                 :            :       }
     262                 :            :     }
     263                 :            : 
     264                 :         78 :     auto run_with_timeout = [&catch_cb, &config,
     265                 :         50 :                              &execution_timeout_cb](Schedule::RetCallback& exe_callback) -> std::optional<bool> {
     266         [ -  + ]:         43 :       if VUNLIKELY (!exe_callback) {
     267                 :            :         return std::nullopt;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     268                 :            :       }
     269                 :            : 
     270                 :         43 :       auto exec_start = std::chrono::steady_clock::now();
     271                 :            : 
     272                 :         43 :       bool result = false;
     273                 :            : 
     274                 :            :       try {
     275         [ +  + ]:         43 :         result = exe_callback();
     276         [ -  + ]:          3 :       } catch (std::exception& e) {
     277         [ +  - ]:          3 :         if (catch_cb) {
     278         [ +  - ]:          3 :           catch_cb(e);
     279                 :            : 
     280                 :          3 :           return std::nullopt;
     281                 :            :         }
     282         [ +  - ]:          3 :       }
     283                 :            : 
     284                 :         40 :       auto exec_end = std::chrono::steady_clock::now();
     285                 :            : 
     286         [ +  + ]:         40 :       if (config.execution_timeout_ms > 0) {
     287   [ +  -  +  - ]:          2 :         auto exec_ms = std::chrono::duration_cast<std::chrono::milliseconds>(exec_end - exec_start).count();
     288                 :            : 
     289         [ +  + ]:          2 :         if (static_cast<uint32_t>(exec_ms) > config.execution_timeout_ms) {
     290         [ +  - ]:          1 :           if (execution_timeout_cb) {
     291         [ +  - ]:          1 :             execution_timeout_cb();
     292                 :            :           }
     293                 :            : 
     294                 :          1 :           return std::nullopt;
     295                 :            :         }
     296                 :            :       }
     297                 :            : 
     298                 :         39 :       return result;
     299                 :         35 :     };
     300                 :            : 
     301         [ +  - ]:         35 :     auto main_ret = run_with_timeout(callback);
     302                 :            : 
     303         [ +  + ]:         35 :     if (!main_ret.has_value()) {
     304                 :          4 :       return;
     305                 :            :     }
     306                 :            : 
     307   [ +  -  +  + ]:         31 :     if (!main_ret.value()) {
     308         [ +  - ]:          2 :       if (else_cb) {
     309         [ +  - ]:          2 :         else_cb();
     310                 :            :       }
     311                 :            : 
     312                 :          2 :       return;
     313                 :            :     }
     314                 :            : 
     315         [ +  + ]:         36 :     for (auto& then_callback : then_callbacks) {
     316         [ +  - ]:          8 :       auto then_ret = run_with_timeout(then_callback);
     317                 :            : 
     318         [ -  + ]:          8 :       if (!then_ret.has_value()) {
     319                 :          1 :         return;
     320                 :            :       }
     321                 :            : 
     322   [ +  -  +  + ]:          8 :       if (!then_ret.value()) {
     323         [ -  + ]:          1 :         if (else_cb) {
     324         [ #  # ]:          0 :           else_cb();
     325                 :            :         }
     326                 :            : 
     327                 :          1 :         return;
     328                 :            :       }
     329                 :            :     }
     330   [ +  -  +  +  :        108 :   };
          +  +  +  +  +  
                +  +  + ]
     331                 :            : 
     332                 :         40 :   return status;
     333                 :            : }  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     334                 :            : 
     335                 :            : }  // namespace vlink