active.h

Go to the documentation of this file.

/*
//@HEADER
// *****************************************************************************
//
//                                   active.h
//                       DARMA/vt => Virtual Transport
//
// Copyright 2019-2024 National Technology & Engineering Solutions of Sandia, LLC
// (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S.
// Government retains certain rights in this software.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
// * Redistributions of source code must retain the above copyright notice,
//   this list of conditions and the following disclaimer.
//
// * Redistributions in binary form must reproduce the above copyright notice,
//   this list of conditions and the following disclaimer in the documentation
//   and/or other materials provided with the distribution.
//
// * Neither the name of the copyright holder nor the names of its
//   contributors may be used to endorse or promote products derived from this
//   software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.
//
// Questions? Contact darma@sandia.gov
//
// *****************************************************************************
//@HEADER
*/
 
#if !defined INCLUDED_VT_MESSAGING_ACTIVE_H
#define INCLUDED_VT_MESSAGING_ACTIVE_H
 
#include <cstdint>
#include <memory>
#include <mpi.h>
 
#include "vt/config.h"
#include "vt/activefn/activefn.h"
#include "vt/messaging/active.fwd.h"
#include "vt/messaging/message/smart_ptr.h"
#include "vt/messaging/pending_send.h"
#include "vt/messaging/request_holder.h"
#include "vt/messaging/send_info.h"
#include "vt/messaging/async_op_wrapper.h"
#include "vt/messaging/param_msg.h"
#include "vt/event/event.h"
#include "vt/registry/auto/auto_registry_interface.h"
#include "vt/trace/trace_common.h"
#include "vt/utils/static_checks/functor.h"
#include "vt/runtime/component/component_pack.h"
#include "vt/elm/elm_id.h"
#include "vt/elm/elm_lb_data.h"
#include "vt/utils/strong/strong_type.h"
#include "vt/utils/fntraits/fntraits.h"
 
#if vt_check_enabled(trace_enabled)
  #include "vt/trace/trace_headers.h"
#endif
 
#include <type_traits>
#include <tuple>
#include <vector>
#include <unordered_map>
#include <limits>
#include <stack>
 
namespace vt {

using PtrLenPairType = std::tuple<std::byte*, ByteType>;

using ContinuationDeleterType =
  std::function<void(PtrLenPairType data, ActionType deleter)>;
 
} /* end namespace vt */
 
namespace vt {
 
struct StrongNodeType { };
using Node = Strong<NodeType, uninitialized_destination, StrongNodeType>;
 
namespace messaging {

 
using MPI_TagType = int;
 
static constexpr TagType const PutPackedTag =
  std::numeric_limits<TagType>::max();
 
enum class MPITag : MPI_TagType {
  ActiveMsgTag = 1,
  DataMsgTag   = 2,
 
  // Size-class active message tags
  ActiveMsgS   = 11,  // <= 512 bytes
  ActiveMsgM   = 12,  // <= 2048 bytes
  ActiveMsgL   = 13,  // <= 8192 bytes
  ActiveMsgXL  = 14   // <= 32768 bytes
};
 
static constexpr TagType const starting_direct_buffer_tag = 1000;
static constexpr MsgSizeType const max_pack_direct_size = 512;

struct PendingRecv {
  int nchunks = 0;
  std::byte* user_buf = nullptr;
  ContinuationDeleterType cont = nullptr;
  ActionType dealloc_user_buf = nullptr;
  NodeType sender = uninitialized_destination;
  PriorityType priority = no_priority;
  bool is_user_buf = false;
 
  PendingRecv(
    int in_nchunks, std::byte* in_user_buf, ContinuationDeleterType in_cont,
    ActionType in_dealloc_user_buf, NodeType node,
    PriorityType in_priority, bool in_is_user_buf
  ) : nchunks(in_nchunks), user_buf(in_user_buf), cont(in_cont),
      dealloc_user_buf(in_dealloc_user_buf), sender(node),
      priority(in_priority), is_user_buf(in_is_user_buf)
  { }
 
  template <typename Serializer>
  void serialize(Serializer& s) {
    s | nchunks
      | user_buf
      | cont
      | dealloc_user_buf
      | sender
      | priority
      | is_user_buf;
  }
};

struct InProgressBase {
  InProgressBase(
    std::byte* in_buf, MsgSizeType in_probe_bytes, NodeType in_sender
  ) : buf(in_buf), probe_bytes(in_probe_bytes), sender(in_sender),
      valid(true)
  { }
 
  template <typename Serializer>
  void serialize(Serializer& s) {
    s | buf
      | probe_bytes
      | sender
      | valid;
  }
 
  std::byte* buf = nullptr;
  MsgSizeType probe_bytes = 0;
  NodeType sender = uninitialized_destination;
  bool valid = false;
};

struct InProgressIRecv : InProgressBase {
 
  InProgressIRecv(
    std::byte* in_buf, MsgSizeType in_probe_bytes, NodeType in_sender,
    MPI_Request in_req = MPI_REQUEST_NULL
  ) : InProgressBase(in_buf, in_probe_bytes, in_sender),
      req(in_req)
  { }
 
  bool test(int& num_mpi_tests) {
    VT_ALLOW_MPI_CALLS; // MPI_Test
 
    int flag = 0;
    MPI_Status stat;
    MPI_Test(&req, &flag, &stat);
    num_mpi_tests++;
    return flag;
  }
 
private:
  MPI_Request req = MPI_REQUEST_NULL;
};

struct InProgressDataIRecv : InProgressBase {
  InProgressDataIRecv(
    std::byte* in_buf, MsgSizeType in_probe_bytes, NodeType in_sender,
    std::vector<MPI_Request> in_reqs, std::byte* const in_user_buf,
    ActionType in_dealloc_user_buf,
    ContinuationDeleterType in_next,
    PriorityType in_priority
  ) : InProgressBase{in_buf, in_probe_bytes, in_sender},
      user_buf(in_user_buf), dealloc_user_buf(in_dealloc_user_buf),
      next(in_next), priority(in_priority), reqs(std::move(in_reqs))
  { }
 
  bool test(int& num_mpi_tests) {
    int flag = 0;
    MPI_Status stat;
    for ( ; cur < reqs.size(); cur++) {
      VT_ALLOW_MPI_CALLS; // MPI_Test
 
      MPI_Test(&reqs[cur], &flag, &stat);
      num_mpi_tests++;
 
      if (flag == 0) {
        return false;
      }
    }
 
    return true;
  }
 
  template <typename Serializer>
  void serialize(Serializer& s) {
    s | user_buf
      | dealloc_user_buf
      | next
      | priority
      | cur
      | reqs;
  }
 
  std::byte* user_buf = nullptr;
  ActionType dealloc_user_buf = nullptr;
  ContinuationDeleterType next = nullptr;
  PriorityType priority = no_priority;
 
private:
  std::size_t cur = 0;
  std::vector<MPI_Request> reqs;
};

struct BufferedActiveMsg {
  using MessageType = MsgSharedPtr<BaseMsgType>;
 
  MessageType buffered_msg;
  NodeType from_node;
  ActionType cont;
 
  BufferedActiveMsg(
    MessageType const& in_buffered_msg, NodeType const& in_from_node,
    ActionType in_cont
  ) : buffered_msg(in_buffered_msg), from_node(in_from_node), cont(in_cont)
  { }
 
  template <typename Serializer>
  void serialize(Serializer& s) {
    s | buffered_msg
      | from_node
      | cont;
  }
};
 
// forward-declare for header
struct MultiMsg;

struct ActiveMessenger : runtime::component::PollableComponent<ActiveMessenger> {
  using BufferedMsgType      = BufferedActiveMsg;
  using MessageType          = ShortMessage;
  using CountType            = int32_t;
  using PendingRecvType      = PendingRecv;
  using EventRecordType      = event::AsyncEvent::EventRecordType;
  using SendFnType           = std::function<SendInfo(PtrLenPairType,NodeType,TagType)>;
  using UserSendFnType       = std::function<void(SendFnType)>;
  using ContainerPendingType = std::unordered_map<TagType,PendingRecvType>;
  using ReadyHanTagType      = std::tuple<HandlerType, TagType>;
  using HandlerManagerType   = HandlerManager;
  using PendingSendType      = PendingSend;

  ActiveMessenger();

  virtual ~ActiveMessenger();
 
  std::string name() override { return "ActiveMessenger"; }
 
  void startup() override;
  void initialize() override;

  template <typename MsgPtrT>
  void markAsTermMessage(MsgPtrT const msg);

  template <typename MsgPtrT>
  void markAsLocationMessage(MsgPtrT const msg);

  template <typename MsgPtrT>
  void markAsSerialMsgMessage(MsgPtrT const msg);

  template <typename MsgPtrT>
  void markAsCollectionMessage(MsgPtrT const msg);

  template <typename MsgT>
  void setEpochMessage(MsgT* msg, EpochType epoch);

  template <typename MsgT>
  void setTagMessage(MsgT* msg, TagType tag);
 
  trace::TraceEventIDType makeTraceCreationSend(
    HandlerType const handler, ByteType serialized_msg_size, bool is_bcast
  );
 
  // With serialization, the correct method is resolved via SFINAE.
  // This also includes additional guards to detect ambiguity.
  template <typename MsgT>
  ActiveMessenger::PendingSendType sendMsgSerializableImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  );
 
  // All messages that do NOT define their own serialization policy
  // and do NOT define their own serialization function are required
  // to be byte-transmittable. This covers basic byte-copyable
  // messages directly inheriting from ActiveMsg. ActivMsg implements
  // a serialize function which is implicitly inherited..
  template <
    typename MsgT,
    std::enable_if_t<true
      and not ::vt::messaging::msg_defines_serialize_mode<MsgT>::value,
      int
    > = 0
  >
  inline ActiveMessenger::PendingSendType sendMsgImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  ) {
#ifndef vt_quirked_serialize_method_detection
    static_assert(
      not ::vt::messaging::has_own_serialize<MsgT>,
      "Message prohibiting serialization must not have a serialization function."
    );
#endif
    return sendMsgCopyableImpl<MsgT>(dest, han, msg, tag);
  }
 
  // Serializable and serialization required on this type.
  template <
    typename MsgT,
    std::enable_if_t<true
      and ::vt::messaging::msg_defines_serialize_mode<MsgT>::value
      and ::vt::messaging::msg_serialization_mode<MsgT>::required,
      int
    > = 0
  >
  inline ActiveMessenger::PendingSendType sendMsgImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  ) {
#ifndef vt_quirked_serialize_method_detection
    static_assert(
      ::vt::messaging::has_own_serialize<MsgT>,
      "Message requiring serialization must have a serialization function."
    );
#endif
    return sendMsgSerializableImpl<MsgT>(dest, han, msg, tag);
  }
 
  // Serializable, but support is only for derived types.
  // This type will still be sent using byte-copy serialization.
  template <
    typename MsgT,
    std::enable_if_t<true
      and ::vt::messaging::msg_defines_serialize_mode<MsgT>::value
      and ::vt::messaging::msg_serialization_mode<MsgT>::supported,
      int
    > = 0
  >
  inline ActiveMessenger::PendingSendType sendMsgImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  ) {
#ifndef vt_quirked_serialize_method_detection
    static_assert(
       ::vt::messaging::has_own_serialize<MsgT>,
       "Message supporting serialization must have a serialization function."
     );
#endif
    return sendMsgCopyableImpl<MsgT>(dest, han, msg, tag);
  }
 
  // Messaged marked as prohibiting serialization cannot define
  // a serialization function and must be sent via byte-transmission.
  template <
    typename MsgT,
    std::enable_if_t<true
      and ::vt::messaging::msg_defines_serialize_mode<MsgT>::value
      and ::vt::messaging::msg_serialization_mode<MsgT>::prohibited,
      int
    > = 0
  >
  inline ActiveMessenger::PendingSendType sendMsgImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  ) {
#ifndef vt_quirked_serialize_method_detection
    static_assert(
      not ::vt::messaging::has_own_serialize<MsgT>,
      "Message prohibiting serialization must not have a serialization function."
    );
#endif
    return sendMsgCopyableImpl<MsgT>(dest, han, msg, tag);
  }
 
  template <typename MsgT>
  ActiveMessenger::PendingSendType sendMsgCopyableImpl(
    NodeType dest,
    HandlerType han,
    MsgSharedPtr<MsgT>& msg,
    TagType tag
  );


  template <typename MsgT>
  [[deprecated("size must be set in makeMessageSz, use regular sendMsg")]]
  PendingSendType sendMsgSz(
    NodeType dest,
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    ByteType msg_size,
    TagType tag = no_tag
  );

  template <typename MsgT>
  PendingSendType sendMsg(
    NodeType dest,
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

  template <typename MsgT>
  PendingSendType sendMsgAuto(
    NodeType dest,
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

 
  /*
   *----------------------------------------------------------------------------
   *          End Basic Active Message Send with Pre-Registered Handler
   *----------------------------------------------------------------------------
   */


  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType broadcastMsgSz(
    MsgPtrThief<MsgT> msg,
    ByteType msg_size,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType broadcastMsg(
    MsgPtrThief<MsgT> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <auto f>
  PendingSendType broadcastMsg(
    MsgPtrThief<typename FuncTraits<decltype(f)>::MsgT> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  ) {
    using MsgT = typename FuncTraits<decltype(f)>::MsgT;
    return broadcastMsg<MsgT, f>(msg, deliver_to_sender, tag);
  }

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

  template <auto f>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<typename FuncTraits<decltype(f)>::MsgT> msg,
    TagType tag = no_tag
  ) {
    using MsgT = typename FuncTraits<decltype(f)>::MsgT;
    return sendMsg<MsgT, f>(dest, msg, tag);
  }

  template <auto f, typename... Params>
  PendingSendType send(Node dest, Params&&... params) {
    using Tuple = typename FuncTraits<decltype(f)>::TupleType;
    using MsgT = ParamMsg<Tuple>;
    auto msg = vt::makeMessage<MsgT>();
    msg->setParams(std::forward<Params>(params)...);
    auto han = auto_registry::makeAutoHandlerParam<decltype(f), f, MsgT>();
    return sendMsg<MsgT>(dest.get(), han, msg, no_tag);
  }

  template <auto f, typename... Params>
  PendingSendType broadcast(Params&&... params) {
    using Tuple = typename FuncTraits<decltype(f)>::TupleType;
    using MsgT = ParamMsg<Tuple>;
    auto msg = vt::makeMessage<MsgT>();
    msg->setParams(std::forward<Params>(params)...);
    auto han = auto_registry::makeAutoHandlerParam<decltype(f), f, MsgT>();
    constexpr bool deliver_to_sender = true;
    return broadcastMsg<MsgT>(han, msg, deliver_to_sender, no_tag);
  }

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  [[deprecated("size must be set in makeMessageSz, use regular sendMsg")]]
  PendingSendType sendMsgSz(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    ByteType msg_size,
    TagType tag = no_tag
  );

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType broadcastMsgAuto(
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType sendMsgAuto(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

 
  /*
   *----------------------------------------------------------------------------
   *             End Send Message Active Function (type-safe handler)
   *----------------------------------------------------------------------------
   */


  template <ActiveFnType* f, typename MsgT>
  PendingSendType broadcastMsg(
    MsgPtrThief<MsgT> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <ActiveFnType* f, typename MsgT>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

 
  /*
   *----------------------------------------------------------------------------
   *              End Send Message BASIC Active Function (deprecated?)
   *----------------------------------------------------------------------------
   */


  template <
    typename FunctorT,
    typename MsgT
  >
  PendingSendType broadcastMsg(
    MsgPtrThief<MsgT> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <typename FunctorT>
  PendingSendType broadcastMsg(
    MsgPtrThief<typename util::FunctorExtractor<FunctorT>::MessageType> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <
    typename FunctorT,
    typename MsgT = typename util::FunctorExtractor<FunctorT>::MessageType
  >
  PendingSendType broadcastMsgAuto(
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

  template <typename FunctorT,typename MsgT>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

  template <typename FunctorT>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<typename util::FunctorExtractor<FunctorT>::MessageType> msg,
    TagType tag = no_tag
  );

  template <
    typename FunctorT,
    typename MsgT = typename util::FunctorExtractor<FunctorT>::MessageType
  >
  PendingSendType sendMsgAuto(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

 
  /*
   *----------------------------------------------------------------------------
   *                      End Send Message Functor Variants
   *----------------------------------------------------------------------------
   */


  template <typename MsgT>
  PendingSendType broadcastMsg(
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    bool deliver_to_sender = true,
    TagType tag = no_tag
  );

  template <typename MsgT>
  PendingSendType sendMsg(
    NodeType dest,
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    UserSendFnType send_payload_fn
  );

  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
  PendingSendType sendMsg(
    NodeType dest,
    MsgPtrThief<MsgT> msg,
    UserSendFnType send_payload_fn
  );

  template <typename MsgT>
  PendingSendType broadcastMsgAuto(
    HandlerType han,
    MsgPtrThief<MsgT> msg,
    TagType tag = no_tag
  );

 
  /*
   *----------------------------------------------------------------------------
   *                           End Send Data Message
   *----------------------------------------------------------------------------
   */

  MsgSizeType packMsg(
    MessageType* msg, MsgSizeType size, std::byte* ptr, MsgSizeType ptr_bytes
  );

  SendInfo sendData(
    PtrLenPairType const& ptr, NodeType const& dest, TagType const& tag
  );

  std::tuple<EventType, int> sendDataMPI(
    PtrLenPairType const& ptr, NodeType const& dest, TagType const& tag
  );

  bool recvDataMsgPriority(
    int nchunks, PriorityType priority, TagType const& tag,
    NodeType const& node, ContinuationDeleterType next = nullptr
  );

  bool recvDataMsg(
    int nchunks, TagType const& tag, NodeType const& node,
    ContinuationDeleterType next = nullptr
  );

  bool recvDataMsg(
    int nchunks, PriorityType priority, TagType const& tag,
    NodeType const& sender, bool const& enqueue,
    ContinuationDeleterType next = nullptr
  );

  bool recvDataMsgBuffer(
    int nchunks, std::byte* const user_buf, PriorityType priority, TagType const& tag,
    NodeType const& node = uninitialized_destination, bool const& enqueue = true,
    ActionType dealloc_user_buf = nullptr,
    ContinuationDeleterType next = nullptr, bool is_user_buf = false
  );

  bool recvDataMsgBuffer(
    int nchunks, std::byte* const user_buf, TagType const& tag,
    NodeType const& node = uninitialized_destination, bool const& enqueue = true,
    ActionType dealloc_user_buf = nullptr,
    ContinuationDeleterType next = nullptr, bool is_user_buf = false
  );

  void recvDataDirect(
    int nchunks, std::byte* const buf, TagType const tag, NodeType const from,
    MsgSizeType len, PriorityType prio, ActionType dealloc = nullptr,
    ContinuationDeleterType next = nullptr, bool is_user_buf = false
#if MPI_VERSION >= 3
    , MPI_Message first_msg = MPI_MESSAGE_NULL, MsgSizeType first_chunk_bytes = 0
#endif
  );

  void recvDataDirect(
    int nchunks, TagType const tag, NodeType const from,
    MsgSizeType len, ContinuationDeleterType next
  );

  EventType doMessageSend(
    MsgSharedPtr<BaseMsgType>& msg
  );

  bool tryProcessIncomingActiveMsg();

  bool tryProcessDataMsgRecv();

  int progress(TimeType current_time) override;

  void processActiveMsg(
    MsgSharedPtr<BaseMsgType> const& base, NodeType const& sender,
    bool insert, ActionType cont = nullptr
  );

  void prepareActiveMsgToRun(
    MsgSharedPtr<BaseMsgType> const& base, NodeType const& from_node,
    bool insert, ActionType cont
  );

  EventType sendMsgBytes(
    NodeType const& dest, MsgSharedPtr<BaseMsgType> const& base,
    MsgSizeType const& msg_size, TagType const& send_tag
  );

  EventType sendMsgBytesWithPut(
    NodeType const& dest, MsgSharedPtr<BaseMsgType> const& base
  );

  EventType sendMsgMPI(
    NodeType const& dest, MsgSharedPtr<BaseMsgType> const& base,
    MsgSizeType const& msg_size, TagType const& send_tag
  );

  inline void pushEpoch(EpochType const& epoch);

  inline EpochType popEpoch(EpochType const& epoch = no_epoch);

  inline EpochType getEpoch() const;

  template <typename MsgT>
  inline EpochType getEpochContextMsg(MsgT* msg);

  template <typename MsgT>
  inline EpochType getEpochContextMsg(MsgSharedPtr<MsgT> const& msg);

  template <typename MsgT>
  inline EpochType setupEpochMsg(MsgT* msg);

  template <typename MsgT>
  inline EpochType setupEpochMsg(MsgSharedPtr<MsgT> const& msg);

  void registerAsyncOp(std::unique_ptr<AsyncOp> op);

  void blockOnAsyncOp(std::unique_ptr<AsyncOp> op);
 
  template <typename SerializerT>
  void serialize(SerializerT& s) {
    s | pending_recvs_
      | cur_direct_buffer_tag_
      | in_progress_active_msg_irecv
      | in_progress_data_irecv
      | in_progress_ops
      | this_node_
      | amForwardCounterGauge
      | amHandlerCount
      | amPollCount
      | amPostedCounterGauge
      | amRecvCounterGauge
      | amSentCounterGauge
      | bcastsSentCount
      | dmPollCount
      | dmPostedCounterGauge
      | dmRecvCounterGauge
      | dmSentCounterGauge
      | tdRecvCount
      | tdSentCount;
 
  # if vt_check_enabled(trace_enabled)
    s | trace_irecv
      | trace_isend
      | trace_irecv_polling_am
      | trace_irecv_polling_dm
      | trace_asyncop;
  # endif
  }
 
private:
  MPI_TagType allocateNewTag();

  void handleChunkedMultiMsg(MultiMsg* msg);

  static void chunkedMultiMsg(MultiMsg* msg);

  bool testPendingActiveMsgAsyncRecv();

  bool testPendingDataMsgAsyncRecv();

  bool testPendingAsyncOps();

  void finishPendingActiveMsgAsyncRecv(InProgressIRecv* irecv);

  void finishPendingDataMsgAsyncRecv(InProgressDataIRecv* irecv);

  void recordLBDataCommForSend(
    NodeType const dest, MsgSharedPtr<BaseMsgType> const& base,
    MsgSizeType const msg_size
  );

  static MPI_TagType selectActiveTag(MsgSizeType size);

  struct ActiveRecvBroker {
    struct Slot {
      int cap = 0;
      MPI_TagType tag = 0;
      std::byte* buf = nullptr;
      MPI_Request req = MPI_REQUEST_NULL;
      bool posted = false;
    };
 
    void setup(ActiveMessenger* self);
    bool progress(ActiveMessenger* self);
 
    static constexpr int num_caps_ = 4;
    static constexpr int caps_[num_caps_] = {512, 2048, 8192, 32768};
  private:
    std::vector<Slot> slots_;
    static constexpr MPI_TagType tags_[num_caps_] = {
      static_cast<MPI_TagType>(MPITag::ActiveMsgS),
      static_cast<MPI_TagType>(MPITag::ActiveMsgM),
      static_cast<MPI_TagType>(MPITag::ActiveMsgL),
      static_cast<MPI_TagType>(MPITag::ActiveMsgXL)
    };
    static constexpr int slots_per_class_ = 4;
 
    void postSlot(ActiveMessenger* self, Slot& s);
  public:
    // Cleanup outstanding posted receives and buffers to avoid leaks at shutdown
    void cleanup();
  };
 
public:
  auto getRankLBData() {
    return std::make_tuple(
      &bare_handler_dummy_elm_id_for_lb_data_,
      &bare_handler_lb_data_
    );
  }
 
private:
# if vt_check_enabled(trace_enabled)
  trace::UserEventIDType trace_irecv             = trace::no_user_event_id;
  trace::UserEventIDType trace_isend             = trace::no_user_event_id;
  trace::UserEventIDType trace_irecv_polling_am  = trace::no_user_event_id;
  trace::UserEventIDType trace_irecv_polling_dm  = trace::no_user_event_id;
  trace::UserEventIDType trace_asyncop           = trace::no_user_event_id;
# endif
 
  ContainerPendingType pending_recvs_                     = {};
  TagType cur_direct_buffer_tag_                          = starting_direct_buffer_tag;
  RequestHolder<InProgressIRecv> in_progress_active_msg_irecv;
  RequestHolder<InProgressDataIRecv> in_progress_data_irecv;
  RequestHolder<AsyncOpWrapper> in_progress_ops;
  NodeType this_node_                                     = uninitialized_destination;
 
private:
  // Diagnostic counter gauge combos for sent counts/bytes
  diagnostic::CounterGauge amSentCounterGauge;
  diagnostic::CounterGauge dmSentCounterGauge;
 
  // Diagnostic counters for recv counts/bytes
  diagnostic::CounterGauge amRecvCounterGauge;
  diagnostic::CounterGauge dmRecvCounterGauge;
 
  // Diagnostic counters for posted irecv counts/bytes
  diagnostic::CounterGauge amPostedCounterGauge;
  diagnostic::CounterGauge dmPostedCounterGauge;
 
  // Diagnostic counters for counting various actions
  diagnostic::Counter amHandlerCount;
  diagnostic::Counter bcastsSentCount;
  diagnostic::Counter amPollCount;
  diagnostic::Counter dmPollCount;
  diagnostic::Counter tdSentCount;
  diagnostic::Counter tdRecvCount;
 
  // Diagnostic counters for counting forwarded messages
  diagnostic::CounterGauge amForwardCounterGauge;
 
private:
  elm::ElementIDStruct bare_handler_dummy_elm_id_for_lb_data_ = {};
  elm::ElementLBData bare_handler_lb_data_;
  MPI_Comm comm_ = MPI_COMM_NULL;
 
  // Active receive broker instance
  ActiveRecvBroker active_broker_;
};
 
}} // end namespace vt::messaging
 
namespace vt {
 
using Active = messaging::ActiveMessenger;
 
extern messaging::ActiveMessenger* theMsg();
 
} // end namespace vt
 
#include "vt/messaging/active.impl.h"
 
#endif /*INCLUDED_VT_MESSAGING_ACTIVE_H*/