diff options
Diffstat (limited to 'fs/ocfs2/vote.c')
| -rw-r--r-- | fs/ocfs2/vote.c | 1202 |
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; +} |