A class describing a message sent between nodes in a cluster. More...

#include <ClusterMessage.h>

Collaboration diagram for ClusterMessage:

Classes
struct	WorkerStatus
struct	DataDescr

Public Types
using	WireMsgHdr = std::array<std::uint32_t, 3>
using	WireMsgBody = std::array<std::uint32_t, 10>
using	WireMsg = std::tuple<WireMsgHdr, std::optional<WireMsgBody>>
using	Payload_t = std::variant<std::monostate, int, WorkerStatus, DataDescr>

Public Member Functions
	ClusterMessage ()
	ClusterMessage (ClusterMessageType mType)
	ClusterMessage (ClusterMessageType mType, int payload)
	ClusterMessage (ClusterMessageType mType, WorkerStatus payload)
	ClusterMessage (ClusterMessageType mType, DataDescr &&payload)
	ClusterMessage (const WireMsg &)
WireMsg	wire_msg () const

Public Attributes
int	source = -1
ClusterMessageType	messageType {ClusterMessageType::EMPTY}
Payload_t	payload {}

Detailed Description

A class describing a message sent between nodes in a cluster.

Definition at line 31 of file ClusterMessage.h.

Member Typedef Documentation

◆ Payload_t

using ClusterMessage::Payload_t = std::variant<std::monostate, int, WorkerStatus, DataDescr>

Definition at line 83 of file ClusterMessage.h.

◆ WireMsg

using ClusterMessage::WireMsg = std::tuple<WireMsgHdr, std::optional<WireMsgBody>>

Definition at line 41 of file ClusterMessage.h.

◆ WireMsgBody

using ClusterMessage::WireMsgBody = std::array<std::uint32_t, 10>

Definition at line 39 of file ClusterMessage.h.

◆ WireMsgHdr

using ClusterMessage::WireMsgHdr = std::array<std::uint32_t, 3>

Definition at line 37 of file ClusterMessage.h.

Constructor & Destructor Documentation

◆ ClusterMessage() [1/6]

ClusterMessage::ClusterMessage ( )

default

◆ ClusterMessage() [2/6]

ClusterMessage::ClusterMessage ( ClusterMessageType mType )

Definition at line 83 of file ClusterMessage.cxx.

                                                       : messageType(mType) {
  switch (mType) {
    case ClusterMessageType::RequestEvent:
    case ClusterMessageType::EventsDone:
    case ClusterMessageType::EmergencyStop:
      // OK
      break;
    default:
      throw std::logic_error{std::format(
          "Incorrect ClusterMessage constructor used for message type {}",
          mType)};
  }
}

◆ ClusterMessage() [3/6]

ClusterMessage::ClusterMessage	(	ClusterMessageType	mType,
		int	payload )

Definition at line 65 of file ClusterMessage.cxx.

    : messageType(mType), payload(payload) {
  if (mType != ClusterMessageType::ProvideEvent) {
    throw std::logic_error{std::format(
        "Incorrect ClusterMessage constructor used for message type {}",
        mType)};
  }
}

◆ ClusterMessage() [4/6]

ClusterMessage::ClusterMessage	(	ClusterMessageType	mType,
		WorkerStatus	payload )

Definition at line 55 of file ClusterMessage.cxx.

    : messageType(mType), payload(payload) {
  if (mType != ClusterMessageType::FinalWorkerStatus &&
      mType != ClusterMessageType::WorkerError) {
    throw std::logic_error{std::format(
        "Incorrect ClusterMessage constructor used for message type {}",
        mType)};
  }
}

◆ ClusterMessage() [5/6]

ClusterMessage::ClusterMessage	(	ClusterMessageType	mType,
		DataDescr &&	payload )

Definition at line 74 of file ClusterMessage.cxx.

    : messageType(mType), payload(std::move(payload)) {
  if (mType != ClusterMessageType::Data) {
    throw std::logic_error{std::format(
        "Incorrect ClusterMessage constructor used for message type {}",
        mType)};
  }
}

◆ ClusterMessage() [6/6]

ClusterMessage::ClusterMessage ( const WireMsg & wire_msg )

Definition at line 99 of file ClusterMessage.cxx.

                                                                    {
  const auto& [header, body] = wire_msg;
  messageType = static_cast<ClusterMessageType>(header[0]);
  source = header[1];
  if (body.has_value()) {
    const auto& body_2 = *body;
    if (messageType == ClusterMessageType::Data) {
      payload = DataDescr(body_2);
    } else {
      WorkerStatus status{};
      status.status = StatusCode(body_2[0]);
      status.createdEvents = body_2[1];
      status.finishedEvents = body_2[2];
      status.skippedEvents = body_2[3];
      payload = status;
    }
  } else {
    if (messageType == ClusterMessageType::ProvideEvent) {
      payload = int(header[2]);
    }
  }
}

Member Function Documentation

◆ wire_msg()

ClusterMessage::WireMsg ClusterMessage::wire_msg ( ) const

nodiscard

Definition at line 122 of file ClusterMessage.cxx.

                                                     {
  constexpr int max_tag = 16383;
  constexpr std::uint64_t lower32 = 0xFFFFFFFF;
 
  static thread_local int next_msg =
      1;  // This is only ever called from one thread per process
  WireMsgHdr header{};
  header[0] = std::uint32_t(messageType);
  header[1] = source;
  if (payload.index() == 3) {
    next_msg = (next_msg % max_tag) + 1;
    header[2] = next_msg;
    WireMsgBody body{};
    const auto& payload_local = std::get<DataDescr>(payload);
    body[0] = std::uint32_t(std::uint64_t(payload_local.ptr) >> 32);
    body[1] = std::uint32_t(std::uint64_t(payload_local.ptr) & lower32);
    body[2] = std::uint32_t(std::uint64_t(payload_local.len) >> 32);
    body[3] = std::uint32_t(std::uint64_t(payload_local.len) & lower32);
    body[4] = std::uint32_t(std::uint64_t(payload_local.align) >> 32);
    body[5] = std::uint32_t(std::uint64_t(payload_local.align) & lower32);
    body[6] = std::uint32_t(std::uint64_t(payload_local.evtNumber) >> 32);
    body[7] = std::uint32_t(std::uint64_t(payload_local.evtNumber) & lower32);
    body[8] = std::uint32_t(std::uint64_t(payload_local.fileNumber) >> 32);
    body[9] = std::uint32_t(std::uint64_t(payload_local.fileNumber) & lower32);
    WireMsg msg{header, std::make_optional(body)};
    return msg;
  }
  if (payload.index() == 2) {
    next_msg = (next_msg % max_tag) + 1;
    header[2] = next_msg;
    WireMsgBody body{};
    const auto& payload_local = std::get<WorkerStatus>(payload);
    body[0] = static_cast<int>(payload_local.status.getCode());
    body[1] = payload_local.createdEvents;
    body[2] = payload_local.finishedEvents;
    body[3] = payload_local.skippedEvents;
    body[4] = body[5] = body[6] = body[7] = body[8] = body[9] = 0;
    WireMsg msg{header, std::make_optional(body)};
    return msg;
  }
  // else
  if (payload.index() == 1) {  // if we have an int payload
    header[2] = std::get<int>(payload);
  } else {
    header[2] = 0;
  }
  WireMsg msg{header, std::nullopt};
  return msg;
}

Member Data Documentation

◆ messageType

ClusterMessageType ClusterMessage::messageType {ClusterMessageType::EMPTY}

Definition at line 44 of file ClusterMessage.h.

44{ClusterMessageType::EMPTY};

◆ payload

Payload_t ClusterMessage::payload {}

Definition at line 84 of file ClusterMessage.h.

84{};

◆ source

int ClusterMessage::source = -1

Definition at line 43 of file ClusterMessage.h.

The documentation for this class was generated from the following files:

Classes

Public Types

Public Member Functions

Public Attributes