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 "./extension/bag_processor.h"
      25                 :            : 
      26                 :            : #include <algorithm>
      27                 :            : #include <atomic>
      28                 :            : #include <chrono>
      29                 :            : #include <deque>
      30                 :            : #include <memory>
      31                 :            : #include <mutex>
      32                 :            : #include <thread>
      33                 :            : #include <utility>
      34                 :            : 
      35                 :            : #include "./base/condition_variable.h"
      36                 :            : #include "./base/elapsed_timer.h"
      37                 :            : #include "./base/logger.h"
      38                 :            : 
      39                 :            : namespace vlink {
      40                 :            : 
      41                 :            : // BagProcessor::Impl
      42                 :            : struct BagProcessor::Impl final {
      43                 :            :   struct CacheEntry final {
      44                 :            :     int64_t data_timestamp{0};
      45                 :            :     int64_t enqueue_time{0};
      46                 :            :     Frame frame;
      47                 :            :     bool data_timestamp_valid{false};
      48                 :            :   };
      49                 :            : 
      50                 :            :   BagProcessor::Config config;
      51                 :            :   BagProcessor::OutputCallback output_callback;
      52                 :            :   std::deque<CacheEntry> data_queue;
      53                 :            :   std::mutex mtx;
      54                 :            :   ConditionVariable cv;
      55                 :            :   std::thread thread;
      56                 :            : 
      57                 :            :   int64_t current_size{0};
      58                 :            :   int64_t last_data_timestamp{0};
      59                 :            :   int64_t last_timestamp{0};
      60                 :            :   int64_t data_timestamp_anchor{0};
      61                 :            :   int64_t timestamp_anchor{0};
      62                 :            :   int64_t last_output_timestamp{0};
      63                 :            : 
      64                 :            :   std::atomic_bool quit_flag{false};
      65                 :            :   bool flush_request{false};
      66                 :            :   bool last_resolved_data_timestamp_valid{false};
      67                 :            :   bool timestamp_anchor_valid{false};
      68                 :            :   bool output_timestamp_valid{false};
      69                 :            : };
      70                 :            : 
      71                 :            : // BagProcessor
      72                 :         28 : BagProcessor::BagProcessor(const Config& config) : impl_(std::make_unique<Impl>()) { impl_->config = config; }
      73                 :            : 
      74                 :         28 : BagProcessor::~BagProcessor() {
      75                 :            :   {
      76                 :         28 :     std::lock_guard lock(impl_->mtx);
      77                 :            : 
      78                 :         28 :     impl_->quit_flag.store(true, std::memory_order_release);
      79                 :         28 :   }
      80                 :            : 
      81                 :         28 :   impl_->cv.notify_all();
      82                 :            : 
      83         [ +  + ]:         28 :   if VLIKELY (impl_->thread.joinable()) {
      84                 :         24 :     impl_->thread.join();
      85                 :            :   }
      86                 :         28 : }
      87                 :            : 
      88                 :         27 : void BagProcessor::register_output_callback(OutputCallback&& output_callback) {
      89         [ +  - ]:         27 :   std::lock_guard lock(impl_->mtx);
      90                 :            : 
      91         [ +  + ]:         27 :   if VUNLIKELY (impl_->output_callback) {
      92   [ +  -  +  - ]:          4 :     VLOG_W("BagProcessor output callback has already been registered.");
      93                 :          2 :     return;
      94                 :            :   }
      95                 :            : 
      96         [ +  + ]:         25 :   if VUNLIKELY (!output_callback) {
      97   [ +  -  -  + ]:          2 :     VLOG_F("BagProcessor output callback is empty.");
      98                 :            :   }
      99                 :            : 
     100                 :         24 :   impl_->output_callback = std::move(output_callback);
     101         [ +  - ]:         24 :   impl_->thread = std::thread(&BagProcessor::on_run, this);
     102         [ +  + ]:         27 : }
     103                 :            : 
     104                 :        155 : void BagProcessor::push(int64_t data_timestamp, const Frame& frame) {
     105         [ +  - ]:        155 :   std::unique_lock lock(impl_->mtx);
     106                 :            : 
     107         [ +  + ]:        155 :   if VUNLIKELY (!impl_->output_callback) {
     108   [ +  -  -  + ]:          2 :     VLOG_F("BagProcessor output callback has not been registered.");
     109                 :            :   }
     110                 :            : 
     111         [ -  + ]:        154 :   if VUNLIKELY (impl_->current_size >= impl_->config.max_cache_size) {
     112                 :            :     VLOG_W("BagProcessor: Cache size is full, waiting to consume.");  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     113                 :            :   }
     114                 :            : 
     115                 :        154 :   impl_->cv.wait(lock, [this]() -> bool {
     116         [ +  + ]:        154 :     if (impl_->data_queue.empty()) {
     117                 :         25 :       return true;
     118                 :            :     }
     119                 :            : 
     120   [ -  +  -  - ]:        129 :     return impl_->current_size < impl_->config.max_cache_size || impl_->quit_flag.load(std::memory_order_acquire);
     121                 :            :   });
     122                 :            : 
     123         [ -  + ]:        154 :   if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
     124                 :            :     return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     125                 :            :   }
     126                 :            : 
     127                 :        154 :   const int64_t enqueue_time = ElapsedTimer::get_sys_timestamp(ElapsedTimer::kMicro);
     128                 :            : 
     129                 :        154 :   bool data_timestamp_valid = true;
     130                 :            : 
     131         [ +  + ]:        154 :   if (data_timestamp < 0) {
     132         [ +  + ]:          7 :     if (!impl_->last_resolved_data_timestamp_valid) {
     133                 :          5 :       data_timestamp = -1;
     134                 :          5 :       data_timestamp_valid = false;
     135                 :            :     } else {
     136                 :          2 :       data_timestamp = impl_->last_data_timestamp + (frame.timestamp - impl_->last_timestamp);
     137                 :            :     }
     138         [ +  + ]:        147 :   } else if (impl_->last_resolved_data_timestamp_valid) {
     139                 :        124 :     const int64_t max_jump = impl_->config.max_jump_time * 1000;
     140                 :        124 :     const int64_t jump = data_timestamp - impl_->last_data_timestamp;
     141                 :            : 
     142   [ +  -  +  +  :        124 :     if (max_jump > 0 && (jump > max_jump || jump < -max_jump)) {
                   +  + ]
     143                 :          2 :       data_timestamp = impl_->last_data_timestamp + (frame.timestamp - impl_->last_timestamp);
     144                 :            :     }
     145                 :            :   }
     146                 :            : 
     147         [ +  + ]:        154 :   if (data_timestamp_valid) {
     148                 :        149 :     impl_->last_resolved_data_timestamp_valid = true;
     149                 :        149 :     impl_->last_data_timestamp = data_timestamp;
     150                 :        149 :     impl_->last_timestamp = frame.timestamp;
     151                 :            :   }
     152                 :            : 
     153                 :        154 :   impl_->current_size += frame.data.size();
     154                 :            : 
     155         [ +  - ]:        154 :   Impl::CacheEntry entry{data_timestamp, enqueue_time, frame, data_timestamp_valid};
     156                 :            : 
     157                 :        154 :   auto iter = std::upper_bound(impl_->data_queue.begin(), impl_->data_queue.end(), entry,
     158                 :        574 :                                [](const Impl::CacheEntry& candidate, const Impl::CacheEntry& queued) {
     159         [ +  + ]:        574 :                                  if (candidate.data_timestamp_valid != queued.data_timestamp_valid) {
     160                 :          2 :                                    return !candidate.data_timestamp_valid;
     161                 :            :                                  }
     162                 :            : 
     163                 :        572 :                                  return candidate.data_timestamp < queued.data_timestamp;
     164         [ +  - ]:        154 :                                });
     165         [ +  - ]:        154 :   impl_->data_queue.emplace(iter, std::move(entry));
     166                 :            : 
     167                 :        154 :   impl_->cv.notify_one();
     168         [ +  - ]:        155 : }
     169                 :            : 
     170                 :         25 : void BagProcessor::flush() {
     171         [ +  - ]:         25 :   std::unique_lock lock(impl_->mtx);
     172                 :            : 
     173   [ +  -  +  +  :         25 :   if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire) || !impl_->thread.joinable()) {
                   +  + ]
     174                 :          2 :     return;
     175                 :            :   }
     176                 :            : 
     177                 :         23 :   impl_->flush_request = true;
     178                 :            : 
     179                 :         23 :   impl_->cv.notify_all();
     180                 :            : 
     181                 :         23 :   impl_->cv.wait(lock, [this]() -> bool {
     182   [ +  +  -  + ]:         46 :     return !impl_->flush_request || impl_->quit_flag.load(std::memory_order_acquire);
     183                 :            :   });  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     184         [ +  + ]:         25 : }
     185                 :            : 
     186                 :         34 : bool BagProcessor::on_check() {
     187         [ -  + ]:         34 :   if (impl_->data_queue.empty()) {
     188                 :            :     return false;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     189                 :            :   }
     190                 :            : 
     191         [ +  + ]:         34 :   if (impl_->current_size >= impl_->config.max_cache_size) {
     192                 :          1 :     return true;
     193                 :            :   }
     194                 :            : 
     195                 :         33 :   const int64_t min_cache_time = impl_->config.min_cache_time * 1000;
     196                 :            : 
     197         [ +  + ]:         33 :   if (impl_->data_queue.back().data_timestamp - impl_->data_queue.front().data_timestamp >= min_cache_time) {
     198                 :          6 :     return true;
     199                 :            :   }
     200                 :            : 
     201                 :         27 :   const int64_t cache_elapsed = static_cast<int64_t>(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kMicro)) -
     202                 :         27 :                                 impl_->data_queue.front().enqueue_time;
     203                 :            : 
     204                 :         27 :   return cache_elapsed >= min_cache_time;
     205                 :            : }
     206                 :            : 
     207                 :         56 : void BagProcessor::on_output(std::unique_lock<std::mutex>& lock, bool at_end) {
     208         [ +  + ]:         56 :   if (impl_->data_queue.empty()) {
     209                 :         24 :     return;
     210                 :            :   }
     211                 :            : 
     212                 :            :   do {
     213                 :        154 :     const int64_t min_cache_time = impl_->config.min_cache_time * 1000;
     214   [ +  +  +  + ]:        154 :     const bool flush_all = at_end || impl_->current_size >= impl_->config.max_cache_size;
     215                 :        154 :     bool should_output = flush_all;
     216                 :            : 
     217         [ +  + ]:        154 :     if (!should_output) {
     218                 :          8 :       const int64_t timestamp_span = impl_->data_queue.back().data_timestamp - impl_->data_queue.front().data_timestamp;
     219                 :            : 
     220         [ +  + ]:          8 :       if (timestamp_span >= min_cache_time) {
     221                 :          6 :         should_output =
     222                 :          6 :             impl_->data_queue.front().data_timestamp <= impl_->data_queue.back().data_timestamp - min_cache_time;
     223                 :            :       } else {
     224                 :          2 :         const int64_t cache_elapsed = static_cast<int64_t>(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kMicro)) -
     225                 :          2 :                                       impl_->data_queue.front().enqueue_time;
     226                 :            : 
     227                 :          2 :         should_output = cache_elapsed >= min_cache_time;
     228                 :            :       }
     229                 :            : 
     230         [ -  + ]:          8 :       if (!should_output) {
     231                 :            :         return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     232                 :            :       }
     233                 :            :     }
     234                 :            : 
     235                 :        154 :     auto entry = std::move(impl_->data_queue.front());
     236                 :        154 :     impl_->data_queue.pop_front();
     237                 :            : 
     238                 :        154 :     impl_->current_size -= entry.frame.data.size();
     239                 :            : 
     240                 :        154 :     Frame frame = std::move(entry.frame);
     241                 :        154 :     int64_t output_timestamp = frame.timestamp;
     242                 :        154 :     bool update_timestamp_anchor = false;
     243                 :            : 
     244   [ +  +  +  +  :        154 :     if (entry.data_timestamp_valid && !impl_->timestamp_anchor_valid) {
                   +  + ]
     245                 :         23 :       impl_->data_timestamp_anchor = entry.data_timestamp;
     246                 :         23 :       update_timestamp_anchor = true;
     247         [ +  + ]:        131 :     } else if (entry.data_timestamp_valid) {
     248                 :        126 :       output_timestamp = impl_->timestamp_anchor + (entry.data_timestamp - impl_->data_timestamp_anchor);
     249                 :            :     }
     250                 :            : 
     251   [ +  +  +  +  :        154 :     if (impl_->output_timestamp_valid && output_timestamp <= impl_->last_output_timestamp) {
                   +  + ]
     252                 :          3 :       output_timestamp = impl_->last_output_timestamp + 1;
     253                 :            :     }
     254                 :            : 
     255         [ +  + ]:        154 :     if (update_timestamp_anchor) {
     256                 :         23 :       impl_->timestamp_anchor_valid = true;
     257                 :         23 :       impl_->timestamp_anchor = output_timestamp;
     258                 :            :     }
     259                 :            : 
     260                 :        154 :     impl_->output_timestamp_valid = true;
     261                 :        154 :     impl_->last_output_timestamp = output_timestamp;
     262                 :        154 :     frame.timestamp = output_timestamp;
     263                 :            : 
     264         [ +  - ]:        154 :     lock.unlock();
     265                 :            : 
     266         [ +  - ]:        154 :     impl_->output_callback(frame);
     267                 :            : 
     268         [ +  - ]:        154 :     lock.lock();
     269   [ +  +  +  +  :        154 :   } while (at_end && !impl_->data_queue.empty());
                   +  + ]
     270                 :            : }
     271                 :            : 
     272                 :         24 : void BagProcessor::on_run() {
     273         [ +  + ]:         80 :   while (!impl_->quit_flag.load(std::memory_order_acquire)) {
     274                 :         56 :     on_exec(false);
     275                 :            :   }
     276                 :            : 
     277                 :         24 :   on_exec(true);
     278                 :         24 : }
     279                 :            : 
     280                 :         80 : void BagProcessor::on_exec(bool at_end) {
     281         [ +  - ]:         80 :   std::unique_lock lock(impl_->mtx);
     282                 :            : 
     283         [ +  + ]:         80 :   if VLIKELY (!at_end) {
     284                 :         56 :     impl_->cv.wait(lock, [this]() -> bool {
     285   [ +  +  +  +  :         84 :       return !impl_->data_queue.empty() || impl_->quit_flag.load(std::memory_order_acquire) || impl_->flush_request;
                   +  + ]
     286                 :            :     });
     287                 :            : 
     288         [ +  + ]:         56 :     if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
     289                 :         22 :       return;
     290                 :            :     }
     291                 :            : 
     292         [ +  + ]:         34 :     if VUNLIKELY (impl_->flush_request) {
     293         [ +  - ]:         17 :       on_output(lock, true);
     294                 :            : 
     295                 :         17 :       impl_->flush_request = false;
     296                 :            : 
     297                 :         17 :       impl_->cv.notify_all();
     298                 :            : 
     299                 :         17 :       return;
     300                 :            :     }
     301                 :            :   }
     302                 :            : 
     303         [ +  + ]:         41 :   if VLIKELY (!at_end) {
     304   [ +  -  +  + ]:         17 :     if (!on_check()) {
     305                 :         10 :       const int64_t min_cache_time = impl_->config.min_cache_time * 1000;
     306                 :         10 :       const int64_t cache_elapsed = static_cast<int64_t>(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kMicro)) -
     307                 :         10 :                                     impl_->data_queue.front().enqueue_time;
     308                 :         10 :       const int64_t wait_time = min_cache_time - cache_elapsed;
     309                 :            : 
     310         [ +  - ]:         10 :       if (wait_time > 0) {
     311                 :         10 :         impl_->cv.wait_for(lock, std::chrono::microseconds(wait_time), [this]() -> bool {
     312   [ +  +  +  +  :         23 :           return impl_->quit_flag.load(std::memory_order_acquire) || impl_->flush_request || on_check();
                   +  + ]
     313                 :            :         });
     314                 :            :       }
     315                 :            : 
     316         [ +  + ]:         10 :       if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
     317                 :          8 :         return;
     318                 :            :       }
     319                 :            : 
     320         [ +  + ]:          8 :       if VUNLIKELY (impl_->flush_request) {
     321         [ +  - ]:          6 :         on_output(lock, true);
     322                 :            : 
     323                 :          6 :         impl_->flush_request = false;
     324                 :            : 
     325                 :          6 :         impl_->cv.notify_all();
     326                 :            : 
     327                 :          6 :         return;
     328                 :            :       }
     329                 :            : 
     330   [ +  -  -  + ]:          2 :       if (!on_check()) {
     331                 :            :         return;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
     332                 :            :       }
     333                 :            :     }
     334                 :            :   }
     335                 :            : 
     336         [ +  - ]:         33 :   on_output(lock, at_end);
     337                 :            : 
     338         [ +  + ]:         33 :   if VLIKELY (!at_end) {
     339                 :          9 :     impl_->cv.notify_all();
     340                 :            :   }
     341         [ +  + ]:         80 : }
     342                 :            : 
     343                 :            : }  // namespace vlink