Merge branch 'dev-multiplex-service'

adds multiplex id feature; putting in master because corresponding
features are in Mercury master now

Makefile.am

@@ -39,5 +39,6 @@ include Make.rules
 
 include $(top_srcdir)/src/Makefile.subdir
 include $(top_srcdir)/examples/Makefile.subdir
+include $(top_srcdir)/examples/multiplex/Makefile.subdir
 include $(top_srcdir)/tests/Makefile.subdir
 

examples/multiplex/Makefile.subdir

+noinst_PROGRAMS += examples/multiplex/margo-example-mp-server examples/multiplex/margo-example-mp-client
+
+examples_multiplex_margo_example_mp_server_SOURCES = \
+ examples/multiplex/margo-example-mp-server.c \
+ examples/multiplex/svc1-server.c \
+ examples/multiplex/svc2-server.c
+
+examples_multiplex_margo_example_mp_client_SOURCES = \
+ examples/multiplex/margo-example-mp-client.c \
+ examples/multiplex/svc1-client.c \
+ examples/multiplex/svc2-client.c

examples/multiplex/margo-example-mp-client.c

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include <abt.h>
+#include <abt-snoozer.h>
+#include <margo.h>
+
+#include "svc1-client.h"
+#include "svc2-client.h"
+
+static hg_id_t my_rpc_shutdown_id;
+
+int main(int argc, char **argv) 
+{
+    int i;
+    int ret;
+    margo_instance_id mid;
+    hg_context_t *hg_context;
+    hg_class_t *hg_class;
+    hg_addr_t svr_addr = HG_ADDR_NULL;
+    hg_handle_t handle;
+    char proto[12] = {0};
+  
+    if(argc != 2)
+    {
+        fprintf(stderr, "Usage: ./client <server_addr>\n");
+        return(-1);
+    }
+       
+    /* boilerplate HG initialization steps */
+    /***************************************/
+
+    /* initialize Mercury using the transport portion of the destination
+     * address (i.e., the part before the first : character if present)
+     */
+    for(i=0; i<11 && argv[1][i] != '\0' && argv[1][i] != ':'; i++)
+        proto[i] = argv[1][i];
+    hg_class = HG_Init(proto, HG_FALSE);
+    if(!hg_class)
+    {
+        fprintf(stderr, "Error: HG_Init()\n");
+        return(-1);
+    }
+    hg_context = HG_Context_create(hg_class);
+    if(!hg_context)
+    {
+        fprintf(stderr, "Error: HG_Context_create()\n");
+        HG_Finalize(hg_class);
+        return(-1);
+    }
+
+    /* set up argobots */
+    /***************************************/
+    ret = ABT_init(argc, argv);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: ABT_init()\n");
+        return(-1);
+    }
+
+    /* set primary ES to idle without polling */
+    ret = ABT_snoozer_xstream_self_set();
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: ABT_snoozer_xstream_self_set()\n");
+        return(-1);
+    }
+
+    /* actually start margo */
+    /***************************************/
+    mid = margo_init(0, 0, hg_context);
+
+    /* register core RPC */
+    my_rpc_shutdown_id = MERCURY_REGISTER(hg_class, "my_shutdown_rpc", void, void, 
+        NULL);
+    /* register service APIs */
+    svc1_register_client(mid);
+    svc2_register_client(mid);
+
+    /* find addr for server */
+    ret = margo_addr_lookup(mid, argv[1], &svr_addr);
+    assert(ret == 0);
+
+    svc1_do_thing(mid, svr_addr, 1);
+    svc1_do_other_thing(mid, svr_addr, 1);
+    svc1_do_thing(mid, svr_addr, 2);
+    svc1_do_other_thing(mid, svr_addr, 2);
+    svc2_do_thing(mid, svr_addr, 3);
+    svc2_do_other_thing(mid, svr_addr, 3);
+
+    /* send one rpc to server to shut it down */
+    /* create handle */
+    ret = HG_Create(hg_context, svr_addr, my_rpc_shutdown_id, &handle);
+    assert(ret == 0);
+
+    margo_forward(mid, handle, NULL);
+
+    HG_Addr_free(hg_class, svr_addr);
+
+    /* shut down everything */
+    margo_finalize(mid);
+    
+    ABT_finalize();
+
+    HG_Context_destroy(hg_context);
+    HG_Finalize(hg_class);
+
+    return(0);
+}
+
+

examples/multiplex/margo-example-mp-server.c

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include <abt.h>
+#include <abt-snoozer.h>
+#include <margo.h>
+
+#include "svc1-server.h"
+#include "svc2-server.h"
+
+/* example server program that starts a skeleton for sub-services within
+ * this process to register with
+ */
+
+/* this is a "common" rpc that is handled by the core daemon
+ */
+
+static void my_rpc_shutdown_ult(hg_handle_t handle)
+{
+    hg_return_t hret;
+    struct hg_info *hgi;
+    margo_instance_id mid;
+
+    //printf("Got RPC request to shutdown\n");
+
+    hgi = HG_Get_info(handle);
+    assert(hgi);
+    mid = margo_hg_class_to_instance(hgi->hg_class);
+
+    hret = margo_respond(mid, handle, NULL);
+    assert(hret == HG_SUCCESS);
+
+    HG_Destroy(handle);
+
+    /* NOTE: we assume that the server daemon is using
+     * margo_wait_for_finalize() to suspend until this RPC executes, so there
+     * is no need to send any extra signal to notify it.
+     */
+    margo_finalize(mid);
+
+    return;
+}
+DEFINE_MARGO_RPC_HANDLER(my_rpc_shutdown_ult)
+
+
+int main(int argc, char **argv) 
+{
+    int ret;
+    margo_instance_id mid;
+    hg_context_t *hg_context;
+    hg_class_t *hg_class;
+    char proto[12] = {0};
+    int i;
+    hg_addr_t addr_self;
+    char addr_self_string[128];
+    hg_size_t addr_self_string_sz = 128;
+    ABT_xstream svc1_xstream2;
+    ABT_pool svc1_pool2;
+    ABT_pool *handler_pool;
+
+    if(argc != 2)
+    {
+        fprintf(stderr, "Usage: ./server <listen_addr>\n");
+        fprintf(stderr, "Example: ./server na+sm://\n");
+        return(-1);
+    }
+
+    /* boilerplate HG initialization steps */
+    /***************************************/
+    hg_class = HG_Init(argv[1], HG_TRUE);
+    if(!hg_class)
+    {
+        fprintf(stderr, "Error: HG_Init()\n");
+        return(-1);
+    }
+    hg_context = HG_Context_create(hg_class);
+    if(!hg_context)
+    {
+        fprintf(stderr, "Error: HG_Context_create()\n");
+        HG_Finalize(hg_class);
+        return(-1);
+    }
+
+    /* figure out what address this server is listening on */
+    ret = HG_Addr_self(hg_class, &addr_self);
+    if(ret != HG_SUCCESS)
+    {
+        fprintf(stderr, "Error: HG_Addr_self()\n");
+        HG_Context_destroy(hg_context);
+        HG_Finalize(hg_class);
+        return(-1);
+    }
+    ret = HG_Addr_to_string(hg_class, addr_self_string, &addr_self_string_sz, addr_self);
+    if(ret != HG_SUCCESS)
+    {
+        fprintf(stderr, "Error: HG_Addr_self()\n");
+        HG_Context_destroy(hg_context);
+        HG_Finalize(hg_class);
+        HG_Addr_free(hg_class, addr_self);
+        return(-1);
+    }
+    HG_Addr_free(hg_class, addr_self);
+
+    for(i=0; i<11 && argv[1][i] != '\0' && argv[1][i] != ':'; i++)
+        proto[i] = argv[1][i];
+    printf("# accepting RPCs on address \"%s://%s\"\n", proto, addr_self_string);
+
+    /* set up argobots */
+    /***************************************/
+    ret = ABT_init(argc, argv);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: ABT_init()\n");
+        return(-1);
+    }
+
+    /* set primary ES to idle without polling */
+    ret = ABT_snoozer_xstream_self_set();
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: ABT_snoozer_xstream_self_set()\n");
+        return(-1);
+    }
+
+    /* actually start margo */
+    /***************************************/
+    mid = margo_init(0, 0, hg_context);
+    assert(mid);
+
+    /* register RPCs and services */
+    /***************************************/
+
+    /* register a shutdown RPC as just a generic handler; not part of a
+     * multiplexed service
+     */
+    MERCURY_REGISTER(hg_class, "my_shutdown_rpc", void, void,
+        my_rpc_shutdown_ult_handler);
+
+    /* register svc1, with mplex_id 1, to execute on the default handler pool
+     * used by Margo
+     */
+    handler_pool = margo_get_handler_pool(mid);
+    ret = svc1_register(mid, *handler_pool, 1);
+    assert(ret == 0);
+
+    /* create a dedicated and pool for another instance of svc1 */
+    ret = ABT_snoozer_xstream_create(1, &svc1_pool2, &svc1_xstream2);
+    assert(ret == 0);
+    /* register svc1, with mplex_id 2, to execute on a separate pool.  This
+     * will result in svc1 being registered twice, with the client being able
+     * to dictate which instance they want to target
+     */
+    ret = svc1_register(mid, svc1_pool2, 2);
+    assert(ret == 0);
+
+    /* register svc2, with mplex_id 3, to execute on the default handler pool
+     * used by Margo
+     */
+    handler_pool = margo_get_handler_pool(mid);
+    ret = svc2_register(mid, *handler_pool, 3);
+    assert(ret == 0);
+
+
+    /* shut things down */
+    /****************************************/
+
+    /* NOTE: there isn't anything else for the server to do at this point
+     * except wait for itself to be shut down.  The
+     * margo_wait_for_finalize() call here yields to let Margo drive
+     * progress until that happens.
+     */
+    margo_wait_for_finalize(mid);
+
+    /*  TODO: rethink this; can't touch mid after wait for finalize */
+#if 0
+    svc1_deregister(mid, *handler_pool, 1);
+    svc1_deregister(mid, svc1_pool2, 2);
+    svc2_deregister(mid, *handler_pool, 3);
+#endif
+
+    ABT_xstream_join(svc1_xstream2);
+    ABT_xstream_free(&svc1_xstream2);
+
+    ABT_finalize();
+
+    HG_Context_destroy(hg_context);
+    HG_Finalize(hg_class);
+
+    return(0);
+}
+

examples/multiplex/svc1-client.c

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include <abt.h>
+
+#include "svc1-client.h"
+#include "svc1-proto.h"
+
+/* NOTE: just making these global for test example, would be cleaner if there
+ * were an instance pointer for the client.
+ */
+static hg_id_t svc1_do_thing_id = -1;
+static hg_id_t svc1_do_other_thing_id = -1;
+
+int svc1_register_client(margo_instance_id mid)
+{
+
+    svc1_do_thing_id = MERCURY_REGISTER(margo_get_class(mid), "svc1_do_thing", 
+        svc1_do_thing_in_t, svc1_do_thing_out_t, NULL);
+
+    svc1_do_other_thing_id = MERCURY_REGISTER(margo_get_class(mid), "svc1_do_other_thing", 
+        svc1_do_other_thing_in_t, svc1_do_other_thing_out_t, NULL);
+
+    return(0);
+}
+
+void svc1_do_thing(margo_instance_id mid, hg_addr_t svr_addr, uint32_t mplex_id)
+{
+    hg_handle_t handle;
+    svc1_do_thing_in_t in;
+    svc1_do_thing_out_t out;
+    int ret;
+    hg_size_t size;
+    void* buffer;
+    struct hg_info *hgi;
+
+    /* allocate buffer for bulk transfer */
+    size = 512;
+    buffer = calloc(1, 512);
+    assert(buffer);
+    sprintf((char*)buffer, "Hello world!\n");
+
+    /* create handle */
+    ret = HG_Create(margo_get_context(mid), svr_addr, svc1_do_thing_id, &handle);
+    assert(ret == 0);
+
+    /* register buffer for rdma/bulk access by server */
+    hgi = HG_Get_info(handle);
+    assert(hgi);
+    ret = HG_Bulk_create(hgi->hg_class, 1, &buffer, &size, 
+        HG_BULK_READ_ONLY, &in.bulk_handle);
+    assert(ret == 0);
+
+    hgi->mplex_id = mplex_id;
+
+    /* Send rpc. Note that we are also transmitting the bulk handle in the
+     * input struct.  It was set above. 
+     */ 
+    in.input_val = 0;
+    margo_forward(mid, handle, &in);
+
+    /* decode response */
+    ret = HG_Get_output(handle, &out);
+    assert(ret == 0);
+
+    /* clean up resources consumed by this rpc */
+    HG_Bulk_free(in.bulk_handle);
+    HG_Free_output(handle, &out);
+    HG_Destroy(handle);
+    free(buffer);
+
+    return;
+}
+
+void svc1_do_other_thing(margo_instance_id mid, hg_addr_t svr_addr, uint32_t mplex_id)
+{
+    hg_handle_t handle;
+    svc1_do_other_thing_in_t in;
+    svc1_do_other_thing_out_t out;
+    int ret;
+    hg_size_t size;
+    void* buffer;
+    struct hg_info *hgi;
+
+    /* allocate buffer for bulk transfer */
+    size = 512;
+    buffer = calloc(1, 512);
+    assert(buffer);
+    sprintf((char*)buffer, "Hello world!\n");
+
+    /* create handle */
+    ret = HG_Create(margo_get_context(mid), svr_addr, svc1_do_other_thing_id, &handle);
+    assert(ret == 0);
+
+    /* register buffer for rdma/bulk access by server */
+    hgi = HG_Get_info(handle);
+    assert(hgi);
+    ret = HG_Bulk_create(hgi->hg_class, 1, &buffer, &size, 
+        HG_BULK_READ_ONLY, &in.bulk_handle);
+    assert(ret == 0);
+
+    hgi->mplex_id = mplex_id;
+
+    /* Send rpc. Note that we are also transmitting the bulk handle in the
+     * input struct.  It was set above. 
+     */ 
+    in.input_val = 0;
+    margo_forward(mid, handle, &in);
+
+    /* decode response */
+    ret = HG_Get_output(handle, &out);
+    assert(ret == 0);
+
+    /* clean up resources consumed by this rpc */
+    HG_Bulk_free(in.bulk_handle);
+    HG_Free_output(handle, &out);
+    HG_Destroy(handle);
+    free(buffer);
+
+    return;
+}
+

examples/multiplex/svc1-client.h

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#ifndef __SVC1_CLIENT
+#define __SVC1_CLIENT
+
+#include <margo.h>
+
+int svc1_register_client(margo_instance_id mid);
+
+void svc1_do_thing(margo_instance_id mid, hg_addr_t svr_addr, uint32_t mplex_id);
+void svc1_do_other_thing(margo_instance_id mid, hg_addr_t svr_addr, uint32_t mplex_id);
+
+#endif /* __SVC1_CLIENT */

examples/multiplex/svc1-proto.h

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#ifndef __SVC1_PROTO
+#define __SVC1_PROTO
+
+#include <margo.h>
+
+MERCURY_GEN_PROC(svc1_do_thing_out_t, ((int32_t)(ret)))
+MERCURY_GEN_PROC(svc1_do_thing_in_t,
+    ((int32_t)(input_val))\
+    ((hg_bulk_t)(bulk_handle)))
+
+MERCURY_GEN_PROC(svc1_do_other_thing_out_t, ((int32_t)(ret)))
+MERCURY_GEN_PROC(svc1_do_other_thing_in_t,
+    ((int32_t)(input_val))\
+    ((hg_bulk_t)(bulk_handle)))
+
+#endif /* __SVC1_PROTO */

examples/multiplex/svc1-server.c

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <assert.h>
+#include <pthread.h>
+#include "svc1-proto.h"
+#include "svc1-server.h"
+
+static void svc1_do_thing_ult(hg_handle_t handle)
+{
+    hg_return_t hret;
+    svc1_do_thing_out_t out;
+    svc1_do_thing_in_t in;
+    int ret;
+    hg_size_t size;
+    void *buffer;
+    hg_bulk_t bulk_handle;
+    struct hg_info *hgi;
+    margo_instance_id mid;
+    ABT_thread my_ult;
+    ABT_xstream my_xstream; 
+    pthread_t my_tid;
+
+    ret = HG_Get_input(handle, &in);
+    assert(ret == HG_SUCCESS);
+    hgi = HG_Get_info(handle);
+    assert(hgi);
+
+    ABT_xstream_self(&my_xstream);
+    ABT_thread_self(&my_ult);
+    my_tid = pthread_self();
+    printf("svc1: do_thing: mplex_id: %u, ult: %p, xstream %p, tid: %lu\n", 
+        hgi->mplex_id, my_ult, my_xstream, my_tid);
+
+    out.ret = 0;
+
+    /* set up target buffer for bulk transfer */
+    size = 512;
+    buffer = calloc(1, 512);
+    assert(buffer);
+
+    /* register local target buffer for bulk access */
+    ret = HG_Bulk_create(hgi->hg_class, 1, &buffer,
+        &size, HG_BULK_WRITE_ONLY, &bulk_handle);
+    assert(ret == 0);
+
+    mid = margo_hg_class_to_instance(hgi->hg_class);
+
+    /* do bulk transfer from client to server */
+    ret = margo_bulk_transfer(mid, HG_BULK_PULL,
+        hgi->addr, in.bulk_handle, 0,
+        bulk_handle, 0, size);
+    assert(ret == 0);
+
+    HG_Free_input(handle, &in);
+
+    hret = HG_Respond(handle, NULL, NULL, &out);
+    assert(hret == HG_SUCCESS);
+
+    HG_Bulk_free(bulk_handle);
+    HG_Destroy(handle);
+    free(buffer);
+
+    return;
+}
+DEFINE_MARGO_RPC_HANDLER(svc1_do_thing_ult)
+
+static void svc1_do_other_thing_ult(hg_handle_t handle)
+{
+    hg_return_t hret;
+    svc1_do_other_thing_out_t out;
+    svc1_do_other_thing_in_t in;
+    int ret;
+    hg_size_t size;
+    void *buffer;
+    hg_bulk_t bulk_handle;
+    struct hg_info *hgi;
+    margo_instance_id mid;
+    ABT_thread my_ult;
+    ABT_xstream my_xstream; 
+    pthread_t my_tid;
+
+    ret = HG_Get_input(handle, &in);
+    assert(ret == HG_SUCCESS);
+    hgi = HG_Get_info(handle);
+    assert(hgi);
+
+    ABT_xstream_self(&my_xstream);
+    ABT_thread_self(&my_ult);
+    my_tid = pthread_self();
+    printf("svc1: do_other_thing: mplex_id: %u, ult: %p, xstream %p, tid: %lu\n", 
+        hgi->mplex_id, my_ult, my_xstream, my_tid);
+
+    out.ret = 0;
+
+    /* set up target buffer for bulk transfer */
+    size = 512;
+    buffer = calloc(1, 512);
+    assert(buffer);
+
+    /* register local target buffer for bulk access */
+    ret = HG_Bulk_create(hgi->hg_class, 1, &buffer,
+        &size, HG_BULK_WRITE_ONLY, &bulk_handle);
+    assert(ret ==