fs/ocfs2/vote.c

   1 /* -*- mode: c; c-basic-offset: 8; -*-
   2  * vim: noexpandtab sw=8 ts=8 sts=0:
   3  *
   4  * vote.c
   5  *
   6  * description here
   7  *
   8  * Copyright (C) 2003, 2004 Oracle.  All rights reserved.
   9  *
  10  * This program is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU General Public
  12  * License as published by the Free Software Foundation; either
  13  * version 2 of the License, or (at your option) any later version.
  14  *
  15  * This program is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18  * General Public License for more details.
  19  *
  20  * You should have received a copy of the GNU General Public
  21  * License along with this program; if not, write to the
  22  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  23  * Boston, MA 021110-1307, USA.
  24  */
  25
  26 #include <linux/types.h>
  27 #include <linux/slab.h>
  28 #include <linux/highmem.h>
  29 #include <linux/smp_lock.h>
  30 #include <linux/kthread.h>
  31
  32 #include <cluster/heartbeat.h>
  33 #include <cluster/nodemanager.h>
  34 #include <cluster/tcp.h>
  35
  36 #include <dlm/dlmapi.h>
  37
  38 #define MLOG_MASK_PREFIX ML_VOTE
  39 #include <cluster/masklog.h>
  40
  41 #include "ocfs2.h"
  42
  43 #include "alloc.h"
  44 #include "dlmglue.h"
  45 #include "extent_map.h"
  46 #include "heartbeat.h"
  47 #include "inode.h"
  48 #include "journal.h"
  49 #include "slot_map.h"
  50 #include "vote.h"
  51
  52 #include "buffer_head_io.h"
  53
  54 #define OCFS2_MESSAGE_TYPE_VOTE     (0x1)
  55 #define OCFS2_MESSAGE_TYPE_RESPONSE (0x2)
  56 struct ocfs2_msg_hdr
  57 {
  58         __be32 h_response_id; /* used to lookup message handle on sending
  59                             * node. */
  60         __be32 h_request;
  61         __be64 h_blkno;
  62         __be32 h_generation;
  63         __be32 h_node_num;    /* node sending this particular message. */
  64 };
  65
  66 /* OCFS2_MAX_FILENAME_LEN is 255 characters, but we want to align this
  67  * for the network. */
  68 #define OCFS2_VOTE_FILENAME_LEN 256
  69 struct ocfs2_vote_msg
  70 {
  71         struct ocfs2_msg_hdr v_hdr;
  72         union {
  73                 __be32 v_generic1;
  74                 __be32 v_orphaned_slot; /* Used during delete votes */
  75                 __be32 v_nlink;         /* Used during unlink votes */
  76         } md1;                          /* Message type dependant 1 */
  77 };
  78
  79 /* Responses are given these values to maintain backwards
  80  * compatibility with older ocfs2 versions */
  81 #define OCFS2_RESPONSE_OK               (0)
  82 #define OCFS2_RESPONSE_BUSY             (-16)
  83 #define OCFS2_RESPONSE_BAD_MSG          (-22)
  84
  85 struct ocfs2_response_msg
  86 {
  87         struct ocfs2_msg_hdr r_hdr;
  88         __be32 r_response;
  89         __be32 r_orphaned_slot;
  90 };
  91
  92 struct ocfs2_vote_work {
  93         struct list_head   w_list;
  94         struct ocfs2_vote_msg w_msg;
  95 };
  96
  97 enum ocfs2_vote_request {
  98         OCFS2_VOTE_REQ_INVALID = 0,
  99         OCFS2_VOTE_REQ_DELETE,
 100         OCFS2_VOTE_REQ_MOUNT,
 101         OCFS2_VOTE_REQ_UMOUNT,
 102         OCFS2_VOTE_REQ_LAST
 103 };
 104
 105 static inline int ocfs2_is_valid_vote_request(int request)
 106 {
 107         return OCFS2_VOTE_REQ_INVALID < request &&
 108                 request < OCFS2_VOTE_REQ_LAST;
 109 }
 110
 111 typedef void (*ocfs2_net_response_callback)(void *priv,
 112                                             struct ocfs2_response_msg *resp);
 113 struct ocfs2_net_response_cb {
 114         ocfs2_net_response_callback     rc_cb;
 115         void                            *rc_priv;
 116 };
 117
 118 struct ocfs2_net_wait_ctxt {
 119         struct list_head        n_list;
 120         u32                     n_response_id;
 121         wait_queue_head_t       n_event;
 122         struct ocfs2_node_map   n_node_map;
 123         int                     n_response; /* an agreggate response. 0 if
 124                                              * all nodes are go, < 0 on any
 125                                              * negative response from any
 126                                              * node or network error. */
 127         struct ocfs2_net_response_cb *n_callback;
 128 };
 129
 130 static void ocfs2_process_mount_request(struct ocfs2_super *osb,
 131                                         unsigned int node_num)
 132 {
 133         mlog(0, "MOUNT vote from node %u\n", node_num);
 134         /* The other node only sends us this message when he has an EX
 135          * on the superblock, so our recovery threads (if having been
 136          * launched) are waiting on it.*/
 137         ocfs2_recovery_map_clear(osb, node_num);
 138         ocfs2_node_map_set_bit(osb, &osb->mounted_map, node_num);
 139
 140         /* We clear the umount map here because a node may have been
 141          * previously mounted, safely unmounted but never stopped
 142          * heartbeating - in which case we'd have a stale entry. */
 143         ocfs2_node_map_clear_bit(osb, &osb->umount_map, node_num);
 144 }
 145
 146 static void ocfs2_process_umount_request(struct ocfs2_super *osb,
 147                                          unsigned int node_num)
 148 {
 149         mlog(0, "UMOUNT vote from node %u\n", node_num);
 150         ocfs2_node_map_clear_bit(osb, &osb->mounted_map, node_num);
 151         ocfs2_node_map_set_bit(osb, &osb->umount_map, node_num);
 152 }
 153
 154 void ocfs2_mark_inode_remotely_deleted(struct inode *inode)
 155 {
 156         struct ocfs2_inode_info *oi = OCFS2_I(inode);
 157
 158         assert_spin_locked(&oi->ip_lock);
 159         /* We set the SKIP_DELETE flag on the inode so we don't try to
 160          * delete it in delete_inode ourselves, thus avoiding
 161          * unecessary lock pinging. If the other node failed to wipe
 162          * the inode as a result of a crash, then recovery will pick
 163          * up the slack. */
 164         oi->ip_flags |= OCFS2_INODE_DELETED|OCFS2_INODE_SKIP_DELETE;
 165 }
 166
 167 static int ocfs2_process_delete_request(struct inode *inode,
 168                                         int *orphaned_slot)
 169 {
 170         int response = OCFS2_RESPONSE_BUSY;
 171
 172         mlog(0, "DELETE vote on inode %lu, read lnk_cnt = %u, slot = %d\n",
 173              inode->i_ino, inode->i_nlink, *orphaned_slot);
 174
 175         spin_lock(&OCFS2_I(inode)->ip_lock);
 176
 177         /* Whatever our vote response is, we want to make sure that
 178          * the orphaned slot is recorded properly on this node *and*
 179          * on the requesting node. Technically, if the requesting node
 180          * did not know which slot the inode is orphaned in but we
 181          * respond with BUSY he doesn't actually need the orphaned
 182          * slot, but it doesn't hurt to do it here anyway. */
 183         if ((*orphaned_slot) != OCFS2_INVALID_SLOT) {
 184                 mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot !=
 185                                 OCFS2_INVALID_SLOT &&
 186                                 OCFS2_I(inode)->ip_orphaned_slot !=
 187                                 (*orphaned_slot),
 188                                 "Inode %llu: This node thinks it's "
 189                                 "orphaned in slot %d, messaged it's in %d\n",
 190                                 (unsigned long long)OCFS2_I(inode)->ip_blkno,
 191                                 OCFS2_I(inode)->ip_orphaned_slot,
 192                                 *orphaned_slot);
 193
 194                 mlog(0, "Setting orphaned slot for inode %llu to %d\n",
 195                      (unsigned long long)OCFS2_I(inode)->ip_blkno,
 196                      *orphaned_slot);
 197
 198                 OCFS2_I(inode)->ip_orphaned_slot = *orphaned_slot;
 199         } else {
 200                 mlog(0, "Sending back orphaned slot %d for inode %llu\n",
 201                      OCFS2_I(inode)->ip_orphaned_slot,
 202                      (unsigned long long)OCFS2_I(inode)->ip_blkno);
 203
 204                 *orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
 205         }
 206
 207         /* vote no if the file is still open. */
 208         if (OCFS2_I(inode)->ip_open_count) {
 209                 mlog(0, "open count = %u\n",
 210                      OCFS2_I(inode)->ip_open_count);
 211                 spin_unlock(&OCFS2_I(inode)->ip_lock);
 212                 goto done;
 213         }
 214         spin_unlock(&OCFS2_I(inode)->ip_lock);
 215
 216         /* directories are a bit ugly... What if someone is sitting in
 217          * it? We want to make sure the inode is removed completely as
 218          * a result of the iput in process_vote. */
 219         if (S_ISDIR(inode->i_mode) && (atomic_read(&inode->i_count) != 1)) {
 220                 mlog(0, "i_count = %u\n", atomic_read(&inode->i_count));
 221                 goto done;
 222         }
 223
 224         if (filemap_fdatawrite(inode->i_mapping)) {
 225                 mlog(ML_ERROR, "Could not sync inode %llu for delete!\n",
 226                      (unsigned long long)OCFS2_I(inode)->ip_blkno);
 227                 goto done;
 228         }
 229         sync_mapping_buffers(inode->i_mapping);
 230         truncate_inode_pages(inode->i_mapping, 0);
 231         ocfs2_extent_map_trunc(inode, 0);
 232
 233         spin_lock(&OCFS2_I(inode)->ip_lock);
 234         /* double check open count - someone might have raced this
 235          * thread into ocfs2_file_open while we were writing out
 236          * data. If we're to allow a wipe of this inode now, we *must*
 237          * hold the spinlock until we've marked it. */
 238         if (OCFS2_I(inode)->ip_open_count) {
 239                 mlog(0, "Raced to wipe! open count = %u\n",
 240                      OCFS2_I(inode)->ip_open_count);
 241                 spin_unlock(&OCFS2_I(inode)->ip_lock);
 242                 goto done;
 243         }
 244
 245         /* Mark the inode as being wiped from disk. */
 246         ocfs2_mark_inode_remotely_deleted(inode);
 247         spin_unlock(&OCFS2_I(inode)->ip_lock);
 248
 249         /* Not sure this is necessary anymore. */
 250         d_prune_aliases(inode);
 251
 252         /* If we get here, then we're voting 'yes', so commit the
 253          * delete on our side. */
 254         response = OCFS2_RESPONSE_OK;
 255 done:
 256         return response;
 257 }
 258
 259 static void ocfs2_process_vote(struct ocfs2_super *osb,
 260                                struct ocfs2_vote_msg *msg)
 261 {
 262         int net_status, vote_response;
 263         int orphaned_slot = 0;
 264         unsigned int node_num, generation;
 265         u64 blkno;
 266         enum ocfs2_vote_request request;
 267         struct inode *inode = NULL;
 268         struct ocfs2_msg_hdr *hdr = &msg->v_hdr;
 269         struct ocfs2_response_msg response;
 270
 271         /* decode the network mumbo jumbo into local variables. */
 272         request = be32_to_cpu(hdr->h_request);
 273         blkno = be64_to_cpu(hdr->h_blkno);
 274         generation = be32_to_cpu(hdr->h_generation);
 275         node_num = be32_to_cpu(hdr->h_node_num);
 276         if (request == OCFS2_VOTE_REQ_DELETE)
 277                 orphaned_slot = be32_to_cpu(msg->md1.v_orphaned_slot);
 278
 279         mlog(0, "processing vote: request = %u, blkno = %llu, "
 280              "generation = %u, node_num = %u, priv1 = %u\n", request,
 281              (unsigned long long)blkno, generation, node_num,
 282              be32_to_cpu(msg->md1.v_generic1));
 283
 284         if (!ocfs2_is_valid_vote_request(request)) {
 285                 mlog(ML_ERROR, "Invalid vote request %d from node %u\n",
 286                      request, node_num);
 287                 vote_response = OCFS2_RESPONSE_BAD_MSG;
 288                 goto respond;
 289         }
 290
 291         vote_response = OCFS2_RESPONSE_OK;
 292
 293         switch (request) {
 294         case OCFS2_VOTE_REQ_UMOUNT:
 295                 ocfs2_process_umount_request(osb, node_num);
 296                 goto respond;
 297         case OCFS2_VOTE_REQ_MOUNT:
 298                 ocfs2_process_mount_request(osb, node_num);
 299                 goto respond;
 300         default:
 301                 /* avoids a gcc warning */
 302                 break;
 303         }
 304
 305         /* We cannot process the remaining message types before we're
 306          * fully mounted. It's perfectly safe however to send a 'yes'
 307          * response as we can't possibly have any of the state they're
 308          * asking us to modify yet. */
 309         if (atomic_read(&osb->vol_state) == VOLUME_INIT)
 310                 goto respond;
 311
 312         /* If we get here, then the request is against an inode. */
 313         inode = ocfs2_ilookup_for_vote(osb, blkno,
 314                                        request == OCFS2_VOTE_REQ_DELETE);
 315
 316         /* Not finding the inode is perfectly valid - it means we're
 317          * not interested in what the other node is about to do to it
 318          * so in those cases we automatically respond with an
 319          * affirmative. Cluster locking ensures that we won't race
 320          * interest in the inode with this vote request. */
 321         if (!inode)
 322                 goto respond;
 323
 324         /* Check generation values. It's possible for us to get a
 325          * request against a stale inode. If so then we proceed as if
 326          * we had not found an inode in the first place. */
 327         if (inode->i_generation != generation) {
 328                 mlog(0, "generation passed %u != inode generation = %u, "
 329                      "ip_flags = %x, ip_blkno = %llu, msg %llu, i_count = %u, "
 330                      "message type = %u\n", generation, inode->i_generation,
 331                      OCFS2_I(inode)->ip_flags,
 332                      (unsigned long long)OCFS2_I(inode)->ip_blkno,
 333                      (unsigned long long)blkno, atomic_read(&inode->i_count),
 334                      request);
 335                 iput(inode);
 336                 inode = NULL;
 337                 goto respond;
 338         }
 339
 340         switch (request) {
 341         case OCFS2_VOTE_REQ_DELETE:
 342                 vote_response = ocfs2_process_delete_request(inode,
 343                                                              &orphaned_slot);
 344                 break;
 345         default:
 346                 mlog(ML_ERROR, "node %u, invalid request: %u\n",
 347                      node_num, request);
 348                 vote_response = OCFS2_RESPONSE_BAD_MSG;
 349         }
 350
 351 respond:
 352         /* Response struture is small so we just put it on the stack
 353          * and stuff it inline. */
 354         memset(&response, 0, sizeof(struct ocfs2_response_msg));
 355         response.r_hdr.h_response_id = hdr->h_response_id;
 356         response.r_hdr.h_blkno = hdr->h_blkno;
 357         response.r_hdr.h_generation = hdr->h_generation;
 358         response.r_hdr.h_node_num = cpu_to_be32(osb->node_num);
 359         response.r_response = cpu_to_be32(vote_response);
 360         response.r_orphaned_slot = cpu_to_be32(orphaned_slot);
 361
 362         net_status = o2net_send_message(OCFS2_MESSAGE_TYPE_RESPONSE,
 363                                         osb->net_key,
 364                                         &response,
 365                                         sizeof(struct ocfs2_response_msg),
 366                                         node_num,
 367                                         NULL);
 368         /* We still want to error print for ENOPROTOOPT here. The
 369          * sending node shouldn't have unregistered his net handler
 370          * without sending an unmount vote 1st */
 371         if (net_status < 0
 372             && net_status != -ETIMEDOUT
 373             && net_status != -ENOTCONN)
 374                 mlog(ML_ERROR, "message to node %u fails with error %d!\n",
 375                      node_num, net_status);
 376
 377         if (inode)
 378                 iput(inode);
 379 }
 380
 381 static void ocfs2_vote_thread_do_work(struct ocfs2_super *osb)
 382 {
 383         unsigned long processed;
 384         struct ocfs2_lock_res *lockres;
 385         struct ocfs2_vote_work *work;
 386
 387         mlog_entry_void();
 388
 389         spin_lock(&osb->vote_task_lock);
 390         /* grab this early so we know to try again if a state change and
 391          * wake happens part-way through our work  */
 392         osb->vote_work_sequence = osb->vote_wake_sequence;
 393
 394         processed = osb->blocked_lock_count;
 395         while (processed) {
 396                 BUG_ON(list_empty(&osb->blocked_lock_list));
 397
 398                 lockres = list_entry(osb->blocked_lock_list.next,
 399                                      struct ocfs2_lock_res, l_blocked_list);
 400                 list_del_init(&lockres->l_blocked_list);
 401                 osb->blocked_lock_count--;
 402                 spin_unlock(&osb->vote_task_lock);
 403
 404                 BUG_ON(!processed);
 405                 processed--;
 406
 407                 ocfs2_process_blocked_lock(osb, lockres);
 408
 409                 spin_lock(&osb->vote_task_lock);
 410         }
 411
 412         while (osb->vote_count) {
 413                 BUG_ON(list_empty(&osb->vote_list));
 414                 work = list_entry(osb->vote_list.next,
 415                                   struct ocfs2_vote_work, w_list);
 416                 list_del(&work->w_list);
 417                 osb->vote_count--;
 418                 spin_unlock(&osb->vote_task_lock);
 419
 420                 ocfs2_process_vote(osb, &work->w_msg);
 421                 kfree(work);
 422
 423                 spin_lock(&osb->vote_task_lock);
 424         }
 425         spin_unlock(&osb->vote_task_lock);
 426
 427         mlog_exit_void();
 428 }
 429
 430 static int ocfs2_vote_thread_lists_empty(struct ocfs2_super *osb)
 431 {
 432         int empty = 0;
 433
 434         spin_lock(&osb->vote_task_lock);
 435         if (list_empty(&osb->blocked_lock_list) &&
 436             list_empty(&osb->vote_list))
 437                 empty = 1;
 438
 439         spin_unlock(&osb->vote_task_lock);
 440         return empty;
 441 }
 442
 443 static int ocfs2_vote_thread_should_wake(struct ocfs2_super *osb)
 444 {
 445         int should_wake = 0;
 446
 447         spin_lock(&osb->vote_task_lock);
 448         if (osb->vote_work_sequence != osb->vote_wake_sequence)
 449                 should_wake = 1;
 450         spin_unlock(&osb->vote_task_lock);
 451
 452         return should_wake;
 453 }
 454
 455 int ocfs2_vote_thread(void *arg)
 456 {
 457         int status = 0;
 458         struct ocfs2_super *osb = arg;
 459
 460         /* only quit once we've been asked to stop and there is no more
 461          * work available */
 462         while (!(kthread_should_stop() &&
 463                  ocfs2_vote_thread_lists_empty(osb))) {
 464
 465                 wait_event_interruptible(osb->vote_event,
 466                                          ocfs2_vote_thread_should_wake(osb) ||
 467                                          kthread_should_stop());
 468
 469                 mlog(0, "vote_thread: awoken\n");
 470
 471                 ocfs2_vote_thread_do_work(osb);
 472         }
 473
 474         osb->vote_task = NULL;
 475         return status;
 476 }
 477
 478 static struct ocfs2_net_wait_ctxt *ocfs2_new_net_wait_ctxt(unsigned int response_id)
 479 {
 480         struct ocfs2_net_wait_ctxt *w;
 481
 482         w = kcalloc(1, sizeof(*w), GFP_NOFS);
 483         if (!w) {
 484                 mlog_errno(-ENOMEM);
 485                 goto bail;
 486         }
 487
 488         INIT_LIST_HEAD(&w->n_list);
 489         init_waitqueue_head(&w->n_event);
 490         ocfs2_node_map_init(&w->n_node_map);
 491         w->n_response_id = response_id;
 492         w->n_callback = NULL;
 493 bail:
 494         return w;
 495 }
 496
 497 static unsigned int ocfs2_new_response_id(struct ocfs2_super *osb)
 498 {
 499         unsigned int ret;
 500
 501         spin_lock(&osb->net_response_lock);
 502         ret = ++osb->net_response_ids;
 503         spin_unlock(&osb->net_response_lock);
 504
 505         return ret;
 506 }
 507
 508 static void ocfs2_dequeue_net_wait_ctxt(struct ocfs2_super *osb,
 509                                         struct ocfs2_net_wait_ctxt *w)
 510 {
 511         spin_lock(&osb->net_response_lock);
 512         list_del(&w->n_list);
 513         spin_unlock(&osb->net_response_lock);
 514 }
 515
 516 static void ocfs2_queue_net_wait_ctxt(struct ocfs2_super *osb,
 517                                       struct ocfs2_net_wait_ctxt *w)
 518 {
 519         spin_lock(&osb->net_response_lock);
 520         list_add_tail(&w->n_list,
 521                       &osb->net_response_list);
 522         spin_unlock(&osb->net_response_lock);
 523 }
 524
 525 static void __ocfs2_mark_node_responded(struct ocfs2_super *osb,
 526                                         struct ocfs2_net_wait_ctxt *w,
 527                                         int node_num)
 528 {
 529         assert_spin_locked(&osb->net_response_lock);
 530
 531         ocfs2_node_map_clear_bit(osb, &w->n_node_map, node_num);
 532         if (ocfs2_node_map_is_empty(osb, &w->n_node_map))
 533                 wake_up(&w->n_event);
 534 }
 535
 536 /* Intended to be called from the node down callback, we fake remove
 537  * the node from all our response contexts */
 538 void ocfs2_remove_node_from_vote_queues(struct ocfs2_super *osb,
 539                                         int node_num)
 540 {
 541         struct list_head *p;
 542         struct ocfs2_net_wait_ctxt *w = NULL;
 543
 544         spin_lock(&osb->net_response_lock);
 545
 546         list_for_each(p, &osb->net_response_list) {
 547                 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
 548
 549                 __ocfs2_mark_node_responded(osb, w, node_num);
 550         }
 551
 552         spin_unlock(&osb->net_response_lock);
 553 }
 554
 555 static int ocfs2_broadcast_vote(struct ocfs2_super *osb,
 556                                 struct ocfs2_vote_msg *request,
 557                                 unsigned int response_id,
 558                                 int *response,
 559                                 struct ocfs2_net_response_cb *callback)
 560 {
 561         int status, i, remote_err;
 562         struct ocfs2_net_wait_ctxt *w = NULL;
 563         int dequeued = 0;
 564
 565         mlog_entry_void();
 566
 567         w = ocfs2_new_net_wait_ctxt(response_id);
 568         if (!w) {
 569                 status = -ENOMEM;
 570                 mlog_errno(status);
 571                 goto bail;
 572         }
 573         w->n_callback = callback;
 574
 575         /* we're pretty much ready to go at this point, and this fills
 576          * in n_response which we need anyway... */
 577         ocfs2_queue_net_wait_ctxt(osb, w);
 578
 579         i = ocfs2_node_map_iterate(osb, &osb->mounted_map, 0);
 580
 581         while (i != O2NM_INVALID_NODE_NUM) {
 582                 if (i != osb->node_num) {
 583                         mlog(0, "trying to send request to node %i\n", i);
 584                         ocfs2_node_map_set_bit(osb, &w->n_node_map, i);
 585
 586                         remote_err = 0;
 587                         status = o2net_send_message(OCFS2_MESSAGE_TYPE_VOTE,
 588                                                     osb->net_key,
 589                                                     request,
 590                                                     sizeof(*request),
 591                                                     i,
 592                                                     &remote_err);
 593                         if (status == -ETIMEDOUT) {
 594                                 mlog(0, "remote node %d timed out!\n", i);
 595                                 status = -EAGAIN;
 596                                 goto bail;
 597                         }
 598                         if (remote_err < 0) {
 599                                 status = remote_err;
 600                                 mlog(0, "remote error %d on node %d!\n",
 601                                      remote_err, i);
 602                                 mlog_errno(status);
 603                                 goto bail;
 604                         }
 605                         if (status < 0) {
 606                                 mlog_errno(status);
 607                                 goto bail;
 608                         }
 609                 }
 610                 i++;
 611                 i = ocfs2_node_map_iterate(osb, &osb->mounted_map, i);
 612                 mlog(0, "next is %d, i am %d\n", i, osb->node_num);
 613         }
 614         mlog(0, "done sending, now waiting on responses...\n");
 615
 616         wait_event(w->n_event, ocfs2_node_map_is_empty(osb, &w->n_node_map));
 617
 618         ocfs2_dequeue_net_wait_ctxt(osb, w);
 619         dequeued = 1;
 620
 621         *response = w->n_response;
 622         status = 0;
 623 bail:
 624         if (w) {
 625                 if (!dequeued)
 626                         ocfs2_dequeue_net_wait_ctxt(osb, w);
 627                 kfree(w);
 628         }
 629
 630         mlog_exit(status);
 631         return status;
 632 }
 633
 634 static struct ocfs2_vote_msg * ocfs2_new_vote_request(struct ocfs2_super *osb,
 635                                                       u64 blkno,
 636                                                       unsigned int generation,
 637                                                       enum ocfs2_vote_request type,
 638                                                       u32 priv)
 639 {
 640         struct ocfs2_vote_msg *request;
 641         struct ocfs2_msg_hdr *hdr;
 642
 643         BUG_ON(!ocfs2_is_valid_vote_request(type));
 644
 645         request = kcalloc(1, sizeof(*request), GFP_NOFS);
 646         if (!request) {
 647                 mlog_errno(-ENOMEM);
 648         } else {
 649                 hdr = &request->v_hdr;
 650                 hdr->h_node_num = cpu_to_be32(osb->node_num);
 651                 hdr->h_request = cpu_to_be32(type);
 652                 hdr->h_blkno = cpu_to_be64(blkno);
 653                 hdr->h_generation = cpu_to_be32(generation);
 654
 655                 request->md1.v_generic1 = cpu_to_be32(priv);
 656         }
 657
 658         return request;
 659 }
 660
 661 /* Complete the buildup of a new vote request and process the
 662  * broadcast return value. */
 663 static int ocfs2_do_request_vote(struct ocfs2_super *osb,
 664                                  struct ocfs2_vote_msg *request,
 665                                  struct ocfs2_net_response_cb *callback)
 666 {
 667         int status, response;
 668         unsigned int response_id;
 669         struct ocfs2_msg_hdr *hdr;
 670
 671         response_id = ocfs2_new_response_id(osb);
 672
 673         hdr = &request->v_hdr;
 674         hdr->h_response_id = cpu_to_be32(response_id);
 675
 676         status = ocfs2_broadcast_vote(osb, request, response_id, &response,
 677                                       callback);
 678         if (status < 0) {
 679                 mlog_errno(status);
 680                 goto bail;
 681         }
 682
 683         status = response;
 684 bail:
 685
 686         return status;
 687 }
 688
 689 static int ocfs2_request_vote(struct inode *inode,
 690                               struct ocfs2_vote_msg *request,
 691                               struct ocfs2_net_response_cb *callback)
 692 {
 693         int status;
 694         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
 695
 696         if (ocfs2_inode_is_new(inode))
 697                 return 0;
 698
 699         status = -EAGAIN;
 700         while (status == -EAGAIN) {
 701                 if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
 702                     signal_pending(current))
 703                         return -ERESTARTSYS;
 704
 705                 status = ocfs2_super_lock(osb, 0);
 706                 if (status < 0) {
 707                         mlog_errno(status);
 708                         break;
 709                 }
 710
 711                 status = 0;
 712                 if (!ocfs2_node_map_is_only(osb, &osb->mounted_map,
 713                                            osb->node_num))
 714                         status = ocfs2_do_request_vote(osb, request, callback);
 715
 716                 ocfs2_super_unlock(osb, 0);
 717         }
 718         return status;
 719 }
 720
 721 static void ocfs2_delete_response_cb(void *priv,
 722                                      struct ocfs2_response_msg *resp)
 723 {
 724         int orphaned_slot, node;
 725         struct inode *inode = priv;
 726
 727         orphaned_slot = be32_to_cpu(resp->r_orphaned_slot);
 728         node = be32_to_cpu(resp->r_hdr.h_node_num);
 729         mlog(0, "node %d tells us that inode %llu is orphaned in slot %d\n",
 730              node, (unsigned long long)OCFS2_I(inode)->ip_blkno,
 731              orphaned_slot);
 732
 733         /* The other node may not actually know which slot the inode
 734          * is orphaned in. */
 735         if (orphaned_slot == OCFS2_INVALID_SLOT)
 736                 return;
 737
 738         /* Ok, the responding node knows which slot this inode is
 739          * orphaned in. We verify that the information is correct and
 740          * then record this in the inode. ocfs2_delete_inode will use
 741          * this information to determine which lock to take. */
 742         spin_lock(&OCFS2_I(inode)->ip_lock);
 743         mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot != orphaned_slot &&
 744                         OCFS2_I(inode)->ip_orphaned_slot
 745                         != OCFS2_INVALID_SLOT, "Inode %llu: Node %d says it's "
 746                         "orphaned in slot %d, we think it's in %d\n",
 747                         (unsigned long long)OCFS2_I(inode)->ip_blkno,
 748                         be32_to_cpu(resp->r_hdr.h_node_num),
 749                         orphaned_slot, OCFS2_I(inode)->ip_orphaned_slot);
 750
 751         OCFS2_I(inode)->ip_orphaned_slot = orphaned_slot;
 752         spin_unlock(&OCFS2_I(inode)->ip_lock);
 753 }
 754
 755 int ocfs2_request_delete_vote(struct inode *inode)
 756 {
 757         int orphaned_slot, status;
 758         struct ocfs2_net_response_cb delete_cb;
 759         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
 760         struct ocfs2_vote_msg *request;
 761
 762         spin_lock(&OCFS2_I(inode)->ip_lock);
 763         orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
 764         spin_unlock(&OCFS2_I(inode)->ip_lock);
 765
 766         delete_cb.rc_cb = ocfs2_delete_response_cb;
 767         delete_cb.rc_priv = inode;
 768
 769         mlog(0, "Inode %llu, we start thinking orphaned slot is %d\n",
 770              (unsigned long long)OCFS2_I(inode)->ip_blkno, orphaned_slot);
 771
 772         status = -ENOMEM;
 773         request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
 774                                          inode->i_generation,
 775                                          OCFS2_VOTE_REQ_DELETE, orphaned_slot);
 776         if (request) {
 777                 status = ocfs2_request_vote(inode, request, &delete_cb);
 778
 779                 kfree(request);
 780         }
 781
 782         return status;
 783 }
 784
 785 int ocfs2_request_mount_vote(struct ocfs2_super *osb)
 786 {
 787         int status;
 788         struct ocfs2_vote_msg *request = NULL;
 789
 790         request = ocfs2_new_vote_request(osb, 0ULL, 0,
 791                                          OCFS2_VOTE_REQ_MOUNT, 0);
 792         if (!request) {
 793                 status = -ENOMEM;
 794                 goto bail;
 795         }
 796
 797         status = -EAGAIN;
 798         while (status == -EAGAIN) {
 799                 if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
 800                     signal_pending(current)) {
 801                         status = -ERESTARTSYS;
 802                         goto bail;
 803                 }
 804
 805                 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
 806                                            osb->node_num)) {
 807                         status = 0;
 808                         goto bail;
 809                 }
 810
 811                 status = ocfs2_do_request_vote(osb, request, NULL);
 812         }
 813
 814 bail:
 815         kfree(request);
 816         return status;
 817 }
 818
 819 int ocfs2_request_umount_vote(struct ocfs2_super *osb)
 820 {
 821         int status;
 822         struct ocfs2_vote_msg *request = NULL;
 823
 824         request = ocfs2_new_vote_request(osb, 0ULL, 0,
 825                                          OCFS2_VOTE_REQ_UMOUNT, 0);
 826         if (!request) {
 827                 status = -ENOMEM;
 828                 goto bail;
 829         }
 830
 831         status = -EAGAIN;
 832         while (status == -EAGAIN) {
 833                 /* Do not check signals on this vote... We really want
 834                  * this one to go all the way through. */
 835
 836                 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
 837                                            osb->node_num)) {
 838                         status = 0;
 839                         goto bail;
 840                 }
 841
 842                 status = ocfs2_do_request_vote(osb, request, NULL);
 843         }
 844
 845 bail:
 846         kfree(request);
 847         return status;
 848 }
 849
 850 /* TODO: This should eventually be a hash table! */
 851 static struct ocfs2_net_wait_ctxt * __ocfs2_find_net_wait_ctxt(struct ocfs2_super *osb,
 852                                                                u32 response_id)
 853 {
 854         struct list_head *p;
 855         struct ocfs2_net_wait_ctxt *w = NULL;
 856
 857         list_for_each(p, &osb->net_response_list) {
 858                 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
 859                 if (response_id == w->n_response_id)
 860                         break;
 861                 w = NULL;
 862         }
 863
 864         return w;
 865 }
 866
 867 /* Translate response codes into local node errno values */
 868 static inline int ocfs2_translate_response(int response)
 869 {
 870         int ret;
 871
 872         switch (response) {
 873         case OCFS2_RESPONSE_OK:
 874                 ret = 0;
 875                 break;
 876
 877         case OCFS2_RESPONSE_BUSY:
 878                 ret = -EBUSY;
 879                 break;
 880
 881         default:
 882                 ret = -EINVAL;
 883         }
 884
 885         return ret;
 886 }
 887
 888 static int ocfs2_handle_response_message(struct o2net_msg *msg,
 889                                          u32 len,
 890                                          void *data)
 891 {
 892         unsigned int response_id, node_num;
 893         int response_status;
 894         struct ocfs2_super *osb = data;
 895         struct ocfs2_response_msg *resp;
 896         struct ocfs2_net_wait_ctxt * w;
 897         struct ocfs2_net_response_cb *resp_cb;
 898
 899         resp = (struct ocfs2_response_msg *) msg->buf;
 900
 901         response_id = be32_to_cpu(resp->r_hdr.h_response_id);
 902         node_num = be32_to_cpu(resp->r_hdr.h_node_num);
 903         response_status =
 904                 ocfs2_translate_response(be32_to_cpu(resp->r_response));
 905
 906         mlog(0, "received response message:\n");
 907         mlog(0, "h_response_id = %u\n", response_id);
 908         mlog(0, "h_request = %u\n", be32_to_cpu(resp->r_hdr.h_request));
 909         mlog(0, "h_blkno = %llu\n",
 910              (unsigned long long)be64_to_cpu(resp->r_hdr.h_blkno));
 911         mlog(0, "h_generation = %u\n", be32_to_cpu(resp->r_hdr.h_generation));
 912         mlog(0, "h_node_num = %u\n", node_num);
 913         mlog(0, "r_response = %d\n", response_status);
 914
 915         spin_lock(&osb->net_response_lock);
 916         w = __ocfs2_find_net_wait_ctxt(osb, response_id);
 917         if (!w) {
 918                 mlog(0, "request not found!\n");
 919                 goto bail;
 920         }
 921         resp_cb = w->n_callback;
 922
 923         if (response_status && (!w->n_response)) {
 924                 /* we only really need one negative response so don't
 925                  * set it twice. */
 926                 w->n_response = response_status;
 927         }
 928
 929         if (resp_cb) {
 930                 spin_unlock(&osb->net_response_lock);
 931
 932                 resp_cb->rc_cb(resp_cb->rc_priv, resp);
 933
 934                 spin_lock(&osb->net_response_lock);
 935         }
 936
 937         __ocfs2_mark_node_responded(osb, w, node_num);
 938 bail:
 939         spin_unlock(&osb->net_response_lock);
 940
 941         return 0;
 942 }
 943
 944 static int ocfs2_handle_vote_message(struct o2net_msg *msg,
 945                                      u32 len,
 946                                      void *data)
 947 {
 948         int status;
 949         struct ocfs2_super *osb = data;
 950         struct ocfs2_vote_work *work;
 951
 952         work = kmalloc(sizeof(struct ocfs2_vote_work), GFP_NOFS);
 953         if (!work) {
 954                 status = -ENOMEM;
 955                 mlog_errno(status);
 956                 goto bail;
 957         }
 958
 959         INIT_LIST_HEAD(&work->w_list);
 960         memcpy(&work->w_msg, msg->buf, sizeof(struct ocfs2_vote_msg));
 961
 962         mlog(0, "scheduling vote request:\n");
 963         mlog(0, "h_response_id = %u\n",
 964              be32_to_cpu(work->w_msg.v_hdr.h_response_id));
 965         mlog(0, "h_request = %u\n", be32_to_cpu(work->w_msg.v_hdr.h_request));
 966         mlog(0, "h_blkno = %llu\n",
 967              (unsigned long long)be64_to_cpu(work->w_msg.v_hdr.h_blkno));
 968         mlog(0, "h_generation = %u\n",
 969              be32_to_cpu(work->w_msg.v_hdr.h_generation));
 970         mlog(0, "h_node_num = %u\n",
 971              be32_to_cpu(work->w_msg.v_hdr.h_node_num));
 972         mlog(0, "v_generic1 = %u\n", be32_to_cpu(work->w_msg.md1.v_generic1));
 973
 974         spin_lock(&osb->vote_task_lock);
 975         list_add_tail(&work->w_list, &osb->vote_list);
 976         osb->vote_count++;
 977         spin_unlock(&osb->vote_task_lock);
 978
 979         ocfs2_kick_vote_thread(osb);
 980
 981         status = 0;
 982 bail:
 983         return status;
 984 }
 985
 986 void ocfs2_unregister_net_handlers(struct ocfs2_super *osb)
 987 {
 988         if (!osb->net_key)
 989                 return;
 990
 991         o2net_unregister_handler_list(&osb->osb_net_handlers);
 992
 993         if (!list_empty(&osb->net_response_list))
 994                 mlog(ML_ERROR, "net response list not empty!\n");
 995
 996         osb->net_key = 0;
 997 }
 998
 999 int ocfs2_register_net_handlers(struct ocfs2_super *osb)
1000 {
1001         int status = 0;
1002
1003         if (ocfs2_mount_local(osb))
1004                 return 0;
1005
1006         status = o2net_register_handler(OCFS2_MESSAGE_TYPE_RESPONSE,
1007                                         osb->net_key,
1008                                         sizeof(struct ocfs2_response_msg),
1009                                         ocfs2_handle_response_message,
1010                                         osb, &osb->osb_net_handlers);
1011         if (status) {
1012                 mlog_errno(status);
1013                 goto bail;
1014         }
1015
1016         status = o2net_register_handler(OCFS2_MESSAGE_TYPE_VOTE,
1017                                         osb->net_key,
1018                                         sizeof(struct ocfs2_vote_msg),
1019                                         ocfs2_handle_vote_message,
1020                                         osb, &osb->osb_net_handlers);
1021         if (status) {
1022                 mlog_errno(status);
1023                 goto bail;
1024         }
1025 bail:
1026         if (status < 0)
1027                 ocfs2_unregister_net_handlers(osb);
1028
1029         return status;
1030 }