text:
#pragma once

#include <memory_resource>
#include <vector>

#include <thrust/execution_policy.h>
#include <thrust/functional.h>
#include <thrust/inner_product.h>
#include <thrust/iterator/zip_iterator.h>
#include <thrust/merge.h>
#include <thrust/reduce.h>
#include <thrust/sort.h>
#include <thrust/tuple.h>

#include <gsl-lite/gsl-lite.hpp>

#include <cuda/define_specifiers.hpp>

#include <thrustshift/copy.h>
#include <thrustshift/equal.h>
#include <thrustshift/managed-vector.h>
#include <thrustshift/math.h>
#include <thrustshift/memory-resource.h>
#include <thrustshift/not-a-vector.h>
#include <thrustshift/transform.h>

namespace thrustshift {

enum class storage_order_t { row_major, col_major, none };

template <typename DataType, typename IndexType>
class COO {

   public:
	using value_type = DataType;
	using index_type = IndexType;

	COO() : num_cols_(0), num_rows_(0) {
	}

	template <class MemoryResource>
	COO(size_t nnz,
	    size_t num_rows,
	    size_t num_cols,
	    MemoryResource& memory_resource)
	    : values_(nnz, &memory_resource),
	      row_indices_(nnz, &memory_resource),
	      col_indices_(nnz, &memory_resource),
	      num_rows_(num_rows),
	      num_cols_(num_cols),
	      storage_order_(storage_order_t::none) {
	}

	COO(size_t nnz, size_t num_rows, size_t num_cols)
	    : COO(nnz, num_rows, num_cols, pmr::default_resource) {
	}

	template <class DataRange,
	          class RowIndRange,
	          class ColIndRange,
	          class MemoryResource>
	COO(DataRange&& values,
	    RowIndRange&& row_indices,
	    ColIndRange&& col_indices,
	    size_t num_rows,
	    size_t num_cols,
	    storage_order_t storage_order,
	    MemoryResource& memory_resource)
	    : values_(values.begin(), values.end(), &memory_resource),
	      row_indices_(row_indices.begin(),
	                   row_indices.end(),
	                   &memory_resource),
	      col_indices_(col_indices.begin(),
	                   col_indices.end(),
	                   &memory_resource),
	      num_rows_(num_rows),
	      num_cols_(num_cols),
	      storage_order_(storage_order) {
		gsl_Expects(values.size() == row_indices.size());
		gsl_Expects(values.size() == col_indices.size());
	}

	template <class DataRange,
	          class RowIndRange,
	          class ColIndRange,
	          class MemoryResource>
	COO(DataRange&& values,
	    RowIndRange&& row_indices,
	    ColIndRange&& col_indices,
	    size_t num_rows,
	    size_t num_cols,
	    MemoryResource& memory_resource)
	    : COO(std::forward<DataRange>(values),
	          std::forward<RowIndRange>(row_indices),
	          std::forward<ColIndRange>(col_indices),
	          num_rows,
	          num_cols,
	          storage_order_t::none,
	          memory_resource) {
	}

	template <class DataRange, class ColIndRange, class RowIndRange>
	COO(DataRange&& values,
	    RowIndRange&& row_indices,
	    ColIndRange&& col_indices,
	    size_t num_rows,
	    size_t num_cols)
	    : COO(std::forward<DataRange>(values),
	          std::forward<RowIndRange>(row_indices),
	          std::forward<ColIndRange>(col_indices),
	          num_rows,
	          num_cols,
	          pmr::default_resource) {
	}

	// The copy constructor is declared explicitly to ensure
	// managed memory is used per default.
	COO(const COO& other)
	    : COO(other.values(),
	          other.row_indices(),
	          other.col_indices(),
	          other.num_rows(),
	          other.num_cols(),
	          other.get_storage_order(),
	          pmr::default_resource) {
	}

	COO(COO&& other) = default;

	void change_storage_order(storage_order_t new_storage_order) {
		IndexType* primary_keys_first;
		IndexType* secondary_keys_first;

		if (new_storage_order == storage_order_) {
			return;
		}

		switch (new_storage_order) {
			case storage_order_t::row_major:
				primary_keys_first = row_indices_.data();
				secondary_keys_first = col_indices_.data();
				break;
			case storage_order_t::col_major:
				primary_keys_first = col_indices_.data();
				secondary_keys_first = row_indices_.data();
				break;
			case storage_order_t::none:
				return;
		}

		// Thrust's relational operators are overloaded for thrust::pair.
		// This ensures that we sort with respect to the second key if the first key is equal.
		auto key_it = thrust::make_zip_iterator(
		    thrust::make_tuple(primary_keys_first, secondary_keys_first));
		thrust::sort_by_key(
		    thrust::cuda::par, key_it, key_it + values_.size(), values_.data());
		storage_order_ = new_storage_order;
	}

	// If the storage order is changed externally
	void set_storage_order(storage_order_t new_storage_order) {
		storage_order_ = new_storage_order;
	}

	void transpose() {
		std::swap(col_indices_, row_indices_);
		std::swap(num_rows_, num_cols_);
		switch (storage_order_) {
			case storage_order_t::none:
				break;
			case storage_order_t::row_major:
				storage_order_ = storage_order_t::col_major;
				break;
			case storage_order_t::col_major:
				storage_order_ = storage_order_t::row_major;
				break;
		}
	}

	gsl_lite::span<DataType> values() {
		return gsl_lite::make_span(values_);
	}

	gsl_lite::span<const DataType> values() const {
		return gsl_lite::make_span(values_);
	}

	gsl_lite::span<IndexType> row_indices() {
		return gsl_lite::make_span(row_indices_);
	}

	gsl_lite::span<const IndexType> row_indices() const {
		return gsl_lite::make_span(row_indices_);
	}

	gsl_lite::span<IndexType> col_indices() {
		return gsl_lite::make_span(col_indices_);
	}

	gsl_lite::span<const IndexType> col_indices() const {
		return gsl_lite::make_span(col_indices_);
	}

	size_t num_rows() const {
		return num_rows_;
	}

	size_t num_cols() const {
		return num_cols_;
	}

	storage_order_t get_storage_order() const {
		return storage_order_;
	}

	//! Return row_ptrs or col_ptrs depending on the current storage order
	thrustshift::managed_vector<index_type> get_ptrs() const {
		gsl_Expects(storage_order_ != storage_order_t::none);
		auto indices = storage_order_ == storage_order_t::row_major
		                   ? row_indices()
		                   : col_indices();
		auto size = storage_order_ == storage_order_t::row_major ? num_rows()
		                                                         : num_cols();

		if (indices.size() > 0) {
			thrustshift::managed_vector<index_type> ptrs(size + 1, -1);
			for (int nns_id = gsl_lite::narrow<int>(indices.size()) - 1;
			     nns_id >= 0;
			     --nns_id) {
				ptrs[indices[nns_id]] = nns_id;
			}
			ptrs[0] = 0;
			ptrs.back() = indices.size();
			index_type k = indices.size();
			for (int id = gsl_lite::narrow<int>(ptrs.size()) - 1; id > 0;
			     --id) {
				if (ptrs[id] == -1) {
					ptrs[id] = k;
				}
				else {
					k = ptrs[id];
				}
			}
			return ptrs;
		}
		else {
			return {0};
		}
	}

   private:
	std::pmr::vector<DataType> values_;
	std::pmr::vector<IndexType> row_indices_;
	std::pmr::vector<IndexType> col_indices_;
	size_t num_rows_;
	size_t num_cols_;
	storage_order_t storage_order_;
};

//! Storage order does not affect comparison operator. Therefore this operator is defined explicitly.
template <typename DataType, typename IndexType>
bool operator==(const COO<DataType, IndexType>& a,
                const COO<DataType, IndexType>& b) {
	return equal(a.values(), b.values()) &&
	       equal(a.row_indices(), b.row_indices()) &&
	       equal(a.col_indices(), b.col_indices()) &&
	       a.num_rows() == b.num_rows() && a.num_cols() == b.num_cols();
}

template <typename DataType, typename IndexType>
class COO_view {

   public:
	using value_type = DataType;
	using index_type = IndexType;

	template <typename OtherDataType, typename OtherIndexType>
	COO_view(COO<OtherDataType, OtherIndexType>& owner)
	    : values_(owner.values()),
	      row_indices_(owner.row_indices()),
	      col_indices_(owner.col_indices()),
	      num_rows_(owner.num_rows()),
	      num_cols_(owner.num_cols()) {
	}

	template <typename OtherDataType, typename OtherIndexType>
	COO_view(const COO<OtherDataType, OtherIndexType>& owner)
	    : values_(owner.values()),
	      row_indices_(owner.row_indices()),
	      col_indices_(owner.col_indices()),
	      num_rows_(owner.num_rows()),
	      num_cols_(owner.num_cols()) {
	}

	template <class ValueRange, class ColIndRange, class RowIndRange>
	COO_view(ValueRange&& values,
	         RowIndRange&& row_indices,
	         ColIndRange&& col_indices,
	         size_t num_rows,
	         size_t num_cols)
	    : values_(values),
	      row_indices_(row_indices),
	      col_indices_(col_indices),
	      num_rows_(num_rows),
	      num_cols_(num_cols) {
		const auto nnz = values.size();
		gsl_Expects(col_indices.size() == nnz);
		gsl_Expects(row_indices.size() == nnz);
	}

	COO_view(const COO_view& other) = default;

	CUDA_FHD gsl_lite::span<DataType> values() {
		return values_;
	}

	CUDA_FHD gsl_lite::span<const DataType> values() const {
		return values_;
	}

	CUDA_FHD gsl_lite::span<IndexType> row_indices() {
		return row_indices_;
	}

	CUDA_FHD gsl_lite::span<const IndexType> row_indices() const {
		return row_indices_;
	}

	CUDA_FHD gsl_lite::span<IndexType> col_indices() {
		return col_indices_;
	}

	CUDA_FHD gsl_lite::span<const IndexType> col_indices() const {
		return col_indices_;
	}

	CUDA_FHD size_t num_rows() const {
		return num_rows_;
	}

	CUDA_FHD size_t num_cols() const {
		return num_cols_;
	}

   private:
	gsl_lite::span<DataType> values_;
	gsl_lite::span<IndexType> row_indices_;
	gsl_lite::span<IndexType> col_indices_;
	size_t num_rows_;
	size_t num_cols_;
};

namespace kernel {

template <typename T, typename T0, typename I0, typename T1, typename I1>
__global__ void diagmm(gsl_lite::span<const T> diag,
                       COO_view<const T0, const I0> mtx,
                       COO_view<T1, I1> result_mtx) {

	const auto nnz = mtx.values().size();
	const auto gtid = threadIdx.x + blockIdx.x * blockDim.x;
	if (gtid < nnz) {
		result_mtx.values()[gtid] =
		    diag[mtx.row_indices()[gtid]] * mtx.values()[gtid];
	}
}

} // namespace kernel

namespace async {

//! Calculate the product `result_coo_mtx = (diag_mtx * coo_mtx)`
//! result_coo_mtx can be equal to coo_mtx. Then the multiplication is in place.
template <class Range, class COO_C0, class COO_C1>
void diagmm(cuda::stream_t& stream,
            Range&& diag_mtx,
            COO_C0&& coo_mtx,
            COO_C1&& result_coo_mtx) {

	gsl_Expects(coo_mtx.values().size() == result_coo_mtx.values().size());
	gsl_Expects(coo_mtx.num_cols() == result_coo_mtx.num_cols());
	gsl_Expects(coo_mtx.num_rows() == result_coo_mtx.num_rows());
	gsl_Expects(diag_mtx.size() == coo_mtx.num_rows());
	// Only square matrices are allowed
	gsl_Expects(coo_mtx.num_rows() == coo_mtx.num_cols());

	using RangeT = typename std::remove_reference<Range>::type::value_type;
	using COO_T0 = typename std::remove_reference<COO_C0>::type::value_type;
	using COO_I0 = typename std::remove_reference<COO_C0>::type::index_type;
	using COO_T1 = typename std::remove_reference<COO_C1>::type::value_type;
	using COO_I1 = typename std::remove_reference<COO_C1>::type::index_type;

	const auto nnz = coo_mtx.values().size();
	constexpr cuda::grid::block_dimension_t block_dim = 128;
	const cuda::grid::dimension_t grid_dim = ceil_divide(nnz, block_dim);
	cuda::enqueue_launch(kernel::diagmm<RangeT, COO_T0, COO_I0, COO_T1, COO_I1>,
	                     stream,
	                     cuda::make_launch_config(grid_dim, block_dim),
	                     diag_mtx,
	                     coo_mtx,
	                     result_coo_mtx);
}

} // namespace async

namespace kernel {

template <typename T0, typename I, typename T1>
__global__ void get_diagonal(COO_view<const T0, const I> mtx,
                             gsl_lite::span<T1> diag) {

	const auto gtid = threadIdx.x + blockIdx.x * blockDim.x;
	const auto values = mtx.values();
	const auto nnz = values.size();
	if (gtid < nnz) {
		const auto row_id = mtx.row_indices()[gtid];
		const auto col_id = mtx.col_indices()[gtid];
		if (row_id == col_id) {
			diag[row_id] = values[gtid];
		}
	}
}

} // namespace kernel

namespace async {

template <typename T, class COO>
void get_diagonal(cuda::stream_t& stream, COO&& mtx, gsl_lite::span<T> diag) {

	gsl_Expects(diag.size() == std::min(mtx.num_rows(), mtx.num_cols()));
	using I = typename std::remove_reference<COO>::type::index_type;
	using T0 = typename std::remove_reference<COO>::type::value_type;
	const auto nnz = mtx.values().size();
	constexpr cuda::grid::block_dimension_t block_dim = 128;
	const cuda::grid::dimension_t grid_dim =
	    ceil_divide(gsl_lite::narrow<cuda::grid::dimension_t>(nnz),
	                gsl_lite::narrow<cuda::grid::dimension_t>(block_dim));
	fill(stream, diag, 0);
	if (nnz != 0) {
		cuda::enqueue_launch(kernel::get_diagonal<T0, I, T>,
		                     stream,
		                     cuda::make_launch_config(grid_dim, block_dim),
		                     mtx,
		                     diag);
	}
}

} // namespace async

/* \brief Transform two sparse COO matrices coefficient-wise `op(op_a(a), op_b(b))`.
 * \param a first sparse COO matrix.
 * \param b second sparse COO matrix.
 * \param op Binary operator how coefficients at the same positions are combined.
 * \param op_a Unary operator which is applied to all non-zero coefficients of `a`.
 * \param op_b Unary operator which is applied to all non-zero coefficients of `b`.
 * \return COO matrix allocated with `memory_resource`
 * \note It might be misleading that the binary operator is **not** applied to coefficients, which
 *   only appear in one of the two matrices. E.g. `op=minus`, `op_a=identity`, `op_b=identity` and `a` is a zero matrix the result
 *   is `result != a - b = -b`. Also note that it cannot be ensured that the left operand of the binary operator
 *   is always an element of `a`. Therefore the unary operators are required.
 */
template <typename DataType,
          typename IndexType,
          class BinaryOperator,
          class UnaryOperatorA,
          class UnaryOperatorB,
          class MemoryResource>
thrustshift::COO<DataType, IndexType> transform(
    thrustshift::COO_view<const DataType, const IndexType> a,
    thrustshift::COO_view<const DataType, const IndexType> b,
    BinaryOperator&& op,
    UnaryOperatorA&& op_a,
    UnaryOperatorB&& op_b,
    MemoryResource& memory_resource) {

	const auto nnz_a = a.values().size();
	const auto nnz_b = b.values().size();
	const auto num_rows = a.num_rows();
	const auto num_cols = b.num_cols();
	gsl_Expects(b.num_rows() == num_rows);
	gsl_Expects(b.num_cols() == num_cols);

	auto [tmp0, values] =
	    make_not_a_vector_and_span<DataType>(nnz_a + nnz_b, memory_resource);
	auto [tmp1, row_indices] =
	    make_not_a_vector_and_span<IndexType>(nnz_a + nnz_b, memory_resource);
	auto [tmp2, col_indices] =
	    make_not_a_vector_and_span<IndexType>(nnz_a + nnz_b, memory_resource);

	auto device = cuda::device::current::get();
	auto stream = device.default_stream();

	async::transform(stream, a.values(), values.first(nnz_a), op_a);
	async::copy(stream, a.row_indices(), row_indices.first(nnz_a));
	async::copy(stream, a.col_indices(), col_indices.first(nnz_a));

	async::transform(stream, b.values(), values.subspan(nnz_a), op_b);
	async::copy(stream, b.row_indices(), row_indices.subspan(nnz_a));
	async::copy(stream, b.col_indices(), col_indices.subspan(nnz_a));

	using KeyT = thrust::tuple<IndexType, IndexType>;
	auto keys_begin = thrust::make_zip_iterator(
	    thrust::make_tuple(row_indices.begin(), col_indices.begin()));
	auto keys_end = keys_begin + nnz_a + nnz_b;

	std::pmr::polymorphic_allocator<KeyT> alloc(&memory_resource);
	thrust::sort_by_key(
	    thrust::cuda::par(alloc), keys_begin, keys_end, values.begin());

	const std::size_t nnz_result =
	    thrust::inner_product(thrust::cuda::par(alloc),
	                          keys_begin,
	                          keys_end - 1,
	                          keys_begin + 1,
	                          std::size_t(0),
	                          thrust::plus<std::size_t>(),
	                          thrust::not_equal_to<KeyT>()) +
	    std::size_t(1);

	// ```cpp
	//   auto memory_resource = ...;
	//   auto res = transform(..., memory_resource);
	//
	// ```
	// should work fine because the dtor of `res` is called before the dtor of `memory_resource`
	COO<DataType, IndexType> result(
	    nnz_result, num_rows, num_cols, memory_resource);
	auto keys_result_begin = thrust::make_zip_iterator(thrust::make_tuple(
	    result.row_indices().begin(), result.col_indices().begin()));
	thrust::reduce_by_key(thrust::cuda::par(alloc),
	                      keys_begin,
	                      keys_end,
	                      values.begin(),
	                      keys_result_begin,
	                      result.values().begin(),
	                      thrust::equal_to<KeyT>(),
	                      op);
	result.set_storage_order(storage_order_t::row_major);
	return result;
}

template <typename DataType, typename IndexType, class MemoryResource>
COO<DataType, IndexType> symmetrize_abs(
    COO_view<const DataType, const IndexType> mtx,
    MemoryResource& memory_resource) {

	if (mtx.values().empty()) {
		return COO<DataType, IndexType>(
		    0, mtx.num_rows(), mtx.num_cols(), memory_resource);
	}

	COO_view<const DataType, const IndexType> mtx_trans(mtx.values(),
	                                                    mtx.col_indices(),
	                                                    mtx.row_indices(),
	                                                    mtx.num_rows(),
	                                                    mtx.num_cols());

	auto abs = [] __device__(DataType x) { return std::abs(x); };

	return transform(
	    mtx,
	    mtx_trans,
	    [] __device__(DataType x, DataType y) { return x + y; },
	    abs,
	    abs,
	    memory_resource);
}

template <typename DataType, typename IndexType, class MemoryResource>
COO<DataType, IndexType> make_pattern_symmetric(
    COO_view<const DataType, const IndexType> mtx,
    MemoryResource& memory_resource) {

	if (mtx.values().empty()) {
		return COO<DataType, IndexType>(
		    0, mtx.num_rows(), mtx.num_cols(), memory_resource);
	}

	COO_view<const DataType, const IndexType> mtx_trans(mtx.values(),
	                                                    mtx.col_indices(),
	                                                    mtx.row_indices(),
	                                                    mtx.num_rows(),
	                                                    mtx.num_cols());

	auto identity = [] __device__(DataType x) { return x; };
	auto make_zero = [] __device__(DataType x) { return DataType(0); };

	return transform(
	    mtx,
	    mtx_trans,
	    [] __device__(DataType x, DataType y) { return x + y; },
	    identity,
	    make_zero,
	    memory_resource);
}

} // namespace thrustshift

max_stars_repo_name: codecircuit/thrustshift

max_issues_repo_name: codecircuit/thrustshift

max_forks_repo_name: codecircuit/thrustshift


text:
/*
   Copyright [2017-2020] [IBM Corporation]
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at
       http://www.apache.org/licenses/LICENSE-2.0
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

#ifndef __API_FABRIC_ITF__
#define __API_FABRIC_ITF__

#include <component/base.h> /* component::IBase, DECLARE_COMPONENT_UUID, DECLARE_INTERFACE_UUID */
#include <common/byte_span.h>
#include <common/string_view.h>

#include <gsl/span>
#include <chrono>
#include <cstddef> /* size_t */
#include <cstdint> /* uint16_t, uint64_t */
#include <functional>
#include <stdexcept> /* runtime_error */
#include <vector>

#ifndef FABRIC_H
#define FI_READ (1ULL << 8)
#define FI_WRITE (1ULL << 9)
#define FI_RECV (1ULL << 10)
#define FI_SEND (1ULL << 11)
#define FI_TRANSMIT FI_SEND
#define FI_REMOTE_READ (1ULL << 12)
#define FI_REMOTE_WRITE (1ULL << 13)

#define FI_MULTI_RECV (1ULL << 16)
#define FI_REMOTE_CQ_DATA (1ULL << 17)
#define FI_MORE (1ULL << 18)
#define FI_PEEK (1ULL << 19)
#define FI_TRIGGER (1ULL << 20)
#define FI_FENCE (1ULL << 21)

#define FI_COMPLETION (1ULL << 24)
#define FI_EVENT FI_COMPLETION
#define FI_INJECT (1ULL << 25)
#define FI_INJECT_COMPLETE (1ULL << 26)
#define FI_TRANSMIT_COMPLETE (1ULL << 27)
#define FI_DELIVERY_COMPLETE (1ULL << 28)
#define FI_AFFINITY (1ULL << 29)
#define FI_COMMIT_COMPLETE (1ULL << 30)

#define FI_VARIABLE_MSG (1ULL << 48)
#define FI_RMA_PMEM (1ULL << 49)
#define FI_SOURCE_ERR (1ULL << 50)
#define FI_LOCAL_COMM (1ULL << 51)
#define FI_REMOTE_COMM (1ULL << 52)
#define FI_SHARED_AV (1ULL << 53)
#define FI_PROV_ATTR_ONLY (1ULL << 54)
#define FI_NUMERICHOST (1ULL << 55)
#define FI_RMA_EVENT (1ULL << 56)
#define FI_SOURCE (1ULL << 57)
#define FI_NAMED_RX_CTX (1ULL << 58)
#define FI_DIRECTED_RECV (1ULL << 59)
#endif

struct iovec; /* definition in <sys/uio.h> */

namespace component
{

class IFabric_runtime_error : public std::runtime_error {
 public:
  explicit IFabric_runtime_error(const common::string_view what_arg);
  /**
   * @return The libfabric error code (enumeration in <rdma/fi_errno.h>)
   */
  virtual unsigned id() const noexcept = 0;
};

/**
 * Fabric/RDMA-based network component
 *
 */
#pragma GCC diagnostic push
#if defined            __GNUC__ && 6 < __GNUC__
#pragma GCC diagnostic ignored "-Wnoexcept-type"
#endif

struct IFabric_memory_region;

class IFabric_op_completer {
 public:
  virtual ~IFabric_op_completer() {}

  /*
   * Functionality which probably belongs to a higher layer, but which landed
   * here. Callbacks which return DEFER are enqueued, to be retried after
   * later callbacks are given a first chance to run.
   */
  enum class cb_acceptance { ACCEPT, DEFER };

  /**
   * Poll completion events; service completion queues and store
   * events not belonging to group (stall them).  This method will
   * BOTH service the completion queues and service those events
   * stalled previously
   *
   *  completion_flags,
   * FI_{SEND,RECV,RMA,ATOMIC,MSG,TAGGED,MULTICAST,(REMOTE_)?READ,(REMOTE_)?WRITE,REMOTE_CQ_DATA,MULTI_RECV},
   *  are described in "man fi_cq" and defined in
   * libfabric/include/rdma/fabric.h
   *
   * @param completion_callback (context_t, completion_flags, std::size_t len,
   * status_t status, void* error_data)
   *
   * @return Number of completions processed
   */
  using complete_old = std::function<void(void *context, ::status_t)>;
  using complete_definite =
      std::function<void(void *context, ::status_t, std::uint64_t completion_flags, std::size_t len, void *error_data)>;
  using complete_tentative = std::function<
      cb_acceptance(void *context, ::status_t, std::uint64_t completion_flags, std::size_t len, void *error_data)>;
  using complete_param_definite = std::function<
      void(void *context, ::status_t, std::uint64_t completion_flags, std::size_t len, void *error_data, void *param)>;
  using complete_param_tentative              = std::function<cb_acceptance(void *context,
                                                               ::status_t,
                                                               std::uint64_t completion_flags,
                                                               std::size_t   len,
                                                               void *        error_data,
                                                               void *        param)>;
  using complete_param_definite_ptr_noexcept  = void (*)(void *context,
                                                        ::status_t,
                                                        std::uint64_t completion_flags,
                                                        std::size_t   len,
                                                        void *        error_data,
                                                        void *        param)
#if 201703L <= __cplusplus
	noexcept
#endif
    ;
  using complete_param_tentative_ptr_noexcept = cb_acceptance (*)(void *context,
                                                                  ::status_t,
                                                                  std::uint64_t completion_flags,
                                                                  std::size_t   len,
                                                                  void *        error_data,
                                                                  void *        param)
#if 201703L <= __cplusplus
	noexcept
#endif
	;

  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions(const complete_old &completion_callback) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions(const complete_definite &completion_callback) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions_tentative(const complete_tentative &completion_callback) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions(const complete_param_definite &completion_callback, void *callback_param) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions_tentative(const complete_param_tentative &completion_callback,
                                                 void *                          callback_param) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions(complete_param_definite_ptr_noexcept completion_callback,
                                       void *                               callback_param) = 0;
  /**
   * @throw IFabric_runtime_error - cq_read unhandled error
   * @throw std::logic_error - called on closed connection
   */
  virtual std::size_t poll_completions_tentative(complete_param_tentative_ptr_noexcept completion_callback,
                                                 void *                                callback_param) = 0;

  /**
   * Get count of stalled completions.
   *
   * @throw std::system_error, e.g. for locking
   */
  virtual std::size_t stalled_completion_count() = 0;

  /**
   * Block and wait for next completion.
   *
   * @param polls_limit Maximum number of polls
   *
   * @return (was next completion context. But poll_completions can retrieve
   * that)
   *
   * @throw IFabric_runtime_error - ::fi_control fail
   * @throw std::system_error - pselect fail
   */
  virtual void wait_for_next_completion(unsigned polls_limit = 0) = 0;
  /**
   * Block and wait for next completion.
   *
   * @param polls_limit Maximum time to wait
   *
   * @return (was next completion context. But poll_completions can retrieve
   * that)
   *
   * @throw IFabric_runtime_error - ::fi_control fail
   * @throw std::system_error - pselect fail
   */
  virtual void wait_for_next_completion(std::chrono::milliseconds timeout) = 0;

  /**
   * Unblock any threads waiting on completions
   *
   * @throw std::system_error, e.g. for locking
   */
  virtual void unblock_completions() = 0;

  /* Additional TODO:
     - support for completion and event counters
     - support for statistics collection
  */
};

class IFabric_memory_control {
 public:
  using const_byte_span = common::const_byte_span;
  virtual ~IFabric_memory_control() {}

  using memory_region_t = IFabric_memory_region *;
  /**
   * Register buffer for RDMA
   *
   * @param contig_addr Pointer to contiguous region
   * @param size Size of buffer in bytes
   * @param key Requested key for the remote memory. Note: if the fabric
   * provider uses the key (i.e., the fabric provider memory region attributes
   *            do not include the FI_MR_PROV_KEY bit), then the key must be
   *            unique among registered memory regions. As this API does not
   *            expose these attributes, the only safe strategy is to assume
   * that the key must be unique among registered memory regsions.
   * @param flags Flags e.g., FI_REMOTE_READ|FI_REMOTE_WRITE. Flag definitions
   * are in <rdma/fabric.h>
   *
   * @return Memory region handle
   *
   * @throw std::range_error - address already registered
   * @throw std::logic_error - inconsistent memory registry
   */
  virtual memory_region_t register_memory(const void *  contig_addr,
                                          std::size_t   size,
                                          std::uint64_t key,
                                          std::uint64_t flags)
  { return register_memory(common::make_const_byte_span(contig_addr, size), key, flags); }
  virtual memory_region_t register_memory(const_byte_span contig_,
                                          std::uint64_t key,
                                          std::uint64_t flags) = 0;

  /**
   * De-register memory region
   *
   * @param memory_region Memory region to de-register
   *
   * @throw std::range_error - address not registered
   * @throw std::logic_error - inconsistent memory registry
   */
  virtual void deregister_memory(memory_region_t memory_region) = 0;

  virtual std::uint64_t get_memory_remote_key(memory_region_t) const noexcept = 0;
  virtual void *        get_memory_descriptor(memory_region_t) const noexcept = 0;

  /**
   * Asynchronously post a buffer to receive data
   *
   * @param buffers Buffer span (containing regions should be registered)
   *
   * @return Work (context) identifier
   *
   * @throw IFabric_runtime_error - ::fi_recvv fail
   */
  virtual void post_recv(gsl::span<const ::iovec> buffers, void **descriptors, void *context) = 0;
  void post_recv(const ::iovec *first, const ::iovec *last, void **descriptors, void *context) { return post_recv( {first, last}, descriptors, context); }
  virtual void post_recv(gsl::span<const ::iovec> buffers, void *context)                           = 0;
};

class IFabric_client;
class IFabric_client_grouped;

/**
 * IFabric_initiator
 *  operations unique to a connected endpoint
 */
class IFabric_initiator
{
 public:
	virtual ~IFabric_initiator() {}
  /**
   * Asynchronously post a buffer to the connection
   *
   * @param buffers Buffer span (containing regions should be registered)
   *
   * @return Work (context) identifier
   *
   * @throw IFabric_runtime_error std::runtime_error - ::fi_sendv fail
   */
  virtual void post_send(gsl::span<const ::iovec> buffers, void **descriptors, void *context) = 0;
  void post_send(const ::iovec *first, const ::iovec *last, void **descriptors, void *context) { return post_send( {first, last}, descriptors, context); }
  virtual void post_send(gsl::span<const ::iovec> buffers, void *context)                     = 0;

  /**
   * Post RDMA read operation
   *
   * @param buffers Destination buffer span
   * @param remote_addr Remote address
   * @param key Key for remote address
   * @param out_context
   *
   * @throw IFabric_runtime_error - ::fi_readv fail
   *
   */
  virtual void post_read(gsl::span<const ::iovec> buffers,
                         void **        descriptors,
                         std::uint64_t  remote_addr,
                         std::uint64_t  key,
                         void *         context) = 0;

  void post_read(const ::iovec *first,
                         const ::iovec *last,
                         void **        descriptors,
                         std::uint64_t  remote_addr,
                         std::uint64_t  key,
                         void *         context) { return post_read( { first, last }, descriptors, remote_addr, key, context); }

  virtual void post_read(gsl::span<const ::iovec> buffers,
                         std::uint64_t               remote_addr,
                         std::uint64_t               key,
                         void *                      context) = 0;

  /**
   * Post RDMA write operation
   *
   * @param buffers Source buffer span
   * @param remote_addr Remote address
   * @param key Key for remote address
   * @param out_context
   *
   * @throw IFabric_runtime_error - ::fi_writev fail
   *
   */
  virtual void post_write(const gsl::span<const ::iovec> buffers,
                          void **        descriptors,
                          std::uint64_t  remote_addr,
                          std::uint64_t  key,
                          void *         context) = 0;

  void post_write(const ::iovec *first,
                          const ::iovec *last,
                          void **        descriptors,
                          std::uint64_t  remote_addr,
                          std::uint64_t  key,
                          void *         context)
  {
    return post_write(
      gsl::span<const ::iovec>(first, last), descriptors, remote_addr, key, context);
  }

  virtual void post_write(const gsl::span<const ::iovec> buffers,
                          std::uint64_t               remote_addr,
                          std::uint64_t               key,
                          void *                      context) = 0;

  /**
   * Send message without completion
   *
   * @param connection Connection to inject on
   * @param buf Data to send
   * @param buf Length of data to send (must not exceed
   * IFabric_connection::max_inject_size())
   *
   * @throw IFabric_runtime_error - ::fi_inject fail
   */
  virtual void inject_send(const void *buf, std::size_t len) = 0;

  /* TODO: atomic RMA operations
  */
};

/**
 * IFabric_endpoint_connected
 *  operations available to a connected endpoint
 */
class IFabric_endpoint_connected
	: public IFabric_memory_control
	, public IFabric_initiator
	, public IFabric_op_completer
{
  /* TODO: statistics collection
  */
};

class IFabric_endpoint_unconnected
	: public IFabric_memory_control
{
};

/**
 * Fabric/RDMA-based network component
 *  an acrive bit not connected endpoint
 */
class IFabric_endpoint_unconnected_client : public IFabric_endpoint_unconnected {
 public:

  /**
   * Open a fabric client (active endpoint) connection to a server. Active
   * endpoints usually correspond with hardware resources, e.g. verbs queue
   * pair. Options may not conflict with those specified for the fabric.
   *
   * @param json_configuration Configuration string in JSON
   * @param remote_endpoint The IP address (URL) of the server
   * @param port The IP port on the server
   *
   * @return the endpoint
   *
   * @throw std::bad_alloc - destination address alloc failed
   * @throw std::bad_alloc - libfabric out of memory
   * @throw std::bad_alloc - libfabric out of memory (creating a new server)
   * @throw std::bad_alloc - out of memory
   * @throw std::domain_error : json file parse-detected error
   * @throw std::logic_error : socket initialized with a negative value (from
   * ::socket) in Fd_control
   * @throw std::logic_error : unexpected event
   * @throw std::system_error : error receiving fabric server name
   * @throw std::system_error : pselect fail (expecting event)
   * @throw std::system_error : resolving address
   * @throw std::system_error : pselect fail
   * @throw std::system_error : read error on event pipe
   * @throw std::system_error - writing event pipe (normal callback)
   * @throw std::system_error - writing event pipe (readerr_eq)
   * @throw std::system_error - receiving data on socket
   * @throw IFabric_runtime_error - ::fi_domain fail
   * @throw IFabric_runtime_error - ::fi_connect fail
   * @throw IFabric_runtime_error - ::fi_ep_bind fail
   * @throw IFabric_runtime_error - ::fi_enable fail
   * @throw IFabric_runtime_error - ::fi_ep_bind fail (event registration)
   *
   */
  virtual IFabric_client *make_open_client() = 0;

  /**
   * Open a fabric endpoint for which communications are divided into smaller
   * entities (groups). Options may not conflict with those specified for the
   * fabric.
   *
   * @param json_configuration Configuration string in JSON
   * @param remote_endpoint The IP address (URL) of the server
   * @param port The IP port on the server
   *
   * @return the endpoint
   *
   * @throw std::bad_alloc - destination address alloc failed
   * @throw std::bad_alloc - libfabric out of memory
   * @throw std::bad_alloc - libfabric out of memory (creating a new server)
   * @throw std::bad_alloc - out of memory
   * @throw std::domain_error : json file parse-detected error
   * @throw std::logic_error : socket initialized with a negative value (from
   * ::socket) in Fd_control
   * @throw std::logic_error : unexpected event
   * @throw std::system_error : error receiving fabric server name
   * @throw std::system_error : pselect fail (expecting event)
   * @throw std::system_error : resolving address
   * @throw std::system_error : pselect fail
   * @throw std::system_error : read error on event pipe
   * @throw std::system_error - writing event pipe (normal callback)
   * @throw std::system_error - writing event pipe (readerr_eq)
   * @throw std::system_error - receiving data on socket
   * @throw IFabric_runtime_error - ::fi_domain fail
   * @throw IFabric_runtime_error - ::fi_connect fail
   * @throw IFabric_runtime_error - ::fi_ep_bind fail
   * @throw IFabric_runtime_error - ::fi_enable fail
   * @throw IFabric_runtime_error - ::fi_ep_bind fail (event registration)
   *
   */
  virtual IFabric_client_grouped *make_open_client_grouped() = 0;
};

class IFabric_connection {
 public:
  virtual ~IFabric_connection() {}

  /**
   * Get address of connected peer (taken from fi_getpeer during
   * connection instantiation). The fi_getpeer documentation says
   * "The address  must  be  in the same format as that specified using
   * fi_info: addr_format when the endpoint was created". Presumably
   * this means that the address returned by fi_getpeer *will* be in
   * that same form, and is thus dependent on the implementation of
   * endpoint creation.
   *
   *
   * @return Peer endpoint address
   * @throw std::bad_alloc, e.g.
   */
  virtual std::string get_peer_addr() = 0;

  /**
   * Get local address of connection (taken from fi_getname during
   * connection instantiation).
   *
   *
   * @return Local endpoint address
   * @throw std::bad_alloc, e.g.
   */
  virtual std::string get_local_addr() = 0;

  /**
   * Get the maximum message size for the provider
   *
   * @return Max message size in bytes
   */
  virtual std::size_t max_message_size() const noexcept = 0;

  /**
   * Get the maximum inject message size for the provider
   *
   * @return Max inject message size in bytes
   */
  virtual std::size_t max_inject_size() const noexcept = 0;

  /* Additional TODO:
     - support for atomic RMA operations
     - support for completion and event counters
     - support for statistics collection
  */
};

/**
 * An endpoint with a communication channel. Distinguished from a connection
 * which can provide grouped communications channels, but does not of itself
 * provide communications.
 *
 */
class IFabric_endpoint_comm
    : public IFabric_connection
    , public IFabric_endpoint_connected {
};

/**
 * An endpoint established as a server.
 *
 */
class IFabric_server : public IFabric_endpoint_comm {
};

/**
 * An endpoint established as a client.
 *
 */
class IFabric_client : public IFabric_endpoint_comm {
};

/* A group: memory control, commands, and completions */
class IFabric_group
	: public IFabric_endpoint_connected
{
};

/**
 * An endpoint "grouped", without a communication channel.
 * Can provide communications channels, but does not initiate operations
 *
 */
class IFabric_endpoint_grouped
    : public IFabric_connection
    , public IFabric_memory_control
    , public IFabric_op_completer
{
 public:
  /**
   * Allocate group (for partitioned completion handling)
   *
   * @throw std::system_error, e.g. for locking
   * @throw std::bad_alloc, e.g.
   */
  virtual IFabric_group *allocate_group() = 0;
};

/**
 * A client endpoint which cannot initiate commands but which
 * can allocate "commuicators" which can initiate commands.
 *
 */
class IFabric_client_grouped : public IFabric_endpoint_grouped {
};

/**
 * A server endpoint which cannot initiate commands but which
 * can allocate "commuicators" which can initiate commands.
 *
 */
class IFabric_server_grouped : public IFabric_endpoint_grouped {
};

/**
 * Fabric passive endpoint. Instantiation of this interface normally creates
 * an active thread that uses the fabric connection manager to handle
 * connections (see man fi_cm).  On release of the interface, the endpoint
 * is destroyed.
 */
class IFabric_passive_endpoint {
 public:
  virtual ~IFabric_passive_endpoint() {}

  /**
   * Get the maximum message size for the provider
   *
   * @return Max message size in bytes
   */
  virtual std::size_t max_message_size() const noexcept = 0;

  /**
   * Get provider name
   *
   *
   * @return Provider name
   *
   * @throw std::bad_alloc, e.g.
   */
  virtual std::string get_provider_name() const = 0;
};

class IFabric_endpoint_unconnected_server
	: public IFabric_endpoint_unconnected
{
};

/**
 * Fabric passive endpoint providing servers with ordinary (not grouped)
 * communicators.
 */
class IFabric_server_factory : public IFabric_passive_endpoint {
 public:
  /**
   * Server/accept side handling of new connections (handled by the
   * active thread) are queued so that they can be taken by a polling
   * thread and integrated into the processing loop.  This method is
   * normally invoked until NULL is returned.
   *
   * @return New connection, or NULL if no new connection.
   *
   * @throw std::system_error, e.g. for locking
   * @throw std::logic_error : unexpected event
   * @throw std::system_error : read error on event pipe
   * @throw std::system_error : failure to listen for connections
   */
	virtual IFabric_endpoint_unconnected_server *get_new_endpoint_unconnected() = 0;
	virtual IFabric_server *open_connection(IFabric_endpoint_unconnected_server *) = 0;

  /**
   * Close connection and release any associated resources
   *
   * @param connection
   *
   * @throw std::system_error, e.g. for locking
   */
  virtual void close_connection(IFabric_server *connection) = 0;

  /**
   * Used to get a vector of active connection belonging to this
   * end point.
   *
   * @return Vector of active connections
   *
   * @throw std::bad_alloc, e.g.
   */
  virtual std::vector<IFabric_server *> connections() = 0;

  /**
   * Get provider name
   *
   *
   * @return Provider name
   *
   * @throw std::bad_alloc, e.g.
   */
  virtual std::string get_provider_name() const = 0;
};

/* Fabric passive endpoint providing servers with grouped communicators.
 */
class IFabric_server_grouped_factory : public IFabric_passive_endpoint {
 public:
  /**
   * Server/accept side handling of new connections (handled by the
   * active thread) are queued so that they can be taken by a polling
   * thread an integrated into the processing loop.  This method is
   * normally invoked until NULL is returned.
   *
   *
   * @return New connection, or NULL if no new connection.
   *
   * @throw std::logic_error : unexpected event
   * @throw std::system_error, e.g. for locking
   * @throw std::system_error : read error on event pipe
   */
	virtual IFabric_endpoint_unconnected_server *get_new_endpoint_unconnected() = 0;
	virtual IFabric_server_grouped *open_connection(IFabric_endpoint_unconnected_server *) = 0;

  /**
   * Close connection and release any associated resources
   *
   * @param connection
   *
   * @throw std::system_error, e.g. for locking
   */
  virtual void close_connection(IFabric_server_grouped *connection) = 0;

  /**
   * Used to get a vector of active connection belonging to this
   * end point.
   *
   * @return Vector of active connections
   * @throw std::bad_alloc, e.g.
   */
  virtual std::vector<IFabric_server_grouped *> connections() = 0;
};

class IFabric {
 public:
  using memory_region_t = IFabric_memory_region *;

  DECLARE_INTERFACE_UUID(0xc373d083, 0xe629, 0x46c9, 0x86fa, 0x6f, 0x96, 0x40, 0x61, 0x10, 0xdf);
  virtual ~IFabric() {}
  /**
   * Open a fabric server factory. Endpoints usually correspond with hardware
   * resources, e.g. verbs queue pair. Options may not conflict with those
   * specified for the fabric.
   *
   * @param json_configuration Configuration string in JSON
   *
   * @return the endpoint
   *
   * @throw std::domain_error : json file parse-detected error
   * @throw IFabric_runtime_error - ::fi_passive_ep fail
   * @throw IFabric_runtime_error - ::fi_pep_bind fail
   * @throw IFabric_runtime_error - ::fi_listen fail
   */
  virtual IFabric_server_factory *open_server_factory(const common::string_view json_configuration, std::uint16_t port) = 0;
  /**
   * Open a fabric "server grouped" factory. Endpoints usually correspond with
   * hardware resources, e.g. verbs queue pair. Options may not conflict with
   * those specified for the fabric. "Server groups" are servers in which each
   * operation is assigned to a "communication group." Each "completion group"
   * may be separately polled for completions.
   *
   * @param json_configuration Configuration string in JSON
   *
   * @return the endpoint
   *
   * @throw std::domain_error : json file parse-detected error
   * @throw IFabric_runtime_error - ::fi_passive_ep fail
   * @throw IFabric_runtime_error - ::fi_pep_bind fail
   * @throw IFabric_runtime_error - ::fi_listen fail
   */
  virtual IFabric_server_grouped_factory *open_server_grouped_factory(const common::string_view json_configuration,
                                                                      std::uint16_t      port) = 0;

  virtual IFabric_endpoint_unconnected_client *make_endpoint(const common::string_view json_configuration, common::string_view remote_endpoint, std::uint16_t port) = 0;

  /*
   * provider name in the fabric.
   */
  virtual const char *prov_name() const noexcept = 0;
};

class IFabric_factory : public component::IBase {
 public:
  DECLARE_INTERFACE_UUID(0xfac3d083, 0xe629, 0x46c9, 0x86fa, 0x6f, 0x96, 0x40, 0x61, 0x10, 0xdf);

  /**
   * Open a fabric endpoint.  Endpoints usually correspond with hardware
   * resources, e.g. verbs queue pair Options should include provider,
   * capabilities, active thread core.
   *
   * @param json_configuration Configuration string in JSON
   * form. e.g. { "caps":["FI_MSG","FI_RMA"], "preferred_provider" : "verbs"}
   * @return
   */
  virtual ~IFabric_factory() {}
  /**
   * @throw std::bad_alloc - out of memory
   * @throw std::domain_error : json file parse-detected error
   * @throw IFabric_runtime_error - ::fi_control fail
   */
  virtual IFabric *make_fabric(const common::string_view json_configuration) = 0;
};

}  // namespace component

#endif  // __API_FABRIC_ITF__

ext: h

max_stars_repo_name: omriarad/mcas

max_issues_repo_name: omriarad/mcas

max_forks_repo_name: omriarad/mcas