<repo_name>daq-db/eRPC<file_sep>/third_party/masstree-beta/README.md
From https://github.com/anujkaliaiitd/masstree-beta.

Do not use bootstrap.sh. Use cmake.

<file_sep>/src/rpc_impl/rpc_reset_handlers.cc
/**
 * @file rpc_reset_handlers.cc
 * @brief Handlers for session resets
 */
#include "rpc.h"

namespace erpc {

template <class TTr>
bool Rpc<TTr>::handle_reset_client_st(Session *session) {
  assert(in_dispatch());

  char issue_msg[kMaxIssueMsgLen];
  sprintf(issue_msg,
          "Rpc %u, lsn %u: Trying to reset client session. State = %s. Issue",
          rpc_id, session->local_session_num,
          session_state_str(session->state).c_str());

  // Erase session slots from credit stall queue
  for (const SSlot &sslot : session->sslot_arr) {
    stallq.erase(std::remove(stallq.begin(), stallq.end(), &sslot),
                 stallq.end());
  }

  // Invoke continuation-with-failure for all active requests
  for (SSlot &sslot : session->sslot_arr) {
    if (sslot.tx_msgbuf == nullptr) {
      sslot.tx_msgbuf = nullptr;
      delete_from_active_rpc_list(sslot);
      session->client_info.sslot_free_vec.push_back(sslot.index);

      MsgBuffer *resp_msgbuf = sslot.client_info.resp_msgbuf;
      resize_msg_buffer(resp_msgbuf, 0);  // 0 response size marks the error
      sslot.client_info.cont_func(context, sslot.client_info.tag);
    }
  }

  assert(session->client_info.sslot_free_vec.size() == kSessionReqWindow);

  // Change state before failure continuations
  session->state = SessionState::kDisconnectInProgress;

  // Act similar to handling a disconnect response
  LOG_INFO("%s: None. Session resetted.\n", issue_msg);
  free_ring_entries();  // Free before callback to allow creating new session
  sm_handler(session->local_session_num, SmEventType::kDisconnected,
             SmErrType::kSrvDisconnected, context);
  bury_session_st(session);
  return true;
}

template <class TTr>
bool Rpc<TTr>::handle_reset_server_st(Session *session) {
  assert(in_dispatch());
  assert(session->is_server());
  session->state = SessionState::kResetInProgress;

  char issue_msg[kMaxIssueMsgLen];
  sprintf(issue_msg,
          "Rpc %u, lsn %u: Trying to reset server session. State = %s. Issue",
          rpc_id, session->local_session_num,
          session_state_str(session->state).c_str());

  size_t pending_enqueue_resps = 0;
  for (const SSlot &sslot : session->sslot_arr) {
    if (sslot.server_info.req_type != kInvalidReqType) pending_enqueue_resps++;
  }

  if (pending_enqueue_resps == 0) {
    // Act similar to handling a disconnect request, but don't send SM response
    LOG_INFO("%s: None. Session resetted.\n", issue_msg);
    free_ring_entries();
    bury_session_st(session);
    return true;
  } else {
    LOG_WARN(
        "Rpc %u, lsn %u: Cannot reset server session. "
        "%zu enqueue_response pending.\n",
        rpc_id, session->local_session_num, pending_enqueue_resps);

    return false;
  }
}

FORCE_COMPILE_TRANSPORTS

}  // namespace erpc
<file_sep>/tests/client_tests/multi_process_test.cc
#include "client_tests.h"

std::atomic<size_t> num_processes_ready;

void req_handler(erpc::ReqHandle *req_handle, void *_c) {
  auto *c = reinterpret_cast<BasicAppContext *>(_c);

  auto &resp = req_handle->pre_resp_msgbuf;
  c->rpc->resize_msg_buffer(&resp, sizeof(size_t));
  c->rpc->enqueue_response(req_handle, &req_handle->pre_resp_msgbuf);
}

void cont_func(void *_c, void *) {
  auto *c = static_cast<BasicAppContext *>(_c);
  c->num_rpc_resps++;
}

// This threads acts as a process with a Nexus
void process_proxy_thread_func(size_t process_id, size_t num_processes) {
  auto uri = "localhost:" + std::to_string(kBaseSmUdpPort + process_id);
  Nexus nexus(uri, 0, 0);
  nexus.register_req_func(kTestReqType, req_handler);

  BasicAppContext c;
  Rpc<CTransport> rpc(&nexus, &c, 0, basic_sm_handler);
  c.rpc = &rpc;

  // Barrier
  num_processes_ready++;
  while (num_processes_ready != num_processes) {
    // Wait for Rpc objects to be registered with their Nexus
  }

  c.session_num_arr = new int[num_processes];
  c.req_msgbufs.resize(num_processes);
  c.resp_msgbufs.resize(num_processes);
  for (size_t i = 0; i < num_processes; i++) {
    if (i == process_id) continue;

    auto remote_uri = "localhost:" + std::to_string(kBaseSmUdpPort + i);
    c.session_num_arr[i] = c.rpc->create_session(remote_uri, 0);

    c.req_msgbufs[i] = c.rpc->alloc_msg_buffer_or_die(sizeof(size_t));
    c.resp_msgbufs[i] = c.rpc->alloc_msg_buffer_or_die(sizeof(size_t));
  }

  wait_for_sm_resps_or_timeout(c, num_processes - 1);
  assert(c.num_sm_resps == num_processes - 1);
  printf("Process %zu: All sessions connected\n", process_id);

  for (size_t i = 0; i < num_processes; i++) {
    if (i == process_id) continue;

    c.rpc->enqueue_request(c.session_num_arr[i], kTestReqType,
                           &c.req_msgbufs[i], &c.resp_msgbufs[i], cont_func,
                           nullptr);
  }

  wait_for_rpc_resps_or_timeout(c, num_processes - 1);
  assert(c.num_rpc_resps == num_processes - 1);
}

TEST(MultiProcessTest, TwoProcesses) {
  num_processes_ready = 0;
  std::thread process_proxy_thread_thread_1(process_proxy_thread_func, 0, 2);
  std::thread process_proxy_thread_thread_2(process_proxy_thread_func, 1, 2);

  process_proxy_thread_thread_1.join();
  process_proxy_thread_thread_2.join();
}

TEST(MultiProcessTest, MaxProcesses) {
  num_processes_ready = 0;
  const size_t num_processes = kMaxNumERpcProcesses;

  std::vector<std::thread> process_proxy_thread_vec(num_processes);

  for (size_t i = 0; i < num_processes; i++) {
    process_proxy_thread_vec[i] =
        std::thread(process_proxy_thread_func, i, num_processes);
  }

  for (size_t i = 0; i < num_processes; i++) {
    process_proxy_thread_vec[i].join();
  }
}

int main(int argc, char **argv) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}
<file_sep>/src/heartbeat_mgr.h
#pragma once

#include <queue>
#include <sstream>
#include <unordered_map>
#include "common.h"
#include "rpc_constants.h"
#include "sm_types.h"
#include "util/autorun_helpers.h"
#include "util/timer.h"
#include "util/udp_client.h"

namespace erpc {

/**
 * @brief A thread-safe heartbeat manager
 *
 * This class has two main tasks: First, it sends heartbeat messages to remote
 * processes at fixed intervals. Second, it checks for timeouts, also at fixed
 * intervals. These tasks are scheduled using a time-based priority queue.
 *
 * For efficiency, if a process has multiple sessions to a remote process,
 * only one instance of the remote URI is tracked.
 *
 * This heartbeat manager is designed to keep the CPU use of eRPC's management
 * thread close to zero in the steady state. An earlier version of eRPC's
 * timeout detection used a reliable UDP library called ENet, which had
 * non-negligible CPU use.
 */
class HeartbeatMgr {
 private:
  static constexpr bool kVerbose = true;
  enum class EventType : bool {
    kSend,  // Send a heartbeat to the remote URI
    kCheck  // Check that we have received a heartbeat from the remote URI
  };

  /// Heartbeat info in the priority queue
  class Event {
   public:
    EventType type;
    std::string rem_uri;  // The remote process for receiving/sending heartbeats
    uint64_t tsc;         // The time at which this event is triggered

    Event(EventType type, const std::string &rem_uri, uint64_t tsc)
        : type(type), rem_uri(rem_uri), tsc(tsc) {}
  };

  struct EventComparator {
    bool operator()(const Event &p1, const Event &p2) {
      return p1.tsc > p2.tsc;
    }
  };

 public:
  HeartbeatMgr(std::string hostname, uint16_t sm_udp_port, double freq_ghz,
               size_t machine_failure_timeout_ms)
      : hostname(hostname),
        sm_udp_port(sm_udp_port),
        freq_ghz(freq_ghz),
        creation_tsc(rdtsc()),
        failure_timeout_tsc(ms_to_cycles(machine_failure_timeout_ms, freq_ghz)),
        hb_send_delta_tsc(failure_timeout_tsc / 10),
        hb_check_delta_tsc(failure_timeout_tsc / 2) {}

  /// Add a remote URI to the tracking set
  void unlocked_add_remote(const std::string &remote_uri) {
    std::lock_guard<std::mutex> lock(heartbeat_mutex);
    if (kVerbose) {
      printf("heartbeat_mgr (%.0f us): Starting tracking URI %s\n",
             us_since_creation(rdtsc()), remote_uri.c_str());
    }

    size_t cur_tsc = rdtsc();

    map_last_hb_rx.emplace(remote_uri, cur_tsc);
    schedule_hb_send(remote_uri);
    schedule_hb_check(remote_uri);
  }

  /// Receive a heartbeat
  void unlocked_receive_hb(const SmPkt &sm_pkt) {
    std::lock_guard<std::mutex> lock(heartbeat_mutex);

    if (map_last_hb_rx.count(sm_pkt.client.uri()) == 0) {
      if (kVerbose) {
        printf("heartbeat_mgr (%.0f us): Ignoring heartbeat from URI %s\n",
               us_since_creation(rdtsc()), sm_pkt.client.uri().c_str());
      }
      return;
    }

    if (kVerbose) {
      printf("heartbeat_mgr (%.0f us): Receiving heartbeat from URI %s\n",
             us_since_creation(rdtsc()), sm_pkt.client.uri().c_str());
    }
    map_last_hb_rx.emplace(sm_pkt.client.uri(), rdtsc());
  }

  /**
   * @brief The main heartbeat work: Send heartbeats, and check expired timers
   *
   * @param failed_uris The list of failed remote URIs to fill-in
   */
  void do_one(std::vector<std::string> &failed_uris) {
    while (true) {
      if (hb_event_pqueue.empty()) break;

      const auto next_ev = hb_event_pqueue.top();  // Copy-out

      // Stop processing the event queue when the top event is in the future
      if (in_future(next_ev.tsc)) {
        if (kVerbose) {
          printf("heartbeat_mgr (%.0f us): Event %s is in the future\n",
                 us_since_creation(rdtsc()), ev_to_string(next_ev).c_str());
        }
        break;
      }

      if (map_last_hb_rx.count(next_ev.rem_uri) == 0) {
        if (kVerbose) {
          printf(
              "heartbeat_mgr (%.0f us): Remote URI for event %s is not "
              "in tracking set. Ignoring.\n",
              us_since_creation(rdtsc()), ev_to_string(next_ev).c_str());
        }
        break;
      }

      if (kVerbose) {
        printf("heartbeat_mgr (%.0f us): Handling event %s\n",
               us_since_creation(rdtsc()), ev_to_string(next_ev).c_str());
      }

      hb_event_pqueue.pop();  // Consume the event

      switch (next_ev.type) {
        case EventType::kSend: {
          SmPkt heartbeat =
              make_heartbeat(hostname, sm_udp_port, next_ev.rem_uri);
          hb_udp_client.send(heartbeat.server.hostname,
                             heartbeat.server.sm_udp_port, heartbeat);

          schedule_hb_send(next_ev.rem_uri);
          break;
        }

        case EventType::kCheck: {
          size_t last_ping_rx = map_last_hb_rx[next_ev.rem_uri];
          if (rdtsc() - last_ping_rx > failure_timeout_tsc) {
            failed_uris.push_back(next_ev.rem_uri);

            if (kVerbose) {
              printf("heartbeat_mgr (%.0f us): Remote URI %s failed\n",
                     us_since_creation(rdtsc()), next_ev.rem_uri.c_str());
            }
            map_last_hb_rx.erase(map_last_hb_rx.find(next_ev.rem_uri));
          } else {
            schedule_hb_check(next_ev.rem_uri.c_str());
          }
          break;
        }
      }
    }
  }

 private:
  // Create a heartbeat packet send by the local URI to the remote URI.
  //
  // A heartbeat packet is a session management packet where most fields are
  // invalid. The local sender acts as the SmPkt's client.
  static SmPkt make_heartbeat(const std::string &local_hostname,
                              uint16_t local_sm_udp_port,
                              const std::string &remote_uri) {
    SmPkt sm_hb;
    sm_hb.pkt_type = SmPktType::kPingReq;
    sm_hb.err_type = SmErrType::kNoError;
    sm_hb.uniq_token = 0;

    std::string remote_hostname;
    uint16_t remote_sm_udp_port;
    split_uri(remote_uri, remote_hostname, remote_sm_udp_port);

    // In sm_hb's session endpoints, the transport type, Rpc ID, and session
    // number are already invalid.
    strcpy(sm_hb.client.hostname, local_hostname.c_str());
    sm_hb.client.sm_udp_port = local_sm_udp_port;

    strcpy(sm_hb.server.hostname, remote_hostname.c_str());
    sm_hb.server.sm_udp_port = remote_sm_udp_port;

    return sm_hb;
  }

  /// Return the microseconds between tsc and this manager's creation time
  double us_since_creation(size_t tsc) const {
    return to_usec(tsc - creation_tsc, freq_ghz);
  }

  /// Return a string representation of a ping event. This is outside the
  /// Event class because we need creation_tsc.
  std::string ev_to_string(const Event &e) const {
    std::ostringstream ret;
    ret << "[Type: " << (e.type == EventType::kSend ? "send" : "check")
        << ", URI " << e.rem_uri << ", time "
        << static_cast<size_t>(us_since_creation(e.tsc)) << " us]";
    return ret.str();
  }

  /// Return true iff a timestamp is in the future
  static bool in_future(size_t tsc) { return tsc > rdtsc(); }

  void schedule_hb_send(const std::string &rem_uri) {
    Event e(EventType::kSend, rem_uri, rdtsc() + hb_send_delta_tsc);
    if (kVerbose) {
      printf("heartbeat_mgr (%.0f us): Scheduling event %s\n",
             us_since_creation(rdtsc()), ev_to_string(e).c_str());
    }
    hb_event_pqueue.push(e);
  }

  void schedule_hb_check(const std::string &rem_uri) {
    Event e(EventType::kCheck, rem_uri, rdtsc() + hb_check_delta_tsc);
    if (kVerbose) {
      printf("heartbeat_mgr (%.0f us): Scheduling event %s\n",
             us_since_creation(rdtsc()), ev_to_string(e).c_str());
    }
    hb_event_pqueue.push(e);
  }

  const std::string hostname;  /// This process's local hostname
  const uint16_t sm_udp_port;  /// This process's management UDP port

  const double freq_ghz;             /// TSC frequency
  const size_t creation_tsc;         /// Time at which this manager was created
  const size_t failure_timeout_tsc;  /// Machine failure timeout in TSC cycles

  /// We send heartbeats every every hb_send_delta_tsc cycles. This duration is
  /// around a tenth of the failure timeout.
  const size_t hb_send_delta_tsc;

  /// We check heartbeats every every hb_check_delta_tsc cycles. This duration
  /// is around half of the failure timeout #nyquist.
  const size_t hb_check_delta_tsc;

  std::priority_queue<Event, std::vector<Event>, EventComparator>
      hb_event_pqueue;

  // This map servers two purposes:
  //
  // 1. Its value for a remote URI is the timestamp when we last received a
  //    heartbeat from the remote URI.
  // 2. The set of remote URIs in the map is our remote tracking set. Since we
  //    cannot delete efficiently from the event priority queue, events for
  //    remote URIs not in this map are ignored when they are dequeued.
  std::unordered_map<std::string, uint64_t> map_last_hb_rx;

  UDPClient<SmPkt> hb_udp_client;

  std::mutex heartbeat_mutex;  // Protects this heartbeat manager
};
}
