Line data    Source code

       1              : /**
       2              :  * @file test_ratelimiter_app.cxx Test application for
       3              :  * ratelimiter implementation
       4              :  *
       5              :  * This is part of the DUNE DAQ Application Framework, copyright 2020.
       6              :  * Licensing/copyright details are in the COPYING file that you should have
       7              :  * received with this code.
       8              :  */
       9              : #
      10              : 
      11              : #include <atomic>
      12              : #include <chrono>
      13              : #include <folly/coro/Baton.h>
      14              : #include <folly/coro/BlockingWait.h>
      15              : #include <folly/coro/CurrentExecutor.h>
      16              : #include <folly/coro/Task.h>
      17              : #include <folly/coro/Timeout.h>
      18              : #include <folly/futures/Future.h>
      19              : #include <folly/futures/ThreadWheelTimekeeper.h>
      20              : #include <iostream>
      21              : #include <thread>
      22              : 
      23              : // using namespace dunedaq::datahandlinglibs;
      24              : using namespace std::chrono_literals;
      25              : 
      26              : folly::coro::Baton baton{ 0 };
      27              : uint32_t max_wait = 500;
      28              : 
      29              : folly::coro::Task<void>
      30            0 : postprocess_schedule()
      31              : {
      32              : 
      33              :   folly::ThreadWheelTimekeeper tk;
      34              : 
      35            0 :   const auto wait_data = [&baton]() -> folly::coro::Task<void> {
      36              :     // folly::coro::timeout cancels the task on timeout.
      37              :     // Baton is not cancellable, so we attach a callback to resume the coroutine.
      38              :     auto token = co_await folly::coro::co_current_cancellation_token;
      39            0 :     folly::CancellationCallback cb(token, [&baton] { baton.post(); });
      40              :     co_await baton; // Wait data
      41            0 :   };
      42              : 
      43              :   uint64_t n_timeouts = 0;
      44              :   uint64_t n_process = 0;
      45              : 
      46              :   while (true) {
      47              :     try {
      48              :       co_await folly::coro::timeout(wait_data(), std::chrono::milliseconds{ max_wait }, &tk);
      49              :       ++n_process;
      50              :     } catch (const folly::FutureTimeout&) {
      51              :       // timeout = true;
      52              :       std::cout << "Timeout " << ++n_timeouts << std::endl;
      53              :     }
      54              :     baton.reset();
      55              :   }
      56              : 
      57              :   co_return;
      58            0 : }
      59              : 
      60              : int
      61            0 : main(int /*argc*/, char** /*argv[]*/)
      62              : {
      63            0 :   std::atomic<bool> run_marker;
      64              : 
      65              :   // A sleepy worker thread
      66            0 :   std::jthread sleepy_worker([&baton](std::stop_token stoken) {
      67            0 :     while (!stoken.stop_requested()) {
      68            0 :       baton.post();
      69              :     }
      70              :     // for (int i = 10; i; --i)
      71              :     // {
      72              :     //     std::this_thread::sleep_for(300ms);
      73              :     //     if (stoken.stop_requested())
      74              :     //     {
      75              :     //         print("Sleepy worker is requested to stop\n");
      76              :     //         return;
      77              :     //     }
      78              :     //     print("Sleepy worker goes back to sleep\n");
      79              :     // }
      80            0 :   });
      81              : 
      82              :   // std::cout << "Sleeping for 3s" << std::endl;
      83              : 
      84              :   // std::this_thread::sleep_for(3s);
      85              : 
      86            0 :   std::cout << "Starting the coroutine" << std::endl;
      87            0 :   folly::coro::blockingWait(postprocess_schedule());
      88              : 
      89            0 :   std::cout << "Requesting stop" << std::endl;
      90              :   // sleepy_worker.request_stop();
      91            0 :   std::cout << "Thread stopped" << std::endl;
      92              : 
      93            0 :   return 0;
      94            0 : } // NOLINT(readability/fn_size)