rpc框架-mercury

// info_string 表示使用 network plugin种类，listen表示是否是target
na_class_t * NA_Initialize(const char *info_string, na_bool_t listen);

na_class_t *NA_Initialize_opt(const char *info_string, na_bool_t listen,const struct na_init_info *na_init_info);

// release na_class_t object
na_return_tNA_Finalize(na_class_t *na_class);

//a context within this plugin must be created, which internally creates and associates a completion queue for the operations
na_context_t *NA_Context_create(na_class_t *na_class);

na_return_t NA_Context_destroy(na_class_t *na_class, na_context_t *context);

//Target Address Lookup
na_return_t NA_Addr_self(na_class_t *na_class, na_addr_t *addr);
//convert that address to a string
na_return_t NA_Addr_to_string(na_class_t *na_class, char *buf, na_size_t buf_size, na_addr_t addr);

//string can then be exchanged to other processes through out-of-band mechanisms (e.g., using a file, etc), which can then look up the target
na_return_t NA_Addr_lookup(na_class_t *na_class, const char *name, na_addr_t *addr);

na_return_t NA_Addr_free(na_class_t *na_class, na_addr_t addr);

na_return_t
NA_Msg_send_unexpected(na_class_t *na_class, na_context_t *context,na_cb_t callback, void *arg, const void *buf, na_size_t buf_size,void *plugin_data, na_addr_t dest_addr, na_uint8_t dest_id, na_tag_t tag,na_op_id_t *op_id);

na_return_t
NA_Msg_send_expected(na_class_t *na_class, na_context_t *context,
    na_cb_t callback, void *arg, const void *buf, na_size_t buf_size,void *plugin_data, na_addr_t dest_addr, na_uint8_t dest_id, na_tag_t tag,na_op_id_t *op_id);

na_return_t
NA_Msg_recv_unexpected(na_class_t *na_class, na_context_t *context,
    na_cb_t callback, void *arg, void *buf, na_size_t buf_size,
    void *plugin_data, na_op_id_t *op_id);

na_return_t
NA_Msg_recv_expected(na_class_t *na_class, na_context_t *context,
    na_cb_t callback, void *arg, void *buf, na_size_t buf_size,
    void *plugin_data, na_addr_t source_addr, na_uint8_t source_id,
    na_tag_t tag, na_op_id_t *op_id);

na_return_t
NA_Mem_handle_create(na_class_t *na_class, void *buf, na_size_t buf_size,unsigned long flags, na_mem_handle_t *mem_handle);

na_return_t
NA_Mem_register(na_class_t *na_class, na_mem_handle_t mem_handle);

na_return_t
NA_Mem_handle_serialize(na_class_t *na_class, void *buf, na_size_t buf_size,na_mem_handle_t mem_handle);

na_return_t
NA_Mem_handle_deserialize(na_class_t *na_class, na_mem_handle_t *mem_handle,const void *buf, na_size_t buf_size);


na_return_t
NA_Put(na_class_t *na_class, na_context_t *context, na_cb_t callback, void *arg,
    na_mem_handle_t local_mem_handle, na_offset_t local_offset,
    na_mem_handle_t remote_mem_handle, na_offset_t remote_offset,
    na_size_t data_size, na_addr_t remote_addr, na_uint8_t remote_id, na_op_id_t *op_id);

na_return_t
NA_Get(na_class_t *na_class, na_context_t *context, na_cb_t callback, void *arg,
    na_mem_handle_t local_mem_handle, na_offset_t local_offset,
    na_mem_handle_t remote_mem_handle, na_offset_t remote_offset,
    na_size_t data_size, na_addr_t remote_addr, na_uint8_t remote_id, na_op_id_t *op_id);

na_return_t
NA_Progress(na_class_t *na_class, na_context_t *context, unsigned int timeout);

na_return_t
NA_Trigger(na_context_t *context, unsigned int timeout, unsigned int max_count,int callback_ret[], unsigned int *actual_count);

na_return_t
NA_Cancel(na_class_t *na_class, na_context_t *context, na_op_id_t *op_id);

hg_class_t *
HG_Init(const char *info_string, hg_bool_t listen);

hg_return_t
HG_Finalize(hg_class_t *hg_class);

hg_context_t *
HG_Context_create(hg_class_t *hg_class);
hg_return_t
HG_Context_destroy(hg_context_t *context);

typedef hg_return_t (*hg_proc_cb_t)(hg_proc_t proc, void *data);
typedef hg_return_t (*hg_rpc_cb_t)(hg_handle_t handle);

hg_id_t
HG_Register_name(hg_class_t *hg_class, const char *func_name,
                 hg_proc_cb_t in_proc_cb, hg_proc_cb_t out_proc_cb,
                 hg_rpc_cb_t rpc_cb);
hg_return_t
HG_Registered_disable_response(hg_class_t *hg_class, hg_id_t id, hg_bool_t disable);

hg_return_t
HG_Deregister(hg_class_t *hg_class, hg_id_t id);

hg_return_t
HG_Addr_self(hg_class_t *hg_class, hg_addr_t *addr);
hg_return_t
HG_Addr_to_string(hg_class_t *hg_class, char *buf, hg_size_t *buf_size, hg_addr_t addr);
hg_return_t
HG_Addr_lookup(hg_class_t *hg_class, const char *name, hg_addr_t *addr);
hg_return_t
HG_Addr_free(hg_class_t *hg_class, hg_addr_t addr);

//Using the RPC ID defined after a call to HG_Register(), one can use the HG_Create() call to define a new hg_handle_t object that can be used (and later re-used without reallocating resources) to set/get input/output arguments.
hg_return_t
HG_Create(hg_context_t *context, hg_addr_t addr, hg_id_t id, hg_handle_t *handle);

//pack the input arguments within a structure, for which a serialization function is provided with the HG_Register() call. The HG_Forward() function can then be used to send that structure (which describes the input arguments). This function is non-blocking. When it completes, the associated callback can be executed by calling HG_Trigger().
typedef hg_return_t (*hg_cb_t)(const struct hg_cb_info *callback_info);
hg_return_t
HG_Forward(hg_handle_t handle, hg_cb_t callback, void *arg, void *in_struct);

//When HG_Forward() completes (i.e., when the user callback can be triggered), the RPC has been remotely executed and a response with the output results has been sent back. 
hg_return_t
HG_Get_output(hg_handle_t handle, void *out_struct);

hg_return_t
HG_Free_output(hg_handle_t handle, void *out_struct);

hg_return_t
HG_Destroy(hg_handle_t handle);

//On the target, the RPC callback function passed to the HG_Register() call must be defined.
typedef hg_return_t (*hg_rpc_cb_t)(hg_handle_t handle);

//Whenever a new RPC is received, that callback will be invoked. 
hg_return_t
HG_Get_input(hg_handle_t handle, void *in_struct);

//When the input has been retrieved, the arguments contained in the input structure can be passed to the actual function call. When the execution is done, an output structure can be filled with the return value and/or the output arguments of the function. 
typedef hg_return_t (*hg_cb_t)(const struct hg_cb_info *callback_info);
hg_return_t
HG_Respond(hg_handle_t handle, hg_cb_t callback, void *arg, void *out_struct);

hg_return_t
HG_Free_input(hg_handle_t handle, void *in_struct);

//Mercury uses a callback model. Callbacks are passed to non-blocking functions and are pushed to the context's completion queue when the operation completes. 
hg_return_t
HG_Progress(hg_context_t *context, unsigned int timeout);

//When an operation completes, calling HG_Trigger() allows the callback execution to be separately controlled from the main progress loop.
hg_return_t
HG_Trigger(hg_context_t *context, unsigned int timeout,
           unsigned int max_count, unsigned int *actual_count);

hg_return_t
HG_Cancel(hg_handle_t handle);

hg_return_t
HG_Bulk_create(hg_class_t *hg_class, hg_uint32_t count,
               void **buf_ptrs, const hg_size_t *buf_sizes,
               hg_uint8_t flags, hg_bulk_t *handle);

hg_return_t HG_Bulk_free(hg_bulk_t handle);

//memory pointers from an existing bulk descriptor can be accessed with:
hg_return_t
HG_Bulk_access(hg_bulk_t handle, hg_size_t offset, hg_size_t size,
               hg_uint8_t flags, hg_uint32_t max_count, void **buf_ptrs,
               hg_size_t *buf_sizes, hg_uint32_t *actual_count);
//通过 HG_Bulk_bind() 函数将源地址绑定到批量句柄，代价是序列化和反序列化寻址信息的额外开销。仅当从 HG_Get_info() 调用检索的源地址与必须用于传输的源地址不同（例如，多个源）时，才需要这样做。
hg_return_t
HG_Bulk_bind(hg_bulk_t handle, hg_context_t *context);

hg_addr_t
HG_Bulk_get_addr(hg_bulk_t handle);

hg_return_t
hg_proc_hg_bulk_t(hg_proc_t proc, void *data);

hg_return_t
HG_Bulk_transfer(hg_context_t *context, hg_bulk_cb_t callback, void *arg,
                 hg_bulk_op_t op, hg_addr_t origin_addr,
                 hg_bulk_t origin_handle, hg_size_t origin_offset,
                 hg_bulk_t local_handle, hg_size_t local_offset,
                 hg_size_t size, hg_op_id_t *op_id);
                 
struct hg_info {
    hg_class_t *hg_class;               /* HG class */
    hg_context_t *context;              /* HG context */
    hg_addr_t addr;                     /* HG address */
    hg_id_t id;                         /* RPC ID */
};

struct hg_info *
HG_Get_info(hg_handle_t handle);       

MERCURY_REGISTER(hg_class, func_name, in_struct_type_name, out_struct_type_name, rpc_cb);

do {
    unsigned int actual_count = 0;
    do {
        ret = HG_Trigger(context, 0, 1, &actual_count);
    } while ((ret == HG_SUCCESS) && actual_count);
    if (done)
        break;
    ret = HG_Progress(context, HG_MAX_IDLE_TIME);
} while (ret == HG_SUCCESS);

open_in_t in_struct;
/* Initialize the interface and get target address */
hg_class = HG_Init("ofi+tcp://eth0:22222", HG_FALSE);
hg_context = HG_Context_create(hg_class);
[...]
HG_Addr_lookup_wait(hg_context, target_name, &target_addr);
/* Register RPC call */
rpc_id = MERCURY_REGISTER(hg_class, "open", open_in_t, open_out_t);
/* Set input parameters */
in_struct.in_param0 = in_param0;
/* Create RPC request */
HG_Create(hg_context, target_addr, rpc_id, &hg_handle);
/* Send RPC request */
HG_Forward(hg_handle, rpc_done_cb, &rpc_done_args, &in_struct);
/* Make progress */

/* cancellation */
// Cancellation: HG Cancel() on handle
// Callback still triggered (canceled = completion)

hg_return_t
    rpc_done_cb(const struct hg_cb_info *callback_info)
{
    open_out_t out_struct;
    /* Get output */
    HG_Get_output(callback_info->handle, &out_struct);
    /* Get output parameters */
    ret = out_struct.ret;
    out_param0 = out_struct.out_param0;
    /* Free output */
    HG_Free_output(callback_info->handle, &out_struct);
    return HG_SUCCESS;
}

int main(int argc, void *argv[])
{
    /* Initialize the interface and listen */
    hg_class = HG_Init("ofi+tcp://eth0:22222", HG_TRUE); 
    [...]
    /* Register RPC call */
    MERCURY_REGISTER(hg_class, "open", open_in_t, open_out_t, open_rpc_cb);
    /* Make progress */
    [...]
    /* Finalize the interface */
    [...]
}

hg_return_t open_rpc_cb(hg_handle_t handle)
{
    open_in_t in_struct;
    open_out_t out_struct;
    /* Get input */
    HG_Get_input(handle, &in_struct);
    in_param0 = in_struct.in_param0;
    /* Execute call */
    out_param0 = open(in_param0, ...);
    /* Set output */
    open_out_struct.out_param0 = out_param0;
    /* Send response back to origin */
    HG_Respond(handle, NULL, NULL, &out_struct);
    /* Free input and destroy handle */
    HG_Free_input(handle, &in_struct);
    HG_Destroy(handle);
    return HG_SUCCESS;
}

/* Initialize the interface and get target address */
[...]
/* Create bulk handle (only change) */
HG_Bulk_create(hg_info->hg_bulk_class, 1, &buf, &buf_size, HG_BULK_READ_ONLY, &bulk_handle);
/* Attach bulk handle to input parameters */
[...]
in_struct.bulk_handle = bulk_handle;
/* Create RPC request */
HG_Create(hg_context, target_addr, rpc_id, &hg_handle);
/* Send RPC request */
HG_Forward(hg_handle, rpc_done_cb, &rpc_done_args, &in_struct);
/* Make progress */

/* Get input parameters and bulk handle */
HG_Get_input(handle, &in_struct);
[...]
origin_bulk_handle = in_struct.bulk_handle;
/* Get size of data and allocate buffer */
nbytes = HG_Bulk_get_size(bulk_handle);
/* Create block handle to read data */
HG_Bulk_create(hg_info->hg_bulk_class, 1, NULL, &nbytes,
               HG_BULK_READWRITE, &local_bulk_handle);
/* Start pulling bulk data (execute call / send response in callback) */
HG_Bulk_transfer(hg_info->bulk_context, bulk_transfer_cb,
                 bulk_args, HG_BULK_PULL, hg_info->addr, origin_bulk_handle, 0,
                 local_bulk_handle, 0, nbytes, HG_OP_ID_IGNORE);

hg_return_t HG_Bulk_create(
    hg_bulk_class_t *hg_bulk_class,
    hg_size_t count,
    void **buf_ptrs,
    const hg_size_t *buf_sizes,
    hg_uint8_t flags,
    hg_bulk_t *handle
);