<commit_before>// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.

#include "caf/io/basp_broker.hpp"

#include <chrono>
#include <limits>

#include "caf/actor_registry.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/after.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/connection_helper.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"

namespace {

#ifdef CAF_MSVC
#  define THREAD_LOCAL thread_local
#else
#  define THREAD_LOCAL __thread
#endif

// Used by make_proxy to detect indirect connections.
THREAD_LOCAL caf::node_id* t_last_hop = nullptr;

#undef THREAD_LOCAL

} // namespace

namespace caf::io {

// -- constructors, destructors, and assignment operators ----------------------

basp_broker::basp_broker(actor_config& cfg)
  : super(cfg),
    basp::instance::callee(super::system(),
                           static_cast<proxy_registry::backend&>(*this)),
    this_context(nullptr) {
  new (&instance) basp::instance(this, *this);
  CAF_ASSERT(this_node() != none);
}

basp_broker::~basp_broker() {
  // nop
}

// -- implementation of local_actor/broker -------------------------------------

void basp_broker::on_exit() {
  // Wait until all pending messages of workers have been shipped.
  // TODO: this blocks the calling thread. This is only safe because we know
  //       that the middleman calls this in its stop() function. However,
  //       ultimately we should find a nonblocking solution here.
  instance.hub().await_workers();
  // All nodes are offline now. We use a default-constructed error code to
  // signal ordinary shutdown.
  for (const auto& [node, observer_list] : node_observers)
    for (const auto& observer : observer_list)
      if (auto hdl = actor_cast<actor>(observer))
        anon_send(hdl, node_down_msg{node, error{}});
  node_observers.clear();
  // Release any obsolete state.
  ctx.clear();
  // Make sure all spawn servers are down before clearing the container.
  for (auto& kvp : spawn_servers)
    anon_send_exit(kvp.second, exit_reason::kill);
  // Clear remaining state.
  spawn_servers.clear();
  monitored_actors.clear();
  proxies().clear();
  instance.~instance();
}

const char* basp_broker::name() const {
  return "caf.system.basp-broker";
}

behavior basp_broker::make_behavior() {
  CAF_LOG_TRACE(CAF_ARG(system().node()));
  set_down_handler([](local_actor* ptr, down_msg& x) {
    static_cast<basp_broker*>(ptr)->handle_down_msg(x);
  });
  if (get_or(config(), "caf.middleman.enable-automatic-connections", false)) {
    CAF_LOG_DEBUG("enable automatic connections");
    // open a random port and store a record for our peers how to
    // connect to this broker directly in the configuration server
    auto res = add_tcp_doorman(uint16_t{0});
    if (res) {
      auto port = res->second;
      auto addrs = network::interfaces::list_addresses(false);
      auto config_server = system().registry().get("ConfigServ");
      send(actor_cast<actor>(config_server), put_atom_v,
           "basp.default-connectivity-tcp",
           make_message(port, std::move(addrs)));
    }
    automatic_connections = true;
  }
  auto heartbeat_interval = get_or(config(), "caf.middleman.heartbeat-interval",
                                   defaults::middleman::heartbeat_interval);
  if (heartbeat_interval.count() > 0) {
    CAF_LOG_DEBUG("enable heartbeat" << CAF_ARG(heartbeat_interval));
    auto now = clock().now();
    auto first_tick = now + heartbeat_interval;
    auto connection_timeout = get_or(config(),
                                     "caf.middleman.connection-timeout",
                                     defaults::middleman::connection_timeout);
    // Note: we send the scheduled time as integer representation to avoid
    //       having to assign a type ID to the time_point type.
    scheduled_send(this, first_tick, tick_atom_v,
                   first_tick.time_since_epoch().count(), heartbeat_interval,
                   connection_timeout);
  }
  return behavior{
    // received from underlying broker implementation
    [=](new_data_msg& msg) {
      CAF_LOG_TRACE(CAF_ARG(msg.handle));
      set_context(msg.handle);
      auto& ctx = *this_context;
      auto next = instance.handle(context(), msg, ctx.hdr,
                                  ctx.cstate == basp::await_payload);
      if (requires_shutdown(next)) {
        connection_cleanup(msg.handle, to_sec(next));
        close(msg.handle);
        return;
      }
      if (next != ctx.cstate) {
        auto rd_size = next == basp::await_payload ? ctx.hdr.payload_len
                                                   : basp::header_size;
        configure_read(msg.handle, receive_policy::exactly(rd_size));
        ctx.cstate = next;
      }
    },
    // received from proxy instances
    [=](forward_atom, strong_actor_ptr& src,
        const std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest,<commit_msg>Add connection_timeout to debug log output<commit_after>// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.

#include "caf/io/basp_broker.hpp"

#include <chrono>
#include <limits>

#include "caf/actor_registry.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/after.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/connection_helper.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"

namespace {

#ifdef CAF_MSVC
#  define THREAD_LOCAL thread_local
#else
#  define THREAD_LOCAL __thread
#endif

// Used by make_proxy to detect indirect connections.
THREAD_LOCAL caf::node_id* t_last_hop = nullptr;

#undef THREAD_LOCAL

} // namespace

namespace caf::io {

// -- constructors, destructors, and assignment operators ----------------------

basp_broker::basp_broker(actor_config& cfg)
  : super(cfg),
    basp::instance::callee(super::system(),
                           static_cast<proxy_registry::backend&>(*this)),
    this_context(nullptr) {
  new (&instance) basp::instance(this, *this);
  CAF_ASSERT(this_node() != none);
}

basp_broker::~basp_broker() {
  // nop
}

// -- implementation of local_actor/broker -------------------------------------

void basp_broker::on_exit() {
  // Wait until all pending messages of workers have been shipped.
  // TODO: this blocks the calling thread. This is only safe because we know
  //       that the middleman calls this in its stop() function. However,
  //       ultimately we should find a nonblocking solution here.
  instance.hub().await_workers();
  // All nodes are offline now. We use a default-constructed error code to
  // signal ordinary shutdown.
  for (const auto& [node, observer_list] : node_observers)
    for (const auto& observer : observer_list)
      if (auto hdl = actor_cast<actor>(observer))
        anon_send(hdl, node_down_msg{node, error{}});
  node_observers.clear();
  // Release any obsolete state.
  ctx.clear();
  // Make sure all spawn servers are down before clearing the container.
  for (auto& kvp : spawn_servers)
    anon_send_exit(kvp.second, exit_reason::kill);
  // Clear remaining state.
  spawn_servers.clear();
  monitored_actors.clear();
  proxies().clear();
  instance.~instance();
}

const char* basp_broker::name() const {
  return "caf.system.basp-broker";
}

behavior basp_broker::make_behavior() {
  CAF_LOG_TRACE(CAF_ARG(system().node()));
  set_down_handler([](local_actor* ptr, down_msg& x) {
    static_cast<basp_broker*>(ptr)->handle_down_msg(x);
  });
  if (get_or(config(), "caf.middleman.enable-automatic-connections", false)) {
    CAF_LOG_DEBUG("enable automatic connections");
    // open a random port and store a record for our peers how to
    // connect to this broker directly in the configuration server
    auto res = add_tcp_doorman(uint16_t{0});
    if (res) {
      auto port = res->second;
      auto addrs = network::interfaces::list_addresses(false);
      auto config_server = system().registry().get("ConfigServ");
      send(actor_cast<actor>(config_server), put_atom_v,
           "basp.default-connectivity-tcp",
           make_message(port, std::move(addrs)));
    }
    automatic_connections = true;
  }
  auto heartbeat_interval = get_or(config(), "caf.middleman.heartbeat-interval",
                                   defaults::middleman::heartbeat_interval);
  if (heartbeat_interval.count() > 0) {
    auto now = clock().now();
    auto first_tick = now + heartbeat_interval;
    auto connection_timeout = get_or(config(),
                                     "caf.middleman.connection-timeout",
                                     defaults::middleman::connection_timeout);
    CAF_LOG_DEBUG("enable heartbeat" << CAF_ARG(heartbeat_interval)
                                     << CAF_ARG(connection_timeout));
    // Note: we send the scheduled time as integer representation to avoid
    //       having to assign a type ID to the time_point type.
    scheduled_send(this, first_tick, tick_atom_v,
                   first_tick.time_since_epoch().count(), heartbeat_interval,
                   connection_timeout);
  }
  return behavior{
    // received from underlying broker implementation
    [=](new_data_msg& msg) {
      CAF_LOG_TRACE(CAF_ARG(msg.handle));
      set_context(msg.handle);
      auto& ctx = *this_context;
      auto next = instance.handle(context(), msg, ctx.hdr,
                                  ctx.cstate == basp::await_payload);
      if (requires_shutdown(next)) {
        connection_cleanup(msg.handle, to_sec(next));
        close(msg.handle);
        return;
      }
      if (next != ctx.cstate) {
        auto rd_size = next == basp::await_payload ? ctx.hdr.payload_len
                                                   : basp::header_size;
        configure_read(msg.handle, receive_policy::exactly(rd_size));
        ctx.cstate = next;
      }
    },
    // received from proxy instances
    [=](forward_atom, strong_actor_ptr& src,
        const std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest,
{'fraction_non_alphanumeric': 0.12981257501615734, 'fraction_numerical': 0.0007386206259809805, 'mean_word_length': 3.0767783214151296, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1342467', 'n_tokens_mistral': 3346, 'n_tokens_neox': 3310, 'n_words': 777}