7712 mandoc -Tlint does always exit with error code 0

[unleashed.git] / usr / src / lib / lvm / libmeta / common / meta_mn_comm.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include <stdlib.h>
#include <unistd.h>
#include <wait.h>
#include <sys/time.h>
#include <strings.h>
#include <meta.h>
#include <syslog.h>

extern md_mn_msg_tbl_entry_t  msg_table[];

/*
 * When contacting the local rpc.mdcommd we always want to do that using
 * the IPv4 version of localhost.
 */
#define LOCALHOST_IPv4  "127.0.0.1"

md_mn_msgclass_t
mdmn_get_message_class(md_mn_msgtype_t msgtype)
{
        return (msg_table[msgtype].mte_class);
}

void (*
mdmn_get_handler(md_mn_msgtype_t msgtype))
        (md_mn_msg_t *msg, uint_t flags, md_mn_result_t *res)
{
        return (msg_table[msgtype].mte_handler);
}

int (*
mdmn_get_submessage_generator(md_mn_msgtype_t msgtype))
        (md_mn_msg_t *msg, md_mn_msg_t **msglist)
{
        return (msg_table[msgtype].mte_smgen);
}

time_t
mdmn_get_timeout(md_mn_msgtype_t msgtype)
{
        return (msg_table[msgtype].mte_timeout);
}


void
ldump_msg(char *prefix, md_mn_msg_t *msg)
{
        (void) fprintf(stderr, "%s &msg       = 0x%x\n", prefix, (uint_t)msg);
        (void) fprintf(stderr, "%s ID         = (%d, 0x%llx-%d)\n", prefix,
            MSGID_ELEMS(msg->msg_msgid));
        (void) fprintf(stderr, "%s sender     = %d\n", prefix, msg->msg_sender);
        (void) fprintf(stderr, "%s flags      = 0x%x\n",
            prefix, msg->msg_flags);
        (void) fprintf(stderr, "%s setno      = %d\n", prefix, msg->msg_setno);
        (void) fprintf(stderr, "%s recipient  = %d\n",
            prefix, msg->msg_recipient);
        (void) fprintf(stderr, "%s type       = %d\n", prefix, msg->msg_type);
        (void) fprintf(stderr, "%s size       = %d\n",
            prefix, msg->msg_event_size);
}

#define COMMD_PROGNAME  "rpc.mdcommd"

extern uint_t meta_rpc_err_mask(void);

/*
 * If a clnt_call gets an RPC error, force the message out here with details.
 * This would be nice to send to commd_debug(), but we can't call rpc.mdcommd
 * code from libmeta.
 */
static void
mdmn_handle_RPC_error(CLIENT *clnt, char *ident, md_mn_nodeid_t nid)
{
        /*
         * This is sized for a max message which would look like this:
         * "mdmn_wakeup_initiator: rpc.mdcommd node 4294967295"
         */
        char errstr[51];
        struct rpc_err e;

        CLNT_GETERR((CLIENT *) clnt, &e);
        if (meta_rpc_err_mask() & (1 << e.re_status)) {
                if (nid == 0) {
                        (void) snprintf(errstr, sizeof (errstr),
                            "%s: %s node (local)", ident, COMMD_PROGNAME);
                } else {
                        (void) snprintf(errstr, sizeof (errstr),
                            "%s: %s node %d", ident, COMMD_PROGNAME, nid);
                }
                syslog(LOG_WARNING, "mdmn_handle_RPC_error: %s",
                    clnt_sperror(clnt, errstr));
        }
}

/* Default timeout can be changed using clnt_control() */
static struct timeval TIMEOUT = { 25, 0 };

md_mn_result_t *
mdmn_send_2(argp, clnt, nid)
        md_mn_msg_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        md_mn_result_t *clnt_res = Zalloc(sizeof (md_mn_result_t));

        res = clnt_call(clnt, mdmn_send,
                (xdrproc_t)xdr_md_mn_msg_t, (caddr_t)argp,
                (xdrproc_t)xdr_md_mn_result_t, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_send", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_work_2(argp, clnt, nid)
        md_mn_msg_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_work,
                (xdrproc_t)xdr_md_mn_msg_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_work", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_wakeup_initiator_2(argp, clnt, nid)
        md_mn_result_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_wakeup_initiator,
                (xdrproc_t)xdr_md_mn_result_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_wakeup_initiator", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_wakeup_master_2(argp, clnt, nid)
        md_mn_result_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_wakeup_master,
                (xdrproc_t)xdr_md_mn_result_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_wakeup_master", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_lock_2(argp, clnt, nid)
        md_mn_set_and_class_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_lock,
                (xdrproc_t)xdr_md_mn_set_and_class_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_lock", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_unlock_2(argp, clnt, nid)
        md_mn_set_and_class_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_unlock,
                (xdrproc_t)xdr_md_mn_set_and_class_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_unlock", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_suspend_2(argp, clnt, nid)
        md_mn_set_and_class_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_suspend,
                (xdrproc_t)xdr_md_mn_set_and_class_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_suspend", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_resume_2(argp, clnt, nid)
        md_mn_set_and_class_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_resume,
                (xdrproc_t)xdr_md_mn_set_and_class_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_resume", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_reinit_set_2(argp, clnt, nid)
        set_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_reinit_set,
                (xdrproc_t)xdr_set_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_reinit_set", nid);
        Free(clnt_res);
        return (NULL);
}

int *
mdmn_comm_msglock_2(argp, clnt, nid)
        md_mn_type_and_lock_t *argp;
        CLIENT *clnt;
        md_mn_nodeid_t nid;
{
        enum clnt_stat  res;
        int *clnt_res = Zalloc(sizeof (int));

        res = clnt_call(clnt, mdmn_comm_msglock,
                (xdrproc_t)xdr_md_mn_type_and_lock_t, (caddr_t)argp,
                (xdrproc_t)xdr_int, (caddr_t)clnt_res, TIMEOUT);

        if (res == RPC_SUCCESS) {
                return (clnt_res);
        }
        mdmn_handle_RPC_error(clnt, "mdmn_comm_msglock", nid);
        Free(clnt_res);
        return (NULL);
}


#define USECS_PER_TICK  10000


/*
 * Let the kernel create a clusterwide unique message ID
 *
 * returns 0 on success
 *         1 on failure
 */

int
mdmn_create_msgid(md_mn_msgid_t *msgid)
{
        md_error_t      mde = mdnullerror;

        if (msgid == NULL) {
                return (1); /* failure */
        }

        if (metaioctl(MD_IOCGUNIQMSGID, msgid, &mde, NULL) != 0) {
                msgid->mid_nid = ~0u;
                msgid->mid_time = 0LL;
                return (1); /* failure */
        }

        /*
         * mid_smid and mid_oclass are only used for submessages.
         * mdmn_create_msgid is never called for submessages, as they inherit
         * the message ID from their parent.
         * Thus we can safely null out the following fields.
         */
        msgid->mid_smid = 0;
        msgid->mid_oclass = 0;

        /* if the node_id is not set yet, somethings seems to be wrong */
        if (msgid->mid_nid == ~0u) {
                return (1); /* failure */
        }

        return (0); /* success */
}

md_mn_result_t *
copy_result(md_mn_result_t *res)
{
        md_mn_result_t *nres;
        nres = Zalloc(sizeof (md_mn_result_t));
        /* It's MSGID_COPY(from, to); */
        MSGID_COPY(&(res->mmr_msgid), &(nres->mmr_msgid));
        nres->mmr_msgtype       = res->mmr_msgtype;
        nres->mmr_setno         = res->mmr_setno;
        nres->mmr_flags         = res->mmr_flags;
        nres->mmr_sender        = res->mmr_sender;
        nres->mmr_failing_node  = res->mmr_failing_node;
        nres->mmr_comm_state    = res->mmr_comm_state;
        nres->mmr_exitval       = res->mmr_exitval;
        nres->mmr_out_size      = res->mmr_out_size;
        nres->mmr_err_size      = res->mmr_err_size;
        if (res->mmr_out_size > 0) {
                nres->mmr_out = Zalloc(res->mmr_out_size);
                bcopy(res->mmr_out, nres->mmr_out, res->mmr_out_size);
        }
        if (res->mmr_err_size > 0) {
                nres->mmr_err = Zalloc(res->mmr_err_size);
                bcopy(res->mmr_err, nres->mmr_err, res->mmr_err_size);
        }
        if (res->mmr_ep.host != '\0') {
                nres->mmr_ep.host = strdup(res->mmr_ep.host);
        }
        if (res->mmr_ep.extra != '\0') {
                nres->mmr_ep.extra = strdup(res->mmr_ep.extra);
        }
        if (res->mmr_ep.name != '\0') {
                nres->mmr_ep.name = strdup(res->mmr_ep.name);
        }
        return (nres);
}

void
free_result(md_mn_result_t *res)
{
        if (res->mmr_out_size > 0) {
                Free(res->mmr_out);
        }
        if (res->mmr_err_size > 0) {
                Free(res->mmr_err);
        }
        if (res->mmr_ep.host != '\0') {
                Free(res->mmr_ep.host);
        }
        if (res->mmr_ep.extra != '\0') {
                Free(res->mmr_ep.extra);
        }
        if (res->mmr_ep.name != '\0') {
                Free(res->mmr_ep.name);
        }
        Free(res);
}


/* allocate a new message and copy a given message into it */
md_mn_msg_t *
copy_msg(md_mn_msg_t *msg, md_mn_msg_t *dest)
{
        md_mn_msg_t *nmsg;

        nmsg = dest;

        if (nmsg == NULL) {
                nmsg = Zalloc(sizeof (md_mn_msg_t));
        }
        if (nmsg->msg_event_data == NULL) {
                nmsg->msg_event_data = Zalloc(msg->msg_event_size);
        }
        /* It's MSGID_COPY(from, to); */
        MSGID_COPY(&(msg->msg_msgid), &(nmsg->msg_msgid));
        nmsg->msg_sender        = msg->msg_sender;
        nmsg->msg_flags         = msg->msg_flags;
        nmsg->msg_setno         = msg->msg_setno;
        nmsg->msg_type          = msg->msg_type;
        nmsg->msg_recipient     = msg->msg_recipient;
        nmsg->msg_event_size    = msg->msg_event_size;
        if (msg->msg_event_size > 0) {
                bcopy(msg->msg_event_data, nmsg->msg_event_data,
                    msg->msg_event_size);
        }
        return (nmsg);
}

void
copy_msg_2(md_mn_msg_t *msg, md_mn_msg_od_t *msgod, int direction)
{
        assert((direction == MD_MN_COPY_TO_ONDISK) ||
            (direction == MD_MN_COPY_TO_INCORE));

        if (direction == MD_MN_COPY_TO_ONDISK) {
                MSGID_COPY(&(msg->msg_msgid), &(msgod->msg_msgid));
                msgod->msg_sender       = msg->msg_sender;
                msgod->msg_flags        = msg->msg_flags;
                msgod->msg_setno        = msg->msg_setno;
                msgod->msg_type         = msg->msg_type;
                msgod->msg_recipient    = msg->msg_recipient;
                msgod->msg_od_event_size = msg->msg_event_size;
                /* paranoid checks */
                if (msg->msg_event_size != 0 && msg->msg_event_data != NULL)
                        bcopy(msg->msg_event_data,
                            &msgod->msg_od_event_data[0], msg->msg_event_size);
        } else {
                MSGID_COPY(&(msgod->msg_msgid), &(msg->msg_msgid));
                msg->msg_sender = msgod->msg_sender;
                msg->msg_flags          = msgod->msg_flags;
                msg->msg_setno          = msgod->msg_setno;
                msg->msg_type           = msgod->msg_type;
                msg->msg_recipient      = msgod->msg_recipient;
                msg->msg_event_size     = msgod->msg_od_event_size;
                if (msg->msg_event_data == NULL)
                        msg->msg_event_data = Zalloc(msg->msg_event_size);

                bcopy(&msgod->msg_od_event_data[0],
                    msg->msg_event_data, msgod->msg_od_event_size);
        }
}

/* Free a message */
void
free_msg(md_mn_msg_t *msg)
{
        if (msg->msg_event_size > 0) {
                Free(msg->msg_event_data);
        }
        Free(msg);
}


/* The following declarations are only for the next two routines */

md_mn_client_list_t *mdmn_clients;

mutex_t mcl_mutex;
#define MNGLC_INIT_ONLY 0x0001
#define MNGLC_FOR_REAL  0x0002
/*
 * mdmn_get_local_clnt(flag)
 * If there is a client in the free pool, get one,
 * If no client is available, create one.
 * Every multithreaded application that uses mdmn_send_message must call it
 * single threaded first with special flags so we do the initialization
 * stuff in a safe environment.
 *
 * Input: MNGLC_INIT_ONLY: just initializes the mutex
 *        MNGLC_FOR_REAL : do real work
 * Output:
 *      An rpc client for sending rpc requests to the local commd
 *      NULL in case of an error
 *
 */
static CLIENT *
mdmn_get_local_clnt(uint_t flag)
{
        CLIENT *local_daemon;
        static int inited = 0;
        md_mn_client_list_t *tmp;

        if (inited == 0) {
                (void) mutex_init(&mcl_mutex, USYNC_THREAD, NULL);
                inited = 1;
        }

        if (flag == MNGLC_INIT_ONLY)
                return ((CLIENT *)NULL);

        (void) mutex_lock(&mcl_mutex);
        if (mdmn_clients == (md_mn_client_list_t *)NULL) {
                /* if there is no entry, create a client and return a it */
                local_daemon = meta_client_create(LOCALHOST_IPv4, MDMN_COMMD,
                    TWO, "tcp");
        } else {
                /*
                 * If there is an entry from a previous put operation,
                 * remove it from the head of the list and free the list stuff
                 * around it. Then return the client
                 */
                local_daemon = mdmn_clients->mcl_clnt;
                tmp = mdmn_clients;
                mdmn_clients = mdmn_clients->mcl_next;
                Free(tmp);
        }
        (void) mutex_unlock(&mcl_mutex);


        if (local_daemon == (CLIENT *)NULL) {
                clnt_pcreateerror("local_daemon");
        }

        return (local_daemon);
}

/*
 * mdmn_put_local_clnt()
 * returns a no longer used client to the pool
 *
 * Input: an RPC client
 * Output: void
 */
static void
mdmn_put_local_clnt(CLIENT *local_daemon)
{
        md_mn_client_list_t *tmp;

        (void) mutex_lock(&mcl_mutex);

        tmp =  mdmn_clients;
        mdmn_clients = (md_mn_client_list_t *)
            malloc(sizeof (md_mn_client_list_t));
        mdmn_clients->mcl_clnt = local_daemon;
        mdmn_clients->mcl_next = tmp;

        (void) mutex_unlock(&mcl_mutex);
}

/*
 * This is the regular interface for sending a message.
 * This function only passes through all arguments to
 * mdmn_send_message_with_msgid() and adds a NULL for the message ID.
 *
 * Normally, you don't have already a message ID for the message you want
 * to send.  Only in case of replaying a previously logged message,
 * a msgid is already attached to it.
 * In that case mdmn_send_message_with_msgid() has to be called directly.
 *
 * The recipient argument is almost always unused, and is therefore typically
 * set to zero, as zero is an invalid cluster nodeid.  The exceptions are the
 * marking and clearing of the DRL from a node that is not currently the
 * owner.  In these cases, the recipient argument will be the nodeid of the
 * mirror owner, and MD_MSGF_DIRECTED will be set in the flags.  Non-owner
 * nodes will not receive these messages.
 *
 * Return values / CAVEAT EMPTOR: see mdmn_send_message_with_msgid()
 */

int
mdmn_send_message(
                set_t setno,
                md_mn_msgtype_t type,
                uint_t flags,
                md_mn_nodeid_t recipient,
                char *data,
                int size,
                md_mn_result_t **result,
                md_error_t *ep)
{
        return (mdmn_send_message_with_msgid(setno, type, flags,
            recipient, data, size, result, MD_NULL_MSGID, ep));
}
/*
 * mdmn_send_message_with_msgid()
 * Create a message from the given pieces of data and hand it over
 * to the local commd.
 * This may fail for various reasons (rpc error / class busy / class locked ...)
 * Some error types are immediately deadly, others will cause retries
 * until the request is fulfilled or until the retries are ecxceeded.
 *
 * In case an error is returned it is up to the user to decide what to do.
 *
 * Returns:
 *      0 on success
 *      1 if retries1 exceeded
 *      2 if retries2 exceeded
 *      -1 if connecting to the local daemon failed
 *      -2 if the RPC call to the local daemon failed
 *      -3 if this node hasn't yet joined the set
 *      -4 if any other problem occured
 *
 * CAVEAT EMPTOR:
 *      The caller is responsible for calling free_result() when finished with
 *      the results!
 */
int
mdmn_send_message_with_msgid(
                set_t setno,
                md_mn_msgtype_t type,
                uint_t flags,
                md_mn_nodeid_t recipient,
                char *data,
                int size,
                md_mn_result_t **result,
                md_mn_msgid_t *msgid,
                md_error_t *ep)
{
        uint_t retry1, ticks1, retry2, ticks2;
        int retval;

        CLIENT *local_daemon;
        struct timeval timeout;

        md_mn_msg_t msg;
        md_mn_result_t *resp;

        /*
         * Special case for multithreaded applications:
         * When starting up, the application should call mdmn_send_message
         * single threaded with all parameters set to NULL.
         * When we detect this we know, we safely can do initialization
         * stuff here.
         * We only check for set and type being zero
         */
        if ((setno == 0) && (type == 0)) {
                /* do all needed initializations here */
                (void) mdmn_get_local_clnt(MNGLC_INIT_ONLY);
                return (0); /* success */
        }


        /* did the caller specify space to store the result pointer? */
        if (result == (md_mn_result_t **)NULL) {
                syslog(LOG_INFO, dgettext(TEXT_DOMAIN,
                    "FATAL, can not allocate result structure\n"));
                return (-4);
        }
        *result = NULL;

        /* Replay messages already have their msgID */
        if ((flags & MD_MSGF_REPLAY_MSG) == 0) {
                if (mdmn_create_msgid(&msg.msg_msgid) != 0) {
                        syslog(LOG_INFO, dgettext(TEXT_DOMAIN,
                            "FATAL, can not create message ID\n"));
                        return (-4);
                }
        } else {
                /* in this case a message ID must be specified */
                assert(msgid != MD_NULL_MSGID);
                MSGID_COPY(msgid, &msg.msg_msgid);
        }


        /*
         * When setting the flags, additionally apply the
         * default flags for this message type.
         */
        msg.msg_flags           = flags;
        msg.msg_setno           = setno;
        msg.msg_recipient       = recipient;
        msg.msg_type            = type;
        msg.msg_event_size      = size;
        msg.msg_event_data      = data;

        /*
         * For the timeout pick the specific timeout for the message times the
         * the maximum number of nodes.
         * This is a better estimate than 1 hour or 3 days or never.
         */
        timeout.tv_sec = mdmn_get_timeout(type) * NNODES;
        timeout.tv_usec = 0;

        if (flags & MD_MSGF_VERBOSE) {
                syslog(LOG_INFO, "send_message: ID=(%d, 0x%llx-%d)\n",
                    MSGID_ELEMS(msg.msg_msgid));
        }

        /* get an RPC client to the local commd */
        local_daemon = mdmn_get_local_clnt(MNGLC_FOR_REAL);
        if (local_daemon == (CLIENT *)NULL) {
                return (-1);
        }
        clnt_control(local_daemon, CLSET_TIMEOUT, (char *)&timeout);

        retry1 = msg_table[type].mte_retry1;
        ticks1 = msg_table[type].mte_ticks1;
        retry2 = msg_table[type].mte_retry2;
        ticks2 = msg_table[type].mte_ticks2;

        /*
         * run that loop until:
         * - commstate is Ok
         * - deadly commstate occured
         * - retries1 or retries2 exceeded
         */
        for (; ; ) {
                *result = mdmn_send_2(&msg, local_daemon, 0);
                resp = *result;
                if (resp != (md_mn_result_t *)NULL) {
                        /* Bingo! */
                        if (resp->mmr_comm_state == MDMNE_ACK) {
                                retval = 0;
                                goto out;
                        }
                        /* Hmm... what if there's no handler? */
                        if (resp->mmr_comm_state == MDMNE_NO_HANDLER) {
                                retval = 0;
                                goto out;

                        }
                        /*
                         * This node didn't yet join the disk set. It is not
                         * supposed to send any messages then.
                         * This is deadly (no retries)
                         */
                        if (resp->mmr_comm_state == MDMNE_NOT_JOINED) {
                                retval = -3;
                                goto out;

                        }
                        /* these two are deadly too (no retries) */
                        if ((resp->mmr_comm_state == MDMNE_NO_WAKEUP_ENTRY) ||
                            (resp->mmr_comm_state == MDMNE_LOG_FAIL)) {
                                retval = -4;
                                goto out;

                        }
                        /* Class busy? Use retry1 */
                        if (resp->mmr_comm_state == MDMNE_CLASS_BUSY) {
                                if (retry1-- == 0) {
                                        retval = 1; /* retry1 exceeded */
                                        goto out;
                                }
                                (void) usleep(ticks1 * USECS_PER_TICK);
                                free_result(resp);

                                if (flags & MD_MSGF_VERBOSE)
                                        (void) printf("#Resend1 ID=(%d, "
                                            "0x%llx-%d)\n",
                                            MSGID_ELEMS(msg.msg_msgid));
                                continue;
                        }
                        if ((resp->mmr_comm_state == MDMNE_CLASS_LOCKED) ||
                            (resp->mmr_comm_state == MDMNE_ABORT)) {
                                /*
                                 * Be patient, wait for 1 secs and try again.
                                 * It's not likely that the ABORT condition ever
                                 * goes away, but it won't hurt to retry
                                 */
                                free_result(resp);
                                (void) sleep(1);
                                continue;
                        }
                        if (resp->mmr_comm_state == MDMNE_SUSPENDED) {
                                if (flags & MD_MSGF_FAIL_ON_SUSPEND) {
                                        /* caller wants us to fail here */
                                        (void) mddserror(ep,
                                            MDE_DS_NOTNOW_RECONFIG, setno,
                                            mynode(), mynode(), NULL);
                                        retval = -4;
                                        goto out;
                                } else {
                                        /* wait for 1 secs and try again. */
                                        free_result(resp);
                                        (void) sleep(1);
                                        continue;
                                }
                        }
                } else {
                        /*
                         * If we get a NULL back from the rpc call, try to
                         * reinitialize the client.
                         * Depending on retries2 we try again, or not.
                         */
                        syslog(LOG_INFO,
                            "send_message: ID=(%d, 0x%llx-%d) resp = NULL\n",
                            MSGID_ELEMS(msg.msg_msgid));

                        clnt_destroy(local_daemon);
                        local_daemon = mdmn_get_local_clnt(MNGLC_FOR_REAL);

                        if (local_daemon == (CLIENT *)NULL) {
                                return (-1);
                        }
                        clnt_control(local_daemon, CLSET_TIMEOUT,
                            (char *)&timeout);
                }

                /*
                 * If we are here, either resp is zero or resp is non-zero
                 * but some commstate not mentioned above occured.
                 * In either case we use retry2
                 */
                if (retry2-- == 0) {
                        syslog(LOG_INFO, dgettext(TEXT_DOMAIN,
                            "send_message: (%d, 0x%llx-%d) retry2 exceeded\n"),
                            MSGID_ELEMS(msg.msg_msgid));

                        retval = 2; /* retry2 exceeded */
                        goto out;
                }
                if (flags & MD_MSGF_VERBOSE) {
                        syslog(LOG_DEBUG, dgettext(TEXT_DOMAIN,
                            "send_message: (%d, 0x%llx-%d) resend on retry2\n"),
                            MSGID_ELEMS(msg.msg_msgid));
                }

                (void) usleep(ticks2 * USECS_PER_TICK);

                if (resp != (md_mn_result_t *)NULL) {
                        free_result(resp);
                }
        }
out:
        mdmn_put_local_clnt(local_daemon);
        return (retval);
}

/*
 * suspend the commd for a given set/class combination.
 *
 * Parameter:
 *      set number or 0 (meaning all sets)
 *      class number or 0 (meaning all classes)
 *
 * Returns:
 *      0 on success (set is suspended and all messages drained)
 *      MDE_DS_COMMDCTL_SUSPEND_NYD if set is not yet drained
 *      MDE_DS_COMMDCTL_SUSPEND_FAIL if any failure occurred
 */
int
mdmn_suspend(set_t setno, md_mn_msgclass_t class, long timeout)
{
        int                     *resp;
        CLIENT                  *local_daemon;
        md_mn_set_and_class_t   msc;
        md_error_t              xep = mdnullerror;

        if ((setno >= MD_MAXSETS) || (class >= MD_MN_NCLASSES)) {
                return (MDE_DS_COMMDCTL_SUSPEND_FAIL);
        }
        local_daemon = meta_client_create(LOCALHOST_IPv4, MDMN_COMMD, TWO,
            "tcp");
        if (local_daemon == (CLIENT *)NULL) {
                clnt_pcreateerror("local_daemon");
                return (MDE_DS_COMMDCTL_SUSPEND_FAIL);
        }

        if (timeout != 0) {
                if (cl_sto(local_daemon, LOCALHOST_IPv4, timeout, &xep) != 0) {
                        clnt_destroy(local_daemon);
                        return (1);
                }
        }

        msc.msc_set = setno;
        msc.msc_class = class;
        msc.msc_flags = 0;

        resp = mdmn_comm_suspend_2(&msc, local_daemon, 0);
        clnt_destroy(local_daemon);

        if (resp == NULL) {
                return (MDE_DS_COMMDCTL_SUSPEND_FAIL);
        }

        if (*resp == MDMNE_ACK) {
                /* set successfully drained, no outstanding messages */
                return (0);
        }
        if (*resp != MDMNE_SET_NOT_DRAINED) {
                /* some error occurred */
                return (MDE_DS_COMMDCTL_SUSPEND_FAIL);
        }

        /* still outstanding messages, return not yet drained failure */
        return (MDE_DS_COMMDCTL_SUSPEND_NYD);
}

/*
 * resume the commd for a given set/class combination.
 *
 * Parameter:
 *      set number or 0 (meaning all sets)
 *      class number or 0 (meaning all classes)
 *
 * Returns:
 *      0 on success
 *      MDE_DS_COMMDCTL_RESUME_FAIL on failure
 */
int
mdmn_resume(set_t setno, md_mn_msgclass_t class, uint_t flags, long timeout)
{
        md_mn_set_and_class_t   msc;
        int                     ret = MDE_DS_COMMDCTL_RESUME_FAIL;
        int                     *resp;
        CLIENT                  *local_daemon;
        md_error_t              xep = mdnullerror;

        if ((setno >= MD_MAXSETS) || (class >= MD_MN_NCLASSES)) {
                return (MDE_DS_COMMDCTL_RESUME_FAIL);
        }
        local_daemon = meta_client_create(LOCALHOST_IPv4, MDMN_COMMD, TWO,
            "tcp");
        if (local_daemon == (CLIENT *)NULL) {
                clnt_pcreateerror("local_daemon");
                return (MDE_DS_COMMDCTL_RESUME_FAIL);
        }

        if (timeout != 0) {
                if (cl_sto(local_daemon, LOCALHOST_IPv4, timeout, &xep) != 0) {
                        clnt_destroy(local_daemon);
                        return (1);
                }
        }

        msc.msc_set = setno;
        msc.msc_class = class;
        msc.msc_flags = flags;

        resp = mdmn_comm_resume_2(&msc, local_daemon, 0);

        if (resp != NULL) {
                if (*resp == MDMNE_ACK) {
                        ret = 0;
                }
                Free(resp);
        }

        clnt_destroy(local_daemon);
        return (ret);
}

/*
 * abort all communication
 *
 * returns void, because: if *this* get's an error what do you want to do?
 */
void
mdmn_abort(void)
{
        char *dummy = "abort";
        md_mn_result_t  *resultp = NULL;
        md_error_t      mdne = mdnullerror;

        (void) mdmn_send_message(0, /* No set is needed for this message */
            MD_MN_MSG_ABORT, MD_MSGF_LOCAL_ONLY, 0,
            dummy, sizeof (dummy), &resultp, &mdne);

        if (resultp != NULL) {
                Free(resultp);
        }
}

/*
 * trigger the reinitialization for a given set.
 *
 * Parameter: set number
 *
 * Returns:
 *      0 on success
 *      1 on failure
 */
int
mdmn_reinit_set(set_t setno, long timeout)
{
        int             ret = 1;
        int             *resp;
        CLIENT          *local_daemon;
        md_error_t      xep = mdnullerror;

        if ((setno == 0) || (setno >= MD_MAXSETS)) {
                return (1);
        }
        local_daemon = meta_client_create(LOCALHOST_IPv4, MDMN_COMMD, TWO,
            "tcp");
        if (local_daemon == (CLIENT *)NULL) {
                clnt_pcreateerror("local_daemon");
                return (1);
        }

        if (timeout != 0) {
                if (cl_sto(local_daemon, LOCALHOST_IPv4, timeout, &xep) != 0) {
                        clnt_destroy(local_daemon);
                        return (1);
                }
        }

        resp = mdmn_comm_reinit_set_2(&setno, local_daemon, 0);

        if (resp != NULL) {
                if (*resp == MDMNE_ACK) {
                        ret = 0;
                }
                Free(resp);
        }

        clnt_destroy(local_daemon);
        return (ret);
}


/*
 * Lock a single message type from being processed on this node
 *
 * Parameter: md_mn_msgtype_t msgtype, uint_t locktype
 *
 * Returns:
 *      0 on success
 *      1 on failure
 */
int
mdmn_msgtype_lock(md_mn_msgtype_t msgtype, uint_t locktype)
{
        int                     ret = 1;
        int                     *resp;
        CLIENT                  *local_daemon;
        md_mn_type_and_lock_t   mmtl;


        if ((msgtype == 0) || (msgtype >= MD_MN_NMESSAGES)) {
                return (1);
        }
        local_daemon = meta_client_create(LOCALHOST_IPv4, MDMN_COMMD, TWO,
            "tcp");
        if (local_daemon == (CLIENT *)NULL) {
                clnt_pcreateerror("local_daemon");
                return (1);
        }
        mmtl.mmtl_type = msgtype;
        mmtl.mmtl_lock = locktype;

        resp = mdmn_comm_msglock_2(&mmtl, local_daemon, 0);

        if (resp != NULL) {
                if (*resp == MDMNE_ACK) {
                        ret = 0;
                }
                Free(resp);
        }

        clnt_destroy(local_daemon);
        return (ret);
}