margo-test-client-timeout.c 5.48 KB
Newer Older
1 2 3 4 5 6 7 8 9
/*
 * (C) 2015 The University of Chicago
 * 
 * See COPYRIGHT in top-level directory.
 */

#include <stdio.h>
#include <assert.h>
#include <unistd.h>
10
#include <mercury.h>
11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40
#include <abt.h>
#include <margo.h>

#include "my-rpc.h"

/* This is an example client program that issues 4 concurrent RPCs, each of
 * which includes a bulk transfer driven by the server.
 *
 * Each client operation executes as an independent ULT in Argobots.
 * The HG forward call is executed using asynchronous operations.
 */

struct run_my_rpc_args
{
    int val;
    margo_instance_id mid;
    hg_addr_t svr_addr;
};

static void run_my_rpc(void *_arg);

static hg_id_t my_rpc_hang_id;
static hg_id_t my_rpc_shutdown_id;

int main(int argc, char **argv) 
{
    struct run_my_rpc_args args[4];
    ABT_thread threads[4];
    int i;
    int ret;
41
    hg_return_t hret;
42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60
    ABT_xstream xstream;
    ABT_pool pool;
    margo_instance_id mid;
    hg_addr_t svr_addr = HG_ADDR_NULL;
    hg_handle_t handle;
    char proto[12] = {0};
      
    if(argc != 2)
    {
        fprintf(stderr, "Usage: ./client-timeout <server_addr>\n");
        return(-1);
    }
   
    /* 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];

61 62 63 64 65 66 67
    /* actually start margo -- margo_init() encapsulates the Mercury &
     * Argobots initialization, so this step must precede their use. */
    /* Use main process to drive progress (it will relinquish control to
     * Mercury during blocking communication calls).  The rpc handler pool 
     * is null in this example program because this is a pure client that 
     * will not be servicing rpc requests.
     */
68
    /***************************************/
69 70
    mid = margo_init(proto, MARGO_CLIENT_MODE, 0, 0);
    if(mid == MARGO_INSTANCE_NULL)
71
    {
72
        fprintf(stderr, "Error: margo_init()\n");
73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
        return(-1);
    }

    /* retrieve current pool to use for ULT creation */
    ret = ABT_xstream_self(&xstream);
    if(ret != 0)
    {
        fprintf(stderr, "Error: ABT_xstream_self()\n");
        return(-1);
    }
    ret = ABT_xstream_get_main_pools(xstream, 1, &pool);
    if(ret != 0)
    {
        fprintf(stderr, "Error: ABT_xstream_get_main_pools()\n");
        return(-1);
    }

90 91
    /* register RPCs */
    my_rpc_hang_id = MARGO_REGISTER(mid, "my_rpc_hang", my_rpc_hang_in_t, my_rpc_hang_out_t, 
92
        NULL);
93
    my_rpc_shutdown_id = MARGO_REGISTER(mid, "my_shutdown_rpc", void, void, 
94 95 96
        NULL);

    /* find addr for server */
97 98
    hret = margo_addr_lookup(mid, argv[1], &svr_addr);
    assert(hret == HG_SUCCESS);
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137

    for(i=0; i<4; i++)
    {
        args[i].val = i;
        args[i].mid = mid;
        args[i].svr_addr = svr_addr;

        /* Each ult gets a pointer to an element of the array to use
         * as input for the run_my_rpc() function.
         */
        ret = ABT_thread_create(pool, run_my_rpc, &args[i],
            ABT_THREAD_ATTR_NULL, &threads[i]);
        if(ret != 0)
        {
            fprintf(stderr, "Error: ABT_thread_create()\n");
            return(-1);
        }

    }

    /* yield to one of the threads */
    ABT_thread_yield_to(threads[0]);

    for(i=0; i<4; i++)
    {
        ret = ABT_thread_join(threads[i]);
        if(ret != 0)
        {
            fprintf(stderr, "Error: ABT_thread_join()\n");
            return(-1);
        }
        ret = ABT_thread_free(&threads[i]);
        if(ret != 0)
        {
            fprintf(stderr, "Error: ABT_thread_join()\n");
            return(-1);
        }
    }

138 139
    /* send one rpc to server to shut it down */

140
    /* create handle */
141 142
    hret = margo_create(mid, svr_addr, my_rpc_shutdown_id, &handle);
    assert(hret == HG_SUCCESS);
143

144 145 146
    hret = margo_forward_timed(mid, handle, NULL, 2000.0);
    assert(hret == HG_SUCCESS);

Shane Snyder's avatar
Shane Snyder committed
147
    margo_destroy(mid, handle);
148
    margo_addr_free(mid, svr_addr);
149 150 151 152 153 154 155 156 157 158 159 160 161

    /* shut down everything */
    margo_finalize(mid);
    
    return(0);
}

static void run_my_rpc(void *_arg)
{
    struct run_my_rpc_args *arg = _arg;
    hg_handle_t handle;
    my_rpc_hang_in_t in;
    my_rpc_hang_out_t out;
162
    hg_return_t hret;
163 164 165 166 167 168 169 170 171 172 173 174
    hg_size_t size;
    void* buffer;

    printf("ULT [%d] running.\n", arg->val);

    /* allocate buffer for bulk transfer */
    size = 512;
    buffer = calloc(1, 512);
    assert(buffer);
    sprintf((char*)buffer, "Hello world!\n");

    /* create handle */
175 176
    hret = margo_create(arg->mid, arg->svr_addr, my_rpc_hang_id, &handle);
    assert(hret == HG_SUCCESS);
177 178

    /* register buffer for rdma/bulk access by server */
179
    hret = margo_bulk_create(arg->mid, 1, &buffer, &size, 
180
        HG_BULK_READ_ONLY, &in.bulk_handle);
181
    assert(hret == HG_SUCCESS);
182 183 184 185 186 187

    /* Send rpc. Note that we are also transmitting the bulk handle in the
     * input struct.  It was set above. 
     */ 
    in.input_val = arg->val;
    /* call with 2 second timeout */
188
    hret = margo_forward_timed(arg->mid, handle, &in, 2000.0);
189

190
    if(hret == HG_SUCCESS)
191 192
    {
        /* decode response */
193 194
        hret = margo_get_output(handle, &out);
        assert(hret == HG_SUCCESS);
195
        printf("Got response ret: %d\n", out.ret);
196
        margo_free_output(handle, &out);
197 198 199
    }
    else
    {
200
        printf("margo_forward returned %d\n", hret);
201 202 203
    }

    /* clean up resources consumed by this rpc */
204
    margo_bulk_free(in.bulk_handle);
Shane Snyder's avatar
Shane Snyder committed
205
    margo_destroy(arg->mid, handle);
206 207 208 209 210
    free(buffer);

    printf("ULT [%d] done.\n", arg->val);
    return;
}