path: root/xlators/nfs/lib/src/rpcsvc.c

/*
  Copyright (c) 2010-2011-2011-2011 Gluster, Inc. <http://www.gluster.com>
  This file is part of GlusterFS.

  GlusterFS is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published
  by the Free Software Foundation; either version 3 of the License,
  or (at your option) any later version.

  GlusterFS is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program.  If not, see
  <http://www.gnu.org/licenses/>.
*/

#ifndef _CONFIG_H
#define _CONFIG_H
#include "config.h"
#endif

#include "rpcsvc.h"
#include "rpc-socket.h"
#include "dict.h"
#include "logging.h"
#include "byte-order.h"
#include "common-utils.h"
#include "compat-errno.h"
#include "list.h"
#include "xdr-rpc.h"
#include "iobuf.h"
#include "globals.h"

#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
#include <rpc/rpc.h>
#include <rpc/pmap_clnt.h>
#include <arpa/inet.h>
#include <rpc/xdr.h>
#include <fnmatch.h>
#include <stdarg.h>
#include <stdio.h>


#define nfs_rpcsvc_alloc_request(con, request)                          \
        do {                                                            \
                request = (rpcsvc_request_t *) mem_get ((con)->rxpool); \
                memset (request, 0, sizeof (rpcsvc_request_t));         \
        } while (0)                                                     \

/* The generic event handler for every stage */
void *
nfs_rpcsvc_stage_proc (void *arg)
{
        rpcsvc_stage_t          *stg = (rpcsvc_stage_t *)arg;

        if (!stg)
                return NULL;

        event_dispatch (stg->eventpool);
        return NULL;
}


rpcsvc_stage_t *
nfs_rpcsvc_stage_init (rpcsvc_t *svc)
{
        rpcsvc_stage_t          *stg = NULL;
        int                     ret = -1;
        size_t                  stacksize = RPCSVC_THREAD_STACK_SIZE;
        pthread_attr_t          stgattr;
        unsigned int            eventpoolsize = 0;

        if (!svc)
                return NULL;

        stg = GF_CALLOC (1, sizeof(*stg), gf_common_mt_rpcsvc_stage_t);
        if (!stg)
                return NULL;

        eventpoolsize = svc->memfactor * RPCSVC_EVENTPOOL_SIZE_MULT;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "event pool size: %d", eventpoolsize);
        stg->eventpool = event_pool_new (eventpoolsize);
        if (!stg->eventpool)
                goto free_stg;

        pthread_attr_init (&stgattr);
        ret = pthread_attr_setstacksize (&stgattr, stacksize);
        if (ret == EINVAL)
                gf_log (GF_RPCSVC, GF_LOG_WARNING,
                                "Using default thread stack size");

        ret = pthread_create (&stg->tid, &stgattr, nfs_rpcsvc_stage_proc,
                              (void *)stg);
        if (ret != 0) {
                ret = -1;
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Stage creation failed");
                goto free_stg;
        }

        stg->svc = svc;
        ret = 0;
free_stg:
        if (ret == -1) {
                GF_FREE (stg);
                stg = NULL;
        }

        return stg;
}


int
nfs_rpcsvc_init_options (rpcsvc_t *svc, dict_t *options)
{
        char            *optstr = NULL;
        int             ret = -1;

        if ((!svc) || (!options))
                return -1;

        svc->memfactor = RPCSVC_DEFAULT_MEMFACTOR;

        svc->register_portmap = _gf_true;
        if (dict_get (options, "rpc.register-with-portmap")) {
                ret = dict_get_str (options, "rpc.register-with-portmap",
                                    &optstr);
                if (ret < 0) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to parse "
                                "dict");
                        goto out;
                }

                ret = gf_string2boolean (optstr, &svc->register_portmap);
                if (ret < 0) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to parse bool "
                                "string");
                        goto out;
                }
        }

        if (!svc->register_portmap)
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Portmap registration "
                        "disabled");

        ret = 0;
out:
        return ret;
}


/* The global RPC service initializer.
 * Starts up the stages and then waits for RPC program registrations
 * to come in.
 */
rpcsvc_t *
nfs_rpcsvc_init (glusterfs_ctx_t *ctx, dict_t *options)
{
        rpcsvc_t        *svc = NULL;
        int             ret = -1;
        int             poolsize = 0;

        if ((!ctx) || (!options))
                return NULL;

        svc = GF_CALLOC (1, sizeof (*svc), gf_common_mt_rpcsvc_t);
        if (!svc)
                return NULL;

        pthread_mutex_init (&svc->rpclock, NULL);
        INIT_LIST_HEAD (&svc->stages);
        INIT_LIST_HEAD (&svc->authschemes);
        INIT_LIST_HEAD (&svc->allprograms);

        ret = nfs_rpcsvc_init_options (svc, options);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to init options");
                goto free_svc;
        }

        ret = nfs_rpcsvc_auth_init (svc, options);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to init "
                        "authentication");
                goto free_svc;
        }

        ret = -1;
        poolsize = RPCSVC_POOLCOUNT_MULT * RPCSVC_DEFAULT_MEMFACTOR;
        svc->connpool = mem_pool_new (rpcsvc_conn_t, poolsize);
        if (!svc->connpool) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to allocate connpool");
                goto free_svc;
        }

        svc->defaultstage = nfs_rpcsvc_stage_init (svc);
        if (!svc->defaultstage) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR,"RPC service init failed.");
                goto free_connpool;
        }
        svc->options = options;
        svc->ctx = ctx;
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC service inited.");

        ret = 0;

free_connpool:
        if (ret == -1)
                mem_pool_destroy (svc->connpool);

free_svc:
        if (ret == -1) {
                GF_FREE (svc);
                svc = NULL;
        }

        return svc;
}


/* Once multi-threaded support is complete, we'll be able to round-robin
 * the various incoming connections over the many available stages. This
 * function selects one from among all the stages.
 */
rpcsvc_stage_t *
nfs_rpcsvc_select_stage (rpcsvc_t *rpcservice)
{
        if (!rpcservice)
                return NULL;

        return rpcservice->defaultstage;
}


int
nfs_rpcsvc_conn_peer_check_search (dict_t *options, char *pattern, char *clstr)
{
        int                     ret = -1;
        char                    *addrtok = NULL;
        char                    *addrstr = NULL;
        char                    *svptr = NULL;

        if ((!options) || (!clstr))
                return -1;

        if (!dict_get (options, pattern))
                return -1;

        ret = dict_get_str (options, pattern, &addrstr);
        if (ret < 0) {
                ret = -1;
                goto err;
        }

        if (!addrstr) {
                ret = -1;
                goto err;
        }

        addrtok = strtok_r (addrstr, ",", &svptr);
        while (addrtok) {

                /* CASEFOLD not present on Solaris */
#ifdef FNM_CASEFOLD
                ret = fnmatch (addrtok, clstr, FNM_CASEFOLD);
#else
                ret = fnmatch (addrtok, clstr, 0);
#endif
                if (ret == 0)
                        goto err;

                addrtok = strtok_r (NULL, ",", &svptr);
        }

        ret = -1;
err:

        return ret;
}


int
nfs_rpcsvc_conn_peer_check_allow (dict_t *options, char *volname, char *clstr)
{
        int     ret = RPCSVC_AUTH_DONTCARE;
        char    *srchstr = NULL;
        char    globalrule[] = "rpc-auth.addr.allow";

        if ((!options) || (!clstr))
                return ret;

        /* If volname is NULL, then we're searching for the general rule to
         * determine the current address in clstr is allowed or not for all
         * subvolumes.
         */
        if (volname) {
                ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.allow", volname);
                if (ret == -1) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
                        ret = RPCSVC_AUTH_DONTCARE;
                        goto out;
                }
        } else
                srchstr = globalrule;

        ret = nfs_rpcsvc_conn_peer_check_search (options, srchstr, clstr);
        if (volname)
                GF_FREE (srchstr);

        if (ret == 0)
                ret = RPCSVC_AUTH_ACCEPT;
        else
                ret = RPCSVC_AUTH_DONTCARE;
out:
        return ret;
}

int
nfs_rpcsvc_conn_peer_check_reject (dict_t *options, char *volname, char *clstr)
{
        int     ret = RPCSVC_AUTH_DONTCARE;
        char    *srchstr = NULL;
        char    generalrule[] = "rpc-auth.addr.reject";

        if ((!options) || (!clstr))
                return ret;

        if (volname) {
                ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.reject",
                                   volname);
                if (ret == -1) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
                        ret = RPCSVC_AUTH_REJECT;
                        goto out;
                }
        } else
                srchstr = generalrule;

        ret = nfs_rpcsvc_conn_peer_check_search (options, srchstr, clstr);
        if (volname)
                GF_FREE (srchstr);

        if (ret == 0)
                ret = RPCSVC_AUTH_REJECT;
        else
                ret = RPCSVC_AUTH_DONTCARE;
out:
        return ret;
}


/* This function tests the results of the allow rule and the reject rule to
 * combine them into a single result that can be used to determine if the
 * connection should be allowed to proceed.
 * Heres the test matrix we need to follow in this function.
 *
 * A -  Allow, the result of the allow test. Never returns R.
 * R - Reject, result of the reject test. Never returns A.
 * Both can return D or dont care if no rule was given.
 *
 * | @allow | @reject | Result |
 * |    A   |   R     | R      |
 * |    D   |   D     | D      |
 * |    A   |   D     | A      |
 * |    D   |   R     | R      |
 */
int
nfs_rpcsvc_combine_allow_reject_volume_check (int allow, int reject)
{
        int     final = RPCSVC_AUTH_REJECT;

        /* If allowed rule allows but reject rule rejects, we stay cautious
         * and reject. */
        if ((allow == RPCSVC_AUTH_ACCEPT) && (reject == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;
        /* if both are dont care, that is user did not specify for either allow
         * or reject, we leave it up to the general rule to apply, in the hope
         * that there is one.
         */
        else if ((allow == RPCSVC_AUTH_DONTCARE) &&
                 (reject == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_DONTCARE;
        /* If one is dont care, the other one applies. */
        else if ((allow == RPCSVC_AUTH_ACCEPT) &&
                 (reject == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((allow == RPCSVC_AUTH_DONTCARE) &&
                 (reject == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;

        return final;
}


/* Combines the result of the general rule test against, the specific rule
 * to determine final permission for the client's address.
 *
 * | @gen   | @spec   | Result |
 * |    A   |   A     | A      |
 * |    A   |   R     | R      |
 * |    A   |   D     | A      |
 * |    D   |   A     | A      |
 * |    D   |   R     | R      |
 * |    D   |   D     | D      |
 * |    R   |   A     | A      |
 * |    R   |   D     | R      |
 * |    R   |   R     | R      |
 */
int
nfs_rpcsvc_combine_gen_spec_addr_checks (int gen, int spec)
{
        int     final = RPCSVC_AUTH_REJECT;

        if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec== RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_DONTCARE;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;

        return final;
}



/* Combines the result of the general rule test against, the specific rule
 * to determine final test for the connection coming in for a given volume.
 *
 * | @gen   | @spec   | Result |
 * |    A   |   A     | A      |
 * |    A   |   R     | R      |
 * |    A   |   D     | A      |
 * |    D   |   A     | A      |
 * |    D   |   R     | R      |
 * |    D   |   D     | R      |, special case, we intentionally disallow this.
 * |    R   |   A     | A      |
 * |    R   |   D     | R      |
 * |    R   |   R     | R      |
 */
int
nfs_rpcsvc_combine_gen_spec_volume_checks (int gen, int spec)
{
        int     final = RPCSVC_AUTH_REJECT;

        if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;
        /* On no rule, we reject. */
        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec== RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_ACCEPT))
                final = RPCSVC_AUTH_ACCEPT;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_DONTCARE))
                final = RPCSVC_AUTH_REJECT;
        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_REJECT))
                final = RPCSVC_AUTH_REJECT;

        return final;
}


int
nfs_rpcsvc_conn_peer_check_name (dict_t *options, char *volname,
                                 rpcsvc_conn_t *conn)
{
        int     ret = RPCSVC_AUTH_REJECT;
        int     aret = RPCSVC_AUTH_REJECT;
        int     rjret = RPCSVC_AUTH_REJECT;
        char    clstr[RPCSVC_PEER_STRLEN];

        if (!conn)
                return ret;

        ret = nfs_rpcsvc_conn_peername (conn, clstr, RPCSVC_PEER_STRLEN);
        if (ret != 0) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get remote addr: "
                        "%s", gai_strerror (ret));
                ret = RPCSVC_AUTH_REJECT;
                goto err;
        }

        aret = nfs_rpcsvc_conn_peer_check_allow (options, volname, clstr);
        rjret = nfs_rpcsvc_conn_peer_check_reject (options, volname, clstr);

        ret = nfs_rpcsvc_combine_allow_reject_volume_check (aret, rjret);

err:
        return ret;
}


int
nfs_rpcsvc_conn_peer_check_addr (dict_t *options, char *volname,
                                 rpcsvc_conn_t *conn)
{
        int     ret = RPCSVC_AUTH_REJECT;
        int     aret = RPCSVC_AUTH_DONTCARE;
        int     rjret = RPCSVC_AUTH_REJECT;
        char    clstr[RPCSVC_PEER_STRLEN];

        if (!conn)
                return ret;

        ret = nfs_rpcsvc_conn_peeraddr (conn, clstr, RPCSVC_PEER_STRLEN, NULL,
                                        0);
        if (ret != 0) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get remote addr: "
                        "%s", gai_strerror (ret));
                ret = RPCSVC_AUTH_REJECT;
                goto err;
        }

        aret = nfs_rpcsvc_conn_peer_check_allow (options, volname, clstr);
        rjret = nfs_rpcsvc_conn_peer_check_reject (options, volname, clstr);

        ret = nfs_rpcsvc_combine_allow_reject_volume_check (aret, rjret);
err:
        return ret;
}


int
nfs_rpcsvc_conn_check_volume_specific (dict_t *options, char *volname,
                                       rpcsvc_conn_t *conn)
{
        int             namechk = RPCSVC_AUTH_REJECT;
        int             addrchk = RPCSVC_AUTH_REJECT;
        gf_boolean_t    namelookup = _gf_true;
        char            *namestr = NULL;
        int             ret = 0;

        if ((!options) || (!volname) || (!conn))
                return RPCSVC_AUTH_REJECT;

        /* Enabled by default */
        if ((dict_get (options, "rpc-auth.addr.namelookup"))) {
                ret = dict_get_str (options, "rpc-auth.addr.namelookup"
                                    , &namestr);
                if (ret == 0)
                        ret = gf_string2boolean (namestr, &namelookup);
        }

        /* We need two separate checks because the rules with addresses in them
         * can be network addresses which can be general and names can be
         * specific which will over-ride the network address rules.
         */
        if (namelookup)
                namechk = nfs_rpcsvc_conn_peer_check_name (options, volname,
                                                           conn);
        addrchk = nfs_rpcsvc_conn_peer_check_addr (options, volname, conn);

        if (namelookup)
                ret = nfs_rpcsvc_combine_gen_spec_addr_checks (addrchk,
                                                               namechk);
        else
                ret = addrchk;

        return ret;
}


int
nfs_rpcsvc_conn_check_volume_general (dict_t *options, rpcsvc_conn_t *conn)
{
        int             addrchk = RPCSVC_AUTH_REJECT;
        int             namechk = RPCSVC_AUTH_REJECT;
        gf_boolean_t    namelookup = _gf_true;
        char            *namestr = NULL;
        int             ret = 0;

        if ((!options) || (!conn))
                return RPCSVC_AUTH_REJECT;

        /* Enabled by default */
        if ((dict_get (options, "rpc-auth.addr.namelookup"))) {
                ret = dict_get_str (options, "rpc-auth.addr.namelookup"
                                    , &namestr);
                if (ret == 0)
                        ret = gf_string2boolean (namestr, &namelookup);
        }

        /* We need two separate checks because the rules with addresses in them
         * can be network addresses which can be general and names can be
         * specific which will over-ride the network address rules.
         */
        if (namelookup)
                namechk = nfs_rpcsvc_conn_peer_check_name (options, NULL, conn);
        addrchk = nfs_rpcsvc_conn_peer_check_addr (options, NULL, conn);

        if (namelookup)
                ret = nfs_rpcsvc_combine_gen_spec_addr_checks (addrchk,
                                                               namechk);
        else
                ret = addrchk;

        return ret;
}

int
nfs_rpcsvc_conn_peer_check (dict_t *options, char *volname, rpcsvc_conn_t *conn)
{
        int     general_chk = RPCSVC_AUTH_REJECT;
        int     specific_chk = RPCSVC_AUTH_REJECT;

        if ((!options) || (!volname) || (!conn))
                return RPCSVC_AUTH_REJECT;

        general_chk = nfs_rpcsvc_conn_check_volume_general (options, conn);
        specific_chk = nfs_rpcsvc_conn_check_volume_specific (options, volname,
                                                              conn);

        return nfs_rpcsvc_combine_gen_spec_volume_checks (general_chk,
                                                          specific_chk);
}


char *
nfs_rpcsvc_volume_allowed (dict_t *options, char *volname)
{
        char    globalrule[] = "rpc-auth.addr.allow";
        char    *srchstr = NULL;
        char    *addrstr = NULL;
        int     ret = -1;

        if ((!options) || (!volname))
                return NULL;

        ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.allow", volname);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
                goto out;
        }

        if (!dict_get (options, srchstr)) {
                GF_FREE (srchstr);
                srchstr = globalrule;
                ret = dict_get_str (options, srchstr, &addrstr);
                srchstr = NULL;
        } else
                ret = dict_get_str (options, srchstr, &addrstr);

out:
        if (srchstr != NULL)
                GF_FREE (srchstr);
        return addrstr;
}


/* Initialize the core of a connection */
rpcsvc_conn_t *
nfs_rpcsvc_conn_init (rpcsvc_t *svc, int sockfd)
{
        rpcsvc_conn_t  *conn = NULL;
        int             ret = -1;
        unsigned int    poolcount = 0;

        conn = mem_get (svc->connpool);
        if (!conn) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "memory allocation failed");
                return NULL;
        }

        conn->sockfd = sockfd;
        INIT_LIST_HEAD (&conn->txbufs);
        poolcount = RPCSVC_POOLCOUNT_MULT * svc->memfactor;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "tx pool: %d", poolcount);
        conn->txpool = mem_pool_new (rpcsvc_txbuf_t, poolcount);
        if (!conn->txpool) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "mem pool allocation failed");
                goto free_conn;
        }

        gf_log (GF_RPCSVC, GF_LOG_TRACE, "rx pool: %d", poolcount);
        conn->rxpool = mem_pool_new (rpcsvc_request_t, poolcount);
        if (!conn->rxpool) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "mem pool allocation failed");
                goto free_txp;
        }

        /* Cannot consider a connection connected unless the user of this
         * connection decides it is ready to use. It is possible that we have
         * to free this connection soon after. That free will not happpen
         * unless the state is disconnected.
         */
        conn->connstate = RPCSVC_CONNSTATE_DISCONNECTED;
        pthread_mutex_init (&conn->connlock, NULL);
        conn->connref = 0;
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "New connection inited: sockfd: %d",
                sockfd);

        ret = 0;
free_txp:
        if (ret == -1)
                mem_pool_destroy (conn->txpool);

free_conn:
        if (ret == -1) {
                GF_FREE (conn);
                conn = NULL;
        }

        return conn;
}


void
nfs_rpcsvc_conn_destroy (rpcsvc_conn_t *conn)
{
        mem_pool_destroy (conn->txpool);
        mem_pool_destroy (conn->rxpool);

        /* Need to destory record state, txlists etc. */
        mem_put (((rpcsvc_t *)conn->stage->svc)->connpool, conn);
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Connection destroyed");
}


int
__nfs_rpcsvc_conn_unref (rpcsvc_conn_t *conn)
{
        --conn->connref;
        return conn->connref;
}


void
__nfs_rpcsvc_conn_deinit (rpcsvc_conn_t *conn)
{
        if (!conn)
                return;

        if ((conn->stage) && (conn->stage->eventpool)) {
                event_unregister (conn->stage->eventpool, conn->sockfd,
                                  conn->eventidx);
        }

        if (nfs_rpcsvc_conn_check_active (conn)) {
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Connection de-activated:"
                        " sockfd: %d", conn->sockfd);
                conn->connstate = RPCSVC_CONNSTATE_DISCONNECTED;
        }

        if (conn->sockfd != -1) {
                close (conn->sockfd);
                conn->sockfd = -1;
        }
}


void
nfs_rpcsvc_conn_deinit (rpcsvc_conn_t *conn)
{
        int ref = 0;

        if (!conn)
                return;

        pthread_mutex_lock (&conn->connlock);
        {
                __nfs_rpcsvc_conn_deinit (conn);
                ref = __nfs_rpcsvc_conn_unref (conn);
        }
        pthread_mutex_unlock (&conn->connlock);

        if (ref == 0)
                nfs_rpcsvc_conn_destroy (conn);

        return;
}


void
nfs_rpcsvc_conn_unref (rpcsvc_conn_t *conn)
{
        int ref = 0;
        if (!conn)
                return;

        pthread_mutex_lock (&conn->connlock);
        {
                ref = __nfs_rpcsvc_conn_unref (conn);
        }
        pthread_mutex_unlock (&conn->connlock);

        if (ref == 0)
                nfs_rpcsvc_conn_destroy (conn);
}


int
nfs_rpcsvc_conn_active (rpcsvc_conn_t *conn)
{
        int     status = 0;

        if (!conn)
                return 0;

        pthread_mutex_lock (&conn->connlock);
        {
                status = nfs_rpcsvc_conn_check_active (conn);
        }
        pthread_mutex_unlock (&conn->connlock);

        return status;
}



void
nfs_rpcsvc_conn_ref (rpcsvc_conn_t *conn)
{
        if (!conn)
                return;

        pthread_mutex_lock (&conn->connlock);
        {
                ++conn->connref;
        }
        pthread_mutex_unlock (&conn->connlock);

        return;
}


void
nfs_rpcsvc_conn_state_init (rpcsvc_conn_t *conn)
{
        if (!conn)
                return;

        ++conn->connref;
        conn->connstate = RPCSVC_CONNSTATE_CONNECTED;
}

/* Builds a rpcsvc_conn_t with the aim of listening on it.
 */
rpcsvc_conn_t *
nfs_rpcsvc_conn_listen_init (rpcsvc_t *svc, rpcsvc_program_t *newprog)
{
        rpcsvc_conn_t  *conn = NULL;
        int             sock = -1;

        if (!newprog)
                return NULL;

        sock = nfs_rpcsvc_socket_listen (newprog->progaddrfamily,
                                         newprog->proghost, newprog->progport);
        if (sock == -1)
                goto err;

        conn = nfs_rpcsvc_conn_init (svc, sock);
        if (!conn)
                goto sock_close_err;

        nfs_rpcsvc_conn_state_init (conn);
sock_close_err:
        if (!conn)
                close (sock);

err:
        return conn;
}

void
nfs_rpcsvc_record_init (rpcsvc_record_state_t *rs, struct iobuf_pool *pool)
{
        if (!rs)
                return;

        rs->state = RPCSVC_READ_FRAGHDR;
        rs->vecstate = 0;
        rs->remainingfraghdr = RPCSVC_FRAGHDR_SIZE;
        rs->remainingfrag = 0;
        rs->fragsize = 0;
        rs->recordsize = 0;
        rs->islastfrag = 0;

        /* If the rs preserves a ref to the iob used by the previous request,
         * we must unref it here to prevent memory leak.
         * If program actor wanted to keep that memory around, it should've
         * refd it on entry into the actor.
         */
        if (rs->activeiob)
                iobuf_unref (rs->activeiob);

        if (rs->vectoriob) {
                iobuf_unref (rs->vectoriob);
                rs->vectoriob = NULL;
        }

        rs->activeiob = iobuf_get (pool);
        rs->fragcurrent = iobuf_ptr (rs->activeiob);

        memset (rs->fragheader, 0, RPCSVC_FRAGHDR_SIZE);
        rs->hdrcurrent = &rs->fragheader[0];

}


int
nfs_rpcsvc_conn_privport_check (rpcsvc_t *svc, char *volname,
                                rpcsvc_conn_t *conn)
{
        struct sockaddr_in      sa;
        int                     ret = RPCSVC_AUTH_REJECT;
        socklen_t               sasize = sizeof (sa);
        char                    *srchstr = NULL;
        char                    *valstr = NULL;
        int                     globalinsecure = RPCSVC_AUTH_REJECT;
        int                     exportinsecure = RPCSVC_AUTH_DONTCARE;
        uint16_t                port = 0;
        gf_boolean_t            insecure = _gf_false;

        if ((!svc) || (!volname) || (!conn))
                return ret;

        ret = nfs_rpcsvc_conn_peeraddr (conn, NULL, 0, (struct sockaddr *)&sa,
                                        sasize);
        if (ret != 0) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get peer addr: %s",
                        gai_strerror (ret));
                ret = RPCSVC_AUTH_REJECT;
                goto err;
        }

        port = ntohs (sa.sin_port);
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Client port: %d", (int)port);
        /* If the port is already a privileged one, dont bother with checking
         * options.
         */
        if (port <= 1024) {
                ret = RPCSVC_AUTH_ACCEPT;
                goto err;
        }

        /* Disabled by default */
        if ((dict_get (svc->options, "rpc-auth.ports.insecure"))) {
                ret = dict_get_str (svc->options, "rpc-auth.ports.insecure"
                                    , &srchstr);
                if (ret == 0) {
                        ret = gf_string2boolean (srchstr, &insecure);
                        if (ret == 0) {
                                if (insecure == _gf_true)
                                        globalinsecure = RPCSVC_AUTH_ACCEPT;
                        } else
                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
                                        " read rpc-auth.ports.insecure value");
                } else
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
                                " read rpc-auth.ports.insecure value");
        }

        /* Disabled by default */
        ret = gf_asprintf (&srchstr, "rpc-auth.ports.%s.insecure", volname);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
                ret = RPCSVC_AUTH_REJECT;
                goto err;
        }

        if (dict_get (svc->options, srchstr)) {
                ret = dict_get_str (svc->options, srchstr, &valstr);
                if (ret == 0) {
                        ret = gf_string2boolean (valstr, &insecure);
                        if (ret == 0) {
                                if (insecure == _gf_true)
                                       exportinsecure = RPCSVC_AUTH_ACCEPT;
                                else
                                        exportinsecure = RPCSVC_AUTH_REJECT;
                        } else
                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
                                        " read rpc-auth.ports.insecure value");
                } else
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
                                " read rpc-auth.ports.insecure value");
        }

        ret = nfs_rpcsvc_combine_gen_spec_volume_checks (globalinsecure,
                                                         exportinsecure);
        if (ret == RPCSVC_AUTH_ACCEPT)
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Unprivileged port allowed");
        else
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Unprivileged port not"
                        " allowed");

err:
        return ret;
}

/* Inits a rpcsvc_conn_t after accepting the connection.
 */
rpcsvc_conn_t *
nfs_rpcsvc_conn_accept_init (rpcsvc_t *svc, int listenfd)
{
        rpcsvc_conn_t   *newconn = NULL;
        int             sock = -1;
        int             ret = -1;

        sock = nfs_rpcsvc_socket_accept (listenfd);
        if (sock == -1)
                goto err;

        newconn = nfs_rpcsvc_conn_init (svc, sock);
        if (!newconn) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to init conn object");
                ret = -1;
                goto err;
        }

        nfs_rpcsvc_record_init (&newconn->rstate, svc->ctx->iobuf_pool);
        nfs_rpcsvc_conn_state_init (newconn);
        ret = 0;

err:
        if (ret == -1)
                close (sock);

        return newconn;
}


/* Once the connection has been created, we need to associate it with
 * a stage so that the selected stage will handle the event on this connection.
 * This function also allows the caller to decide which handler should
 * be executed in the context of the stage, and also which specific events
 * should be handed to the handler when running in this particular stage.
 */
int
nfs_rpcsvc_stage_conn_associate (rpcsvc_stage_t *stg, rpcsvc_conn_t *conn,
                                 event_handler_t handler, void *data)
{
        int     ret = -1;

        if ((!stg) || (!conn))
                return -1;

        conn->stage = stg;
        conn->eventidx = event_register (stg->eventpool, conn->sockfd, handler,
                                         data, 1, 0);
        if (conn->eventidx == -1)
                goto err;

        ret = 0;
err:
        return ret;
}


/* Depending on the state we're in, return the size of the next read request. */
size_t
nfs_rpcsvc_record_read_size (rpcsvc_record_state_t *rs)
{
        size_t  toread = -1;

        if (!rs)
                return -1;

        if (nfs_rpcsvc_record_readfraghdr (rs))
                toread = rs->remainingfraghdr;
        else if (nfs_rpcsvc_record_readfrag (rs))
                toread = rs->remainingfrag;
        else
                toread = RPCSVC_CONN_READ;

        return toread;
}


uint32_t
nfs_rpcsvc_record_extract_fraghdr (char *fraghdr)
{
        uint32_t        hdr = 0;
        if (!fraghdr)
                return 0;

        memcpy ((void *)&hdr, fraghdr, sizeof (hdr));

        hdr = ntohl (hdr);
        return hdr;
}


ssize_t
nfs_rpcsvc_record_read_complete_fraghdr (rpcsvc_record_state_t *rs,
                                         ssize_t dataread)
{
        uint32_t        remhdr = 0;
        char            *fraghdrstart = NULL;
        uint32_t        fraghdr = 0;

        fraghdrstart = &rs->fragheader[0];
        remhdr = rs->remainingfraghdr;
        fraghdr = nfs_rpcsvc_record_extract_fraghdr (fraghdrstart);
        rs->fragsize = RPCSVC_FRAGSIZE (fraghdr);
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Received fragment size: %d",
                rs->fragsize);
        if (nfs_rpcsvc_record_vectored (rs)) {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC header,"
                        " remaining: %d", RPCSVC_BARERPC_MSGSZ);
                rs->remainingfrag = RPCSVC_BARERPC_MSGSZ;
        } else {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Regular RPC header,"
                        " remaining: %d", rs->fragsize);
                rs->remainingfrag = rs->fragsize;
        }

        rs->state = RPCSVC_READ_FRAG;
        dataread -= remhdr;
        rs->remainingfraghdr -= remhdr;
        rs->islastfrag = RPCSVC_LASTFRAG (fraghdr);

        return dataread;
}


ssize_t
nfs_rpcsvc_record_read_partial_fraghdr (rpcsvc_record_state_t *rs,
                                        ssize_t dataread)
{

        /* In case we got less than even the remaining header size,
         * we need to consume it all and wait for remaining frag hdr
         * bytes to come in.
         */
         rs->remainingfraghdr -= dataread;
         nfs_rpcsvc_record_update_currenthdr (rs, dataread);
         dataread = 0;
         gf_log (GF_RPCSVC, GF_LOG_TRACE, "Fragment header remaining: %d",
                 rs->remainingfraghdr);

         return dataread;
}


ssize_t
nfs_rpcsvc_record_update_fraghdr (rpcsvc_record_state_t *rs, ssize_t dataread)
{
        if ((!rs) || (dataread <= 0))
                return -1;

        /* Why are we even here, we're not supposed to be in the fragment
         * header processing state.
         */
        if (!nfs_rpcsvc_record_readfraghdr(rs)) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "record state inconsistent"
                        ": request to update frag header when state is not"
                        "RPCSVC_READ_FRAGHDR");
                return -1;
        }

        /* Again, if header has been read then the state member above should've
         * been different, this is crazy. We should not be here.
         */
        if (rs->remainingfraghdr == 0) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "record state inconsistent"
                        ": request to update frag header when frag header"
                        "remaining is 0.");
                return -1;
        }

        /* We've definitely got the full header now and may be even more. */
        if (dataread >= rs->remainingfraghdr)
                dataread = nfs_rpcsvc_record_read_complete_fraghdr (rs,
                                                                    dataread);
        else
                dataread = nfs_rpcsvc_record_read_partial_fraghdr (rs,
                                                                   dataread);

        return dataread;
}

ssize_t
nfs_rpcsvc_record_read_complete_frag (rpcsvc_record_state_t *rs,
                                      ssize_t dataread)
{
        uint32_t        remfrag;

        /* Since the frag is now complete, change the state to the next
         * one, i.e. to read the header of the next fragment.
         */
        remfrag = rs->remainingfrag;
        rs->state = RPCSVC_READ_FRAGHDR;
        dataread -= remfrag;

        /* This will be 0 now. */
        rs->remainingfrag -= remfrag;

        /* Now that the fragment is complete, we must update the
         * record size. Recall that fragsize was got from the frag
         * header.
         */
        rs->recordsize += rs->fragsize;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Fragment remaining: %d",
                rs->remainingfrag);

        return dataread;
}


ssize_t
nfs_rpcsvc_record_read_partial_frag (rpcsvc_record_state_t *rs,
                                     ssize_t dataread)
{
        /* Just take whatever has come through the current network buffer. */
        rs->remainingfrag -= dataread;

        nfs_rpcsvc_record_update_currentfrag (rs, dataread);
        /* Since we know we're consuming the whole buffer from dataread
         * simply setting to 0 zero is fine.
         */
        dataread = 0;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Fragment remaining: %d",
                rs->remainingfrag);
        return dataread;
}


ssize_t
nfs_rpcsvc_record_update_frag (rpcsvc_record_state_t *rs, ssize_t dataread)
{
        if ((!rs) || (dataread <= 0))
                return -1;

        if (!nfs_rpcsvc_record_readfrag (rs)) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "record state inconsistent"
                        ": request to update fragment when record state is not"
                        "RPCSVC_READ_FRAG.");
                return -1;
        }

        if (rs->remainingfrag == 0) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "record state inconsistent"
                        ": request to update fragment when there is no fragment"
                        " data remaining to be read.");
                return -1;
        }

        /* We've read in more data than the current fragment requires. */
        if (dataread >= rs->remainingfrag)
                dataread = nfs_rpcsvc_record_read_complete_frag (rs, dataread);
        else
                dataread = nfs_rpcsvc_record_read_partial_frag (rs, dataread);

        return dataread;
}



/* This needs to change to returning errors, since
 * we need to return RPC specific error messages when some
 * of the pointers below are NULL.
 */
int
__nfs_rpcsvc_program_actor (rpcsvc_request_t *req, rpcsvc_program_t **prg)
{
        rpcsvc_program_t        *program = NULL;
        int                     ret = PROG_UNAVAIL;
        rpcsvc_actor_t          *actor = NULL;
        struct list_head        *prglist = NULL;

        if (!req)
                return ret;

        prglist = &((nfs_rpcsvc_request_service (req))->allprograms);
        if (list_empty (prglist))
                goto err;

        list_for_each_entry (program, prglist, proglist) {
                ret = PROG_UNAVAIL;
                if (req->prognum != program->prognum)
                        continue;

                if (!program->actors) {
                        ret = SYSTEM_ERR;
                        goto err;
                }

                ret = PROG_MISMATCH;
                if (req->progver != program->progver)
                        continue;

                ret = PROC_UNAVAIL;
                if ((req->procnum < 0) || (req->procnum >= program->numactors))
                        goto err;

                actor = &program->actors[req->procnum];
                if (!actor->actor) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "RPC Program procedure"
                                " not defined");
                        actor = NULL;
                        goto err;
                } else {
                        ret = SUCCESS;
                        break;
                }
        }

        *prg = program;
err:
        switch (ret) {

        case PROG_UNAVAIL:
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC program not available");
                break;

        case PROG_MISMATCH:
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC program version "
                        "not available");
                break;

        case PROC_UNAVAIL:
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC Program procedure"
                        " not available");
                break;

        case SUCCESS:
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "RPC Program found");
                break;

        default:
                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "System error");
                break;
        }

        /* If the error is not RPC_MISMATCH, we consider the call as accepted
         * since we are not handling authentication failures for now.
         */
        req->rpc_stat = MSG_ACCEPTED;
        req->rpc_err = ret;

        return ret;
}

/* This needs to change to returning errors, since
 * we need to return RPC specific error messages when some
 * of the pointers below are NULL.
 */
rpcsvc_actor_t *
nfs_rpcsvc_program_actor (rpcsvc_request_t *req)
{
        int                     err = SYSTEM_ERR;
        rpcsvc_actor_t          *actor = NULL;

        if (!req)
                goto err;

        actor = &req->program->actors[req->procnum];
        err = SUCCESS;
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Actor found: %s - %s",
                req->program->progname, actor->procname);
err:
        if (req)
                req->rpc_err = err;

        return actor;
}


rpcsvc_txbuf_t *
nfs_rpcsvc_init_txbuf (rpcsvc_conn_t *conn, struct iovec msg, struct iobuf *iob,
                       struct iobref *iobref, int txflags)
{
        rpcsvc_txbuf_t  *txbuf = NULL;

        txbuf = (rpcsvc_txbuf_t *) mem_get(conn->txpool);
        if (!txbuf) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get txbuf");
                return NULL;
        }

        memset (txbuf, 0, sizeof (*txbuf));
        INIT_LIST_HEAD (&txbuf->txlist);
        txbuf->buf = msg;

        /* If it was required, this iob must've been ref'd already
         *  so I dont have to bother here.
         */
        txbuf->iob = iob;
        txbuf->iobref = iobref;
        txbuf->offset = 0;
        txbuf->txbehave = txflags;

        return txbuf;
}


int
nfs_rpcsvc_conn_append_txlist (rpcsvc_conn_t *conn, struct iovec msg,
                               struct iobuf *iob, int txflags)
{
        rpcsvc_txbuf_t          *txbuf = NULL;

        if ((!conn) || (!msg.iov_base) || (!iob))
                return -1;

        txbuf = nfs_rpcsvc_init_txbuf (conn, msg, iob, NULL, txflags);
        if (!txbuf)
                return -1;

        list_add_tail (&txbuf->txlist, &conn->txbufs);
        return 0;
}


void
nfs_rpcsvc_set_lastfrag (uint32_t *fragsize) {
        (*fragsize) |= 0x80000000U;
}

void
nfs_rpcsvc_set_frag_header_size (uint32_t size, char *haddr)
{
        size = htonl (size);
        memcpy (haddr, &size, sizeof (size));
}

void
nfs_rpcsvc_set_last_frag_header_size (uint32_t size, char *haddr)
{
        nfs_rpcsvc_set_lastfrag (&size);
        nfs_rpcsvc_set_frag_header_size (size, haddr);
}


/* Given the RPC reply structure and the payload handed by the RPC program,
 * encode the RPC record header into the buffer pointed by recordstart.
 */
struct iovec
nfs_rpcsvc_record_build_header (char *recordstart, size_t rlen,
                                struct rpc_msg reply, size_t payload)
{
        struct iovec    replyhdr;
        struct iovec    txrecord = {0, 0};
        size_t          fraglen = 0;
        int             ret = -1;

        /* After leaving aside the 4 bytes for the fragment header, lets
         * encode the RPC reply structure into the buffer given to us.
         */
        ret = nfs_rpc_reply_to_xdr (&reply,(recordstart + RPCSVC_FRAGHDR_SIZE),
                                    rlen, &replyhdr);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to create RPC reply");
                goto err;
        }

        fraglen = payload + replyhdr.iov_len;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Reply fraglen %zu, payload: %zu, "
                "rpc hdr: %zu", fraglen, payload, replyhdr.iov_len);

        /* Since we're not spreading RPC records over mutiple fragments
         * we just set this fragment as the first and last fragment for this
         * record.
         */
        nfs_rpcsvc_set_last_frag_header_size (fraglen, recordstart);

        /* Even though the RPC record starts at recordstart+RPCSVC_FRAGHDR_SIZE
         * we need to transmit the record with the fragment header, which starts
         * at recordstart.
         */
        txrecord.iov_base = recordstart;

        /* Remember, this is only the vec for the RPC header and does not
         * include the payload above. We needed the payload only to calculate
         * the size of the full fragment. This size is sent in the fragment
         * header.
         */
        txrecord.iov_len = RPCSVC_FRAGHDR_SIZE + replyhdr.iov_len;

err:
        return txrecord;
}


int
nfs_rpcsvc_conn_submit (rpcsvc_conn_t *conn, struct iovec hdr,
                        struct iobuf *hdriob, struct iovec msgvec,
                        struct iobuf *msgiob)
{
        int     ret = -1;

        if ((!conn) || (!hdr.iov_base) || (!hdriob))
                return -1;

        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Tx Header: %zu, payload: %zu",
                hdr.iov_len, msgvec.iov_len);
        /* Now that we have both the RPC and Program buffers in xdr format
         * lets hand it to the transmission layer.
         */
        pthread_mutex_lock (&conn->connlock);
        {
                if (!nfs_rpcsvc_conn_check_active (conn)) {
                        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Connection inactive");
                        goto unlock_err;
                }

                ret = nfs_rpcsvc_conn_append_txlist (conn, hdr, hdriob,
                                                     RPCSVC_TXB_FIRST);
                if (ret == -1) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to append "
                                "header to transmission list");
                        goto unlock_err;
                }

                /* It is possible that this RPC reply is an error reply. In that
                 * case we might not have been handed a payload.
                 */
                ret = 0;
                if (msgiob)
                        ret = nfs_rpcsvc_conn_append_txlist (conn, msgvec,
                                                             msgiob,
                                                             RPCSVC_TXB_LAST);
                if (ret == -1) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to append"
                                " payload to transmission list");
                        goto unlock_err;
                }
        }
unlock_err:
        pthread_mutex_unlock (&conn->connlock);

        if (ret == -1)
                goto err;

        /* Tell event pool, we're interested in poll_out to trigger flush
         * of our tx buffers.
         */
        conn->eventidx = event_select_on (conn->stage->eventpool, conn->sockfd,
                                          conn->eventidx, -1, 1);
        ret = 0;
err:

        return ret;
}


int
nfs_rpcsvc_fill_reply (rpcsvc_request_t *req, struct rpc_msg *reply)
{
        rpcsvc_program_t        *prog = NULL;
        if ((!req) || (!reply))
                return -1;

        prog = nfs_rpcsvc_request_program (req);
        nfs_rpc_fill_empty_reply (reply, req->xid);

        if (req->rpc_stat == MSG_DENIED)
                nfs_rpc_fill_denied_reply (reply, req->rpc_err, req->auth_err);
        else if (req->rpc_stat == MSG_ACCEPTED) {
                if (!prog)
                        nfs_rpc_fill_accepted_reply (reply, req->rpc_err, 0, 0,
                                                     req->verf.flavour,
                                                     req->verf.datalen,
                                                     req->verf.authdata);
                else
                        nfs_rpc_fill_accepted_reply (reply, req->rpc_err,
                                                     prog->proglowvers,
                                                     prog->proghighvers,
                                                     req->verf.flavour,
                                                     req->verf.datalen,
                                                     req->verf.authdata);

        } else
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Invalid rpc_stat value");

        return 0;
}


/* Given a request and the reply payload, build a reply and encodes the reply
 * into a record header. This record header is encoded into the vector pointed
 * to be recbuf.
 * msgvec is the buffer that points to the payload of the RPC program.
 * This buffer can be NULL, if an RPC error reply is being constructed.
 * The only reason it is needed here is that in case the buffer is provided,
 * we should account for the length of that buffer in the RPC fragment header.
 */
struct iobuf *
nfs_rpcsvc_record_build_record (rpcsvc_request_t *req, size_t payload,
                                struct iovec *recbuf)
{
        struct rpc_msg          reply;
        struct iobuf            *replyiob = NULL;
        char                    *record = NULL;
        struct iovec            recordhdr = {0, };
        size_t                  pagesize = 0;
        rpcsvc_conn_t           *conn = NULL;
        rpcsvc_t                *svc = NULL;

        if ((!req) || (!req->conn) || (!recbuf))
                return NULL;

        /* First, try to get a pointer into the buffer which the RPC
         * layer can use.
         */
        conn = req->conn;
        svc = nfs_rpcsvc_conn_rpcsvc (conn);
        replyiob = iobuf_get (svc->ctx->iobuf_pool);
        pagesize = iobpool_pagesize ((struct iobuf_pool *)svc->ctx->iobuf_pool);
        if (!replyiob) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get iobuf");
                goto err_exit;
        }

        record = iobuf_ptr (replyiob);  /* Now we have it. */

        /* Fill the rpc structure and XDR it into the buffer got above. */
        nfs_rpcsvc_fill_reply (req, &reply);
        recordhdr = nfs_rpcsvc_record_build_header (record, pagesize, reply,
                                                    payload);
        if (!recordhdr.iov_base) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to build record "
                        " header");
                iobuf_unref (replyiob);
                replyiob = NULL;
                recbuf->iov_base = NULL;
                goto err_exit;
        }

        recbuf->iov_base = recordhdr.iov_base;
        recbuf->iov_len = recordhdr.iov_len;
err_exit:
        return replyiob;
}


/*
 * The function to submit a program message to the RPC service.
 * This message is added to the transmission queue of the
 * conn.
 *
 * Program callers are not expected to use the msgvec->iov_base
 * address for anything else.
 * Nor are they expected to free it once this function returns.
 * Once the transmission of the buffer is completed by the RPC service,
 * the memory area as referenced through @msg will be unrefed.
 * If a higher layer does not want anything to do with this iobuf
 * after this function returns, it should call unref on it. For keeping
 * it around till the transmission is actually complete, rpcsvc also refs it.
 *  *
 * If this function returns an error by returning -1, the
 * higher layer programs should assume that a disconnection happened
 * and should know that the conn memory area as well as the req structure
 * has been freed internally.
 *
 * For now, this function assumes that a submit is always called
 * to send a new record. Later, if there is a situation where different
 * buffers for the same record come from different sources, then we'll
 * need to change this code to account for multiple submit calls adding
 * the buffers into a single record.
 */

int
nfs_rpcsvc_submit_generic (rpcsvc_request_t *req, struct iovec msgvec,
                           struct iobuf *msg)
{
        int                     ret = -1;
        struct iobuf            *replyiob = NULL;
        struct iovec            recordhdr = {0, };
        rpcsvc_conn_t           *conn = NULL;
        int                     rpc_status = 0;
        int                     rpc_error = 0;

        if ((!req) || (!req->conn))
                return -1;

        conn = req->conn;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Tx message: %zu", msgvec.iov_len);
        /* Build the buffer containing the encoded RPC reply. */
        replyiob = nfs_rpcsvc_record_build_record (req, msgvec.iov_len,
                                                   &recordhdr);
        if (!replyiob) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR,"Reply record creation failed");
                goto disconnect_exit;
        }

        /* Must ref the iobuf got from higher layer so that the higher layer
         * can rest assured that it can unref it and leave the final freeing
         * of the buffer to us. Note msg can be NULL if an RPC-only message
         * was being sent. Happens when an RPC error reply is being sent.
         */
        if (msg)
                iobuf_ref (msg);
        ret = nfs_rpcsvc_conn_submit (conn, recordhdr, replyiob, msgvec, msg);

        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to submit message");
                iobuf_unref (replyiob);
        }

disconnect_exit:
        /* Note that a unref is called everytime a reply is sent. This is in
         * response to the ref that is performed on the conn when a request is
         * handed to the RPC program.
         *
         * The catch, however, is that if the reply is an rpc error, we must
         * not unref. This is because the ref only contains
         * references for the actors to which the request was handed plus one
         * reference maintained by the RPC layer. By unrefing for a case where
         * no actor was called, we will be losing the ref held for the RPC
         * layer.
         */

        /* If the request succeeded and was handed to the actor, then unref the
         * conn.
         */

        rpc_status = req->rpc_stat;
        rpc_error = req->rpc_err;
        /* Must mem_put req back to rxpool before the possibility of destroying
         * conn in conn_unref, where the rxpool itself is destroyed.
         */
        mem_put (conn->rxpool, req);
        if ((rpc_status == MSG_ACCEPTED) && (rpc_error == SUCCESS))
                nfs_rpcsvc_conn_unref (conn);

        return ret;
}


int
nfs_rpcsvc_request_attach_vector (rpcsvc_request_t *req, struct iovec msgvec,
                                  struct iobuf *iob, struct iobref *iobref,
                                  int finalvector)
{
        rpcsvc_txbuf_t          *txb = NULL;
        int                     txflags = 0;

        if ((!req) || (!msgvec.iov_base))
                return -1;

        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Tx Vector: %zu", msgvec.iov_len);
        if (finalvector)
                txflags |= RPCSVC_TXB_LAST;
        /* We only let the user decide whether this is the last vector for the
         * record, since the first vector is always the RPC header.
         */
        txb = nfs_rpcsvc_init_txbuf (req->conn, msgvec, iob, iobref, txflags);
        if (!txb) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Could not init tx buf");
                return -1;
        }

        req->payloadsize += msgvec.iov_len;
        if (iob)
                iobuf_ref (iob);
        if (iobref)
                iobref_ref (iobref);
        list_add_tail (&txb->txlist, &req->txlist);

        return 0;
}


int
nfs_rpcsvc_request_attach_vectors (rpcsvc_request_t *req, struct iovec *payload,
                                   int vcount, struct iobref *piobref)
{
        int     c = 0;
        int     ret = -1;

        for (;c < (vcount-1); c++) {
                ret = nfs_rpcsvc_request_attach_vector (req, payload[c], NULL,
                                                        piobref, 0);
                if (ret < 0) {
                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to attach "
                                "vector");
                        goto out;
                }
        }

        ret = nfs_rpcsvc_request_attach_vector (req, payload[vcount-1], NULL,
                                                piobref, 1);
        if (ret < 0)
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to attach final vec");

out:
	return ret;
}


int
nfs_rpcsvc_submit_vectors (rpcsvc_request_t *req)
{
        int                     ret = -1;
        struct iobuf            *replyiob = NULL;
        struct iovec            recordhdr = {0, };
        rpcsvc_txbuf_t          *rpctxb = NULL;
        rpcsvc_conn_t           *conn = NULL;

        if ((!req) || (!req->conn))
                return -1;

        /* Build the buffer containing the encoded RPC reply. */
        replyiob = nfs_rpcsvc_record_build_record (req, req->payloadsize,
                                                   &recordhdr);
        if (!replyiob) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR,"Reply record creation failed");
                goto disconnect_exit;
        }

        rpctxb = nfs_rpcsvc_init_txbuf (req->conn, recordhdr, replyiob, NULL,
                                        RPCSVC_TXB_FIRST);
        if (!rpctxb) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to create tx buf");
                goto disconnect_exit;
        }

        pthread_mutex_lock (&req->conn->connlock);
        {
                list_add_tail (&rpctxb->txlist, &req->conn->txbufs);
                list_append_init (&req->txlist, &req->conn->txbufs);
        }
        pthread_mutex_unlock (&req->conn->connlock);

        ret = 0;
        req->conn->eventidx = event_select_on (req->conn->stage->eventpool,
                                               req->conn->sockfd,
                                               req->conn->eventidx, -1, 1);
disconnect_exit:
        /* Note that a unref is called everytime a reply is sent. This is in
         * response to the ref that is performed on the conn when a request is
         * handed to the RPC program.
         */
        conn = req->conn;
        /* Must mem_put req back to rxpool before the possibility of destroying
         * conn in conn_unref, where the rxpool itself is destroyed.
         */
        mem_put (conn->rxpool, req);

        nfs_rpcsvc_conn_unref (conn);
        if (ret == -1)
                iobuf_unref (replyiob);

        return ret;
}


int
nfs_rpcsvc_error_reply (rpcsvc_request_t *req)
{
        struct iovec    dummyvec = {0, };

        if (!req)
                return -1;

        /* At this point the req should already have been filled with the
         * appropriate RPC error numbers.
         */
        return nfs_rpcsvc_submit_generic (req, dummyvec, NULL);
}


rpcsvc_request_t *
nfs_rpcsvc_request_init (rpcsvc_conn_t *conn, struct rpc_msg *callmsg,
                         struct iovec progmsg, rpcsvc_request_t *req)
{
        if ((!conn) || (!callmsg)|| (!req))
                return NULL;


        /* We start a RPC request as always denied. */
        req->rpc_stat = MSG_DENIED;
        req->xid = nfs_rpc_call_xid (callmsg);
        req->prognum = nfs_rpc_call_program (callmsg);
        req->progver = nfs_rpc_call_progver (callmsg);
        req->procnum = nfs_rpc_call_progproc (callmsg);
        req->conn = conn;
        req->msg = progmsg;
        req->recordiob = conn->rstate.activeiob;
        INIT_LIST_HEAD (&req->txlist);
        req->payloadsize = 0;

        /* By this time, the data bytes for the auth scheme would have already
         * been copied into the required sections of the req structure,
         * we just need to fill in the meta-data about it now.
         */
        req->cred.flavour = nfs_rpc_call_cred_flavour (callmsg);
        req->cred.datalen = nfs_rpc_call_cred_len (callmsg);
        req->verf.flavour = nfs_rpc_call_verf_flavour (callmsg);
        req->verf.datalen = nfs_rpc_call_verf_len (callmsg);

        /* AUTH */
        nfs_rpcsvc_auth_request_init (req);
        return req;
}


rpcsvc_request_t *
nfs_rpcsvc_request_create (rpcsvc_conn_t *conn)
{
        char                    *msgbuf = NULL;
        struct rpc_msg          rpcmsg;
        struct iovec            progmsg;        /* RPC Program payload */
        rpcsvc_request_t        *req = NULL;
        int                     ret = -1;
        rpcsvc_program_t        *program = NULL;

        if (!conn)
                return NULL;

        /* We need to allocate the request before actually calling
         * rpcsvc_request_init on the request so that we, can fill the auth
         * data directly into the request structure from the message iobuf.
         * This avoids a need to keep a temp buffer into which the auth data
         * would've been copied otherwise.
         */
        nfs_rpcsvc_alloc_request (conn, req);
        if (!req) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to alloc request");
                goto err;
        }

        msgbuf = iobuf_ptr (conn->rstate.activeiob);
        ret = nfs_xdr_to_rpc_call (msgbuf, conn->rstate.recordsize, &rpcmsg,
                                   &progmsg, req->cred.authdata,
                                   req->verf.authdata);

        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "RPC call decoding failed");
                nfs_rpcsvc_request_seterr (req, GARBAGE_ARGS);
                req->conn = conn;
                goto err;
        }

        ret = -1;
        nfs_rpcsvc_request_init (conn, &rpcmsg, progmsg, req);

        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC XID: %lx, Ver: %ld, Program: %ld,"
                " ProgVers: %ld, Proc: %ld", nfs_rpc_call_xid (&rpcmsg),
                nfs_rpc_call_rpcvers (&rpcmsg), nfs_rpc_call_program (&rpcmsg),
                nfs_rpc_call_progver (&rpcmsg),
                nfs_rpc_call_progproc (&rpcmsg));

        if (nfs_rpc_call_rpcvers (&rpcmsg) != 2) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "RPC version not supported");
                nfs_rpcsvc_request_seterr (req, RPC_MISMATCH);
                goto err;
        }

        ret = __nfs_rpcsvc_program_actor (req, &program);
        if (ret != SUCCESS) {
                ret = -1;
                goto err;
        }

        req->program = program;
        ret = nfs_rpcsvc_authenticate (req);
        if (ret == RPCSVC_AUTH_REJECT) {
                /* No need to set auth_err, that is the responsibility of
                 * the authentication handler since only that know what exact
                 * error happened.
                 */
                nfs_rpcsvc_request_seterr (req, AUTH_ERROR);
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed authentication");
                ret = -1;
                goto err;
        }

        ret = 0;
err:
        if (ret == -1) {
                ret = nfs_rpcsvc_error_reply (req);
                req = NULL;
        }

        return req;
}


int
nfs_rpcsvc_handle_rpc_call (rpcsvc_conn_t *conn)
{
        rpcsvc_actor_t          *actor = NULL;
        rpcsvc_request_t        *req = NULL;
        int                     ret = -1;

        if (!conn)
                return -1;

        req = nfs_rpcsvc_request_create (conn);
        if (!req)
                goto err;

        if (!nfs_rpcsvc_request_accepted (req))
                goto err_reply;

        actor = nfs_rpcsvc_program_actor (req);
        if (!actor)
                goto err_reply;

        if ((actor) && (actor->actor)) {
                THIS = nfs_rpcsvc_request_actorxl (req);
                nfs_rpcsvc_conn_ref (conn);
                ret = actor->actor (req);
        }

err_reply:
        if (ret == RPCSVC_ACTOR_ERROR)
                ret = nfs_rpcsvc_error_reply (req);
        else if (ret == RPCSVC_ACTOR_IGNORE)
                mem_put (conn->rxpool, req);

        /* No need to propagate error beyond this function since the reply
         * has now been queued. */
        ret = 0;
err:
        return ret;
}

#define nfs_rpc_call_cred_addr(rs) (iobuf_ptr ((rs)->activeiob) + RPCSVC_BARERPC_MSGSZ - 4)

uint32_t
nfs_rpcsvc_call_credlen (rpcsvc_record_state_t *rs)
{
        char                    *credaddr = NULL;
        uint32_t                credlen_nw = 0;
        uint32_t                credlen_host = 0;

        /* Position to the start of the credential length field. */
        credaddr = nfs_rpc_call_cred_addr (rs);
        credlen_nw = *(uint32_t *)credaddr;
        credlen_host = ntohl (credlen_nw);

        return credlen_host;
}

uint32_t
nfs_rpcsvc_call_verflen (rpcsvc_record_state_t *rs)
{
        char            *verfaddr = NULL;
        uint32_t        verflen_nw = 0;
        uint32_t        verflen_host = 0;
        uint32_t        credlen = 0;

        /* Position to the start of the verifier length field. */
        credlen = nfs_rpcsvc_call_credlen (rs);
        verfaddr = (nfs_rpc_call_cred_addr (rs) + 4 + credlen);
        verflen_nw = *(uint32_t *)verfaddr;
        verflen_host = ntohl (verflen_nw);

        return verflen_host;
}


void
nfs_rpcsvc_update_vectored_verf (rpcsvc_record_state_t *rs)
{
        if (!rs)
                return;

        rs->recordsize += nfs_rpcsvc_call_verflen (rs);
        return;
}


void
nfs_rpcsvc_handle_vectored_prep_rpc_call (rpcsvc_conn_t *conn)
{
        rpcsvc_actor_t          *actor = NULL;
        rpcsvc_request_t        *req = NULL;
        rpcsvc_record_state_t   *rs = NULL;
        rpcsvc_t                *svc = NULL;
        int                     ret = -1;
        ssize_t                 remfrag = RPCSVC_ACTOR_ERROR;
        int                     newbuf = 0;

        if (!conn)
                return;

        rs = &conn->rstate;

        /* In case one of the steps below fails, we need to make sure that the
         * remaining frag in the kernel's buffers are read-out so that the
         * requests that follow can be served.
         */
        rs->remainingfrag = rs->fragsize - rs->recordsize;
        rs->vecstate = RPCSVC_VECTOR_IGNORE;
        req = nfs_rpcsvc_request_create (conn);
        svc = nfs_rpcsvc_conn_rpcsvc (conn);
        if (!req)
                goto err;

        if (!nfs_rpcsvc_request_accepted (req))
                goto err_reply;

        actor = nfs_rpcsvc_program_actor (req);
        if (!actor)
                goto err_reply;

        if (!actor->vector_sizer) {
                ret = -1;
                nfs_rpcsvc_request_seterr (req, PROC_UNAVAIL);
                goto err_reply;
        }

        nfs_rpcsvc_conn_ref (conn);
        THIS = nfs_rpcsvc_request_actorxl (req);
        ret = actor->vector_sizer (req, &remfrag, &newbuf);
        nfs_rpcsvc_conn_unref (conn);

        if (ret == RPCSVC_ACTOR_ERROR) {
                ret = -1;
                nfs_rpcsvc_request_seterr (req, SYSTEM_ERR);
                goto err_reply;
        }

        rs->remainingfrag = remfrag;
        rs->vecstate = RPCSVC_VECTOR_READPROCHDR;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC proc header remaining:"
                " %d", rs->remainingfrag);
        conn->vectoredreq = req;

        /* Store the reference to the current frag pointer. This is where the
         * proc header will be read into.
         */
        req->msg.iov_base = rs->fragcurrent;
        req->msg.iov_len = rs->remainingfrag;
        ret = 0;

err_reply:
        if (ret == -1)
                ret = nfs_rpcsvc_error_reply (req);

        /* No need to propagate error beyond this function since the reply
         * has now been queued. */
        ret = 0;
err:
        return;
}


void
nfs_rpcsvc_update_vectored_verfsz (rpcsvc_conn_t *conn)
{
        rpcsvc_record_state_t   *rs = NULL;
        uint32_t                verflen = 0;

        if (!conn)
                return;

        rs = &conn->rstate;

        verflen = nfs_rpcsvc_call_verflen (rs);
        rs->recordsize += 8;
        if (verflen > 0) {
                rs->remainingfrag = verflen;
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC verf remaining: "
                        " %d", rs->remainingfrag);
                rs->vecstate = RPCSVC_VECTOR_READVERF;
        } else {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC preparing call");
                nfs_rpcsvc_handle_vectored_prep_rpc_call (conn);
        }

        return;
}


void
nfs_rpcsvc_update_vectored_cred (rpcsvc_record_state_t *rs)
{
        uint32_t                credlen = 0;

        if (!rs)
                return;

        credlen = nfs_rpcsvc_call_credlen (rs);
        /* Update remainingfrag to read the 8 bytes needed for
         * reading verf flavour and verf len.
         */
        rs->remainingfrag = (2 * sizeof (uint32_t));
        rs->vecstate = RPCSVC_VECTOR_READVERFSZ;
        rs->recordsize += credlen;
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC verfsz remaining: %d",
                rs->remainingfrag);

        return;
}

void
nfs_rpcsvc_update_vectored_barerpc (rpcsvc_record_state_t *rs)
{
        uint32_t                credlen = 0;

        if (!rs)
                return;

        credlen = nfs_rpcsvc_call_credlen (rs);
        rs->recordsize = RPCSVC_BARERPC_MSGSZ;
        if (credlen == 0) {
                rs->remainingfrag = 8;
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC verfsz remaining"
                        ": %d", rs->remainingfrag);
                rs->vecstate = RPCSVC_VECTOR_READVERFSZ;
        } else {
                rs->remainingfrag = credlen;
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC cred remaining: "
                        "%d", rs->remainingfrag);
                rs->vecstate = RPCSVC_VECTOR_READCRED;
        }

        return;
}


void
nfs_rpcsvc_handle_vectored_rpc_call (rpcsvc_conn_t *conn)
{
        rpcsvc_actor_t          *actor = NULL;
        rpcsvc_request_t        *req = NULL;
        rpcsvc_record_state_t   *rs = NULL;
        rpcsvc_t                *svc = NULL;
        int                     ret = -1;
        ssize_t                 remfrag = -1;
        int                     newbuf = 0;

        if (!conn)
                return;

        rs = &conn->rstate;

        req = conn->vectoredreq;
        svc = nfs_rpcsvc_conn_rpcsvc (conn);

        if (!req)
                goto err;

        actor = nfs_rpcsvc_program_actor (req);
        if (!actor)
                goto err_reply;

        if (!actor->vector_sizer) {
                ret = -1;
                nfs_rpcsvc_request_seterr (req, PROC_UNAVAIL);
                goto err_reply;
        }

        req->msg.iov_len = (unsigned long)((long)rs->fragcurrent - (long)req->msg.iov_base);
        nfs_rpcsvc_conn_ref (conn);
        THIS = nfs_rpcsvc_request_actorxl (req);
        ret = actor->vector_sizer (req, &remfrag, &newbuf);
        nfs_rpcsvc_conn_unref (conn);
        if (ret == RPCSVC_ACTOR_ERROR) {
                ret = -1;
                nfs_rpcsvc_request_seterr (req, SYSTEM_ERR);
                goto err_reply;
        }

        if (newbuf) {
                rs->vectoriob = iobuf_get (svc->ctx->iobuf_pool);
                rs->fragcurrent = iobuf_ptr (rs->vectoriob);
                rs->vecstate = RPCSVC_VECTOR_READVEC;
                rs->remainingfrag = remfrag;
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC buf remaining:"
                        " %d", rs->remainingfrag);
        } else {
                rs->remainingfrag = remfrag;
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC proc remaining:"
                        " %d", rs->remainingfrag);
        }

        ret = 0;
err_reply:
        if (ret == -1)
                ret = nfs_rpcsvc_error_reply (req);

        /* No need to propagate error beyond this function since the reply
         * has now been queued. */
        ret = 0;
err:
        return;
}



void
nfs_rpcsvc_record_vectored_call_actor (rpcsvc_conn_t *conn)
{
        rpcsvc_actor_t          *actor = NULL;
        rpcsvc_request_t        *req = NULL;
        rpcsvc_record_state_t   *rs = NULL;
        rpcsvc_t                *svc = NULL;
        int                     ret = -1;

        if (!conn)
                return;

        rs = &conn->rstate;
        req = conn->vectoredreq;
        svc = nfs_rpcsvc_conn_rpcsvc (conn);

        if (!req)
                goto err;

        actor = nfs_rpcsvc_program_actor (req);
        if (!actor)
                goto err_reply;

        if (actor->vector_actor) {
                nfs_rpcsvc_conn_ref (conn);
                THIS = nfs_rpcsvc_request_actorxl (req);
                ret = actor->vector_actor (req, rs->vectoriob);
        } else {
                nfs_rpcsvc_request_seterr (req, PROC_UNAVAIL);
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "No vectored handler present");
                ret = RPCSVC_ACTOR_ERROR;
        }

err_reply:
        if (ret == RPCSVC_ACTOR_ERROR)
                ret = nfs_rpcsvc_error_reply (req);
        else if (ret == RPCSVC_ACTOR_IGNORE)
                mem_put (conn->rxpool, req);

        /* No need to propagate error beyond this function since the reply
         * has now been queued. */
        ret = 0;
err:
        return;
}



ssize_t
nfs_rpcsvc_update_vectored_state (rpcsvc_conn_t *conn)
{
        rpcsvc_record_state_t   *rs = NULL;
        rpcsvc_t                *svc = NULL;

        if (!conn)
                return 0;

        /* At this point, we can be confident that the activeiob contains
         * exactly the first RPCSVC_BARERPC_MSGSZ bytes needed in order to
         * determine the program and actor. So the next state will be
         * to read the credentials.
         *
         * But first, try to determine how many more bytes do we need from the
         * network to complete the RPC message including the credentials.
         */

        rs = &conn->rstate;
        if (nfs_rpcsvc_record_vectored_baremsg (rs))
                nfs_rpcsvc_update_vectored_barerpc (rs);
        else if (nfs_rpcsvc_record_vectored_cred (rs))
                nfs_rpcsvc_update_vectored_cred (rs);
        else if (nfs_rpcsvc_record_vectored_verfsz (rs))
                nfs_rpcsvc_update_vectored_verfsz (conn);
        else if (nfs_rpcsvc_record_vectored_verfread (rs)) {
                nfs_rpcsvc_update_vectored_verf (rs);
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC preparing call");
                nfs_rpcsvc_handle_vectored_prep_rpc_call (conn);
        } else if (nfs_rpcsvc_record_vectored_readprochdr (rs))
                nfs_rpcsvc_handle_vectored_rpc_call (conn);
        else if (nfs_rpcsvc_record_vectored_ignore (rs)) {
                svc = nfs_rpcsvc_conn_rpcsvc (conn);
                nfs_rpcsvc_record_init (rs, svc->ctx->iobuf_pool);
        } else if (nfs_rpcsvc_record_vectored_readvec (rs)) {
                svc = nfs_rpcsvc_conn_rpcsvc (conn);
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored RPC vector read");
                nfs_rpcsvc_record_vectored_call_actor (conn);
                nfs_rpcsvc_record_init (rs, svc->ctx->iobuf_pool);
        }

        return 0;
}


ssize_t
nfs_rpcsvc_record_read_partial_frag (rpcsvc_record_state_t *rs,
                                     ssize_t dataread);

ssize_t
nfs_rpcsvc_update_vectored_msg (rpcsvc_conn_t *conn, ssize_t dataread)
{

        if (!conn)
                return dataread;

        /* find out how much of the bare msg is pending and set that up to be
         * read into the updated fragcurrent along with the updated size into
         * remainingfrag.
         */


        /* Incidently, the logic needed here is similar to a regular partial
         * fragment read since we've already set the remainingfrag member in
         * rstate to be RPCSVC_BARERPC_MSGSZ for the purpose of a vectored
         * fragment.
         */
        return nfs_rpcsvc_record_read_partial_frag (&conn->rstate, dataread);
}

/* FIX: As a first version of vectored reading, I am assuming dataread will
 * always be equal to RPCSVC_BARERPC_MSGSZ for the sake of simplicity on the
 * belief that we're never actually reading more bytes than needed in each
 * poll_in.
 */
ssize_t
nfs_rpcsvc_handle_vectored_frag (rpcsvc_conn_t *conn, ssize_t dataread)
{
        if (!conn)
                return dataread;

        /* At this point we can be confident that only the frag size has been
         * read from the network. Now it is up to us to have the remaining RPC
         * fields given to us here.
         */

        /* Since the poll_in handler uses the remainingfrag field to determine
         * how much to read from the network, we'll hack this scheme to tell
         * the poll_in to read at most RPCSVC_BARERPC_MSGSZ bytes. This is done
         * to, as a first step, identify which (program, actor) we need to call.
         */

        dataread = nfs_rpcsvc_update_vectored_msg (conn, dataread);

        if (conn->rstate.remainingfrag == 0) {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored frag complete");
                dataread = nfs_rpcsvc_update_vectored_state (conn);
        }

        return dataread;
}


int
nfs_rpcsvc_record_update_state (rpcsvc_conn_t *conn, ssize_t dataread)
{
        rpcsvc_record_state_t   *rs = NULL;
        rpcsvc_t                *svc = NULL;

        if (!conn)
                return -1;

        rs = &conn->rstate;
        /* At entry into this function, fragcurrent will be pointing to the\
         * start of the area into which dataread number of bytes were read.
         */

        if (nfs_rpcsvc_record_readfraghdr(rs))
                dataread = nfs_rpcsvc_record_update_fraghdr (rs, dataread);

        if (nfs_rpcsvc_record_readfrag(rs)) {
                /* The minimum needed for triggering the vectored handler is
                 * the frag size field. The fragsize member remains set to this
                 * size till this request is completely extracted from the
                 * network. Once all the data has been read from the network,
                 * the request structure would've been created. The point being
                 * that even if it takes multiple calls to network IO for
                 * getting the vectored fragment, we can continue to use this
                 * condition as the flag to tell us that this is a vectored
                 * fragment.
                 */
                if ((dataread > 0) && (nfs_rpcsvc_record_vectored (rs))) {
                        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Vectored frag");
                        dataread = nfs_rpcsvc_handle_vectored_frag (conn,
                                                                    dataread);
                } else if (dataread > 0) {
                        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Regular frag");
                        dataread = nfs_rpcsvc_record_update_frag (rs, dataread);
                }
        }

        /* This should not happen. We are never reading more than the current
         * fragment needs. Something is seriously wrong.
         */
        if (dataread > 0) {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Data Left: %zd", dataread);
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Unwanted data read from "
                        " connection.");
        }

        /* If we're now supposed to wait for a new fragment header and if the
         * fragment that we just completed in the previous call to
         * rpcsvc_record_update_frag was the last fragment for the current
         * RPC record, then, it is time to perform the translation from
         * XDR formatted buffer in activeiob followed by the upcall to the
         * protocol actor.
         */
        if ((nfs_rpcsvc_record_readfraghdr(rs)) && (rs->islastfrag)) {
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "Full Record Received.");
                nfs_rpcsvc_handle_rpc_call (conn);
                svc = nfs_rpcsvc_conn_rpcsvc (conn);
                nfs_rpcsvc_record_init (rs, svc->ctx->iobuf_pool);
        }

        return 0;
}


char *
nfs_rpcsvc_record_read_addr (rpcsvc_record_state_t *rs)
{

        if (nfs_rpcsvc_record_readfraghdr (rs))
                return nfs_rpcsvc_record_currenthdr_addr (rs);
        else if (nfs_rpcsvc_record_readfrag (rs))
                return nfs_rpcsvc_record_currentfrag_addr (rs);

        return NULL;
}


int
nfs_rpcsvc_conn_data_poll_in (rpcsvc_conn_t *conn)
{
        ssize_t         dataread = -1;
        size_t          readsize = 0;
        char            *readaddr = NULL;
        int             ret = -1;

        readaddr = nfs_rpcsvc_record_read_addr (&conn->rstate);
        if (!readaddr)
                goto err;

        readsize = nfs_rpcsvc_record_read_size (&conn->rstate);
        if (readsize == -1)
                goto err;

        dataread = nfs_rpcsvc_socket_read (conn->sockfd, readaddr, readsize);
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "conn: 0x%lx, readsize: %zu, dataread:"
                "%zd", (long)conn, readsize, dataread);

        if (dataread > 0)
                ret = nfs_rpcsvc_record_update_state (conn, dataread);

err:
        return ret;
}


int
nfs_rpcsvc_conn_data_poll_err (rpcsvc_conn_t *conn)
{
        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Received error event");
        nfs_rpcsvc_conn_deinit (conn);
        return 0;
}


int
__nfs_rpcsvc_conn_data_poll_out (rpcsvc_conn_t *conn)
{
        rpcsvc_txbuf_t          *txbuf = NULL;
        rpcsvc_txbuf_t          *tmp = NULL;
        ssize_t                 written = -1;
        char                    *writeaddr = NULL;
        size_t                  writesize = -1;
        int                     eagain = 0;

        if (!conn)
                return -1;

        /* Attempt transmission of each of the pending buffers */
        list_for_each_entry_safe (txbuf, tmp, &conn->txbufs, txlist) {
tx_remaining:
                eagain = 0;
                writeaddr = (char *)(txbuf->buf.iov_base + txbuf->offset);
                writesize = (txbuf->buf.iov_len - txbuf->offset);

                if (txbuf->txbehave & RPCSVC_TXB_FIRST) {
                        gf_log (GF_RPCSVC, GF_LOG_TRACE, "First Tx Buf");
                        nfs_rpcsvc_socket_block_tx (conn->sockfd);
                }

                written = nfs_rpcsvc_socket_write (conn->sockfd, writeaddr,
                                                   writesize, &eagain);
                if (txbuf->txbehave & RPCSVC_TXB_LAST) {
                        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Last Tx Buf");
                        nfs_rpcsvc_socket_unblock_tx (conn->sockfd);
                }
                gf_log (GF_RPCSVC, GF_LOG_TRACE, "conn: 0x%lx, Tx request: %zu,"
                        " Tx sent: %zd", (long)conn, writesize, written);

                /* There was an error transmitting this buffer */
                if (written == -1)
                        break;

                if (written >= 0)
                        txbuf->offset += written;

                /* If the current buffer has been completely transmitted,
                 * delete it from the list and move on to the next buffer.
                 */
                if (txbuf->offset == txbuf->buf.iov_len) {
                        /* It doesnt matter who ref'ed this iobuf, rpcsvc for
                         * its own header or a RPC program.
                         */
                        if (txbuf->iob)
                                iobuf_unref (txbuf->iob);
                        if (txbuf->iobref)
                                iobref_unref (txbuf->iobref);

                        list_del (&txbuf->txlist);
                        mem_put (conn->txpool, txbuf);
                } else {
                        /* If the current buffer is incompletely tx'd, do not
                         * go to the head of the loop, since that moves us to
                         * the next buffer.
                         *
                         * BUT, if the current transmission exited due to EAGAIN
                         * we need to leave the buffers where they are and come
                         * back later for retransmission.
                         */
                        if (!eagain)
                                goto tx_remaining;
                        else
                                break;
                }

        }

        /* If we've broken out of the loop above then we must unblock
         * the transmission now.
         */
        nfs_rpcsvc_socket_unblock_tx (conn->sockfd);
        if (list_empty (&conn->txbufs))
                conn->eventidx = event_select_on (conn->stage->eventpool,
                                                  conn->sockfd, conn->eventidx,
                                                  -1, 0);

        return 0;
}


int
nfs_rpcsvc_conn_data_poll_out (rpcsvc_conn_t *conn)
{
        if (!conn)
                return -1;


        pthread_mutex_lock (&conn->connlock);
        {
                __nfs_rpcsvc_conn_data_poll_out (conn);
        }
        pthread_mutex_unlock (&conn->connlock);

        return 0;
}


int
nfs_rpcsvc_conn_data_handler (int fd, int idx, void *data, int poll_in,
                              int poll_out, int poll_err)
{
        rpcsvc_conn_t   *conn = NULL;
        int             ret = 0;

        if (!data)
                return 0;

        conn = (rpcsvc_conn_t *)data;

        if (poll_out)
                ret = nfs_rpcsvc_conn_data_poll_out (conn);

        if (poll_err) {
                ret = nfs_rpcsvc_conn_data_poll_err (conn);
                return 0;
        }

        if (poll_in) {
                ret = 0;
                ret = nfs_rpcsvc_conn_data_poll_in (conn);
        }

        if (ret == -1)
                nfs_rpcsvc_conn_data_poll_err (conn);

        return 0;
}


int
nfs_rpcsvc_conn_listening_handler (int fd, int idx, void *data, int poll_in,
                                   int poll_out, int poll_err)
{
        rpcsvc_conn_t           *newconn = NULL;
        rpcsvc_stage_t          *selectedstage = NULL;
        int                     ret = -1;
        rpcsvc_conn_t           *conn = NULL;
        rpcsvc_t                *svc = NULL;

        if (!poll_in)
                return 0;

        conn = (rpcsvc_conn_t *)data;
        svc = nfs_rpcsvc_conn_rpcsvc (conn);
        newconn = nfs_rpcsvc_conn_accept_init (svc, fd);
        if (!newconn) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "failed to accept connection");
                goto err;
        }

        selectedstage = nfs_rpcsvc_select_stage (svc);
        if (!selectedstage)
                goto close_err;

        /* Now that we've accepted the connection, we need to associate
         * its events to a stage.
         */
        ret = nfs_rpcsvc_stage_conn_associate (selectedstage, newconn,
                                               nfs_rpcsvc_conn_data_handler,
                                               newconn);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "could not associated stage "
                        " with new connection");
                goto close_err;
        }
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "New Connection");
        ret = 0;
close_err:
        if (ret == -1)
                nfs_rpcsvc_conn_unref (newconn);

err:
        return ret;
}


/* Register the program with the local portmapper service. */
int
nfs_rpcsvc_program_register_portmap (rpcsvc_t *svc, rpcsvc_program_t *newprog)
{
        if (!newprog)
                return -1;

        if (!svc->register_portmap)
                return 0;

        if (!(pmap_set(newprog->prognum, newprog->progver, IPPROTO_TCP,
                       newprog->progport))) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Could not register with"
                        " portmap");
                return -1;
        }

        return 0;
}


int
nfs_rpcsvc_program_unregister_portmap (rpcsvc_t *svc, rpcsvc_program_t *prog)
{
        if (!prog)
                return -1;

        if (!svc->register_portmap)
                return 0;

        if (!(pmap_unset(prog->prognum, prog->progver))) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Could not unregister with"
                        " portmap");
                return -1;
        }

        return 0;
}


int
nfs_rpcsvc_stage_program_register (rpcsvc_stage_t *stg,
                                   rpcsvc_program_t *newprog)
{
        rpcsvc_conn_t           *newconn = NULL;
        rpcsvc_t                *svc = NULL;

        if ((!stg) || (!newprog))
                return -1;

        svc = nfs_rpcsvc_stage_service (stg);
        /* Create a listening socket */
        newconn = nfs_rpcsvc_conn_listen_init (svc, newprog);
        if (!newconn) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "could not create listening"
                        " connection");
                return -1;
        }

        if ((nfs_rpcsvc_stage_conn_associate (stg, newconn,
                                              nfs_rpcsvc_conn_listening_handler,
                                              newconn)) == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR,"could not associate stage with"
                        " listening connection");
                return -1;
        }

        return 0;
}


int
nfs_rpcsvc_program_register (rpcsvc_t *svc, rpcsvc_program_t program)
{
        rpcsvc_program_t        *newprog = NULL;
        rpcsvc_stage_t          *selectedstage = NULL;
        int                     ret = -1;

        if (!svc)
                return -1;

        newprog = GF_CALLOC (1, sizeof(*newprog),gf_common_mt_rpcsvc_program_t);
        if (!newprog)
                return -1;

        if (!program.actors)
                goto free_prog;

        memcpy (newprog, &program, sizeof (program));
        INIT_LIST_HEAD (&newprog->proglist);
        list_add_tail (&newprog->proglist, &svc->allprograms);
        selectedstage = nfs_rpcsvc_select_stage (svc);

        ret = nfs_rpcsvc_stage_program_register (selectedstage, newprog);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "stage registration of program"
                        " failed");
                goto free_prog;
        }

        ret = nfs_rpcsvc_program_register_portmap (svc, newprog);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "portmap registration of"
                        " program failed");
                goto free_prog;
        }

        ret = 0;
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "New program registered: %s, Num: %d,"
                " Ver: %d, Port: %d", newprog->progname, newprog->prognum,
                newprog->progver, newprog->progport);

free_prog:
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Program registration failed:"
                        " %s, Num: %d, Ver: %d, Port: %d", newprog->progname,
                        newprog->prognum, newprog->progver, newprog->progport);
                list_del (&newprog->proglist);
                GF_FREE (newprog);
        }

        return ret;
}

/* The only difference between the generic submit and this one is that the
 * generic submit is also used for submitting RPC error replies in where there
 * are no payloads so the msgvec and msgbuf can be NULL.
 * Since RPC programs should be using this function along with their payloads
 * we must perform NULL checks before calling the generic submit.
 */
int
nfs_rpcsvc_submit_message (rpcsvc_request_t *req, struct iovec msgvec,
                           struct iobuf *msg)
{
        if ((!req) || (!req->conn) || (!msg) || (!msgvec.iov_base))
                return -1;

        return nfs_rpcsvc_submit_generic (req, msgvec, msg);
}


int
nfs_rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t prog)
{
        int                     ret = -1;

        if (!svc)
                return -1;

        /* TODO: De-init the listening connection for this program. */
        ret = nfs_rpcsvc_program_unregister_portmap (svc, &prog);
        if (ret == -1) {
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "portmap unregistration of"
                        " program failed");
                goto err;
        }

        ret = 0;
        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Program unregistered: %s, Num: %d,"
                " Ver: %d, Port: %d", prog.progname, prog.prognum,
                prog.progver, prog.progport);

err:
        if (ret == -1)
                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Program unregistration failed"
                        ": %s, Num: %d, Ver: %d, Port: %d", prog.progname,
                        prog.prognum, prog.progver, prog.progport);

        return ret;
}


int
nfs_rpcsvc_conn_peername (rpcsvc_conn_t *conn, char *hostname, int hostlen)
{
        if (!conn)
                return -1;

        return nfs_rpcsvc_socket_peername (conn->sockfd, hostname, hostlen);
}


int
nfs_rpcsvc_conn_peeraddr (rpcsvc_conn_t *conn, char *addrstr, int addrlen,
                          struct sockaddr *sa, socklen_t sasize)
{
        if (!conn)
                return -1;

        return nfs_rpcsvc_socket_peeraddr (conn->sockfd, addrstr, addrlen, sa,
                                           sasize);
}