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                 :         26 : 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                 :         46 : Schedule::Status::Status() : impl_(MemoryResource::make_shared<Schedule::Status::Impl>()) {
      51                 :         46 :   impl_->is_valid.store(true, std::memory_order_relaxed);
      52                 :         46 : }
      53                 :            : 
      54                 :         76 : 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                 :          4 : void Schedule::Status::set_valid(bool valid) {
      81         [ +  - ]:          4 :   if VLIKELY (impl_) {
      82                 :          4 :     impl_->is_valid.store(valid, std::memory_order_relaxed);
      83                 :            :   }
      84                 :          4 : }
      85                 :            : 
      86   [ +  -  +  + ]:         50 : bool Schedule::Status::is_valid() const { return impl_ && impl_->is_valid.load(std::memory_order_relaxed); }
      87                 :            : 
      88                 :          9 : bool Schedule::Status::dispatch() {
      89                 :          9 :   commit();
      90                 :            : 
      91                 :          9 :   return is_valid();
      92                 :            : }
      93                 :            : 
      94                 :          4 : Schedule::Status& Schedule::Status::on_schedule_timeout(Callback&& callback) {
      95         [ +  - ]:          4 :   if VLIKELY (is_valid()) {
      96         [ +  - ]:          4 :     std::lock_guard lock(impl_->mtx);
      97                 :            : 
      98         [ -  + ]:          4 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
      99                 :            :       VLOG_E("Schedule: on_schedule_timeout registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     100                 :            :       return *this;                                                                 // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     101                 :            :     }
     102                 :            : 
     103         [ -  + ]:          4 :     if VUNLIKELY (impl_->schedule_timeout_callback) {
     104                 :            :       VLOG_E("Schedule: Schedule timeout callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     105                 :            :       return *this;                                                   // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     106                 :            :     }
     107                 :            : 
     108                 :          4 :     impl_->schedule_timeout_callback = std::move(callback);
     109         [ +  - ]:          4 :   }
     110                 :            : 
     111                 :          4 :   return *this;
     112                 :            : }
     113                 :            : 
     114                 :          3 : Schedule::Status& Schedule::Status::on_execution_timeout(Callback&& callback) {
     115         [ +  - ]:          3 :   if VLIKELY (is_valid()) {
     116         [ +  - ]:          3 :     std::lock_guard lock(impl_->mtx);
     117                 :            : 
     118         [ -  + ]:          3 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     119                 :            :       VLOG_E("Schedule: on_execution_timeout registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     120                 :            :       return *this;                                                                  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     121                 :            :     }
     122                 :            : 
     123         [ -  + ]:          3 :     if VUNLIKELY (impl_->execution_timeout_callback) {
     124                 :            :       VLOG_E("Schedule: Execution timeout callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     125                 :            :       return *this;                                                    // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     126                 :            :     }
     127                 :            : 
     128                 :          3 :     impl_->execution_timeout_callback = std::move(callback);
     129         [ +  - ]:          3 :   }
     130                 :            : 
     131                 :          3 :   return *this;
     132                 :            : }
     133                 :            : 
     134                 :          5 : Schedule::Status& Schedule::Status::on_catch(CatchCallback&& callback) {
     135         [ +  - ]:          5 :   if VLIKELY (is_valid()) {
     136         [ +  - ]:          5 :     std::lock_guard lock(impl_->mtx);
     137                 :            : 
     138         [ -  + ]:          5 :     if VUNLIKELY (impl_->dispatched.load(std::memory_order_acquire)) {
     139                 :            :       VLOG_E("Schedule: on_catch registered after dispatch; ignored.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     140                 :            :       return *this;                                                      // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     141                 :            :     }
     142                 :            : 
     143         [ -  + ]:          5 :     if VUNLIKELY (impl_->catch_callback) {
     144                 :            :       VLOG_E("Schedule: Catch callback is already set.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     145                 :            :       return *this;                                        // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     146                 :            :     }
     147                 :            : 
     148                 :          5 :     impl_->catch_callback = std::move(callback);
     149         [ +  - ]:          5 :   }
     150                 :            : 
     151                 :          5 :   return *this;
     152                 :            : }
     153                 :            : 
     154                 :         87 : void Schedule::Status::commit() noexcept {
     155   [ +  +  +  +  :         87 :   if (!impl_ || committed_) {
                   +  + ]
     156                 :         41 :     return;
     157                 :            :   }
     158                 :            : 
     159                 :         46 :   committed_ = true;
     160                 :            : 
     161         [ +  + ]:         46 :   if VLIKELY (launcher_) {
     162                 :         29 :     launcher_();
     163                 :            :   }
     164                 :            : }
     165                 :            : 
     166                 :            : // Schedule::RetStatus
     167                 :          7 : Schedule::Status& Schedule::RetStatus::on_else(Callback&& callback) {
     168         [ +  - ]:          7 :   if VLIKELY (is_valid()) {
     169         [ +  - ]:          7 :     std::lock_guard lock(impl_->mtx);
     170                 :            : 
     171         [ -  + ]:          7 :     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         [ -  + ]:          7 :     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                 :          7 :     impl_->else_callback = std::move(callback);
     182         [ +  - ]:          7 :   }
     183                 :            : 
     184                 :          7 :   return *this;
     185                 :            : }
     186                 :            : 
     187                 :         16 : Schedule::RetStatus& Schedule::RetStatus::on_then(RetCallback&& callback) {
     188         [ +  - ]:         16 :   if VLIKELY (is_valid()) {
     189         [ +  - ]:         16 :     std::lock_guard lock(impl_->mtx);
     190                 :            : 
     191         [ +  + ]:         16 :     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         [ +  - ]:         15 :     impl_->then_callback_list.emplace_back(std::move(callback));
     197         [ +  + ]:         16 :   }
     198                 :            : 
     199                 :         15 :   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                 :         12 : Schedule::RetStatus Schedule::process_with_ret(const Config& config, RetCallback&& callback,
     216                 :            :                                                Callback& wrapper_callback) {
     217                 :         12 :   return internal_process_with_ret(config, std::move(callback), wrapper_callback);
     218                 :            : }
     219                 :            : 
     220                 :         41 : Schedule::RetStatus Schedule::internal_process_with_ret(const Config& config, RetCallback&& callback,
     221                 :            :                                                         Callback& wrapper_callback) {
     222                 :         41 :   Schedule::RetStatus status;
     223                 :            : 
     224                 :         41 :   wrapper_callback = [callback = std::move(callback), config, impl = status.impl_]() mutable {
     225                 :         37 :     Schedule::Callback schedule_timeout_cb;
     226                 :         37 :     Schedule::Callback execution_timeout_cb;
     227                 :         37 :     Schedule::CatchCallback catch_cb;
     228                 :         37 :     Schedule::Callback else_cb;
     229                 :         37 :     std::chrono::steady_clock::time_point submit_time;
     230                 :         37 :     std::vector<Schedule::RetCallback> then_callbacks;
     231                 :            : 
     232                 :            :     {
     233         [ +  - ]:         37 :       std::lock_guard lock(impl->mtx);
     234                 :            : 
     235         [ -  + ]:         37 :       if VUNLIKELY (!impl->is_valid.load(std::memory_order_relaxed)) {
     236                 :            :         return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     237                 :            :       }
     238                 :            : 
     239                 :         37 :       schedule_timeout_cb = std::move(impl->schedule_timeout_callback);
     240                 :         37 :       execution_timeout_cb = std::move(impl->execution_timeout_callback);
     241                 :         37 :       catch_cb = std::move(impl->catch_callback);
     242                 :         37 :       else_cb = std::move(impl->else_callback);
     243                 :         37 :       submit_time = impl->submit_time;
     244                 :         37 :       then_callbacks = std::move(impl->then_callback_list);
     245                 :            : 
     246                 :         37 :       impl->dispatched.store(true, std::memory_order_release);
     247         [ +  - ]:         37 :     }
     248                 :            : 
     249         [ +  + ]:         37 :     if (config.schedule_timeout_ms > 0) {
     250                 :          3 :       auto now = std::chrono::steady_clock::now();
     251   [ +  -  +  - ]:          3 :       auto wait_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - submit_time).count();
     252                 :          3 :       auto timeout_ms = static_cast<uint64_t>(config.schedule_timeout_ms) + config.delay_ms;
     253                 :            : 
     254         [ +  + ]:          3 :       if (static_cast<uint64_t>(wait_ms) > timeout_ms) {
     255         [ +  - ]:          1 :         if (schedule_timeout_cb) {
     256         [ +  - ]:          1 :           schedule_timeout_cb();
     257                 :            :         }
     258                 :            : 
     259                 :          1 :         return;
     260                 :            :       }
     261                 :            :     }
     262                 :            : 
     263                 :         80 :     auto run_with_timeout = [&catch_cb, &config,
     264                 :         95 :                              &execution_timeout_cb](Schedule::RetCallback& exe_callback) -> std::optional<bool> {
     265         [ -  + ]:         44 :       if VUNLIKELY (!exe_callback) {
     266                 :            :         return std::nullopt;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     267                 :            :       }
     268                 :            : 
     269                 :         44 :       std::chrono::steady_clock::time_point exec_start;
     270                 :            : 
     271         [ +  + ]:         44 :       if (config.execution_timeout_ms > 0) {
     272                 :          2 :         exec_start = std::chrono::steady_clock::now();
     273                 :            :       }
     274                 :            : 
     275                 :         44 :       bool result = false;
     276                 :            : 
     277                 :            :       try {
     278         [ +  + ]:         44 :         result = exe_callback();
     279         [ -  + ]:          4 :       } catch (std::exception& e) {
     280         [ +  + ]:          4 :         if (catch_cb) {
     281         [ +  - ]:          3 :           catch_cb(e);
     282                 :            :         } else {
     283   [ +  -  +  -  :          2 :           CLOG_E("Schedule: Task threw an exception and no on_catch handler is registered: %s.", e.what());
                   +  - ]
     284                 :            :         }
     285                 :            : 
     286                 :          4 :         return std::nullopt;
     287                 :          4 :       }
     288                 :            : 
     289         [ +  + ]:         40 :       if (config.execution_timeout_ms > 0) {
     290                 :          2 :         auto exec_end = std::chrono::steady_clock::now();
     291   [ +  -  +  - ]:          2 :         auto exec_ms = std::chrono::duration_cast<std::chrono::milliseconds>(exec_end - exec_start).count();
     292                 :            : 
     293         [ +  + ]:          2 :         if (static_cast<uint32_t>(exec_ms) > config.execution_timeout_ms) {
     294         [ +  - ]:          1 :           if (execution_timeout_cb) {
     295         [ +  - ]:          1 :             execution_timeout_cb();
     296                 :            :           }
     297                 :            : 
     298                 :          1 :           return std::nullopt;
     299                 :            :         }
     300                 :            :       }
     301                 :            : 
     302                 :         39 :       return result;
     303                 :         36 :     };
     304                 :            : 
     305         [ +  - ]:         36 :     auto main_ret = run_with_timeout(callback);
     306                 :            : 
     307         [ +  + ]:         36 :     if (!main_ret.has_value()) {
     308                 :          5 :       return;
     309                 :            :     }
     310                 :            : 
     311   [ +  -  +  + ]:         31 :     if (!main_ret.value()) {
     312         [ +  - ]:          2 :       if (else_cb) {
     313         [ +  - ]:          2 :         else_cb();
     314                 :            :       }
     315                 :            : 
     316                 :          2 :       return;
     317                 :            :     }
     318                 :            : 
     319         [ +  + ]:         36 :     for (auto& then_callback : then_callbacks) {
     320         [ +  - ]:          8 :       auto then_ret = run_with_timeout(then_callback);
     321                 :            : 
     322         [ -  + ]:          8 :       if (!then_ret.has_value()) {
     323                 :          1 :         return;
     324                 :            :       }
     325                 :            : 
     326   [ +  -  +  + ]:          8 :       if (!then_ret.value()) {
     327         [ -  + ]:          1 :         if (else_cb) {
     328         [ #  # ]:          0 :           else_cb();
     329                 :            :         }
     330                 :            : 
     331                 :          1 :         return;
     332                 :            :       }
     333                 :            :     }
     334   [ +  -  +  +  :        114 :   };
          +  +  +  +  +  
                +  +  + ]
     335                 :            : 
     336                 :         41 :   return status;
     337                 :            : }  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     338                 :            : 
     339                 :            : }  // namespace vlink