/*!
 * \internal
 * \brief Send an XML message with process list of all known peers to client(s)
 *
 * \param[in] client  Send message to this client, or all clients if NULL
 */
void
update_process_clients(crm_client_t *client)
{
    GHashTableIter iter;
    crm_node_t *node = NULL;
    xmlNode *update = create_xml_node(NULL, "nodes");

    if (is_corosync_cluster()) {
        crm_xml_add_int(update, "quorate", pcmk_quorate);
    }

    g_hash_table_iter_init(&iter, crm_peer_cache);
    while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & node)) {
        xmlNode *xml = create_xml_node(update, "node");

        crm_xml_add_int(xml, "id", node->id);
        crm_xml_add(xml, "uname", node->uname);
        crm_xml_add(xml, "state", node->state);
        crm_xml_add_int(xml, "processes", node->processes);
    }

    if(client) {
        crm_trace("Sending process list to client %s", client->id);
        crm_ipcs_send(client, 0, update, crm_ipc_server_event);

    } else {
        crm_trace("Sending process list to %d clients", crm_hash_table_size(client_connections));
        g_hash_table_iter_init(&iter, client_connections);
        while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & client)) {
            crm_ipcs_send(client, 0, update, crm_ipc_server_event);
        }
    }

    free_xml(update);
}

static int32_t
ipc_proxy_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
{
    uint32_t id = 0;
    uint32_t flags = 0;
    crm_client_t *client = crm_client_get(c);
    crm_client_t *ipc_proxy = crm_client_get_by_id(client->userdata);
    xmlNode *request = NULL;
    xmlNode *msg = NULL;

    if (!ipc_proxy) {
        qb_ipcs_disconnect(client->ipcs);
        return 0;
    }

    /* This is a request from the local ipc client going
     * to the ipc provider.
     *
     * Looking at the chain of events.
     *
     * -----remote node----------------|---- cluster node ------
     * ipc_client <--1--> this code <--2--> crmd:remote_proxy_dispatch_internal() <----3----> ipc server
     *
     * This function is receiving a request from connection
     * 1 and forwarding it to connection 2.
     */
    request = crm_ipcs_recv(client, data, size, &id, &flags);

    if (!request) {
        return 0;
    }

    CRM_CHECK(client != NULL, crm_err("Invalid client");
              return FALSE);
    CRM_CHECK(client->id != NULL, crm_err("Invalid client: %p", client);
              return FALSE);

    /* this ensures that synced request/responses happen over the event channel
     * in the crmd, allowing the crmd to process the messages async */
    set_bit(flags, crm_ipc_proxied);
    client->request_id = id;

    msg = create_xml_node(NULL, T_LRMD_IPC_PROXY);
    crm_xml_add(msg, F_LRMD_IPC_OP, "request");
    crm_xml_add(msg, F_LRMD_IPC_SESSION, client->id);
    crm_xml_add(msg, F_LRMD_IPC_CLIENT, crm_client_name(client));
    crm_xml_add(msg, F_LRMD_IPC_USER, client->user);
    crm_xml_add_int(msg, F_LRMD_IPC_MSG_ID, id);
    crm_xml_add_int(msg, F_LRMD_IPC_MSG_FLAGS, flags);
    add_message_xml(msg, F_LRMD_IPC_MSG, request);
    lrmd_server_send_notify(ipc_proxy, msg);
    free_xml(msg);

    return 0;
}

int
stonith_fence_history(xmlNode * msg, xmlNode ** output)
{
    int rc = 0;
    const char *target = NULL;
    xmlNode *dev = get_xpath_object("//@" F_STONITH_TARGET, msg, LOG_TRACE);

    if (dev) {
        int options = 0;

        target = crm_element_value(dev, F_STONITH_TARGET);
        crm_element_value_int(msg, F_STONITH_CALLOPTS, &options);
        if (target && (options & st_opt_cs_nodeid)) {
            int nodeid = crm_atoi(target, NULL);
            crm_node_t *node = crm_get_peer(nodeid, NULL);

            if (node) {
                target = node->uname;
            }
        }
    }

    crm_trace("Looking for operations on %s in %p", target, remote_op_list);

    *output = create_xml_node(NULL, F_STONITH_HISTORY_LIST);
    if (remote_op_list) {
        GHashTableIter iter;
        remote_fencing_op_t *op = NULL;

        g_hash_table_iter_init(&iter, remote_op_list);
        while (g_hash_table_iter_next(&iter, NULL, (void **)&op)) {
            xmlNode *entry = NULL;

            if (target && strcmp(op->target, target) != 0) {
                continue;
            }

            rc = 0;
            crm_trace("Attaching op %s", op->id);
            entry = create_xml_node(*output, STONITH_OP_EXEC);
            crm_xml_add(entry, F_STONITH_TARGET, op->target);
            crm_xml_add(entry, F_STONITH_ACTION, op->action);
            crm_xml_add(entry, F_STONITH_ORIGIN, op->originator);
            crm_xml_add(entry, F_STONITH_DELEGATE, op->delegate);
            crm_xml_add(entry, F_STONITH_CLIENTNAME, op->client_name);
            crm_xml_add_int(entry, F_STONITH_DATE, op->completed);
            crm_xml_add_int(entry, F_STONITH_STATE, op->state);
        }
    }

    return rc;
}

static void
throttle_send_command(enum throttle_state_e mode)
{
    xmlNode *xml = NULL;

    xml = create_request(CRM_OP_THROTTLE, NULL, NULL, CRM_SYSTEM_CRMD, CRM_SYSTEM_CRMD, NULL);
    crm_xml_add_int(xml, F_CRM_THROTTLE_MODE, mode);
    crm_xml_add_int(xml, F_CRM_THROTTLE_MAX, throttle_job_max);

    send_cluster_message(NULL, crm_msg_crmd, xml, TRUE);
    free_xml(xml);

    crm_info("Updated throttle state to %.4x", mode);
}

/*!
 * \internal
 * \brief Craft xml difference between local fence-history and a history
 *        coming from remote
 *
 * \param[in] remote_history    Fence-history as hash-table (may be NULL)
 * \param[in] add_id            If crafting the answer for an API
 *                              history-request there is no need for the id
 * \param[in] target            Optionally limit to certain fence-target
 *
 * \return The fence-history as xml
 */
static xmlNode *
stonith_local_history_diff(GHashTable *remote_history,
                           gboolean add_id,
                           const char *target)
{
    xmlNode *history = NULL;
    int cnt = 0;

    if (stonith_remote_op_list) {
            GHashTableIter iter;
            remote_fencing_op_t *op = NULL;

            history = create_xml_node(NULL, F_STONITH_HISTORY_LIST);

            g_hash_table_iter_init(&iter, stonith_remote_op_list);
            while (g_hash_table_iter_next(&iter, NULL, (void **)&op)) {
                xmlNode *entry = NULL;

                if (remote_history &&
                    g_hash_table_lookup(remote_history, op->id)) {
                    continue; /* skip entries broadcasted already */
                }

                if (target && strcmp(op->target, target) != 0) {
                    continue;
                }

                cnt++;
                crm_trace("Attaching op %s", op->id);
                entry = create_xml_node(history, STONITH_OP_EXEC);
                if (add_id) {
                    crm_xml_add(entry, F_STONITH_REMOTE_OP_ID, op->id);
                }
                crm_xml_add(entry, F_STONITH_TARGET, op->target);
                crm_xml_add(entry, F_STONITH_ACTION, op->action);
                crm_xml_add(entry, F_STONITH_ORIGIN, op->originator);
                crm_xml_add(entry, F_STONITH_DELEGATE, op->delegate);
                crm_xml_add(entry, F_STONITH_CLIENTNAME, op->client_name);
                crm_xml_add_int(entry, F_STONITH_DATE, op->completed);
                crm_xml_add_int(entry, F_STONITH_STATE, op->state);
            }
    }

    if (cnt == 0) {
        free_xml(history);
        return NULL;
    } else {
        return history;
    }
}

/*!
 * \brief Send a request to pacemaker-attrd to clear resource failure
 *
 * \param[in] ipc           Connection to pacemaker-attrd (NULL to use local connection)
 * \param[in] host          Affect only this host (or NULL for all hosts)
 * \param[in] resource      Name of resource to clear (or NULL for all)
 * \param[in] operation     Name of operation to clear (or NULL for all)
 * \param[in] interval_spec If operation is not NULL, its interval
 * \param[in] user_name     ACL user to pass to pacemaker-attrd
 * \param[in] options       attrd_opt_remote if host is a Pacemaker Remote node
 *
 * \return pcmk_ok if request was successfully submitted to pacemaker-attrd, else -errno
 */
int
attrd_clear_delegate(crm_ipc_t *ipc, const char *host, const char *resource,
                     const char *operation, const char *interval_spec,
                     const char *user_name, int options)
{
    int rc = pcmk_ok;
    xmlNode *clear_op = create_attrd_op(user_name);
    const char *interval_desc = NULL;
    const char *op_desc = NULL;

    crm_xml_add(clear_op, F_ATTRD_TASK, ATTRD_OP_CLEAR_FAILURE);
    crm_xml_add(clear_op, F_ATTRD_HOST, host);
    crm_xml_add(clear_op, F_ATTRD_RESOURCE, resource);
    crm_xml_add(clear_op, F_ATTRD_OPERATION, operation);
    crm_xml_add(clear_op, F_ATTRD_INTERVAL, interval_spec);
    crm_xml_add_int(clear_op, F_ATTRD_IS_REMOTE, is_set(options, attrd_opt_remote));

    rc = send_attrd_op(ipc, clear_op);
    free_xml(clear_op);

    if (operation) {
        interval_desc = interval_spec? interval_spec : "nonrecurring";
        op_desc = operation;
    } else {
        interval_desc = "all";
        op_desc = "operations";
    }
    crm_debug("Asked pacemaker-attrd to clear failure of %s %s for %s on %s: %s (%d)",
              interval_desc, op_desc, (resource? resource : "all resources"),
              (host? host : "all nodes"), pcmk_strerror(rc), rc);
    return rc;
}

void
attrd_peer_sync(crm_node_t *peer, xmlNode *xml)
{
    GHashTableIter aIter;
    GHashTableIter vIter;

    attribute_t *a = NULL;
    attribute_value_t *v = NULL;
    xmlNode *sync = create_xml_node(NULL, __FUNCTION__);

    crm_xml_add(sync, F_ATTRD_TASK, "sync-response");

    g_hash_table_iter_init(&aIter, attributes);
    while (g_hash_table_iter_next(&aIter, NULL, (gpointer *) & a)) {
        g_hash_table_iter_init(&vIter, a->values);
        while (g_hash_table_iter_next(&vIter, NULL, (gpointer *) & v)) {
            crm_debug("Syncing %s[%s] = %s to %s", a->id, v->nodename, v->current, peer?peer->uname:"everyone");
            build_attribute_xml(sync, a->id, a->set, a->uuid, a->timeout_ms, a->user, v->nodename, v->nodeid, v->current);
        }
    }

    crm_debug("Syncing values to %s", peer?peer->uname:"everyone");
    crm_xml_add_int(sync, F_ATTRD_WRITER, election_state(writer));
    send_cluster_message(peer, crm_msg_attrd, sync, TRUE);
    free_xml(sync);
}

void
do_stonith_async_timeout_update(const char *client_id, const char *call_id, int timeout)
{
    crm_client_t *client = NULL;
    xmlNode *notify_data = NULL;

    if (!timeout || !call_id || !client_id) {
        return;
    }

    client = crm_client_get_by_id(client_id);
    if (!client) {
        return;
    }

    notify_data = create_xml_node(NULL, T_STONITH_TIMEOUT_VALUE);
    crm_xml_add(notify_data, F_TYPE, T_STONITH_TIMEOUT_VALUE);
    crm_xml_add(notify_data, F_STONITH_CALLID, call_id);
    crm_xml_add_int(notify_data, F_STONITH_TIMEOUT, timeout);

    crm_trace("timeout update is %d for client %s and call id %s", timeout, client_id, call_id);

    if (client) {
        crm_ipcs_send(client, 0, notify_data, crm_ipc_server_event);
    }

    free_xml(notify_data);
}

static void
handle_local_reply_and_notify(remote_fencing_op_t * op, xmlNode * data, int rc)
{
    xmlNode *notify_data = NULL;
    xmlNode *reply = NULL;

    if (op->notify_sent == TRUE) {
        /* nothing to do */
        return;
    }

    /* Do notification with a clean data object */
    notify_data = create_op_done_notify(op, rc);
    crm_xml_add_int(data, "state", op->state);
    crm_xml_add(data, F_STONITH_TARGET, op->target);
    crm_xml_add(data, F_STONITH_OPERATION, op->action);

    reply = stonith_construct_reply(op->request, NULL, data, rc);
    crm_xml_add(reply, F_STONITH_DELEGATE, op->delegate);

    /* Send fencing OP reply to local client that initiated fencing */
    do_local_reply(reply, op->client_id, op->call_options & st_opt_sync_call, FALSE);

    /* bcast to all local clients that the fencing operation happend */
    do_stonith_notify(0, T_STONITH_NOTIFY_FENCE, rc, notify_data);

    /* mark this op as having notify's already sent */
    op->notify_sent = TRUE;
    free_xml(reply);
    free_xml(notify_data);
}

xmlNode *
cib_construct_reply(xmlNode * request, xmlNode * output, int rc)
{
    int lpc = 0;
    xmlNode *reply = NULL;
    const char *name = NULL;
    const char *value = NULL;

    const char *names[] = {
        F_CIB_OPERATION,
        F_CIB_CALLID,
        F_CIB_CLIENTID,
        F_CIB_CALLOPTS
    };
    static int max = DIMOF(names);

    crm_trace("Creating a basic reply");
    reply = create_xml_node(NULL, "cib-reply");
    crm_xml_add(reply, F_TYPE, T_CIB);

    for (lpc = 0; lpc < max; lpc++) {
        name = names[lpc];
        value = crm_element_value(request, name);
        crm_xml_add(reply, name, value);
    }

    crm_xml_add_int(reply, F_CIB_RC, rc);

    if (output != NULL) {
        crm_trace("Attaching reply output");
        add_message_xml(reply, F_CIB_CALLDATA, output);
    }
    return reply;
}

static int
lrmd_remote_client_msg(gpointer data)
{
    int id = 0;
    int rc = 0;
    int disconnected = 0;
    xmlNode *request = NULL;
    crm_client_t *client = data;

    if (client->remote->tls_handshake_complete == FALSE) {
        return remoted__read_handshake_data(client);
    }

    rc = crm_remote_ready(client->remote, 0);
    if (rc == 0) {
        /* no msg to read */
        return 0;
    } else if (rc < 0) {
        crm_info("Remote client disconnected while polling it");
        return -1;
    }

    crm_remote_recv(client->remote, -1, &disconnected);

    request = crm_remote_parse_buffer(client->remote);
    while (request) {
        crm_element_value_int(request, F_LRMD_REMOTE_MSG_ID, &id);
        crm_trace("Processing remote client request %d", id);
        if (!client->name) {
            const char *value = crm_element_value(request, F_LRMD_CLIENTNAME);

            if (value) {
                client->name = strdup(value);
            }
        }

        lrmd_call_id++;
        if (lrmd_call_id < 1) {
            lrmd_call_id = 1;
        }

        crm_xml_add(request, F_LRMD_CLIENTID, client->id);
        crm_xml_add(request, F_LRMD_CLIENTNAME, client->name);
        crm_xml_add_int(request, F_LRMD_CALLID, lrmd_call_id);

        process_lrmd_message(client, id, request);
        free_xml(request);

        /* process all the messages in the current buffer */
        request = crm_remote_parse_buffer(client->remote);
    }

    if (disconnected) {
        crm_info("Remote client disconnected while reading from it");
        return -1;
    }

    return 0;
}

static xmlNode *
create_op_done_notify(remote_fencing_op_t * op, int rc)
{
    xmlNode *notify_data = create_xml_node(NULL, T_STONITH_NOTIFY_FENCE);

    crm_xml_add_int(notify_data, "state", op->state);
    crm_xml_add_int(notify_data, F_STONITH_RC, rc);
    crm_xml_add(notify_data, F_STONITH_TARGET, op->target);
    crm_xml_add(notify_data, F_STONITH_ACTION, op->action);
    crm_xml_add(notify_data, F_STONITH_DELEGATE, op->delegate);
    crm_xml_add(notify_data, F_STONITH_REMOTE_OP_ID, op->id);
    crm_xml_add(notify_data, F_STONITH_ORIGIN, op->originator);
    crm_xml_add(notify_data, F_STONITH_CLIENTID, op->client_id);
    crm_xml_add(notify_data, F_STONITH_CLIENTNAME, op->client_name);

    return notify_data;
}

static void
cib_handle_remote_msg(crm_client_t * client, xmlNode * command)
{
    const char *value = NULL;

    value = crm_element_name(command);
    if (safe_str_neq(value, "cib_command")) {
        crm_log_xml_trace(command, "Bad command: ");
        return;
    }

    if (client->name == NULL) {
        value = crm_element_value(command, F_CLIENTNAME);
        if (value == NULL) {
            client->name = strdup(client->id);
        } else {
            client->name = strdup(value);
        }
    }

    if (client->userdata == NULL) {
        value = crm_element_value(command, F_CIB_CALLBACK_TOKEN);
        if (value != NULL) {
            client->userdata = strdup(value);
            crm_trace("Callback channel for %s is %s", client->id, (char*)client->userdata);

        } else {
            client->userdata = strdup(client->id);
        }
    }

    /* unset dangerous options */
    xml_remove_prop(command, F_ORIG);
    xml_remove_prop(command, F_CIB_HOST);
    xml_remove_prop(command, F_CIB_GLOBAL_UPDATE);

    crm_xml_add(command, F_TYPE, T_CIB);
    crm_xml_add(command, F_CIB_CLIENTID, client->id);
    crm_xml_add(command, F_CIB_CLIENTNAME, client->name);
#if ENABLE_ACL
    crm_xml_add(command, F_CIB_USER, client->user);
#endif

    if (crm_element_value(command, F_CIB_CALLID) == NULL) {
        char *call_uuid = crm_generate_uuid();

        /* fix the command */
        crm_xml_add(command, F_CIB_CALLID, call_uuid);
        free(call_uuid);
    }

    if (crm_element_value(command, F_CIB_CALLOPTS) == NULL) {
        crm_xml_add_int(command, F_CIB_CALLOPTS, 0);
    }

    crm_log_xml_trace(command, "Remote command: ");
    cib_common_callback_worker(0, 0, command, client, TRUE);
}

xmlNode *
build_attribute_xml(
    xmlNode *parent, const char *name, const char *set, const char *uuid, unsigned int timeout_ms, const char *user,
    const char *peer, uint32_t peerid, const char *value)
{
    xmlNode *xml = create_xml_node(parent, __FUNCTION__);

    crm_xml_add(xml, F_ATTRD_ATTRIBUTE, name);
    crm_xml_add(xml, F_ATTRD_SET, set);
    crm_xml_add(xml, F_ATTRD_KEY, uuid);
    crm_xml_add(xml, F_ATTRD_USER, user);
    crm_xml_add(xml, F_ATTRD_HOST, peer);
    crm_xml_add_int(xml, F_ATTRD_HOST_ID, peerid);
    crm_xml_add(xml, F_ATTRD_VALUE, value);
    crm_xml_add_int(xml, F_ATTRD_DAMPEN, timeout_ms/1000);

    return xml;
}

static void
throttle_send_command(enum throttle_state_e mode)
{
    xmlNode *xml = NULL;
    static enum throttle_state_e last = -1;

    if(mode != last) {
        crm_info("New throttle mode: %.4x (was %.4x)", mode, last);
        last = mode;

        xml = create_request(CRM_OP_THROTTLE, NULL, NULL, CRM_SYSTEM_CRMD, CRM_SYSTEM_CRMD, NULL);
        crm_xml_add_int(xml, F_CRM_THROTTLE_MODE, mode);
        crm_xml_add_int(xml, F_CRM_THROTTLE_MAX, throttle_job_max);

        send_cluster_message(NULL, crm_msg_crmd, xml, TRUE);
        free_xml(xml);
    }
}

void
master_append_meta(resource_t * rsc, xmlNode * xml)
{
    char *name = NULL;
    clone_variant_data_t *clone_data = NULL;

    get_clone_variant_data(clone_data, rsc);

    clone_append_meta(rsc, xml);

    name = crm_meta_name(XML_RSC_ATTR_MASTER_MAX);
    crm_xml_add_int(xml, name, clone_data->master_max);
    crm_free(name);

    name = crm_meta_name(XML_RSC_ATTR_MASTER_NODEMAX);
    crm_xml_add_int(xml, name, clone_data->master_node_max);
    crm_free(name);
}

static gboolean
send_attrd_message(crm_node_t * node, xmlNode * data)
{
    crm_xml_add(data, F_TYPE, T_ATTRD);
    crm_xml_add(data, F_ATTRD_IGNORE_LOCALLY, "atomic-version"); /* Tell older versions to ignore our messages */
    crm_xml_add(data, F_ATTRD_VERSION, ATTRD_PROTOCOL_VERSION);
    crm_xml_add_int(data, F_ATTRD_WRITER, election_state(writer));

    return send_cluster_message(node, crm_msg_attrd, data, TRUE);
}

gboolean
finalize_join_for(gpointer key, gpointer value, gpointer user_data)
{
    const char *join_to = NULL;
    const char *join_state = NULL;
    xmlNode *acknak = NULL;
    crm_node_t *join_node = NULL;

    if (key == NULL || value == NULL) {
        return TRUE;
    }

    join_to = (const char *)key;
    join_state = (const char *)value;

    /* make sure the node exists in the config section */
    create_node_entry(join_to, join_to, NORMALNODE);

    join_node = crm_get_peer(0, join_to);
    if (crm_is_peer_active(join_node) == FALSE) {
        /*
         * NACK'ing nodes that the membership layer doesn't know about yet
         * simply creates more churn
         *
         * Better to leave them waiting and let the join restart when
         * the new membership event comes in
         *
         * All other NACKs (due to versions etc) should still be processed
         */
        return TRUE;
    }

    /* send the ack/nack to the node */
    acknak = create_request(CRM_OP_JOIN_ACKNAK, NULL, join_to,
                            CRM_SYSTEM_CRMD, CRM_SYSTEM_DC, NULL);
    crm_xml_add_int(acknak, F_CRM_JOIN_ID, current_join_id);

    /* set the ack/nack */
    if (safe_str_eq(join_state, CRMD_JOINSTATE_MEMBER)) {
        crm_debug("join-%d: ACK'ing join request from %s, state %s",
                  current_join_id, join_to, join_state);
        crm_xml_add(acknak, CRM_OP_JOIN_ACKNAK, XML_BOOLEAN_TRUE);
        g_hash_table_insert(finalized_nodes,
                            strdup(join_to), strdup(CRMD_JOINSTATE_MEMBER));
    } else {
        crm_warn("join-%d: NACK'ing join request from %s, state %s",
                 current_join_id, join_to, join_state);

        crm_xml_add(acknak, CRM_OP_JOIN_ACKNAK, XML_BOOLEAN_FALSE);
    }

    send_cluster_message(join_to, crm_msg_crmd, acknak, TRUE);
    free_xml(acknak);
    return TRUE;
}

/* ELECTION投票処理 */
void
election_vote(election_t *e)
{
    struct timeval age;
    xmlNode *vote = NULL;
    crm_node_t *our_node;

    if(e == NULL) {
        crm_trace("Not voting in election: not initialized");
        return;
    }

    our_node = crm_get_peer(0, e->uname);
    if (our_node == NULL || crm_is_peer_active(our_node) == FALSE) {
        crm_trace("Cannot vote yet: %p", our_node);
        return;
    }
	/* 自ノードの情報を生成する */
    e->state = election_in_progress;
    /* CRM_OP_VOTE(F_ATTRD_TASKフィールド)メッセージを生成する */
    vote = create_request(CRM_OP_VOTE, NULL, NULL, CRM_SYSTEM_CRMD, CRM_SYSTEM_CRMD, NULL);

    e->count++;
    crm_xml_add(vote, F_CRM_ELECTION_OWNER, our_node->uuid);
    crm_xml_add_int(vote, F_CRM_ELECTION_ID, e->count);

    crm_uptime(&age);
    crm_xml_add_int(vote, F_CRM_ELECTION_AGE_S, age.tv_sec);
    crm_xml_add_int(vote, F_CRM_ELECTION_AGE_US, age.tv_usec);
	/* クラスタに投票メッセージを送信する */
    send_cluster_message(NULL, crm_msg_crmd, vote, TRUE);
    free_xml(vote);

    crm_debug("Started election %d", e->count);
    if (e->voted) {
        g_hash_table_destroy(e->voted);
        e->voted = NULL;
    }
	/* ELECTIONのタイマーを開始する */
    election_timeout_start(e);
    return;
}

void
finalize_join_for(gpointer key, gpointer value, gpointer user_data)
{
    xmlNode *acknak = NULL;
    xmlNode *tmp1 = NULL;
    crm_node_t *join_node = value;
    const char *join_to = join_node->uname;

    if(join_node->join != crm_join_integrated) {
        crm_trace("Skipping %s in state %d", join_to, join_node->join);
        return;
    }

    /* make sure a node entry exists for the new node */
    crm_trace("Creating node entry for %s", join_to);

    tmp1 = create_xml_node(NULL, XML_CIB_TAG_NODE);
    set_uuid(tmp1, XML_ATTR_UUID, join_node);
    crm_xml_add(tmp1, XML_ATTR_UNAME, join_to);

    fsa_cib_anon_update(XML_CIB_TAG_NODES, tmp1,
                        cib_scope_local | cib_quorum_override | cib_can_create);
    free_xml(tmp1);

    join_node = crm_get_peer(0, join_to);
    if (crm_is_peer_active(join_node) == FALSE) {
        /*
         * NACK'ing nodes that the membership layer doesn't know about yet
         * simply creates more churn
         *
         * Better to leave them waiting and let the join restart when
         * the new membership event comes in
         *
         * All other NACKs (due to versions etc) should still be processed
         */
        crm_update_peer_expected(__FUNCTION__, join_node, CRMD_JOINSTATE_PENDING);
        return;
    }

    /* send the ack/nack to the node */
    acknak = create_request(CRM_OP_JOIN_ACKNAK, NULL, join_to,
                            CRM_SYSTEM_CRMD, CRM_SYSTEM_DC, NULL);
    crm_xml_add_int(acknak, F_CRM_JOIN_ID, current_join_id);

    crm_debug("join-%d: ACK'ing join request from %s",
              current_join_id, join_to);
    crm_xml_add(acknak, CRM_OP_JOIN_ACKNAK, XML_BOOLEAN_TRUE);
    crm_update_peer_join(__FUNCTION__, join_node, crm_join_finalized);
    crm_update_peer_expected(__FUNCTION__, join_node, CRMD_JOINSTATE_MEMBER);

    send_cluster_message(crm_get_peer(0, join_to), crm_msg_crmd, acknak, TRUE);
    free_xml(acknak);
    return;
}