/**
* Copyright (C) Mellanox Technologies Ltd. 2001-2019.  ALL RIGHTS RESERVED.
*
* See file LICENSE for terms.
*/

#ifndef UCT_IFACE_H_
#define UCT_IFACE_H_

#include "uct_worker.h"

#include <uct/api/uct.h>
#include <uct/base/uct_component.h>
#include <ucs/config/parser.h>
#include <ucs/datastruct/arbiter.h>
#include <ucs/datastruct/mpool.h>
#include <ucs/datastruct/queue.h>
#include <ucs/debug/log.h>
#include <ucs/stats/stats.h>
#include <ucs/sys/compiler.h>
#include <ucs/sys/sys.h>
#include <ucs/type/class.h>

#include <ucs/datastruct/mpool.inl>


enum {
    UCT_EP_STAT_AM,
    UCT_EP_STAT_PUT,
    UCT_EP_STAT_GET,
    UCT_EP_STAT_ATOMIC,
#if IBV_HW_TM
    UCT_EP_STAT_TAG,
#endif
    UCT_EP_STAT_BYTES_SHORT,
    UCT_EP_STAT_BYTES_BCOPY,
    UCT_EP_STAT_BYTES_ZCOPY,
    UCT_EP_STAT_NO_RES,
    UCT_EP_STAT_FLUSH,
    UCT_EP_STAT_FLUSH_WAIT,  /* number of times flush called while in progress */
    UCT_EP_STAT_PENDING,
    UCT_EP_STAT_FENCE,
    UCT_EP_STAT_LAST
};

enum {
    UCT_IFACE_STAT_RX_AM,
    UCT_IFACE_STAT_RX_AM_BYTES,
    UCT_IFACE_STAT_TX_NO_DESC,
    UCT_IFACE_STAT_FLUSH,
    UCT_IFACE_STAT_FLUSH_WAIT,  /* number of times flush called while in progress */
    UCT_IFACE_STAT_FENCE,
    UCT_IFACE_STAT_LAST
};


/*
 * Statistics macros
 */
#define UCT_TL_EP_STAT_OP(_ep, _op, _method, _size) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_##_op, 1); \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_BYTES_##_method, _size);
#define UCT_TL_EP_STAT_OP_IF_SUCCESS(_status, _ep, _op, _method, _size) \
    if (_status >= 0) { \
        UCT_TL_EP_STAT_OP(_ep, _op, _method, _size) \
    }
#define UCT_TL_EP_STAT_ATOMIC(_ep) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_ATOMIC, 1);
#define UCT_TL_EP_STAT_FLUSH(_ep) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_FLUSH, 1);
#define UCT_TL_EP_STAT_FLUSH_WAIT(_ep) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_FLUSH_WAIT, 1);
#define UCT_TL_EP_STAT_FENCE(_ep) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_FENCE, 1);
#define UCT_TL_EP_STAT_PEND(_ep) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCT_EP_STAT_PENDING, 1);

#define UCT_TL_IFACE_STAT_FLUSH(_iface) \
    UCS_STATS_UPDATE_COUNTER((_iface)->stats, UCT_IFACE_STAT_FLUSH, 1);
#define UCT_TL_IFACE_STAT_FLUSH_WAIT(_iface) \
    UCS_STATS_UPDATE_COUNTER((_iface)->stats, UCT_IFACE_STAT_FLUSH_WAIT, 1);
#define UCT_TL_IFACE_STAT_FENCE(_iface) \
    UCS_STATS_UPDATE_COUNTER((_iface)->stats, UCT_IFACE_STAT_FENCE, 1);
#define UCT_TL_IFACE_STAT_TX_NO_DESC(_iface) \
    UCS_STATS_UPDATE_COUNTER((_iface)->stats, UCT_IFACE_STAT_TX_NO_DESC, 1);


#define UCT_CB_FLAGS_CHECK(_flags) \
    do { \
        if ((_flags) & UCT_CB_FLAG_RESERVED) { \
            ucs_error("Unsupported callback flag 0x%x", UCT_CB_FLAG_RESERVED); \
            return UCS_ERR_INVALID_PARAM; \
        } \
    } while (0)


/**
 * In release mode - do nothing.
 *
 * In debug mode, if _condition is not true, return an error. This could be less
 * optimal because of additional checks, and that compiler needs to generate code
 * for error flow as well.
 */
#define UCT_CHECK_PARAM(_condition, _err_message, ...) \
    if (ENABLE_PARAMS_CHECK && !(_condition)) { \
        ucs_error(_err_message, ## __VA_ARGS__); \
        return UCS_ERR_INVALID_PARAM; \
    }


/**
 * In release mode - do nothing.
 *
 * In debug mode, if @a _params field mask does not have set
 * @ref UCT_EP_PARAM_FIELD_DEV_ADDR and @ref UCT_EP_PARAM_FIELD_IFACE_ADDR
 * flags, return an error.
 */
#define UCT_EP_PARAMS_CHECK_DEV_IFACE_ADDRS(_params) \
    UCT_CHECK_PARAM(ucs_test_all_flags((_params)->field_mask, \
                                       UCT_EP_PARAM_FIELD_DEV_ADDR | \
                                       UCT_EP_PARAM_FIELD_IFACE_ADDR), \
                    "UCT_EP_PARAM_FIELD_DEV_ADDR and UCT_EP_PARAM_FIELD_IFACE_ADDR are not defined")


#define UCT_EP_PARAMS_GET_PATH_INDEX(_params) \
    (((_params)->field_mask & UCT_EP_PARAM_FIELD_PATH_INDEX) ? \
     (_params)->path_index : 0)

/**
 * Check the condition and return status as a pointer if not true.
 */
#define UCT_CHECK_PARAM_PTR(_condition, _err_message, ...) \
    if (ENABLE_PARAMS_CHECK && !(_condition)) { \
        ucs_error(_err_message, ## __VA_ARGS__); \
        return UCS_STATUS_PTR(UCS_ERR_INVALID_PARAM); \
    }


/**
 * Check the size of the IOV array
 */
#define UCT_CHECK_IOV_SIZE(_iovcnt, _max_iov, _name) \
    UCT_CHECK_PARAM((_iovcnt) <= (_max_iov), \
                    "iovcnt(%lu) should be limited by %lu in %s", \
                    _iovcnt, _max_iov, _name)


/**
 * In debug mode, if _condition is not true, generate 'Invalid length' error.
 */
#define UCT_CHECK_LENGTH(_length, _min_length, _max_length, _name) \
    { \
        typeof(_length) __length = _length; \
        UCT_CHECK_PARAM((_length) <= (_max_length), \
                        "Invalid %s length: %zu (expected: <= %zu)", \
                        _name, (size_t)(__length), (size_t)(_max_length)); \
        UCT_CHECK_PARAM((ssize_t)(_length) >= (_min_length), \
                        "Invalid %s length: %zu (expected: >= %zu)", \
                        _name, (size_t)(__length), (size_t)(_min_length)); \
    }

/**
 * Skip if this is a zero-length operation.
 */
#define UCT_SKIP_ZERO_LENGTH(_length, ...) \
    if (0 == (_length)) { \
        ucs_trace_data("Zero length request: skip it"); \
        UCS_PP_FOREACH(_UCT_RELEASE_DESC, _, __VA_ARGS__)  \
        return UCS_OK; \
    }
#define _UCT_RELEASE_DESC(_, _desc) \
    ucs_mpool_put(_desc);


/**
 * In debug mode, check that active message ID is valid.
 */
#define UCT_CHECK_AM_ID(_am_id) \
    UCT_CHECK_PARAM((_am_id) < UCT_AM_ID_MAX, \
                    "Invalid active message id (valid range: 0..%d)", \
                    (int)UCT_AM_ID_MAX - 1)


/**
 * In debug mode, check that keepalive params are valid
 */
#define UCT_EP_KEEPALIVE_CHECK_PARAM(_flags, _comp) \
    UCT_CHECK_PARAM((_comp) == NULL, "Unsupported completion on ep_check"); \
    UCT_CHECK_PARAM((_flags) == 0, "Unsupported flags: %x", (_flags));


/**
 * Declare classes for structures defined in api/tl.h
 */
UCS_CLASS_DECLARE(uct_iface_h, uct_iface_ops_t, uct_md_h);
UCS_CLASS_DECLARE(uct_ep_t, uct_iface_h);


/**
 * Active message handle table entry
 */
typedef struct uct_am_handler {
    uct_am_callback_t cb;
    void              *arg;
    uint32_t          flags;
} uct_am_handler_t;


/**
 * Base structure of all interfaces.
 * Includes the AM table which we don't want to expose.
 */
typedef struct uct_base_iface {
    uct_iface_t             super;
    uct_md_h                md;               /* MD this interface is using */
    uct_priv_worker_t       *worker;          /* Worker this interface is on */
    uct_am_handler_t        am[UCT_AM_ID_MAX];/* Active message table */
    uct_am_tracer_t         am_tracer;        /* Active message tracer */
    void                    *am_tracer_arg;   /* Tracer argument */
    uct_error_handler_t     err_handler;      /* Error handler */
    void                    *err_handler_arg; /* Error handler argument */
    uint32_t                err_handler_flags; /* Error handler callback flags */
    uct_worker_progress_t   prog;             /* Will be removed once all transports
                                                 support progress control */
    unsigned                progress_flags;   /* Which progress is currently enabled */

    struct {
        unsigned            num_alloc_methods;
        uct_alloc_method_t  alloc_methods[UCT_ALLOC_METHOD_LAST];
        ucs_log_level_t     failure_level;
        size_t              max_num_eps;
    } config;

    UCS_STATS_NODE_DECLARE(stats)            /* Statistics */
} uct_base_iface_t;

UCS_CLASS_DECLARE(uct_base_iface_t, uct_iface_ops_t*,  uct_md_h, uct_worker_h,
                  const uct_iface_params_t*, const uct_iface_config_t*
                  UCS_STATS_ARG(ucs_stats_node_t*) UCS_STATS_ARG(const char*));


/**
 * Stub interface used for failed endpoints
 */
typedef struct uct_failed_iface {
    uct_iface_t       super;
    ucs_queue_head_t  pend_q;
} uct_failed_iface_t;


/**
 * Base structure of all endpoints.
 */
typedef struct uct_base_ep {
    uct_ep_t          super;
    UCS_STATS_NODE_DECLARE(stats)
} uct_base_ep_t;
UCS_CLASS_DECLARE(uct_base_ep_t, uct_base_iface_t*);


/**
 * Internal resource descriptor of a transport device
 */
typedef struct uct_tl_device_resource {
    char                     name[UCT_DEVICE_NAME_MAX]; /**< Hardware device name */
    uct_device_type_t        type;       /**< The device represented by this resource
                                              (e.g. UCT_DEVICE_TYPE_NET for a network interface) */
    ucs_sys_device_t         sys_device; /**< The identifier associated with the device
                                              bus_id as captured in ucs_sys_bus_id_t struct */
} uct_tl_device_resource_t;


/**
 * UCT transport definition. This structure should not be used directly; use
 * @ref UCT_TL_DEFINE macro to define a transport.
 */
typedef struct uct_tl {
    char                   name[UCT_TL_NAME_MAX]; /**< Transport name */

    ucs_status_t           (*query_devices)(uct_md_h md,
                                            uct_tl_device_resource_t **tl_devices_p,
                                            unsigned *num_tl_devices_p);

    ucs_status_t           (*iface_open)(uct_md_h md, uct_worker_h worker,
                                         const uct_iface_params_t *params,
                                         const uct_iface_config_t *config,
                                         uct_iface_h *iface_p);

    ucs_config_global_list_entry_t config; /**< Transport configuration entry */
    ucs_list_link_t                list;   /**< Entry in component's transports list */
} uct_tl_t;


/**
 * Define a transport
 *
 * @param _component      Component to add the transport to
 * @param _name           Name of the transport (should be a token, not a string)
 * @param _query_devices  Function to query the list of available devices
 * @param _iface_class    Struct type defining the uct_iface class
 */
#define UCT_TL_DEFINE(_component, _name, _query_devices, _iface_class, \
                      _cfg_prefix, _cfg_table, _cfg_struct) \
    \
    uct_tl_t uct_##_name##_tl = { \
        .name               = #_name, \
        .query_devices      = _query_devices, \
        .iface_open         = UCS_CLASS_NEW_FUNC_NAME(_iface_class), \
        .config = { \
            .name           = #_name" transport", \
            .prefix         = _cfg_prefix, \
            .table          = _cfg_table, \
            .size           = sizeof(_cfg_struct), \
         } \
    }; \
    UCS_CONFIG_REGISTER_TABLE_ENTRY(&(uct_##_name##_tl).config, &ucs_config_global_list); \
    UCS_STATIC_INIT { \
        ucs_list_add_tail(&(_component)->tl_list, &(uct_##_name##_tl).list); \
    }


/**
 * "Base" structure which defines interface configuration options.
 * Specific transport extend this structure.
 */
struct uct_iface_config {
    struct {
        uct_alloc_method_t  *methods;
        unsigned            count;
    } alloc_methods;

    int               failure;   /* Level of failure reports */
    size_t            max_num_eps;
};


/**
 * Memory pool configuration.
 */
typedef struct uct_iface_mpool_config {
    unsigned          max_bufs;  /* Upper limit to number of buffers */
    unsigned          bufs_grow; /* How many buffers (approx.) are allocated every time */
} uct_iface_mpool_config_t;


/**
 * Define configuration fields for memory pool parameters.
 */
#define UCT_IFACE_MPOOL_CONFIG_FIELDS(_prefix, _dfl_max, _dfl_grow, _mp_name, _offset, _desc) \
    {_prefix "MAX_BUFS", UCS_PP_QUOTE(_dfl_max), \
     "Maximal number of " _mp_name " buffers for the interface. -1 is infinite." \
     _desc, \
     (_offset) + ucs_offsetof(uct_iface_mpool_config_t, max_bufs), UCS_CONFIG_TYPE_INT}, \
    \
    {_prefix "BUFS_GROW", UCS_PP_QUOTE(_dfl_grow), \
     "How much buffers are added every time the " _mp_name " memory pool grows.\n" \
     "0 means the value is chosen by the transport.", \
     (_offset) + ucs_offsetof(uct_iface_mpool_config_t, bufs_grow), UCS_CONFIG_TYPE_UINT}


/**
 * Get a descriptor from memory pool, tell valgrind it's already defined, return
 * error if the memory pool is empty.
 *
 * @param _mp       Memory pool to get descriptor from.
 * @param _desc     Variable to assign descriptor to.
 * @param _failure  What do to if memory poll is empty.
 *
 * @return TX descriptor fetched from memory pool.
 */
#define UCT_TL_IFACE_GET_TX_DESC(_iface, _mp, _desc, _failure) \
    { \
        _desc = ucs_mpool_get_inline(_mp); \
        if (ucs_unlikely((_desc) == NULL)) { \
            UCT_TL_IFACE_STAT_TX_NO_DESC(_iface); \
            _failure; \
        } \
        \
        VALGRIND_MAKE_MEM_DEFINED(_desc, sizeof(*(_desc))); \
    }


#define UCT_TL_IFACE_GET_RX_DESC(_iface, _mp, _desc, _failure) \
    { \
        _desc = ucs_mpool_get_inline(_mp); \
        if (ucs_unlikely((_desc) == NULL)) { \
            uct_iface_mpool_empty_warn(_iface, _mp); \
            _failure; \
        } \
        \
        VALGRIND_MAKE_MEM_DEFINED(_desc, sizeof(*(_desc))); \
    }


#define UCT_TL_IFACE_PUT_DESC(_desc) \
    { \
        ucs_mpool_put_inline(_desc); \
        VALGRIND_MAKE_MEM_UNDEFINED(_desc, sizeof(*(_desc))); \
    }


/**
 * TL Memory pool object initialization callback.
 */
typedef void (*uct_iface_mpool_init_obj_cb_t)(uct_iface_h iface, void *obj,
                                              uct_mem_h memh);


/**
 * Base structure for private data held inside a pending request for TLs
 * which use ucs_arbiter_t to progress pending requests.
 */
typedef struct {
    ucs_arbiter_elem_t  arb_elem;
} uct_pending_req_priv_arb_t;


static UCS_F_ALWAYS_INLINE ucs_arbiter_elem_t *
uct_pending_req_priv_arb_elem(uct_pending_req_t *req)
{
    uct_pending_req_priv_arb_t *priv_arb_p =
        (uct_pending_req_priv_arb_t *)&req->priv;

    return &priv_arb_p->arb_elem;
}


/**
 * Add a pending request to the arbiter.
 */
#define uct_pending_req_arb_group_push(_arbiter_group, _req) \
    do { \
        ucs_arbiter_elem_init(uct_pending_req_priv_arb_elem(_req)); \
        ucs_arbiter_group_push_elem_always(_arbiter_group, \
                                           uct_pending_req_priv_arb_elem(_req)); \
    } while (0)


/**
 * Add a pending request to the head of group in arbiter.
 */
#define uct_pending_req_arb_group_push_head(_arbiter, _arbiter_group, _req) \
    do { \
        ucs_arbiter_elem_init(uct_pending_req_priv_arb_elem(_req)); \
        ucs_arbiter_group_push_head_elem_always(_arbiter_group, \
                                                uct_pending_req_priv_arb_elem(_req)); \
    } while (0)


/**
 * Base structure for private data held inside a pending request for TLs
 * which use ucs_queue_t to progress pending requests.
 */
typedef struct {
    ucs_queue_elem_t    queue_elem;
} uct_pending_req_priv_queue_t;


static UCS_F_ALWAYS_INLINE ucs_queue_elem_t *
uct_pending_req_priv_queue_elem(uct_pending_req_t* req)
{
    uct_pending_req_priv_queue_t *priv_queue_p =
        (uct_pending_req_priv_queue_t *)&req->priv;

    return &priv_queue_p->queue_elem;
}


/**
 * Add a pending request to the queue.
 */
#define uct_pending_req_queue_push(_queue, _req) \
    ucs_queue_push((_queue), uct_pending_req_priv_queue_elem(_req))


typedef struct {
    uct_pending_purge_callback_t cb;
    void                         *arg;
} uct_purge_cb_args_t;


/**
 * Dispatch all requests in the pending queue, as long as _cond holds true.
 * _cond is an expression which can use "_priv" variable.
 *
 * @param _priv   Variable which will hold a pointer to request private data.
 * @param _queue  The pending queue.
 * @param _cond   Condition which should be true in order to keep dispatching.
 *
 * TODO support a callback returning UCS_INPROGRESS.
 */
#define uct_pending_queue_dispatch(_priv, _queue, _cond) \
    while (!ucs_queue_is_empty(_queue)) { \
        uct_pending_req_priv_queue_t *_base_priv; \
        uct_pending_req_t *_req; \
        ucs_status_t _status; \
        \
        _base_priv = \
            ucs_queue_head_elem_non_empty((_queue), \
                                          uct_pending_req_priv_queue_t, \
                                          queue_elem); \
        \
        UCS_STATIC_ASSERT(sizeof(*(_priv)) <= UCT_PENDING_REQ_PRIV_LEN); \
        _priv = (typeof(_priv))(_base_priv); \
        \
        if (!(_cond)) { \
            break; \
        } \
        \
        _req = ucs_container_of(_priv, uct_pending_req_t, priv); \
        ucs_queue_pull_non_empty(_queue); \
        _status = _req->func(_req); \
        if ((_status == UCS_ERR_NO_RESOURCE) || (_status == UCS_INPROGRESS)) { \
            ucs_queue_push_head(_queue, &_base_priv->queue_elem); \
        } \
    }


/**
 * Purge messages from the pending queue.
 *
 * @param _priv   Variable which will hold a pointer to request private data.
 * @param _queue  The pending queue.
 * @param _cond   Condition which should be true in order to remove a request.
 * @param _cb     Callback for purging the request.
 * @return Callback return value.
 */
#define uct_pending_queue_purge(_priv, _queue, _cond, _cb, _arg) \
    { \
        uct_pending_req_priv_queue_t *_base_priv; \
        ucs_queue_iter_t             _iter; \
        \
        ucs_queue_for_each_safe(_base_priv, _iter, _queue, queue_elem) { \
            _priv = (typeof(_priv))(_base_priv); \
            if (_cond) { \
                ucs_queue_del_iter(_queue, _iter); \
                (void)_cb(ucs_container_of(_base_priv, uct_pending_req_t, priv), _arg); \
            } \
        } \
    }


/**
 * Helper macro to trace active message send/receive.
 *
 * @param _iface    Interface.
 * @param _type     Message type (send/receive)
 * @param _am_id    Active message ID.
 * @param _payload  Active message payload.
 * @paral _length   Active message length
 */
#define uct_iface_trace_am(_iface, _type, _am_id, _payload, _length, _fmt, ...) \
    if (ucs_log_is_enabled(UCS_LOG_LEVEL_TRACE_DATA)) { \
        char buf[256] = {0}; \
        uct_iface_dump_am(_iface, _type, _am_id, _payload, _length, \
                          buf, sizeof(buf) - 1); \
        ucs_trace_data(_fmt " am_id %d len %zu %s", ## __VA_ARGS__, \
                       _am_id, (size_t)(_length), buf); \
    }


/**
 * Helper macro to invoke the function from iface operations.
 *
 * @param _iface    UCT interface.
 * @param _ops_type Type of iface operations.
 * @param _func     Function to call.
 * @param ...       Parameters that is passed to the function.
 */
#define uct_iface_invoke_ops_func(_iface, _ops_type, _func, ...) \
    ({ \
        _ops_type *__ops = ucs_derived_of((_iface)->ops, _ops_type); \
        __ops->_func(__VA_ARGS__); \
    })


extern ucs_config_field_t uct_iface_config_table[];


/**
 * Initialize a memory pool for buffers used by TL interface.
 *
 * @param mp
 * @param elem_size
 * @param align_offset
 * @param alignment    Data will be aligned to these units.
 * @param config       Memory pool configuration.
 * @param grow         Default number of buffers added for every chunk.
 * @param init_obj_cb  Object constructor.
 * @param name         Memory pool name.
 */
ucs_status_t uct_iface_mpool_init(uct_base_iface_t *iface, ucs_mpool_t *mp,
                                  size_t elem_size, size_t align_offset, size_t alignment,
                                  const uct_iface_mpool_config_t *config, unsigned grow,
                                  uct_iface_mpool_init_obj_cb_t init_obj_cb,
                                  const char *name);


/**
 * Dump active message contents using the user-defined tracer callback.
 */
void uct_iface_dump_am(uct_base_iface_t *iface, uct_am_trace_type_t type,
                       uint8_t id, const void *data, size_t length,
                       char *buffer, size_t max);

void uct_iface_mpool_empty_warn(uct_base_iface_t *iface, ucs_mpool_t *mp);

void uct_iface_set_async_event_params(const uct_iface_params_t *params,
                                      uct_async_event_cb_t *event_cb,
                                      void **event_arg);

ucs_status_t uct_iface_handle_ep_err(uct_iface_h iface, uct_ep_h ep,
                                      ucs_status_t status);

void uct_base_iface_query(uct_base_iface_t *iface, uct_iface_attr_t *iface_attr);

ucs_status_t uct_single_device_resource(uct_md_h md, const char *dev_name,
                                        uct_device_type_t dev_type,
                                        ucs_sys_device_t sys_device,
                                        uct_tl_device_resource_t **tl_devices_p,
                                        unsigned *num_tl_devices_p);

ucs_status_t uct_base_iface_flush(uct_iface_h tl_iface, unsigned flags,
                                  uct_completion_t *comp);

ucs_status_t uct_base_iface_fence(uct_iface_h tl_iface, unsigned flags);

void uct_base_iface_progress_enable(uct_iface_h tl_iface, unsigned flags);

void uct_base_iface_progress_enable_cb(uct_base_iface_t *iface,
                                       ucs_callback_t cb, unsigned flags);

void uct_base_iface_progress_disable(uct_iface_h tl_iface, unsigned flags);

ucs_status_t uct_base_ep_flush(uct_ep_h tl_ep, unsigned flags,
                               uct_completion_t *comp);

ucs_status_t uct_base_ep_fence(uct_ep_h tl_ep, unsigned flags);

/*
 * Invoke active message handler.
 *
 * @param iface    Interface to invoke the handler for.
 * @param id       Active message ID.
 * @param data     Received data.
 * @param length   Length of received data.
 * @param flags    Mask with @ref uct_cb_param_flags
 */
static inline ucs_status_t
uct_iface_invoke_am(uct_base_iface_t *iface, uint8_t id, void *data,
                    unsigned length, unsigned flags)
{
    ucs_status_t     status;
    uct_am_handler_t *handler;

    ucs_assertv(id < UCT_AM_ID_MAX, "invalid am id: %d (max: %lu)",
                id, UCT_AM_ID_MAX - 1);

    UCS_STATS_UPDATE_COUNTER(iface->stats, UCT_IFACE_STAT_RX_AM, 1);
    UCS_STATS_UPDATE_COUNTER(iface->stats, UCT_IFACE_STAT_RX_AM_BYTES, length);

    handler = &iface->am[id];
    status = handler->cb(handler->arg, data, length, flags);
    ucs_assert((status == UCS_OK) ||
               ((status == UCS_INPROGRESS) && (flags & UCT_CB_PARAM_FLAG_DESC)));
    return status;
}


/**
 * Invoke send completion.
 *
 * @param comp   Completion to invoke.
 * @param status Status of completed operation.
 */
static UCS_F_ALWAYS_INLINE
void uct_invoke_completion(uct_completion_t *comp, ucs_status_t status)
{
    ucs_trace_func("comp=%p, count=%d, status=%d", comp, comp->count, status);
    ucs_assertv(comp->count > 0, "comp=%p count=%d", comp, comp->count);

    uct_completion_update_status(comp, status);
    if (--comp->count == 0) {
        comp->func(comp);
    }
}


/**
 * Copy data to target am_short buffer
 */
static UCS_F_ALWAYS_INLINE
void uct_am_short_fill_data(void *buffer, uint64_t header, const void *payload,
                            size_t length)
{
    /**
     * Helper structure to fill send buffer of short messages for
     * non-accelerated transports
     */
    struct uct_am_short_packet {
        uint64_t header;
        char     payload[];
    } UCS_S_PACKED *packet = (struct uct_am_short_packet*)buffer;

    packet->header = header;
    /* suppress false positive diagnostic from uct_mm_ep_am_common_send call */
    /* cppcheck-suppress ctunullpointer */
    memcpy(packet->payload, payload, length);
}

#endif
