[PATCH] OCFS2: The Second Oracle Cluster Filesystem

The OCFS2 file system module.

Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>
Signed-off-by: Kurt Hackel <kurt.hackel@oracle.com>

1 files changed, 1202 insertions, 0 deletions

diff --git a/fs/ocfs2/vote.c b/fs/ocfs2/vote.c
new file mode 100644
index 000000000000..021978e0576b
--- /dev/null
+++ b/fs/ocfs2/vote.c
@@ -0,0 +1,1202 @@
+/* -*- 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/smp_lock.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. */
+};
+
+/* OCFS2_MAX_FILENAME_LEN is 255 characters, but we want to align this
+ * for the network. */
+#define OCFS2_VOTE_FILENAME_LEN 256
+struct ocfs2_vote_msg
+{
+        struct ocfs2_msg_hdr v_hdr;
+        union {
+                __be32 v_generic1;
+                __be32 v_orphaned_slot; /* Used during delete votes */
+                __be32 v_nlink;         /* Used during unlink votes */
+        } md1;                          /* Message type dependant 1 */
+        __be32 v_unlink_namelen;
+        __be64 v_unlink_parent;
+        u8  v_unlink_dirent[OCFS2_VOTE_FILENAME_LEN];
+};
+
+/* 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;
+        __be32 r_orphaned_slot;
+};
+
+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_DELETE,
+        OCFS2_VOTE_REQ_UNLINK,
+        OCFS2_VOTE_REQ_RENAME,
+        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);
+}
+
+void ocfs2_mark_inode_remotely_deleted(struct inode *inode)
+{
+        struct ocfs2_inode_info *oi = OCFS2_I(inode);
+
+        assert_spin_locked(&oi->ip_lock);
+        /* We set the SKIP_DELETE flag on the inode so we don't try to
+         * delete it in delete_inode ourselves, thus avoiding
+         * unecessary lock pinging. If the other node failed to wipe
+         * the inode as a result of a crash, then recovery will pick
+         * up the slack. */
+        oi->ip_flags |= OCFS2_INODE_DELETED|OCFS2_INODE_SKIP_DELETE;
+}
+
+static int ocfs2_process_delete_request(struct inode *inode,
+                                        int *orphaned_slot)
+{
+        int response = OCFS2_RESPONSE_BUSY;
+
+        mlog(0, "DELETE vote on inode %lu, read lnk_cnt = %u, slot = %d\n",
+             inode->i_ino, inode->i_nlink, *orphaned_slot);
+
+        spin_lock(&OCFS2_I(inode)->ip_lock);
+
+        /* Whatever our vote response is, we want to make sure that
+         * the orphaned slot is recorded properly on this node *and*
+         * on the requesting node. Technically, if the requesting node
+         * did not know which slot the inode is orphaned in but we
+         * respond with BUSY he doesn't actually need the orphaned
+         * slot, but it doesn't hurt to do it here anyway. */
+        if ((*orphaned_slot) != OCFS2_INVALID_SLOT) {
+                mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot !=
+                                OCFS2_INVALID_SLOT &&
+                                OCFS2_I(inode)->ip_orphaned_slot !=
+                                (*orphaned_slot),
+                                "Inode %"MLFu64": This node thinks it's "
+                                "orphaned in slot %d, messaged it's in %d\n",
+                                OCFS2_I(inode)->ip_blkno,
+                                OCFS2_I(inode)->ip_orphaned_slot,
+                                *orphaned_slot);
+
+                mlog(0, "Setting orphaned slot for inode %"MLFu64" to %d\n",
+                     OCFS2_I(inode)->ip_blkno, *orphaned_slot);
+
+                OCFS2_I(inode)->ip_orphaned_slot = *orphaned_slot;
+        } else {
+                mlog(0, "Sending back orphaned slot %d for inode %"MLFu64"\n",
+                     OCFS2_I(inode)->ip_orphaned_slot,
+                     OCFS2_I(inode)->ip_blkno);
+
+                *orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
+        }
+
+        /* vote no if the file is still open. */
+        if (OCFS2_I(inode)->ip_open_count) {
+                mlog(0, "open count = %u\n",
+                     OCFS2_I(inode)->ip_open_count);
+                spin_unlock(&OCFS2_I(inode)->ip_lock);
+                goto done;
+        }
+        spin_unlock(&OCFS2_I(inode)->ip_lock);
+
+        /* directories are a bit ugly... What if someone is sitting in
+         * it? We want to make sure the inode is removed completely as
+         * a result of the iput in process_vote. */
+        if (S_ISDIR(inode->i_mode) && (atomic_read(&inode->i_count) != 1)) {
+                mlog(0, "i_count = %u\n", atomic_read(&inode->i_count));
+                goto done;
+        }
+
+        if (filemap_fdatawrite(inode->i_mapping)) {
+                mlog(ML_ERROR, "Could not sync inode %"MLFu64" for delete!\n",
+                     OCFS2_I(inode)->ip_blkno);
+                goto done;
+        }
+        sync_mapping_buffers(inode->i_mapping);
+        truncate_inode_pages(inode->i_mapping, 0);
+        ocfs2_extent_map_trunc(inode, 0);
+
+        spin_lock(&OCFS2_I(inode)->ip_lock);
+        /* double check open count - someone might have raced this
+         * thread into ocfs2_file_open while we were writing out
+         * data. If we're to allow a wipe of this inode now, we *must*
+         * hold the spinlock until we've marked it. */
+        if (OCFS2_I(inode)->ip_open_count) {
+                mlog(0, "Raced to wipe! open count = %u\n",
+                     OCFS2_I(inode)->ip_open_count);
+                spin_unlock(&OCFS2_I(inode)->ip_lock);
+                goto done;
+        }
+
+        /* Mark the inode as being wiped from disk. */
+        ocfs2_mark_inode_remotely_deleted(inode);
+        spin_unlock(&OCFS2_I(inode)->ip_lock);
+
+        /* Not sure this is necessary anymore. */
+        d_prune_aliases(inode);
+
+        /* If we get here, then we're voting 'yes', so commit the
+         * delete on our side. */
+        response = OCFS2_RESPONSE_OK;
+done:
+        return response;
+}
+
+static int ocfs2_match_dentry(struct dentry *dentry,
+                              u64 parent_blkno,
+                              unsigned int namelen,
+                              const char *name)
+{
+        struct inode *parent;
+
+        if (!dentry->d_parent) {
+                mlog(0, "Detached from parent.\n");
+                return 0;
+        }
+
+        parent = dentry->d_parent->d_inode;
+        /* Negative parent dentry? */
+        if (!parent)
+                return 0;
+
+        /* Name is in a different directory. */
+        if (OCFS2_I(parent)->ip_blkno != parent_blkno)
+                return 0;
+
+        if (dentry->d_name.len != namelen)
+                return 0;
+
+        /* comparison above guarantees this is safe. */
+        if (memcmp(dentry->d_name.name, name, namelen))
+                return 0;
+
+        return 1;
+}
+
+static void ocfs2_process_dentry_request(struct inode *inode,
+                                         int rename,
+                                         unsigned int new_nlink,
+                                         u64 parent_blkno,
+                                         unsigned int namelen,
+                                         const char *name)
+{
+        struct dentry *dentry = NULL;
+        struct list_head *p;
+        struct ocfs2_inode_info *oi = OCFS2_I(inode);
+
+        mlog(0, "parent %"MLFu64", namelen = %u, name = %.*s\n", parent_blkno,
+             namelen, namelen, name);
+
+        spin_lock(&dcache_lock);
+
+        /* Another node is removing this name from the system. It is
+         * up to us to find the corresponding dentry and if it exists,
+         * unhash it from the dcache. */
+        list_for_each(p, &inode->i_dentry) {
+                dentry = list_entry(p, struct dentry, d_alias);
+
+                if (ocfs2_match_dentry(dentry, parent_blkno, namelen, name)) {
+                        mlog(0, "dentry found: %.*s\n",
+                             dentry->d_name.len, dentry->d_name.name);
+
+                        dget_locked(dentry);
+                        break;
+                }
+
+                dentry = NULL;
+        }
+
+        spin_unlock(&dcache_lock);
+
+        if (dentry) {
+                d_delete(dentry);
+                dput(dentry);
+        }
+
+        /* rename votes don't send link counts */
+        if (!rename) {
+                mlog(0, "new_nlink = %u\n", new_nlink);
+
+                /* We don't have the proper locks here to directly
+                 * change i_nlink and besides, the vote is sent
+                 * *before* the operation so it may have failed on the
+                 * other node. This passes a hint to ocfs2_drop_inode
+                 * to force ocfs2_delete_inode, who will take the
+                 * proper cluster locks to sort things out. */
+                if (new_nlink == 0) {
+                        spin_lock(&oi->ip_lock);
+                        oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
+                        spin_unlock(&OCFS2_I(inode)->ip_lock);
+                }
+        }
+}
+
+static void ocfs2_process_vote(struct ocfs2_super *osb,
+                               struct ocfs2_vote_msg *msg)
+{
+        int net_status, vote_response;
+        int orphaned_slot = 0;
+        int rename = 0;
+        unsigned int node_num, generation, new_nlink, namelen;
+        u64 blkno, parent_blkno;
+        enum ocfs2_vote_request request;
+        struct inode *inode = NULL;
+        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);
+        generation = be32_to_cpu(hdr->h_generation);
+        node_num = be32_to_cpu(hdr->h_node_num);
+        if (request == OCFS2_VOTE_REQ_DELETE)
+                orphaned_slot = be32_to_cpu(msg->md1.v_orphaned_slot);
+
+        mlog(0, "processing vote: request = %u, blkno = %"MLFu64", "
+             "generation = %u, node_num = %u, priv1 = %u\n", request,
+             blkno, generation, node_num, be32_to_cpu(msg->md1.v_generic1));
+
+        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;
+        }
+
+        /* We cannot process the remaining message types before we're
+         * fully mounted. It's perfectly safe however to send a 'yes'
+         * response as we can't possibly have any of the state they're
+         * asking us to modify yet. */
+        if (atomic_read(&osb->vol_state) == VOLUME_INIT)
+                goto respond;
+
+        /* If we get here, then the request is against an inode. */
+        inode = ocfs2_ilookup_for_vote(osb, blkno,
+                                       request == OCFS2_VOTE_REQ_DELETE);
+
+        /* Not finding the inode is perfectly valid - it means we're
+         * not interested in what the other node is about to do to it
+         * so in those cases we automatically respond with an
+         * affirmative. Cluster locking ensures that we won't race
+         * interest in the inode with this vote request. */
+        if (!inode)
+                goto respond;
+
+        /* Check generation values. It's possible for us to get a
+         * request against a stale inode. If so then we proceed as if
+         * we had not found an inode in the first place. */
+        if (inode->i_generation != generation) {
+                mlog(0, "generation passed %u != inode generation = %u, "
+                     "ip_flags = %x, ip_blkno = %"MLFu64", msg %"MLFu64", "
+                     "i_count = %u, message type = %u\n",
+                     generation, inode->i_generation, OCFS2_I(inode)->ip_flags,
+                     OCFS2_I(inode)->ip_blkno, blkno,
+                     atomic_read(&inode->i_count), request);
+                iput(inode);
+                inode = NULL;
+                goto respond;
+        }
+
+        switch (request) {
+        case OCFS2_VOTE_REQ_DELETE:
+                vote_response = ocfs2_process_delete_request(inode,
+                                                             &orphaned_slot);
+                break;
+        case OCFS2_VOTE_REQ_RENAME:
+                rename = 1;
+                /* fall through */
+        case OCFS2_VOTE_REQ_UNLINK:
+                parent_blkno = be64_to_cpu(msg->v_unlink_parent);
+                namelen = be32_to_cpu(msg->v_unlink_namelen);
+                /* new_nlink will be ignored in case of a rename vote */
+                new_nlink = be32_to_cpu(msg->md1.v_nlink);
+                ocfs2_process_dentry_request(inode, rename, new_nlink,
+                                             parent_blkno, namelen,
+                                             msg->v_unlink_dirent);
+                break;
+        default:
+                mlog(ML_ERROR, "node %u, invalid request: %u\n",
+                     node_num, request);
+                vote_response = OCFS2_RESPONSE_BAD_MSG;
+        }
+
+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);
+        response.r_orphaned_slot = cpu_to_be32(orphaned_slot);
+
+        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);
+
+        if (inode)
+                iput(inode);
+}
+
+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 = kcalloc(1, sizeof(*w), GFP_KERNEL);
+        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,
+                                                      u32 priv)
+{
+        struct ocfs2_vote_msg *request;
+        struct ocfs2_msg_hdr *hdr;
+
+        BUG_ON(!ocfs2_is_valid_vote_request(type));
+
+        request = kcalloc(1, sizeof(*request), GFP_KERNEL);
+        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);
+
+                request->md1.v_generic1 = cpu_to_be32(priv);
+        }
+
+        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;
+        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;
+}
+
+static int ocfs2_request_vote(struct inode *inode,
+                              struct ocfs2_vote_msg *request,
+                              struct ocfs2_net_response_cb *callback)
+{
+        int status;
+        struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
+
+        if (ocfs2_inode_is_new(inode))
+                return 0;
+
+        status = -EAGAIN;
+        while (status == -EAGAIN) {
+                if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
+                    signal_pending(current))
+                        return -ERESTARTSYS;
+
+                status = ocfs2_super_lock(osb, 0);
+                if (status < 0) {
+                        mlog_errno(status);
+                        break;
+                }
+
+                status = 0;
+                if (!ocfs2_node_map_is_only(osb, &osb->mounted_map,
+                                           osb->node_num))
+                        status = ocfs2_do_request_vote(osb, request, callback);
+
+                ocfs2_super_unlock(osb, 0);
+        }
+        return status;
+}
+
+static void ocfs2_delete_response_cb(void *priv,
+                                     struct ocfs2_response_msg *resp)
+{
+        int orphaned_slot, node;
+        struct inode *inode = priv;
+
+        orphaned_slot = be32_to_cpu(resp->r_orphaned_slot);
+        node = be32_to_cpu(resp->r_hdr.h_node_num);
+        mlog(0, "node %d tells us that inode %"MLFu64" is orphaned in slot "
+             "%d\n", node, OCFS2_I(inode)->ip_blkno, orphaned_slot);
+
+        /* The other node may not actually know which slot the inode
+         * is orphaned in. */
+        if (orphaned_slot == OCFS2_INVALID_SLOT)
+                return;
+
+        /* Ok, the responding node knows which slot this inode is
+         * orphaned in. We verify that the information is correct and
+         * then record this in the inode. ocfs2_delete_inode will use
+         * this information to determine which lock to take. */
+        spin_lock(&OCFS2_I(inode)->ip_lock);
+        mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot != orphaned_slot &&
+                        OCFS2_I(inode)->ip_orphaned_slot
+                        != OCFS2_INVALID_SLOT, "Inode %"MLFu64": Node %d "
+                        "says it's orphaned in slot %d, we think it's in %d\n",
+                        OCFS2_I(inode)->ip_blkno,
+                        be32_to_cpu(resp->r_hdr.h_node_num),
+                        orphaned_slot, OCFS2_I(inode)->ip_orphaned_slot);
+
+        OCFS2_I(inode)->ip_orphaned_slot = orphaned_slot;
+        spin_unlock(&OCFS2_I(inode)->ip_lock);
+}
+
+int ocfs2_request_delete_vote(struct inode *inode)
+{
+        int orphaned_slot, status;
+        struct ocfs2_net_response_cb delete_cb;
+        struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
+        struct ocfs2_vote_msg *request;
+
+        spin_lock(&OCFS2_I(inode)->ip_lock);
+        orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
+        spin_unlock(&OCFS2_I(inode)->ip_lock);
+
+        delete_cb.rc_cb = ocfs2_delete_response_cb;
+        delete_cb.rc_priv = inode;
+
+        mlog(0, "Inode %"MLFu64", we start thinking orphaned slot is %d\n",
+             OCFS2_I(inode)->ip_blkno, orphaned_slot);
+
+        status = -ENOMEM;
+        request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
+                                         inode->i_generation,
+                                         OCFS2_VOTE_REQ_DELETE, orphaned_slot);
+        if (request) {
+                status = ocfs2_request_vote(inode, request, &delete_cb);
+
+                kfree(request);
+        }
+
+        return status;
+}
+
+static void ocfs2_setup_unlink_vote(struct ocfs2_vote_msg *request,
+                                    struct dentry *dentry)
+{
+        struct inode *parent = dentry->d_parent->d_inode;
+
+        /* We need some values which will uniquely identify a dentry
+         * on the other nodes so that they can find it and run
+         * d_delete against it. Parent directory block and full name
+         * should suffice. */
+
+        mlog(0, "unlink/rename request: parent: %"MLFu64" name: %.*s\n",
+             OCFS2_I(parent)->ip_blkno, dentry->d_name.len,
+             dentry->d_name.name);
+
+        request->v_unlink_parent = cpu_to_be64(OCFS2_I(parent)->ip_blkno);
+        request->v_unlink_namelen = cpu_to_be32(dentry->d_name.len);
+        memcpy(request->v_unlink_dirent, dentry->d_name.name,
+               dentry->d_name.len);
+}
+
+int ocfs2_request_unlink_vote(struct inode *inode,
+                              struct dentry *dentry,
+                              unsigned int nlink)
+{
+        int status;
+        struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
+        struct ocfs2_vote_msg *request;
+
+        if (dentry->d_name.len > OCFS2_VOTE_FILENAME_LEN)
+                return -ENAMETOOLONG;
+
+        status = -ENOMEM;
+        request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
+                                         inode->i_generation,
+                                         OCFS2_VOTE_REQ_UNLINK, nlink);
+        if (request) {
+                ocfs2_setup_unlink_vote(request, dentry);
+
+                status = ocfs2_request_vote(inode, request, NULL);
+
+                kfree(request);
+        }
+        return status;
+}
+
+int ocfs2_request_rename_vote(struct inode *inode,
+                              struct dentry *dentry)
+{
+        int status;
+        struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
+        struct ocfs2_vote_msg *request;
+
+        if (dentry->d_name.len > OCFS2_VOTE_FILENAME_LEN)
+                return -ENAMETOOLONG;
+
+        status = -ENOMEM;
+        request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
+                                         inode->i_generation,
+                                         OCFS2_VOTE_REQ_RENAME, 0);
+        if (request) {
+                ocfs2_setup_unlink_vote(request, dentry);
+
+                status = ocfs2_request_vote(inode, request, NULL);
+
+                kfree(request);
+        }
+        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, 0);
+        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:
+        if (request)
+                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, 0);
+        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:
+        if (request)
+                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)
+{
+        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 = %"MLFu64"\n", 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)
+{
+        int status;
+        struct ocfs2_super *osb = data;
+        struct ocfs2_vote_work *work;
+
+        work = kmalloc(sizeof(struct ocfs2_vote_work), GFP_KERNEL);
+        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 = %"MLFu64"\n",
+             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));
+        mlog(0, "v_generic1 = %u\n", be32_to_cpu(work->w_msg.md1.v_generic1));
+
+        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;
+
+        status = o2net_register_handler(OCFS2_MESSAGE_TYPE_RESPONSE,
+                                        osb->net_key,
+                                        sizeof(struct ocfs2_response_msg),
+                                        ocfs2_handle_response_message,
+                                        osb, &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, &osb->osb_net_handlers);
+        if (status) {
+                mlog_errno(status);
+                goto bail;
+        }
+bail:
+        if (status < 0)
+                ocfs2_unregister_net_handlers(osb);
+
+        return status;
+}