root/branches/orange-next/src/common/misc/msgpairarray.sm @ 8959

/* 
 * (C) 2003 Clemson University and The University of Chicago 
 *
 * See COPYING in top-level directory.
 */

/* pvfs2_msgpairarray_sm
 *
 * The purpose of this state machine is to prepare, send, and
 * receive a collection of request/response pairs (msgpairs).
 */

#include <string.h>
#include <assert.h>

#include "msgpairarray.h"
#include "pvfs2-debug.h"
#include "pint-cached-config.h"
#include "job.h"
#include "gossip.h"
#include "PINT-reqproto-encode.h"
#include "pvfs2-util.h"
#include "pint-util.h"
#include "server-config-mgr.h"
#include "pvfs2-internal.h"
#include "state-machine.h"

#define gossip_err_unless_quiet(format, f...) \
do {\
    if(mop->params.quiet_flag)\
        gossip_debug(GOSSIP_MSGPAIR_DEBUG, format, ##f); \
    else \
        gossip_err(format, ##f); \
} while(0)

enum
{
    MSGPAIRS_COMPLETE = 190,
    MSGPAIRS_RETRY = 191,
    MSGPAIRS_RETRY_NODELAY = 192
};

%%

nested machine pvfs2_msgpairarray_sm
{
    state init
    {
        run msgpairarray_init;
        default => post;
    }

    state post
    {
        run msgpairarray_post;
        MSGPAIRS_COMPLETE => completion_fn;
        default => complete;
    }

    state post_retry
    {
        run msgpairarray_post_retry;
        default => post;
    }

    state complete
    {
        run msgpairarray_complete;
        MSGPAIRS_COMPLETE => completion_fn;
        default => complete;
    }

    state completion_fn
    {
        run msgpairarray_completion_fn;
        MSGPAIRS_RETRY => post_retry;
        MSGPAIRS_RETRY_NODELAY => post;
        default => done;
    }

    state done
    {
        run msgpairarray_done;
        default => return;
    }
}

%%

static PINT_sm_action msgpairarray_init(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    PINT_sm_msgarray_op *mop = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int i = 0;
    PINT_sm_msgpair_state *msg_p = NULL;

    gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing msgpairarray_init...\n");
    gossip_debug(GOSSIP_MIRROR_DEBUG,"\tbase frame:%d\tframe count:%d\n"
                                    ,smcb->base_frame,smcb->frame_count);

    gossip_debug(GOSSIP_MSGPAIR_DEBUG, "(%p) msgpairarray state: init "
                 "(%d msgpair(s))\n", smcb, mop->count);

    assert(mop->count > 0);

    js_p->error_code = 0;

    /* set number of operations that must complete.  */
    mop->params.comp_ct = (2 * mop->count);

    for(i = 0; i < mop->count; i++)
    {
        msg_p = &mop->msgarray[i];
        assert(msg_p);

        assert((msg_p->retry_flag == PVFS_MSGPAIR_RETRY) ||
               (msg_p->retry_flag == PVFS_MSGPAIR_NO_RETRY));

        msg_p->encoded_resp_p = NULL;
        msg_p->retry_count = 0;
        msg_p->complete = 0;

    }
    return SM_ACTION_COMPLETE;
}

/* msgpairarray_post()
 *
 * The following elements of the PINT_sm_msgpair_state
 * should be valid prior to this state (for each msgpair in array):
 * - req (unencoded request)
 * - srv_addr of each element in msg array
 *
 * This state performs the following operations for each msgpair,
 * one at a time:
 * (1) encodes request
 * (2) calculates maximum response size
 * (3) allocates BMI memory for response data (encoded)
 * (4) gets a session tag for the pair of messages
 * (5) posts the receive of the response
 * (6) posts the send of the request
 * (7) stores job ids for later matching
 *
 */
static PINT_sm_action msgpairarray_post(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    PINT_sm_msgarray_op *mop = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL, i = 0, tmp = 0;
    struct server_configuration_s *server_config = NULL;
    PVFS_msg_tag_t session_tag;
    PINT_sm_msgpair_state *msg_p = NULL;
    struct filesystem_configuration_s *cur_fs = NULL;
    int must_loop_encodings = 0;
    int local_enc_and_alloc = 0;

    gossip_debug(
        GOSSIP_MSGPAIR_DEBUG, "%s: sm %p "
        "%d total message(s) with %d incomplete\n", __func__, smcb,
        mop->count * 2, mop->params.comp_ct);

    js_p->error_code = 0;

    assert(mop->count > 0);
    assert(mop->params.comp_ct >= 2);

    for (i = 0; i < mop->count; i++)
    {
        msg_p = &mop->msgarray[i];
        assert(msg_p);

        /*
          here we skip over the msgs that have already completed in
          the case of being in the retry code path when it's ok
        */
        if (msg_p->complete)
        {
            continue;
        }

        msg_p->op_status = 0;

        if (msg_p->encoded_resp_p == NULL)
        {
            if (msg_p->fs_id != PVFS_FS_ID_NULL)
            {
                server_config = PINT_server_config_mgr_get_config(
                    msg_p->fs_id);
                assert(server_config);

                cur_fs = PINT_config_find_fs_id(
                    server_config, msg_p->fs_id);
                PINT_server_config_mgr_put_config(server_config);
                assert(cur_fs);
                msg_p->enc_type = cur_fs->encoding;
            }

            if (!ENCODING_IS_VALID(msg_p->enc_type))
            {
                PRINT_ENCODING_ERROR("supported", msg_p->enc_type);
                must_loop_encodings = 1;
                msg_p->enc_type = (ENCODING_INVALID_MIN + 1);
            }
            else if (!ENCODING_IS_SUPPORTED(msg_p->enc_type))
            {
                PRINT_ENCODING_ERROR("supported", msg_p->enc_type);
                must_loop_encodings = 1;
                msg_p->enc_type = ENCODING_SUPPORTED_MIN;
            }

          try_next_encoding:
            assert(ENCODING_IS_VALID(msg_p->enc_type));

            ret = PINT_encode(&msg_p->req, PINT_ENCODE_REQ,
                              &msg_p->encoded_req, msg_p->svr_addr,
                              msg_p->enc_type);
            if (ret != 0)
            {
                if (must_loop_encodings)
                {
                    gossip_debug(GOSSIP_MSGPAIR_DEBUG, "Looping through "
                                 "encodings [%d/%d]\n", msg_p->enc_type,
                                 ENCODING_INVALID_MAX);

                    msg_p->enc_type++;
                    if (ENCODING_IS_VALID(msg_p->enc_type))
                    {
                        goto try_next_encoding;
                    }
                }
                gossip_lerr("msgpairarray_post: PINT_encode failed\n");
                js_p->error_code = ret;
                return SM_ACTION_COMPLETE;
            }

            /* calculate max response msg size and allocate space */
            msg_p->max_resp_sz = PINT_encode_calc_max_size(
                PINT_ENCODE_RESP, msg_p->req.op, msg_p->enc_type);


            msg_p->encoded_resp_p = BMI_memalloc(
                msg_p->svr_addr, msg_p->max_resp_sz, BMI_RECV);

            if (msg_p->encoded_resp_p == NULL)
            {
                js_p->error_code = -PVFS_ENOMEM;
                return SM_ACTION_COMPLETE;
            }
            local_enc_and_alloc = 1;
        }

        session_tag = PINT_util_get_next_tag();

        /*store the session tag for this msgpair, so the msgpair completion */
        /*function can pass it to the caller of msgpairarray.               */
        msg_p->session_tag = session_tag;

        gossip_debug(GOSSIP_MSGPAIR_DEBUG, "%s: sm %p msgpair %d: "
                     "posting recv\n", __func__, smcb, i);

        /* post receive of response; job_id stored in recv_id */
        ret = job_bmi_recv(msg_p->svr_addr,
                           msg_p->encoded_resp_p,
                           msg_p->max_resp_sz,
                           session_tag,
                           BMI_PRE_ALLOC,
                           smcb,
                           i,
                           &msg_p->recv_status,
                           &msg_p->recv_id,
                           mop->params.job_context,
                           mop->params.job_timeout,
                           msg_p->req.hints);
        if (ret == 0)
        {
            /* perform a quick test to see if the recv failed before posting
             * the send; if it reports an error quickly then we can save the
             * confusion of sending a request for which we can't recv a
             * response
             */
            ret = job_test(msg_p->recv_id, &tmp, NULL,
                           &msg_p->recv_status, 0,
                           mop->params.job_context);
        }

        if ((ret < 0) || (ret == 1))
        {
            /* it is impossible for this recv to complete at this point
             * without errors; we haven't sent the request yet!
             */
            assert(ret < 0 || msg_p->recv_status.error_code != 0);
            if (ret < 0)
            {
                PVFS_perror_gossip("Post of receive failed", ret);
            }
            else
            {
                PVFS_perror_gossip("Receive immediately failed",
                            msg_p->recv_status.error_code);
            }

            msg_p->recv_id = 0;
            msg_p->send_id = 0;

            /* mark send as bad too and don't post it */
            msg_p->send_status.error_code = msg_p->recv_status.error_code;
            msg_p->op_status = msg_p->recv_status.error_code;
            mop->params.comp_ct -= 2;

            if (local_enc_and_alloc)
            {
                PINT_encode_release(&msg_p->encoded_req, PINT_ENCODE_REQ);
                memset(&msg_p->encoded_req,0,sizeof(msg_p->encoded_req));
                BMI_memfree(msg_p->svr_addr,msg_p->encoded_resp_p,
                            msg_p->max_resp_sz, BMI_RECV);
                msg_p->encoded_resp_p = NULL;
                local_enc_and_alloc = 0;
            }

            /* continue to send other array entries if possible */
            continue;
        }

        /* if we reach here, the recv has been posted without failure, but
         * has not completed yet
         */
        assert(ret == 0);

        gossip_debug(GOSSIP_MSGPAIR_DEBUG, "%s: sm %p msgpair %d: "
                     "posting send\n", __func__, smcb, i);

        /* post send of request; job_id stored in send_id */
        ret = job_bmi_send_list(msg_p->encoded_req.dest,
                                msg_p->encoded_req.buffer_list,
                                msg_p->encoded_req.size_list,
                                msg_p->encoded_req.list_count,
                                msg_p->encoded_req.total_size,
                                session_tag,
                                msg_p->encoded_req.buffer_type,
                                1,
                                smcb,
                                mop->count+i,
                                &msg_p->send_status,
                                &msg_p->send_id,
                                mop->params.job_context,
                                mop->params.job_timeout,
                                msg_p->req.hints);

         if ((ret < 0) ||
            ((ret == 1) && (msg_p->send_status.error_code != 0)))
        {
            if (ret < 0)
            {
                PVFS_perror_gossip("Post of send failed", ret);
            }
            else
            {
                PVFS_perror_gossip("Send immediately failed",
                    msg_p->send_status.error_code);
            }

            gossip_err_unless_quiet("Send error: cancelling recv.\n");

            job_bmi_cancel(msg_p->recv_id, mop->params.job_context);
            
            /* we still have to wait for recv completion, so just decrement
             * comp_ct by one and keep going
             */
            msg_p->op_status = msg_p->send_status.error_code;
            msg_p->send_id = 0;
            mop->params.comp_ct--;
        }
        else if (ret == 1)
        {
            /* immediate completion */
            msg_p->send_id = 0;
            /* decrement our count, since send is already done. */
            mop->params.comp_ct--;
        }
        /* else: successful post, no immediate completion */
    }

    if (mop->params.comp_ct == 0)
    {
        /* everything is completed already (could happen in some failure
         * cases); jump straight to final completion function.
         */
         js_p->error_code = MSGPAIRS_COMPLETE;
         return SM_ACTION_COMPLETE;
    }

    /* we are still waiting on operations to complete, next state
     * transition will handle them
     */
    return SM_ACTION_DEFERRED;
}

static PINT_sm_action msgpairarray_post_retry(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    PINT_sm_msgarray_op *mop = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    job_id_t tmp_id;

    gossip_debug(GOSSIP_MSGPAIR_DEBUG, "%s: sm %p, wait %d ms\n",
      __func__, smcb, mop->params.retry_delay);

    js_p->error_code = 0;  /* do not leak MSGPAIRS_RETRY through to wait */
    return job_req_sched_post_timer(
        mop->params.retry_delay,
        smcb, 0, js_p, &tmp_id,
        mop->params.job_context);
}

static PINT_sm_action msgpairarray_complete(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    PINT_sm_msgarray_op *mop = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    gossip_debug(GOSSIP_MSGPAIR_DEBUG,
      "%s: sm %p status_user_tag %d msgarray_count %d\n",
      __func__, smcb, (int) js_p->status_user_tag, mop->count);


    /* match operation with something in the msgpair array */
    /* the first N tags are receives, the second N are sends */
    assert(js_p->status_user_tag < mop->count*2);

    if (js_p->status_user_tag < mop->count)
    {
        PINT_sm_msgpair_state *msg_p = 
            &mop->msgarray[js_p->status_user_tag];

        msg_p->recv_id = 0;
        msg_p->recv_status = *js_p;

        /* save error (if we don't already have one) in op_status */
        if(msg_p->op_status == 0)
            msg_p->op_status = msg_p->recv_status.error_code;

        if(msg_p->recv_status.error_code && msg_p->send_id != 0)
        {
            /* we got a receive error, but send is still pending.  Cancel
             * the send
             */
            job_bmi_cancel(msg_p->send_id, mop->params.job_context);
        }
    }
    else
    {
        PINT_sm_msgpair_state *msg_p = &mop->msgarray[
            js_p->status_user_tag - mop->count];

        msg_p->send_id = 0;
        msg_p->send_status = *js_p;

        /* save error (if we don't already have one) in op_status */
        if(msg_p->op_status == 0)
            msg_p->op_status = msg_p->send_status.error_code;

        if(msg_p->send_status.error_code && msg_p->recv_id != 0)
        {
            /* we got a send error, but recv is still pending.  Cancel
             * the recv 
             */
            job_bmi_cancel(msg_p->recv_id, mop->params.job_context);
        }
    }

    /* decrement comp_ct until all operations have completed */
    if (--mop->params.comp_ct > 0)
    {
        gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                 "  msgpairarray: %d operations remain\n",
                mop->params.comp_ct);
        return SM_ACTION_DEFERRED;
    }

    assert(mop->params.comp_ct == 0);
    gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                 "  msgpairarray: all operations complete\n");

    js_p->error_code = MSGPAIRS_COMPLETE;
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action msgpairarray_completion_fn(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    PINT_sm_msgarray_op *mop = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL, i = 0;
    int need_retry = 0;
    struct PINT_decoded_msg decoded_resp;
    const char* server_string = NULL;

    /* response structure (decoded) */
    struct PVFS_server_resp *resp_p = NULL;

    js_p->error_code = 0;

    gossip_debug(GOSSIP_MSGPAIR_DEBUG, "(%p) msgpairarray state: "
                 "completion_fn\n", smcb);
    gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing msgpairarray_completion_fn..\n");

    for (i = 0; i < mop->count; i++)
    {
        PINT_sm_msgpair_state *msg_p = &mop->msgarray[i];
        assert(msg_p);

        /*
         * Can take multiple trips through this function as we retry
         * ones that failed.
         */
        if (msg_p->complete)
            continue;

        if (msg_p->op_status != 0)
        {
            char s[1024];
            PVFS_strerror_r(msg_p->op_status, s, sizeof(s));
            server_string = PINT_cached_config_map_addr(
                msg_p->fs_id, msg_p->svr_addr);
            if(!server_string)
            {
                server_string = BMI_addr_rev_lookup(msg_p->svr_addr);
            }

            gossip_err("Warning: msgpair failed to %s, will retry: %s\n", server_string, s);
                       
            ++need_retry;
            continue;
        }

        ret = PINT_serv_decode_resp(msg_p->fs_id,
                                    msg_p->encoded_resp_p,
                                    &decoded_resp,
                                    &msg_p->svr_addr,
                                    msg_p->recv_status.actual_size,
                                    &resp_p);
        if (ret != 0)
        {
            PVFS_perror_gossip("msgpairarray decode error", ret);
            msg_p->op_status = ret;
        }
        else
        {
            /* if we've made it this far, the server response status is
             * meaningful, so we save it.
             */
            msg_p->op_status = resp_p->status;
        }

        /* NOTE: we call the function associated with each message,
         *       not just the one from the first array element.  so
         *       there could in theory be different functions for each
         *       message (to handle different types of messages all in
         *       the same array).
         */
        if (msg_p->comp_fn != NULL)
        {
            gossip_debug(GOSSIP_MIRROR_DEBUG,"\texecuting msg_p->comp_fn..\n");
            /* If we call the completion function, store the result on
             * a per message pair basis.  Also store some non-zero
             * (failure) value in js_p->error_code if we see one.
             */
            msg_p->op_status = msg_p->comp_fn(smcb, resp_p, i);
            if (msg_p->op_status != 0)
            {
                js_p->error_code = msg_p->op_status;
            }

            /* even if we see a failure, continue to process with the
             * completion function. -- RobR
             */
        }
        else if( (msg_p->op_status != 0) && (resp_p == NULL) )
        {
            /* before we reference resp_p we need to catch the case where 
             * PINT_serv_decode_resp didn't give us a struct. We error out
             * the same without referencing resp_p */
            gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                         "%s: error code %d from PINT_serv_decode_resp, "
                         "from server %d\n", __func__, msg_p->op_status, i);
            js_p->error_code = msg_p->op_status;
            break;
        }
        else if (resp_p->status != 0)
        {
            /* no comp_fn specified and status non-zero */
            gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                         "notice: msgpairarray_complete: error %d "
                         "from server %d\n", resp_p->status, i);

            /* save a non-zero status to return if we see one */
            js_p->error_code = resp_p->status;

            /* If we don't have a completion function, there is no point
             * in continuing to process after seeing a failure.
             */
            if (js_p->error_code)
            {
                break;
            }
        }

        /* free all the resources that we used to send and receive. */
        ret = PINT_serv_free_msgpair_resources(
            &msg_p->encoded_req, msg_p->encoded_resp_p, &decoded_resp,
            &msg_p->svr_addr, msg_p->max_resp_sz);
        if (ret)
        {
            PVFS_perror_gossip("Failed to free msgpair resources", ret);
            js_p->error_code = ret;
            return SM_ACTION_COMPLETE;
        }

        memset(&msg_p->encoded_req,0,sizeof(msg_p->encoded_req));
        msg_p->encoded_resp_p = NULL;
        msg_p->max_resp_sz = 0;

        /*
          mark that this msgpair has been completed and should not be
          retried in the case of possible future retries
        */
        msg_p->complete = 1;

        gossip_debug(GOSSIP_MSGPAIR_DEBUG, "%s: sm %p msgpair %d "
                     "marked complete\n", __func__, smcb, i);
    }/*end for*/

    if (need_retry) {
        /*
         * We only retry msgpairs that are not yet complete.  Factor
         * of two since they are pairs.  If over the count, do not
         * retry, just return one of the error codes.
         */
        mop->params.comp_ct = 0;
        js_p->error_code = 0;
        for (i=0; i < mop->count; i++) {

            PINT_sm_msgpair_state *msg_p = &mop->msgarray[i];

            if (msg_p->complete)
                continue;

            if (msg_p->retry_flag == PVFS_MSGPAIR_RETRY
             && PVFS_ERROR_CLASS(-msg_p->op_status) == PVFS_ERROR_BMI
             && msg_p->retry_count < mop->params.retry_limit) {

                ++msg_p->retry_count;
                mop->params.comp_ct += 2;
                gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                  "*** %s: msgpair %d failed, retry %d\n",
                  __func__, i, msg_p->retry_count);
                if(msg_p->op_status == -BMI_ECANCEL)
                {
                    /* if the error code indicates cancel, then skip the
                     * delay.  We have probably already been waiting a while
                     */
                    gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                       "*** %s: msgpair skipping retry delay.\n", __func__);
                    js_p->error_code = MSGPAIRS_RETRY_NODELAY;
                }
                else
                {
                    gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                       "*** %s: msgpair retrying after delay.\n", __func__);
                    js_p->error_code = MSGPAIRS_RETRY;
                }

            } else {
                char s[1024];
                server_string = PINT_cached_config_map_addr(
                    msg_p->fs_id, msg_p->svr_addr);
                if(!server_string)
                {
                    server_string = "[UNKNOWN]";
                }
                PVFS_strerror_r(msg_p->op_status, s, sizeof(s));
                gossip_err_unless_quiet("*** %s: msgpair to server %s failed: %s\n",
                        __func__, server_string, s);
                if(msg_p->retry_flag != PVFS_MSGPAIR_RETRY)
                {
                    gossip_err_unless_quiet("*** No retries requested.\n");
                }
                else if(PVFS_ERROR_CLASS(-msg_p->op_status) !=
                    PVFS_ERROR_BMI)
                {
                    gossip_err_unless_quiet("*** Non-BMI failure.\n");
                }
                else
                {
                    gossip_err_unless_quiet("*** Out of retries.\n");
                }
                if (js_p->error_code == 0)
                    js_p->error_code = msg_p->op_status;
            }

        }
    }
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action msgpairarray_done(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    int task_id, error_code, remaining;
    PINT_sm_pop_frame(smcb, &task_id, &error_code, &remaining);
    return SM_ACTION_COMPLETE;
}

/*********************************************************************
 * helper functions used in conjunction with state machine defined above
 */

int PINT_msgpairarray_init(
    PINT_sm_msgarray_op *op,
    int count)
{
    op->msgarray = (PINT_sm_msgpair_state *)malloc(
        count * sizeof(PINT_sm_msgpair_state));
    if(!op->msgarray)
    {
        return -PVFS_ENOMEM;
    }
    memset(op->msgarray, 0, (count * sizeof(PINT_sm_msgpair_state)));
    op->count = count;

    return 0;
}

/* we pass in a pointer to the array so that we can set it to NULL */
void PINT_msgpairarray_destroy(
    PINT_sm_msgarray_op *op)
{
    if(op->msgarray && (&op->msgpair) != op->msgarray)
    {
        free(op->msgarray);
    }
    op->msgarray = NULL;
    op->count = 0;
}

int PINT_msgarray_status(PINT_sm_msgarray_op *op)
{
    int i;
    for (i = 0; i < op->count; i++) 
    {
        if (op->msgarray[i].op_status != 0)
        {
            return op->msgarray[i].op_status;
        }
    }
    return 0;
}

int PINT_serv_decode_resp(PVFS_fs_id fs_id,
                          void *encoded_resp_p,
                          struct PINT_decoded_msg *decoded_resp_p,
                          PVFS_BMI_addr_t *svr_addr_p,
                          int actual_resp_sz,
                          struct PVFS_server_resp **resp_out_pp)
{
    int ret = -1;
    const char *server_string;

    ret = PINT_decode(encoded_resp_p, PINT_DECODE_RESP,
                      decoded_resp_p, /* holds data on decoded resp */
                      *svr_addr_p, actual_resp_sz);
    if (ret > -1)
    {
        *resp_out_pp = (struct PVFS_server_resp *)decoded_resp_p->buffer;
        if ((*resp_out_pp)->op == PVFS_SERV_PROTO_ERROR)
        {

            gossip_err("Error: server does not seem to understand "
                       "the protocol that this client is using.\n");
            gossip_err("   Please check server logs for more "
                       "information.\n");

            if (fs_id != PVFS_FS_ID_NULL)
            {
                server_string = PINT_cached_config_map_addr(
                    fs_id, *svr_addr_p);
                gossip_err("   Server: %s.\n", server_string);
            }
            else
            {
                gossip_err("   Server: unknown; probably an error "
                           "contacting server listed in pvfs2tab "
                           "file.\n");
            }
            return(-EPROTONOSUPPORT);
        }
    }
    return ret;
}

int PINT_serv_free_msgpair_resources(
    struct PINT_encoded_msg *encoded_req_p,
    void *encoded_resp_p,
    struct PINT_decoded_msg *decoded_resp_p,
    PVFS_BMI_addr_t *svr_addr_p,
    int max_resp_sz)
{
    int ret = -PVFS_EINVAL;

    if (encoded_req_p && decoded_resp_p && svr_addr_p)
    {
        PINT_encode_release(encoded_req_p, PINT_ENCODE_REQ);
        memset(encoded_req_p,0,sizeof(*encoded_req_p));

        PINT_decode_release(decoded_resp_p, PINT_DECODE_RESP);
        memset(decoded_resp_p,0,sizeof(*decoded_resp_p));

        BMI_memfree(*svr_addr_p, encoded_resp_p, max_resp_sz, BMI_RECV);
        encoded_resp_p = NULL;

        ret = 0;
    }
    return ret;
}

/* PINT_serv_msgpair_array_resolve_addrs()
 *
 * fills in BMI address of server for each entry in the msgpair array,
 * based on the handle and fsid
 *
 * returns 0 on success, -PVFS_error on failure
 */
int PINT_serv_msgpairarray_resolve_addrs(
    PINT_sm_msgarray_op *mop)
{
    int i = 0;
    int ret = -PVFS_EINVAL;

    if ((mop->count > 0) && mop->msgarray)
    {
        for(i = 0; i < mop->count; i++)
        {
            PINT_sm_msgpair_state *msg_p = &mop->msgarray[i];
            assert(msg_p);

            ret = PINT_cached_config_map_to_server(
                &msg_p->svr_addr, msg_p->handle, msg_p->fs_id);

            if (ret != 0)
            {
                gossip_err("Failed to map server address to handle\n");
                break;
            }

            gossip_debug(GOSSIP_MSGPAIR_DEBUG,
                         " mapped handle %llu to server %lld\n",
                         llu(msg_p->handle), lld(msg_p->svr_addr));
        }
    }
    return ret;
}

/*
 * Local variables:
 *  mode: c
 *  c-indent-level: 4
 *  c-basic-offset: 4
 * End:
 *
 * vim: ft=c ts=8 sts=4 sw=4 expandtab
 */