lci.hpp

Go to the documentation of this file.

// Copyright (c) 2025 The LCI Project Authors
// SPDX-License-Identifier: MIT
 
#ifndef LCI_API_LCI_HPP
#define LCI_API_LCI_HPP
 
#include <memory>
#include <stdexcept>
#include <vector>
#include <string>
#include <cstring>
#include <functional>
 
#include "lci_config.hpp"









namespace lci
{
enum class attr_backend_t {
  none,
  ibv,
  ofi,
  ucx,
};
 
enum attr_net_lock_mode_t {
  LCI_NET_TRYLOCK_SEND = 1,
  LCI_NET_TRYLOCK_RECV = 1 << 1,
  LCI_NET_TRYLOCK_POLL = 1 << 2,
  LCI_NET_TRYLOCK_MAX = 1 << 3,
};
 
// mimic std::optional as we don't want to force c++17 for now
template <typename T>
struct option_t {
  option_t() : m_value(), m_is_set(false) {}
  option_t(T value_) : m_value(value_), m_is_set(true) {}
  option_t(T value_, bool is_set_)
      : m_value(value_), m_is_set(is_set_) {}  // set default value
  T get_value_or(T default_value) const
  {
    return m_is_set ? m_value : default_value;
  }
  bool get_set_value(T* value) const
  {
    if (m_is_set) {
      *value = this->m_value;
      return true;
    }
    return false;
  }
  T get_value() const { return m_value; }
  bool is_set() const { return m_is_set; }
  operator T() const { return m_value; }
  T m_value;
  bool m_is_set;
};
}  // namespace lci
 
#include "lci_binding_pre.hpp"
 
namespace lci
{
enum class errorcode_t {
  done_min,     
  done,         
  done_backlog, 
  done_max,     
  posted_min,   
  posted, 
  posted_backlog, 
  posted_max,     
  retry_min,      
  retry, 
  retry_init, 
  retry_lock, 
  retry_nopacket, 
  retry_nomem,   
  retry_backlog, 
  retry_max,     
  fatal, 
};

const char* get_errorcode_str(errorcode_t errorcode);

struct error_t {
  errorcode_t errorcode;
  error_t() : errorcode(errorcode_t::retry_init) {}
  error_t(errorcode_t errorcode_) : errorcode(errorcode_) {}
  void reset_retry() { errorcode = errorcode_t::retry; }
  bool is_done() const
  {
    return errorcode > errorcode_t::done_min &&
           errorcode < errorcode_t::done_max;
  }

  bool is_posted() const
  {
    return errorcode > errorcode_t::posted_min &&
           errorcode < errorcode_t::posted_max;
  }

  bool is_retry() const
  {
    return errorcode > errorcode_t::retry_min &&
           errorcode < errorcode_t::retry_max;
  }

  const char* get_str() const { return lci::get_errorcode_str(errorcode); }
};

enum class net_opcode_t {
  SEND,         
  RECV,         
  WRITE,        
  REMOTE_WRITE, 
  READ,         
};

const char* get_net_opcode_str(net_opcode_t opcode);

enum class broadcast_algorithm_t {
  none,   
  direct, 
  tree,   
  ring,   
};

const char* get_broadcast_algorithm_str(broadcast_algorithm_t algorithm);

enum class reduce_scatter_algorithm_t {
  none,   
  direct, 
  tree,   
  ring,   
};

const char* get_reduce_scatter_algorithm_str(broadcast_algorithm_t algorithm);

enum class allreduce_algorithm_t {
  none,   
  direct, 
  tree,   
  ring,   
};

const char* get_allreduce_algorithm_str(broadcast_algorithm_t algorithm);

using net_imm_data_t = uint32_t;

struct net_status_t {
  net_opcode_t opcode;
  int rank;
  void* user_context;
  size_t length;
  net_imm_data_t imm_data;
};

const mr_t MR_HOST = mr_t();

const mr_t MR_DEVICE = mr_t(reinterpret_cast<void*>(0x1));

const mr_t MR_UNKNOWN = mr_t(reinterpret_cast<void*>(0x2));
 
inline bool mr_t::is_empty() const
{
  return reinterpret_cast<uintptr_t>(p_impl) < 3;
}

struct rmr_t {
  uintptr_t base;
  uint64_t opaque_rkey;
  rmr_t() : base(0), opaque_rkey(0) {}
  bool is_empty() const { return base == 0 && opaque_rkey == 0; }
};

const rmr_t RMR_NULL = rmr_t();

using tag_t = uint64_t;

enum class direction_t {
  OUT, 
  IN,  
};

using rcomp_t = uint32_t;

const int ANY_SOURCE = -1;

const tag_t ANY_TAG = static_cast<tag_t>(-1);

enum class matching_entry_type_t : unsigned {
  send = 0, 
  recv = 1, 
};

enum class matching_policy_t : unsigned {
  none = 0,      
  rank_only = 1, 
  tag_only = 2,  
  rank_tag = 3,  
  max = 4,       
};

using matching_entry_key_t = uint64_t;
using matching_entry_val_t = void*;

enum class comp_semantic_t {
  buffer,  
  network, 
};

struct buffer_t {
  void* base;  
  size_t size; 
  mr_t mr;     
  buffer_t() : base(nullptr), size(0), mr() {}
  buffer_t(void* base_, size_t size_) : base(base_), size(size_) {}
  buffer_t(void* base_, size_t size_, mr_t mr_)
      : base(base_), size(size_), mr(mr_)
  {
  }
};

struct rbuffer_t {
  uintptr_t
      disp;  
  rmr_t rmr; 
  rbuffer_t() : disp(0) {}
  rbuffer_t(uintptr_t disp_) : disp(disp_) {}
  rbuffer_t(uintptr_t disp_, rmr_t rmr_) : disp(disp_), rmr(rmr_) {}
};
using buffers_t = std::vector<buffer_t>;
using rbuffers_t = std::vector<rbuffer_t>;

struct allocator_base_t {
  virtual void* allocate(size_t size) = 0;
  virtual void deallocate(void* ptr) = 0;
  virtual ~allocator_base_t() = default;
};

struct data_t {
  static const int MAX_SCALAR_SIZE = 23;
  enum type_t {
    LCI_DATA_TYPE_NONE,
    LCI_DATA_TYPE_SCALAR,
    LCI_DATA_TYPE_BUFFER,
    LCI_DATA_TYPE_BUFFERS,
  };
  // FIXME: It is a undefined behavior to access multiple union members at the
  // same time
#if defined(__clang__)
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wnested-anon-types"
#endif
  union {
    struct {
      // common fields
      type_t type : 2;
      bool own_data : 1;
    } common;
    struct {
      type_t type : 2;
      bool own_data : 1;
      uint8_t size : 5;
      char data[MAX_SCALAR_SIZE];
    } scalar;
    struct {
      type_t type : 2;
      bool own_data : 1;
      size_t size : 61;
      void* base;
      mr_t mr;
    } buffer;
    struct {
      type_t type : 2;
      bool own_data : 1;
      size_t count;
      buffer_t* buffers;
    } buffers;
  };
#if defined(__clang__)
#pragma clang diagnostic pop
#endif
  data_t();
  data_t(buffer_t buffer_, bool own_data_ = false);
  data_t(buffers_t buffers_, bool own_data_ = false);
  data_t(size_t size, allocator_base_t* allocator = nullptr);
  data_t(size_t sizes[], int count, allocator_base_t* allocator = nullptr);
  data_t(const data_t& other);
  data_t(data_t&& other);
  data_t& operator=(data_t other);
  friend void swap(data_t& first, data_t& second);
 
  type_t get_type() const { return common.type; }
  void set_type(type_t type_) { common.type = type_; }
  bool get_own_data() const { return common.own_data; }
  void set_own_data(bool own_data_) { common.own_data = own_data_; }
 
  bool is_scalar() const { return get_type() == LCI_DATA_TYPE_SCALAR; }
  bool is_buffer() const { return get_type() == LCI_DATA_TYPE_BUFFER; }
  bool is_buffers() const { return get_type() == LCI_DATA_TYPE_BUFFERS; }
 
  void copy_from(const void* data_, size_t size,
                 allocator_base_t* allocator_ = nullptr);
  void copy_from(const buffers_t& buffers_,
                 allocator_base_t* allocator_ = nullptr);

  enum class get_semantic_t {
    move, 
    copy, 
    view, 
  };

  template <typename T>
  T get_scalar(/* always copy semantic */) const;
  buffer_t get_buffer(get_semantic_t semantic = get_semantic_t::move);
  size_t get_buffers_count() const;
  buffers_t get_buffers(get_semantic_t semantic = get_semantic_t::move);
};

struct status_t {
  error_t error = errorcode_t::retry_init;
  int rank = -1;
  data_t data;
  tag_t tag = 0;
  void* user_context = nullptr;
  status_t() = default;
  status_t(errorcode_t error_) : error(error_) {}
  explicit status_t(void* user_context_)
      : error(errorcode_t::done), user_context(user_context_)
  {
  }
  void set_done() { error = errorcode_t::done; }
  void set_posted() { error = errorcode_t::posted; }
  void set_retry() { error = errorcode_t::retry; }
  bool is_done() const { return error.is_done(); }
  bool is_posted() const { return error.is_posted(); }
  bool is_retry() const { return error.is_retry(); }
  template <typename T>
  T get_scalar() const
  {
    return data.get_scalar<T>();
  }
  buffer_t get_buffer() { return data.get_buffer(); }
  buffers_t get_buffers() { return data.get_buffers(); }
  bool is_scalar() const { return data.is_scalar(); }
  bool is_buffer() const { return data.is_buffer(); }
  bool is_buffers() const { return data.is_buffers(); }
};

const comp_t COMP_NULL = comp_t(reinterpret_cast<comp_impl_t*>(0x0));

const comp_t COMP_NULL_EXPECT_DONE =
    comp_t(reinterpret_cast<comp_impl_t*>(0x0));

const comp_t COMP_NULL_RETRY = comp_t(reinterpret_cast<comp_impl_t*>(0x1));

const comp_t COMP_NULL_EXPECT_DONE_OR_RETRY =
    comp_t(reinterpret_cast<comp_impl_t*>(0x1));
 
inline bool comp_t::is_empty() const
{
  return reinterpret_cast<uintptr_t>(p_impl) <= 1;
}

class comp_impl_t
{
 public:
  using attr_t = comp_attr_t;
  comp_impl_t(const attr_t& attr_) : attr(attr_) {}
  virtual ~comp_impl_t() = default;
  virtual void signal(status_t) = 0;
  comp_attr_t attr;
};

using comp_handler_t = void (*)(status_t status);

using reduce_op_t = void (*)(const void* left, const void* right, void* dst,
                             size_t n);
using graph_node_t = void*;

const graph_node_t GRAPH_START = reinterpret_cast<graph_node_t>(0x1);
const graph_node_t GRAPH_END = reinterpret_cast<graph_node_t>(0x2);

using graph_node_run_cb_t = status_t (*)(void* value);

const graph_node_run_cb_t GRAPH_NODE_DUMMY_CB = nullptr;

using graph_node_free_cb_t = void (*)(void* value);

using graph_edge_run_cb_t = void (*)(status_t status, void* src_value,
                                     void* dst_value);
 
}  // namespace lci
 
#include "lci_binding_post.hpp"
 
namespace lci
{
/***********************************************************************
 * Overloading graph_add_node for functor
 **********************************************************************/
#if __cplusplus >= 201703L
template <typename T>
status_t graph_execute_op_fn(void* value)
{
  auto op = static_cast<T*>(value);
  using result_t = std::invoke_result_t<T>;
 
  if constexpr (std::is_same_v<result_t, status_t>) {
    status_t result = (*op)();
    return result;
  } else if constexpr (std::is_same_v<result_t, errorcode_t>) {
    errorcode_t result = (*op)();
    return result;
  } else {
    (*op)();
    return errorcode_t::done;
  }
}
#else
// Specialization for status_t return type
template <typename T>
typename std::enable_if<
    std::is_same<typename std::result_of<T()>::type, status_t>::value,
    status_t>::type
graph_execute_op_fn(void* value)
{
  auto op = static_cast<T*>(value);
  status_t result = (*op)();
  return result;
}
 
// Specialization for errorcode_t return type
template <typename T>
typename std::enable_if<
    std::is_same<typename std::result_of<T()>::type, errorcode_t>::value,
    status_t>::type
graph_execute_op_fn(void* value)
{
  auto op = static_cast<T*>(value);
  errorcode_t result = (*op)();
  return result;
}
 
// Specialization for all other return types
template <typename T>
typename std::enable_if<
    !std::is_same<typename std::result_of<T()>::type, status_t>::value &&
        !std::is_same<typename std::result_of<T()>::type, errorcode_t>::value,
    status_t>::type
graph_execute_op_fn(void* value)
{
  auto op = static_cast<T*>(value);
  (*op)();
  return errorcode_t::done;
}
#endif
 
template <typename T>
void graph_free_op_fn(void* value)
{
  auto op = static_cast<T*>(value);
  delete op;
}

template <typename T>
graph_node_t graph_add_node_op(comp_t graph, const T& op)
{
  graph_node_run_cb_t wrapper = graph_execute_op_fn<T>;
  T* fn = new T(op);
  graph_node_free_cb_t free_cb = graph_free_op_fn<T>;
  auto ret = graph_add_node_x(graph, wrapper)
                 .value(reinterpret_cast<void*>(fn))
                 .free_cb(free_cb)();
  fn->user_context(ret);
  return ret;
}
 
/***********************************************************************
 * Some inline implementation
 **********************************************************************/
inline data_t::data_t()
{
  set_type(LCI_DATA_TYPE_NONE);
  set_own_data(false);
  static_assert(sizeof(data_t) == 24, "data_t size is not 24 bytes");
}
inline data_t::data_t(buffer_t buffer_, bool own_data_)
{
  set_type(LCI_DATA_TYPE_BUFFER);
  set_own_data(own_data_);
  buffer.base = buffer_.base;
  buffer.size = buffer_.size;
  buffer.mr = buffer_.mr;
}
inline data_t::data_t(buffers_t buffers_, bool own_data_)
{
  set_type(LCI_DATA_TYPE_BUFFERS);
  set_own_data(own_data_);
  buffers.count = buffers_.size();
  buffers.buffers = new buffer_t[buffers.count];
  for (size_t i = 0; i < buffers.count; i++) {
    buffers.buffers[i] = buffers_[i];
  }
}
inline data_t::data_t(size_t size, allocator_base_t* allocator)
{
  set_type(LCI_DATA_TYPE_BUFFER);
  set_own_data(true);
  if (allocator) {
    buffer.base = allocator->allocate(size);
  } else {
    buffer.base = malloc(size);
  }
  buffer.size = size;
  buffer.mr = mr_t();
}
inline data_t::data_t(size_t sizes[], int count, allocator_base_t* allocator)
{
  set_type(LCI_DATA_TYPE_BUFFERS);
  set_own_data(true);
  buffers.count = count;
  buffers.buffers = new buffer_t[buffers.count];
  for (int i = 0; i < count; i++) {
    if (allocator) {
      buffers.buffers[i].base = allocator->allocate(sizes[i]);
    } else {
      buffers.buffers[i].base = malloc(sizes[i]);
    }
    buffers.buffers[i].size = sizes[i];
    buffers.buffers[i].mr = mr_t();
  }
}
// copy constructor
inline data_t::data_t(const data_t& other)
{
  set_type(other.get_type());
  bool own_data = other.get_own_data();
  set_own_data(own_data);
  if (own_data)
    fprintf(stderr, "Copying buffer with own_data=true is not recommended\n");
  if (other.is_scalar()) {
    scalar.size = other.scalar.size;
    memcpy(scalar.data, other.scalar.data, scalar.size);
  } else if (other.is_buffer()) {
    buffer = other.buffer;
    if (own_data) {
      buffer.base = malloc(other.buffer.size);
      memcpy(buffer.base, other.buffer.base, other.buffer.size);
    }
  } else if (other.is_buffers()) {
    buffers.count = other.buffers.count;
    buffers.buffers = new buffer_t[buffers.count];
    for (size_t i = 0; i < buffers.count; i++) {
      buffers.buffers[i] = other.buffers.buffers[i];
      if (own_data) {
        buffers.buffers[i].base = malloc(other.buffers.buffers[i].size);
        memcpy(buffers.buffers[i].base, other.buffers.buffers[i].base,
               other.buffers.buffers[i].size);
      }
    }
  }
}
// move constructor
inline data_t::data_t(data_t&& other)
{
  set_type(other.get_type());
  set_own_data(other.get_own_data());
  if (other.is_scalar()) {
    scalar.size = other.scalar.size;
    memcpy(scalar.data, other.scalar.data, scalar.size);
  } else if (other.is_buffer()) {
    buffer = other.buffer;
    other.set_own_data(false);
  } else if (other.is_buffers()) {
    buffers.count = other.buffers.count;
    buffers.buffers = other.buffers.buffers;
    other.set_own_data(false);
    other.buffers.buffers = nullptr;
  }
}
// generic assignment operator
inline data_t& data_t::operator=(data_t other)
{
  swap(*this, other);
  return *this;
}
 
inline void swap(data_t& first, data_t& second)
{
  char* buf = (char*)malloc(sizeof(data_t));
#if defined(__GNUC__) && !defined(__clang__) && !defined(__NVCC__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wclass-memaccess"
#endif
  memcpy(buf, &first, sizeof(data_t));
  memcpy(&first, &second, sizeof(data_t));
  memcpy(&second, buf, sizeof(data_t));
#if defined(__GNUC__) && !defined(__clang__)
#pragma GCC diagnostic pop
#endif
  free(buf);
}
 
inline void data_t::copy_from(const void* data_, size_t size,
                              allocator_base_t* allocator)
{
  if (size <= MAX_SCALAR_SIZE) {
    set_type(LCI_DATA_TYPE_SCALAR);
    set_own_data(true);
    scalar.size = size;
    memcpy(scalar.data, data_, size);
  } else {
    set_type(LCI_DATA_TYPE_BUFFER);
    set_own_data(true);
    if (allocator) {
      buffer.base = allocator->allocate(size);
    } else {
      buffer.base = malloc(size);
    }
    memcpy(buffer.base, data_, size);
    buffer.size = size;
  }
}
 
inline void data_t::copy_from(const buffers_t& buffers_,
                              allocator_base_t* allocator)
{
  set_type(LCI_DATA_TYPE_BUFFERS);
  set_own_data(true);
  buffers.count = buffers_.size();
  buffers.buffers = new buffer_t[buffers.count];
  for (size_t i = 0; i < buffers.count; i++) {
    buffers.buffers[i].size = buffers_[i].size;
    if (allocator) {
      buffers.buffers[i].base = allocator->allocate(buffers_[i].size);
    } else {
      buffers.buffers[i].base = malloc(buffers_[i].size);
    }
    memcpy(buffers.buffers[i].base, buffers_[i].base, buffers_[i].size);
  }
}
 
template <typename T>
T data_t::get_scalar(/* always copy semantic */) const
{
  // We still keep the ownership of the data
  if (is_buffer()) {
    if (buffer.size != sizeof(T)) {
      throw std::runtime_error("Buffer size does not match scalar size");
    }
    return *reinterpret_cast<const T*>(buffer.base);
  } else if (is_scalar()) {
    if (sizeof(T) > scalar.size) {
      throw std::runtime_error("No enough data to fit the scalar.");
    }
    return *reinterpret_cast<const T*>(scalar.data);
  } else {
    throw std::runtime_error("Cannot convert to a scalar");
  }
}
 
inline buffer_t data_t::get_buffer(get_semantic_t semantic)
{
  buffer_t ret;
  if (is_scalar()) {
    if (semantic == get_semantic_t::move || semantic == get_semantic_t::copy) {
      ret.size = scalar.size;
      ret.base = malloc(scalar.size);
      memcpy(ret.base, scalar.data, scalar.size);
    } else {
      ret = buffer_t(scalar.data, scalar.size);
    }
  } else if (is_buffer()) {
    if (semantic == get_semantic_t::copy && get_own_data()) {
      ret.size = buffer.size;
      ret.base = malloc(buffer.size);
      memcpy(ret.base, buffer.base, buffer.size);
    } else {
      ret = buffer_t(buffer.base, buffer.size, buffer.mr);
    }
  } else {
    throw std::runtime_error("Cannot convert to a buffer");
  }
  if (semantic == get_semantic_t::move) {
    set_own_data(false);
    set_type(LCI_DATA_TYPE_NONE);
  }
  return ret;
}
 
inline size_t data_t::get_buffers_count() const
{
  if (!is_buffers()) {
    throw std::runtime_error("Not a buffers");
  }
  return buffers.count;
}
 
inline buffers_t data_t::get_buffers(get_semantic_t semantic)
{
  if (!is_buffers()) {
    throw std::runtime_error("Not buffers");
  }
  if (semantic == get_semantic_t::move) {
    set_own_data(false);
    set_type(LCI_DATA_TYPE_NONE);
  }
  buffers_t ret;
  for (size_t i = 0; i < buffers.count; i++) {
    if (semantic == get_semantic_t::copy && get_own_data()) {
      buffer_t buffer;
      buffer.size = buffers.buffers[i].size;
      buffer.base = malloc(buffer.size);
      memcpy(buffer.base, buffers.buffers[i].base, buffer.size);
      ret.push_back(buffer);
    } else {
      ret.push_back(buffers.buffers[i]);
    }
  }
  return ret;
}
 
}  // namespace lci
 
#endif  // LCI_API_LCI_HPP