Examples of hg_thread_pool_init

[mashoodk]

Are there simpler examples of using hg_thread_pool_init ? I am facing difficulty in programming with thread pools. Basically my bulk transfer call backs are waiting until threads which invoked bulk transfer has completed. For me the bulk transfer (PULL) is working - I verified by adding a sleep and I can find the data transferred.
I am using socket provider.

Adding a thread pool to one of the programs in examples folder would work for me.
As of now I added the thread pool to my hg_engine_init code, I can see more threads getting created, but it doesn't help in callbacks.

void
hg_engine_init(hg_bool_t listen, const char *local_addr)
{
int ret;

HG_Set_log_level("warning");

/* boilerplate HG initialization steps */
hg_class = HG_Init(local_addr, listen);
assert(hg_class);

**hg_thread_pool_t *thread_pool;
hg_thread_pool_init(4, &thread_pool);**

hg_context = HG_Context_create(hg_class);
assert(hg_context);

/* start up thread to drive progress */
ret = pthread_create(&hg_progress_tid, NULL, hg_progress_fn, NULL);
assert(ret == 0);
(void) ret;

return;

}

[soumagne]

Ah alright, you could look at https://github.com/mercury-hpc/mercury/blob/master/Testing/util/test_threadpool.c to give you an example. For your purpose you would need to wrap the callback that is passed to HG_Register() so that it calls hg_thread_pool_post() using the callback that you would normally pass to HG_Register(). Here is also a more complex example of doing just that: https://github.com/mercury-hpc/mercury/blob/master/Testing/mercury_rpc_cb.c#L35
In the long run however we usually do not encourage using pthreads or that thread pool code directly but instead have a layer called margo (see https://github.com/mochi-hpc/mochi-margo) that lets you do all that using user-level threads instead.

[mashoodk]

Ok, let me try this :
Given below is my code logic, I have a function send_my_data() in client and process_my_data() in server.

Client :
I have a function send_my_data() in client -
send_my_data() {
...
my_rpc_id = mercuryRPC->register_process_mydata();
...
hg_engine_create_handle(svr_addr, my_rpc_id, &my_handle);

/* register buffer for rdma/bulk access by server */
const struct hg_info *hgi = HG_Get_info(my_handle);
int ret = HG_Bulk_create(hgi->hg_class, 1, &buffer, &size,
HG_BULK_READ_ONLY, &req.bulk_buffer);
...
ret = HG_Forward(my_handle, client_cb, rpcState, &req);
...
wait_for_response(); // This works fine
}

Server:
I have a function process_my_data()

process_my_data() {
...
ret = HG_Bulk_create(hgi->hg_class, 1, &buffer, &size,
HG_BULK_WRITE_ONLY, &bulk_buffer);

   // Do bulk transfer
   ret = HG_Bulk_transfer(hgi->context, my_rpc_handler_bulk_cb, &rpcState,
                            HG_BULK_PULL, hgi->addr,
                            request.bulk_buffer, 0, bulk_buffer, 0,
                            size, HG_OP_ID_IGNORE);
   ...
   // Process the data
   ret = HG_Respond(handle, NULL, NULL, &response);
   ...

}

So here I will wrap process_my_data with another function (process_my_data_wrapper()) which will do hg_thread_pool_post() and call process_my_data().

[mashoodk]

For now I am experimenting with mercury. But for incorporating into actual product, i will consider using margo. Thank you!