/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * vote.c
 *
 * description here
 *
 * Copyright (C) 2003, 2004 Oracle.  All rights reserved.
 *
 * This program 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 2 of the License, or (at your option) any later version.
 *
 * This program 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, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 */

#include <linux/types.h>
#include <linux/slab.h>
#include <linux/highmem.h>
#include <linux/kthread.h>

#include <cluster/heartbeat.h>
#include <cluster/nodemanager.h>
#include <cluster/tcp.h>

#include <dlm/dlmapi.h>

#define MLOG_MASK_PREFIX ML_VOTE
#include <cluster/masklog.h>

#include "ocfs2.h"

#include "alloc.h"
#include "dlmglue.h"
#include "extent_map.h"
#include "heartbeat.h"
#include "inode.h"
#include "journal.h"
#include "slot_map.h"
#include "vote.h"

#include "buffer_head_io.h"

#define OCFS2_MESSAGE_TYPE_VOTE     (0x1)
#define OCFS2_MESSAGE_TYPE_RESPONSE (0x2)
struct ocfs2_msg_hdr
{
        __be32 h_response_id; /* used to lookup message handle on sending
                            * node. */
        __be32 h_request;
        __be64 h_blkno;
        __be32 h_generation;
        __be32 h_node_num;    /* node sending this particular message. */
};

struct ocfs2_vote_msg
{
        struct ocfs2_msg_hdr v_hdr;
        __be32 v_reserved1;
} __attribute__ ((packed));

/* Responses are given these values to maintain backwards
 * compatibility with older ocfs2 versions */
#define OCFS2_RESPONSE_OK               (0)
#define OCFS2_RESPONSE_BUSY             (-16)
#define OCFS2_RESPONSE_BAD_MSG          (-22)

struct ocfs2_response_msg
{
        struct ocfs2_msg_hdr r_hdr;
        __be32 r_response;
} __attribute__ ((packed));

struct ocfs2_vote_work {
        struct list_head   w_list;
        struct ocfs2_vote_msg w_msg;
};

enum ocfs2_vote_request {
        OCFS2_VOTE_REQ_INVALID = 0,
        OCFS2_VOTE_REQ_MOUNT,
        OCFS2_VOTE_REQ_UMOUNT,
        OCFS2_VOTE_REQ_LAST
};

static inline int ocfs2_is_valid_vote_request(int request)
{
        return OCFS2_VOTE_REQ_INVALID < request &&
                request < OCFS2_VOTE_REQ_LAST;
}

typedef void (*ocfs2_net_response_callback)(void *priv,
                                            struct ocfs2_response_msg *resp);
struct ocfs2_net_response_cb {
        ocfs2_net_response_callback     rc_cb;
        void                            *rc_priv;
};

struct ocfs2_net_wait_ctxt {
        struct list_head        n_list;
        u32                     n_response_id;
        wait_queue_head_t       n_event;
        struct ocfs2_node_map   n_node_map;
        int                     n_response; /* an agreggate response. 0 if
                                             * all nodes are go, < 0 on any
                                             * negative response from any
                                             * node or network error. */
        struct ocfs2_net_response_cb *n_callback;
};

static void ocfs2_process_mount_request(struct ocfs2_super *osb,
                                        unsigned int node_num)
{
        mlog(0, "MOUNT vote from node %u\n", node_num);
        /* The other node only sends us this message when he has an EX
         * on the superblock, so our recovery threads (if having been
         * launched) are waiting on it.*/
        ocfs2_recovery_map_clear(osb, node_num);
        ocfs2_node_map_set_bit(osb, &osb->mounted_map, node_num);

        /* We clear the umount map here because a node may have been
         * previously mounted, safely unmounted but never stopped
         * heartbeating - in which case we'd have a stale entry. */
        ocfs2_node_map_clear_bit(osb, &osb->umount_map, node_num);
}

static void ocfs2_process_umount_request(struct ocfs2_super *osb,
                                         unsigned int node_num)
{
        mlog(0, "UMOUNT vote from node %u\n", node_num);
        ocfs2_node_map_clear_bit(osb, &osb->mounted_map, node_num);
        ocfs2_node_map_set_bit(osb, &osb->umount_map, node_num);
}

static void ocfs2_process_vote(struct ocfs2_super *osb,
                               struct ocfs2_vote_msg *msg)
{
        int net_status, vote_response;
        unsigned int node_num;
        u64 blkno;
        enum ocfs2_vote_request request;
        struct ocfs2_msg_hdr *hdr = &msg->v_hdr;
        struct ocfs2_response_msg response;

        /* decode the network mumbo jumbo into local variables. */
        request = be32_to_cpu(hdr->h_request);
        blkno = be64_to_cpu(hdr->h_blkno);
        node_num = be32_to_cpu(hdr->h_node_num);

        mlog(0, "processing vote: request = %u, blkno = %llu, node_num = %u\n",
             request, (unsigned long long)blkno, node_num);

        if (!ocfs2_is_valid_vote_request(request)) {
                mlog(ML_ERROR, "Invalid vote request %d from node %u\n",
                     request, node_num);
                vote_response = OCFS2_RESPONSE_BAD_MSG;
                goto respond;
        }

        vote_response = OCFS2_RESPONSE_OK;

        switch (request) {
        case OCFS2_VOTE_REQ_UMOUNT:
                ocfs2_process_umount_request(osb, node_num);
                goto respond;
        case OCFS2_VOTE_REQ_MOUNT:
                ocfs2_process_mount_request(osb, node_num);
                goto respond;
        default:
                /* avoids a gcc warning */
                break;
        }

respond:
        /* Response struture is small so we just put it on the stack
         * and stuff it inline. */
        memset(&response, 0, sizeof(struct ocfs2_response_msg));
        response.r_hdr.h_response_id = hdr->h_response_id;
        response.r_hdr.h_blkno = hdr->h_blkno;
        response.r_hdr.h_generation = hdr->h_generation;
        response.r_hdr.h_node_num = cpu_to_be32(osb->node_num);
        response.r_response = cpu_to_be32(vote_response);

        net_status = o2net_send_message(OCFS2_MESSAGE_TYPE_RESPONSE,
                                        osb->net_key,
                                        &response,
                                        sizeof(struct ocfs2_response_msg),
                                        node_num,
                                        NULL);
        /* We still want to error print for ENOPROTOOPT here. The
         * sending node shouldn't have unregistered his net handler
         * without sending an unmount vote 1st */
        if (net_status < 0
            && net_status != -ETIMEDOUT
            && net_status != -ENOTCONN)
                mlog(ML_ERROR, "message to node %u fails with error %d!\n",
                     node_num, net_status);
}

static void ocfs2_vote_thread_do_work(struct ocfs2_super *osb)
{
        unsigned long processed;
        struct ocfs2_lock_res *lockres;
        struct ocfs2_vote_work *work;

        mlog_entry_void();

        spin_lock(&osb->vote_task_lock);
        /* grab this early so we know to try again if a state change and
         * wake happens part-way through our work  */
        osb->vote_work_sequence = osb->vote_wake_sequence;

        processed = osb->blocked_lock_count;
        while (processed) {
                BUG_ON(list_empty(&osb->blocked_lock_list));

                lockres = list_entry(osb->blocked_lock_list.next,
                                     struct ocfs2_lock_res, l_blocked_list);
                list_del_init(&lockres->l_blocked_list);
                osb->blocked_lock_count--;
                spin_unlock(&osb->vote_task_lock);

                BUG_ON(!processed);
                processed--;

                ocfs2_process_blocked_lock(osb, lockres);

                spin_lock(&osb->vote_task_lock);
        }

        while (osb->vote_count) {
                BUG_ON(list_empty(&osb->vote_list));
                work = list_entry(osb->vote_list.next,
                                  struct ocfs2_vote_work, w_list);
                list_del(&work->w_list);
                osb->vote_count--;
                spin_unlock(&osb->vote_task_lock);

                ocfs2_process_vote(osb, &work->w_msg);
                kfree(work);

                spin_lock(&osb->vote_task_lock);
        }
        spin_unlock(&osb->vote_task_lock);

        mlog_exit_void();
}

static int ocfs2_vote_thread_lists_empty(struct ocfs2_super *osb)
{
        int empty = 0;

        spin_lock(&osb->vote_task_lock);
        if (list_empty(&osb->blocked_lock_list) &&
            list_empty(&osb->vote_list))
                empty = 1;

        spin_unlock(&osb->vote_task_lock);
        return empty;
}

static int ocfs2_vote_thread_should_wake(struct ocfs2_super *osb)
{
        int should_wake = 0;

        spin_lock(&osb->vote_task_lock);
        if (osb->vote_work_sequence != osb->vote_wake_sequence)
                should_wake = 1;
        spin_unlock(&osb->vote_task_lock);

        return should_wake;
}

int ocfs2_vote_thread(void *arg)
{
        int status = 0;
        struct ocfs2_super *osb = arg;

        /* only quit once we've been asked to stop and there is no more
         * work available */
        while (!(kthread_should_stop() &&
                 ocfs2_vote_thread_lists_empty(osb))) {

                wait_event_interruptible(osb->vote_event,
                                         ocfs2_vote_thread_should_wake(osb) ||
                                         kthread_should_stop());

                mlog(0, "vote_thread: awoken\n");

                ocfs2_vote_thread_do_work(osb);
        }

        osb->vote_task = NULL;
        return status;
}

static struct ocfs2_net_wait_ctxt *ocfs2_new_net_wait_ctxt(unsigned int response_id)
{
        struct ocfs2_net_wait_ctxt *w;

        w = kzalloc(sizeof(*w), GFP_NOFS);
        if (!w) {
                mlog_errno(-ENOMEM);
                goto bail;
        }

        INIT_LIST_HEAD(&w->n_list);
        init_waitqueue_head(&w->n_event);
        ocfs2_node_map_init(&w->n_node_map);
        w->n_response_id = response_id;
        w->n_callback = NULL;
bail:
        return w;
}

static unsigned int ocfs2_new_response_id(struct ocfs2_super *osb)
{
        unsigned int ret;

        spin_lock(&osb->net_response_lock);
        ret = ++osb->net_response_ids;
        spin_unlock(&osb->net_response_lock);

        return ret;
}

static void ocfs2_dequeue_net_wait_ctxt(struct ocfs2_super *osb,
                                        struct ocfs2_net_wait_ctxt *w)
{
        spin_lock(&osb->net_response_lock);
        list_del(&w->n_list);
        spin_unlock(&osb->net_response_lock);
}

static void ocfs2_queue_net_wait_ctxt(struct ocfs2_super *osb,
                                      struct ocfs2_net_wait_ctxt *w)
{
        spin_lock(&osb->net_response_lock);
        list_add_tail(&w->n_list,
                      &osb->net_response_list);
        spin_unlock(&osb->net_response_lock);
}

static void __ocfs2_mark_node_responded(struct ocfs2_super *osb,
                                        struct ocfs2_net_wait_ctxt *w,
                                        int node_num)
{
        assert_spin_locked(&osb->net_response_lock);

        ocfs2_node_map_clear_bit(osb, &w->n_node_map, node_num);
        if (ocfs2_node_map_is_empty(osb, &w->n_node_map))
                wake_up(&w->n_event);
}

/* Intended to be called from the node down callback, we fake remove
 * the node from all our response contexts */
void ocfs2_remove_node_from_vote_queues(struct ocfs2_super *osb,
                                        int node_num)
{
        struct list_head *p;
        struct ocfs2_net_wait_ctxt *w = NULL;

        spin_lock(&osb->net_response_lock);

        list_for_each(p, &osb->net_response_list) {
                w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);

                __ocfs2_mark_node_responded(osb, w, node_num);
        }

        spin_unlock(&osb->net_response_lock);
}

static int ocfs2_broadcast_vote(struct ocfs2_super *osb,
                                struct ocfs2_vote_msg *request,
                                unsigned int response_id,
                                int *response,
                                struct ocfs2_net_response_cb *callback)
{
        int status, i, remote_err;
        struct ocfs2_net_wait_ctxt *w = NULL;
        int dequeued = 0;

        mlog_entry_void();

        w = ocfs2_new_net_wait_ctxt(response_id);
        if (!w) {
                status = -ENOMEM;
                mlog_errno(status);
                goto bail;
        }
        w->n_callback = callback;

        /* we're pretty much ready to go at this point, and this fills
         * in n_response which we need anyway... */
        ocfs2_queue_net_wait_ctxt(osb, w);

        i = ocfs2_node_map_iterate(osb, &osb->mounted_map, 0);

        while (i != O2NM_INVALID_NODE_NUM) {
                if (i != osb->node_num) {
                        mlog(0, "trying to send request to node %i\n", i);
                        ocfs2_node_map_set_bit(osb, &w->n_node_map, i);

                        remote_err = 0;
                        status = o2net_send_message(OCFS2_MESSAGE_TYPE_VOTE,
                                                    osb->net_key,
                                                    request,
                                                    sizeof(*request),
                                                    i,
                                                    &remote_err);
                        if (status == -ETIMEDOUT) {
                                mlog(0, "remote node %d timed out!\n", i);
                                status = -EAGAIN;
                                goto bail;
                        }
                        if (remote_err < 0) {
                                status = remote_err;
                                mlog(0, "remote error %d on node %d!\n",
                                     remote_err, i);
                                mlog_errno(status);
                                goto bail;
                        }
                        if (status < 0) {
                                mlog_errno(status);
                                goto bail;
                        }
                }
                i++;
                i = ocfs2_node_map_iterate(osb, &osb->mounted_map, i);
                mlog(0, "next is %d, i am %d\n", i, osb->node_num);
        }
        mlog(0, "done sending, now waiting on responses...\n");

        wait_event(w->n_event, ocfs2_node_map_is_empty(osb, &w->n_node_map));

        ocfs2_dequeue_net_wait_ctxt(osb, w);
        dequeued = 1;

        *response = w->n_response;
        status = 0;
bail:
        if (w) {
                if (!dequeued)
                        ocfs2_dequeue_net_wait_ctxt(osb, w);
                kfree(w);
        }

        mlog_exit(status);
        return status;
}

static struct ocfs2_vote_msg * ocfs2_new_vote_request(struct ocfs2_super *osb,
                                                      u64 blkno,
                                                      unsigned int generation,
                                                      enum ocfs2_vote_request type)
{
        struct ocfs2_vote_msg *request;
        struct ocfs2_msg_hdr *hdr;

        BUG_ON(!ocfs2_is_valid_vote_request(type));

        request = kzalloc(sizeof(*request), GFP_NOFS);
        if (!request) {
                mlog_errno(-ENOMEM);
        } else {
                hdr = &request->v_hdr;
                hdr->h_node_num = cpu_to_be32(osb->node_num);
                hdr->h_request = cpu_to_be32(type);
                hdr->h_blkno = cpu_to_be64(blkno);
                hdr->h_generation = cpu_to_be32(generation);
        }

        return request;
}

/* Complete the buildup of a new vote request and process the
 * broadcast return value. */
static int ocfs2_do_request_vote(struct ocfs2_super *osb,
                                 struct ocfs2_vote_msg *request,
                                 struct ocfs2_net_response_cb *callback)
{
        int status, response = -EBUSY;
        unsigned int response_id;
        struct ocfs2_msg_hdr *hdr;

        response_id = ocfs2_new_response_id(osb);

        hdr = &request->v_hdr;
        hdr->h_response_id = cpu_to_be32(response_id);

        status = ocfs2_broadcast_vote(osb, request, response_id, &response,
                                      callback);
        if (status < 0) {
                mlog_errno(status);
                goto bail;
        }

        status = response;
bail:

        return status;
}

int ocfs2_request_mount_vote(struct ocfs2_super *osb)
{
        int status;
        struct ocfs2_vote_msg *request = NULL;

        request = ocfs2_new_vote_request(osb, 0ULL, 0, OCFS2_VOTE_REQ_MOUNT);
        if (!request) {
                status = -ENOMEM;
                goto bail;
        }

        status = -EAGAIN;
        while (status == -EAGAIN) {
                if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
                    signal_pending(current)) {
                        status = -ERESTARTSYS;
                        goto bail;
                }

                if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
                                           osb->node_num)) {
                        status = 0;
                        goto bail;
                }

                status = ocfs2_do_request_vote(osb, request, NULL);
        }

bail:
        kfree(request);
        return status;
}

int ocfs2_request_umount_vote(struct ocfs2_super *osb)
{
        int status;
        struct ocfs2_vote_msg *request = NULL;

        request = ocfs2_new_vote_request(osb, 0ULL, 0, OCFS2_VOTE_REQ_UMOUNT);
        if (!request) {
                status = -ENOMEM;
                goto bail;
        }

        status = -EAGAIN;
        while (status == -EAGAIN) {
                /* Do not check signals on this vote... We really want
                 * this one to go all the way through. */

                if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
                                           osb->node_num)) {
                        status = 0;
                        goto bail;
                }

                status = ocfs2_do_request_vote(osb, request, NULL);
        }

bail:
        kfree(request);
        return status;
}

/* TODO: This should eventually be a hash table! */
static struct ocfs2_net_wait_ctxt * __ocfs2_find_net_wait_ctxt(struct ocfs2_super *osb,
                                                               u32 response_id)
{
        struct list_head *p;
        struct ocfs2_net_wait_ctxt *w = NULL;

        list_for_each(p, &osb->net_response_list) {
                w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
                if (response_id == w->n_response_id)
                        break;
                w = NULL;
        }

        return w;
}

/* Translate response codes into local node errno values */
static inline int ocfs2_translate_response(int response)
{
        int ret;

        switch (response) {
        case OCFS2_RESPONSE_OK:
                ret = 0;
                break;

        case OCFS2_RESPONSE_BUSY:
                ret = -EBUSY;
                break;

        default:
                ret = -EINVAL;
        }

        return ret;
}

static int ocfs2_handle_response_message(struct o2net_msg *msg,
                                         u32 len,
                                         void *data, void **ret_data)
{
        unsigned int response_id, node_num;
        int response_status;
        struct ocfs2_super *osb = data;
        struct ocfs2_response_msg *resp;
        struct ocfs2_net_wait_ctxt * w;
        struct ocfs2_net_response_cb *resp_cb;

        resp = (struct ocfs2_response_msg *) msg->buf;

        response_id = be32_to_cpu(resp->r_hdr.h_response_id);
        node_num = be32_to_cpu(resp->r_hdr.h_node_num);
        response_status = 
                ocfs2_translate_response(be32_to_cpu(resp->r_response));

        mlog(0, "received response message:\n");
        mlog(0, "h_response_id = %u\n", response_id);
        mlog(0, "h_request = %u\n", be32_to_cpu(resp->r_hdr.h_request));
        mlog(0, "h_blkno = %llu\n",
             (unsigned long long)be64_to_cpu(resp->r_hdr.h_blkno));
        mlog(0, "h_generation = %u\n", be32_to_cpu(resp->r_hdr.h_generation));
        mlog(0, "h_node_num = %u\n", node_num);
        mlog(0, "r_response = %d\n", response_status);

        spin_lock(&osb->net_response_lock);
        w = __ocfs2_find_net_wait_ctxt(osb, response_id);
        if (!w) {
                mlog(0, "request not found!\n");
                goto bail;
        }
        resp_cb = w->n_callback;

        if (response_status && (!w->n_response)) {
                /* we only really need one negative response so don't
                 * set it twice. */
                w->n_response = response_status;
        }

        if (resp_cb) {
                spin_unlock(&osb->net_response_lock);

                resp_cb->rc_cb(resp_cb->rc_priv, resp);

                spin_lock(&osb->net_response_lock);
        }

        __ocfs2_mark_node_responded(osb, w, node_num);
bail:
        spin_unlock(&osb->net_response_lock);

        return 0;
}

static int ocfs2_handle_vote_message(struct o2net_msg *msg,
                                     u32 len,
                                     void *data, void **ret_data)
{
        int status;
        struct ocfs2_super *osb = data;
        struct ocfs2_vote_work *work;

        work = kmalloc(sizeof(struct ocfs2_vote_work), GFP_NOFS);
        if (!work) {
                status = -ENOMEM;
                mlog_errno(status);
                goto bail;
        }

        INIT_LIST_HEAD(&work->w_list);
        memcpy(&work->w_msg, msg->buf, sizeof(struct ocfs2_vote_msg));

        mlog(0, "scheduling vote request:\n");
        mlog(0, "h_response_id = %u\n",
             be32_to_cpu(work->w_msg.v_hdr.h_response_id));
        mlog(0, "h_request = %u\n", be32_to_cpu(work->w_msg.v_hdr.h_request));
        mlog(0, "h_blkno = %llu\n",
             (unsigned long long)be64_to_cpu(work->w_msg.v_hdr.h_blkno));
        mlog(0, "h_generation = %u\n",
             be32_to_cpu(work->w_msg.v_hdr.h_generation));
        mlog(0, "h_node_num = %u\n",
             be32_to_cpu(work->w_msg.v_hdr.h_node_num));

        spin_lock(&osb->vote_task_lock);
        list_add_tail(&work->w_list, &osb->vote_list);
        osb->vote_count++;
        spin_unlock(&osb->vote_task_lock);

        ocfs2_kick_vote_thread(osb);

        status = 0;
bail:
        return status;
}

void ocfs2_unregister_net_handlers(struct ocfs2_super *osb)
{
        if (!osb->net_key)
                return;

        o2net_unregister_handler_list(&osb->osb_net_handlers);

        if (!list_empty(&osb->net_response_list))
                mlog(ML_ERROR, "net response list not empty!\n");

        osb->net_key = 0;
}

int ocfs2_register_net_handlers(struct ocfs2_super *osb)
{
        int status = 0;

        if (ocfs2_mount_local(osb))
                return 0;

        status = o2net_register_handler(OCFS2_MESSAGE_TYPE_RESPONSE,
                                        osb->net_key,
                                        sizeof(struct ocfs2_response_msg),
                                        ocfs2_handle_response_message,
                                        osb, NULL, &osb->osb_net_handlers);
        if (status) {
                mlog_errno(status);
                goto bail;
        }

        status = o2net_register_handler(OCFS2_MESSAGE_TYPE_VOTE,
                                        osb->net_key,
                                        sizeof(struct ocfs2_vote_msg),
                                        ocfs2_handle_vote_message,
                                        osb, NULL, &osb->osb_net_handlers);
        if (status) {
                mlog_errno(status);
                goto bail;
        }
bail:
        if (status < 0)
                ocfs2_unregister_net_handlers(osb);

        return status;
}