103-asio_ping-pong-2-threads.cpp 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149
  1. //
  2. // Copyright (c) 2019-2020 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
  3. //
  4. // Distributed under the MIT Software License
  5. //
  6. #include "catch.hpp"
  7. #include "rotor.hpp"
  8. #include "rotor/asio.hpp"
  9. #include "supervisor_asio_test.h"
  10. #include "actor_test.h"
  11. #include <iostream>
  12. namespace r = rotor;
  13. namespace ra = rotor::asio;
  14. namespace rt = r::test;
  15. namespace asio = boost::asio;
  16. namespace pt = boost::posix_time;
  17. struct ping_t {};
  18. struct pong_t {};
  19. struct pinger_t : public rt::actor_test_t {
  20. std::uint32_t ping_sent = 0;
  21. std::uint32_t pong_received = 0;
  22. using rt::actor_test_t::actor_test_t;
  23. void set_ponger_addr(const r::address_ptr_t &addr) { ponger_addr = addr; }
  24. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  25. plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) { p.subscribe_actor(&pinger_t::on_pong); });
  26. plugin.with_casted<r::plugin::link_client_plugin_t>(
  27. [&](auto &p) { p.link(ponger_addr, true, [&](auto &ec) mutable { REQUIRE(!ec); }); });
  28. }
  29. void on_start() noexcept override {
  30. r::actor_base_t::on_start();
  31. do_send_ping();
  32. std::cout << "pinger start\n";
  33. }
  34. void shutdown_finish() noexcept override {
  35. r::actor_base_t::shutdown_finish();
  36. supervisor->shutdown();
  37. ponger_addr->supervisor.shutdown();
  38. std::cout << "pinger shutdown finish\n";
  39. }
  40. void on_pong(r::message_t<pong_t> &) noexcept {
  41. ++pong_received;
  42. do_shutdown();
  43. }
  44. void do_send_ping() {
  45. ++ping_sent;
  46. send<ping_t>(ponger_addr);
  47. }
  48. r::address_ptr_t ponger_addr;
  49. };
  50. struct ponger_t : public rt::actor_test_t {
  51. std::uint32_t ping_received = 0;
  52. std::uint32_t pong_sent = 0;
  53. using rt::actor_test_t::actor_test_t;
  54. void set_pinger_addr(const r::address_ptr_t &addr) { pinger_addr = addr; }
  55. void on_start() noexcept override {
  56. r::actor_base_t::on_start();
  57. std::cout << "ponger start\n";
  58. }
  59. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  60. plugin.with_casted<r::plugin::starter_plugin_t>([](auto &p) { p.subscribe_actor(&ponger_t::on_ping); });
  61. }
  62. void on_ping(r::message_t<ping_t> &) noexcept {
  63. ++ping_received;
  64. send<pong_t>(pinger_addr);
  65. ++pong_sent;
  66. }
  67. private:
  68. r::address_ptr_t pinger_addr;
  69. };
  70. TEST_CASE("ping/pong on 2 threads", "[supervisor][asio]") {
  71. using sup_t = rt::supervisor_asio_test_t;
  72. asio::io_context io_ctx1;
  73. asio::io_context io_ctx2;
  74. auto timeout = r::pt::milliseconds{10};
  75. auto sys_ctx1 = ra::system_context_asio_t::ptr_t{new ra::system_context_asio_t(io_ctx1)};
  76. auto sys_ctx2 = ra::system_context_asio_t::ptr_t{new ra::system_context_asio_t(io_ctx2)};
  77. auto strand1 = std::make_shared<asio::io_context::strand>(io_ctx1);
  78. auto strand2 = std::make_shared<asio::io_context::strand>(io_ctx2);
  79. auto sup1 = sys_ctx1->create_supervisor<sup_t>().strand(strand1).timeout(timeout).guard_context(true).finish();
  80. auto sup2 = sys_ctx2->create_supervisor<sup_t>().strand(strand2).timeout(timeout).guard_context(true).finish();
  81. auto pinger = sup1->create_actor<pinger_t>().timeout(timeout).finish();
  82. auto ponger = sup2->create_actor<ponger_t>().timeout(timeout).finish();
  83. std::cout << "sup1: " << sup1.get() << ", sup2: " << sup2.get() << "\n";
  84. std::cout << "pinger: " << pinger.get() << ", ponger: " << ponger.get() << "\n";
  85. auto &pinger_addr = static_cast<r::actor_base_t *>(pinger.get())->get_address();
  86. auto &ponger_addr = static_cast<r::actor_base_t *>(ponger.get())->get_address();
  87. std::cout << "pinger addr: " << pinger_addr.get() << ", ponger addr: " << ponger_addr.get() << "\n";
  88. pinger->set_ponger_addr(static_cast<r::actor_base_t *>(ponger.get())->get_address());
  89. ponger->set_pinger_addr(static_cast<r::actor_base_t *>(pinger.get())->get_address());
  90. sup1->start();
  91. sup2->start();
  92. auto t1 = std::thread([&] { io_ctx1.run(); });
  93. auto t2 = std::thread([&] { io_ctx2.run(); });
  94. t1.join();
  95. t2.join();
  96. if (pinger->ping_sent == 1) {
  97. CHECK(pinger->ping_sent == 1);
  98. CHECK(pinger->pong_received == 1);
  99. CHECK(ponger->ping_received == 1);
  100. CHECK(ponger->pong_sent == 1);
  101. } else {
  102. // this might happen due to unavoidable race between threads, system scheduler etc.
  103. // can be artificially archived launching tool like `stress-ng --cpu 8`
  104. // and/or set small timeout
  105. // no special acetion is needed, as the pinger will trigger shutdown for the
  106. // both threads
  107. }
  108. CHECK(static_cast<r::actor_base_t *>(sup1.get())->access<rt::to::state>() == r::state_t::SHUT_DOWN);
  109. CHECK(sup1->get_leader_queue().size() == 0);
  110. CHECK(rt::empty(sup1->get_subscription()));
  111. CHECK(static_cast<r::actor_base_t *>(sup2.get())->access<rt::to::state>() == r::state_t::SHUT_DOWN);
  112. CHECK(sup2->get_leader_queue().size() == 0);
  113. CHECK(rt::empty(sup2->get_subscription()));
  114. REQUIRE(pinger->get_state() == r::state_t::SHUT_DOWN);
  115. REQUIRE(ponger->get_state() == r::state_t::SHUT_DOWN);
  116. }