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 struct ocfs2_vote_msg
  67 {
  68         struct ocfs2_msg_hdr v_hdr;
  69         __be32 v_reserved1;
  70 };
  71
  72 /* Responses are given these values to maintain backwards
  73  * compatibility with older ocfs2 versions */
  74 #define OCFS2_RESPONSE_OK               (0)
  75 #define OCFS2_RESPONSE_BUSY             (-16)
  76 #define OCFS2_RESPONSE_BAD_MSG          (-22)
  77
  78 struct ocfs2_response_msg
  79 {
  80         struct ocfs2_msg_hdr r_hdr;
  81         __be32 r_response;
  82 };
  83
  84 struct ocfs2_vote_work {
  85         struct list_head   w_list;
  86         struct ocfs2_vote_msg w_msg;
  87 };
  88
  89 enum ocfs2_vote_request {
  90         OCFS2_VOTE_REQ_INVALID = 0,
  91         OCFS2_VOTE_REQ_MOUNT,
  92         OCFS2_VOTE_REQ_UMOUNT,
  93         OCFS2_VOTE_REQ_LAST
  94 };
  95
  96 static inline int ocfs2_is_valid_vote_request(int request)
  97 {
  98         return OCFS2_VOTE_REQ_INVALID < request &&
  99                 request < OCFS2_VOTE_REQ_LAST;
 100 }
 101
 102 typedef void (*ocfs2_net_response_callback)(void *priv,
 103                                             struct ocfs2_response_msg *resp);
 104 struct ocfs2_net_response_cb {
 105         ocfs2_net_response_callback     rc_cb;
 106         void                            *rc_priv;
 107 };
 108
 109 struct ocfs2_net_wait_ctxt {
 110         struct list_head        n_list;
 111         u32                     n_response_id;
 112         wait_queue_head_t       n_event;
 113         struct ocfs2_node_map   n_node_map;
 114         int                     n_response; /* an agreggate response. 0 if
 115                                              * all nodes are go, < 0 on any
 116                                              * negative response from any
 117                                              * node or network error. */
 118         struct ocfs2_net_response_cb *n_callback;
 119 };
 120
 121 static void ocfs2_process_mount_request(struct ocfs2_super *osb,
 122                                         unsigned int node_num)
 123 {
 124         mlog(0, "MOUNT vote from node %u\n", node_num);
 125         /* The other node only sends us this message when he has an EX
 126          * on the superblock, so our recovery threads (if having been
 127          * launched) are waiting on it.*/
 128         ocfs2_recovery_map_clear(osb, node_num);
 129         ocfs2_node_map_set_bit(osb, &osb->mounted_map, node_num);
 130
 131         /* We clear the umount map here because a node may have been
 132          * previously mounted, safely unmounted but never stopped
 133          * heartbeating - in which case we'd have a stale entry. */
 134         ocfs2_node_map_clear_bit(osb, &osb->umount_map, node_num);
 135 }
 136
 137 static void ocfs2_process_umount_request(struct ocfs2_super *osb,
 138                                          unsigned int node_num)
 139 {
 140         mlog(0, "UMOUNT vote from node %u\n", node_num);
 141         ocfs2_node_map_clear_bit(osb, &osb->mounted_map, node_num);
 142         ocfs2_node_map_set_bit(osb, &osb->umount_map, node_num);
 143 }
 144
 145 static void ocfs2_process_vote(struct ocfs2_super *osb,
 146                                struct ocfs2_vote_msg *msg)
 147 {
 148         int net_status, vote_response;
 149         unsigned int node_num;
 150         u64 blkno;
 151         enum ocfs2_vote_request request;
 152         struct ocfs2_msg_hdr *hdr = &msg->v_hdr;
 153         struct ocfs2_response_msg response;
 154
 155         /* decode the network mumbo jumbo into local variables. */
 156         request = be32_to_cpu(hdr->h_request);
 157         blkno = be64_to_cpu(hdr->h_blkno);
 158         node_num = be32_to_cpu(hdr->h_node_num);
 159
 160         mlog(0, "processing vote: request = %u, blkno = %llu, node_num = %u\n",
 161              request, (unsigned long long)blkno, node_num);
 162
 163         if (!ocfs2_is_valid_vote_request(request)) {
 164                 mlog(ML_ERROR, "Invalid vote request %d from node %u\n",
 165                      request, node_num);
 166                 vote_response = OCFS2_RESPONSE_BAD_MSG;
 167                 goto respond;
 168         }
 169
 170         vote_response = OCFS2_RESPONSE_OK;
 171
 172         switch (request) {
 173         case OCFS2_VOTE_REQ_UMOUNT:
 174                 ocfs2_process_umount_request(osb, node_num);
 175                 goto respond;
 176         case OCFS2_VOTE_REQ_MOUNT:
 177                 ocfs2_process_mount_request(osb, node_num);
 178                 goto respond;
 179         default:
 180                 /* avoids a gcc warning */
 181                 break;
 182         }
 183
 184 respond:
 185         /* Response struture is small so we just put it on the stack
 186          * and stuff it inline. */
 187         memset(&response, 0, sizeof(struct ocfs2_response_msg));
 188         response.r_hdr.h_response_id = hdr->h_response_id;
 189         response.r_hdr.h_blkno = hdr->h_blkno;
 190         response.r_hdr.h_generation = hdr->h_generation;
 191         response.r_hdr.h_node_num = cpu_to_be32(osb->node_num);
 192         response.r_response = cpu_to_be32(vote_response);
 193
 194         net_status = o2net_send_message(OCFS2_MESSAGE_TYPE_RESPONSE,
 195                                         osb->net_key,
 196                                         &response,
 197                                         sizeof(struct ocfs2_response_msg),
 198                                         node_num,
 199                                         NULL);
 200         /* We still want to error print for ENOPROTOOPT here. The
 201          * sending node shouldn't have unregistered his net handler
 202          * without sending an unmount vote 1st */
 203         if (net_status < 0
 204             && net_status != -ETIMEDOUT
 205             && net_status != -ENOTCONN)
 206                 mlog(ML_ERROR, "message to node %u fails with error %d!\n",
 207                      node_num, net_status);
 208 }
 209
 210 static void ocfs2_vote_thread_do_work(struct ocfs2_super *osb)
 211 {
 212         unsigned long processed;
 213         struct ocfs2_lock_res *lockres;
 214         struct ocfs2_vote_work *work;
 215
 216         mlog_entry_void();
 217
 218         spin_lock(&osb->vote_task_lock);
 219         /* grab this early so we know to try again if a state change and
 220          * wake happens part-way through our work  */
 221         osb->vote_work_sequence = osb->vote_wake_sequence;
 222
 223         processed = osb->blocked_lock_count;
 224         while (processed) {
 225                 BUG_ON(list_empty(&osb->blocked_lock_list));
 226
 227                 lockres = list_entry(osb->blocked_lock_list.next,
 228                                      struct ocfs2_lock_res, l_blocked_list);
 229                 list_del_init(&lockres->l_blocked_list);
 230                 osb->blocked_lock_count--;
 231                 spin_unlock(&osb->vote_task_lock);
 232
 233                 BUG_ON(!processed);
 234                 processed--;
 235
 236                 ocfs2_process_blocked_lock(osb, lockres);
 237
 238                 spin_lock(&osb->vote_task_lock);
 239         }
 240
 241         while (osb->vote_count) {
 242                 BUG_ON(list_empty(&osb->vote_list));
 243                 work = list_entry(osb->vote_list.next,
 244                                   struct ocfs2_vote_work, w_list);
 245                 list_del(&work->w_list);
 246                 osb->vote_count--;
 247                 spin_unlock(&osb->vote_task_lock);
 248
 249                 ocfs2_process_vote(osb, &work->w_msg);
 250                 kfree(work);
 251
 252                 spin_lock(&osb->vote_task_lock);
 253         }
 254         spin_unlock(&osb->vote_task_lock);
 255
 256         mlog_exit_void();
 257 }
 258
 259 static int ocfs2_vote_thread_lists_empty(struct ocfs2_super *osb)
 260 {
 261         int empty = 0;
 262
 263         spin_lock(&osb->vote_task_lock);
 264         if (list_empty(&osb->blocked_lock_list) &&
 265             list_empty(&osb->vote_list))
 266                 empty = 1;
 267
 268         spin_unlock(&osb->vote_task_lock);
 269         return empty;
 270 }
 271
 272 static int ocfs2_vote_thread_should_wake(struct ocfs2_super *osb)
 273 {
 274         int should_wake = 0;
 275
 276         spin_lock(&osb->vote_task_lock);
 277         if (osb->vote_work_sequence != osb->vote_wake_sequence)
 278                 should_wake = 1;
 279         spin_unlock(&osb->vote_task_lock);
 280
 281         return should_wake;
 282 }
 283
 284 int ocfs2_vote_thread(void *arg)
 285 {
 286         int status = 0;
 287         struct ocfs2_super *osb = arg;
 288
 289         /* only quit once we've been asked to stop and there is no more
 290          * work available */
 291         while (!(kthread_should_stop() &&
 292                  ocfs2_vote_thread_lists_empty(osb))) {
 293
 294                 wait_event_interruptible(osb->vote_event,
 295                                          ocfs2_vote_thread_should_wake(osb) ||
 296                                          kthread_should_stop());
 297
 298                 mlog(0, "vote_thread: awoken\n");
 299
 300                 ocfs2_vote_thread_do_work(osb);
 301         }
 302
 303         osb->vote_task = NULL;
 304         return status;
 305 }
 306
 307 static struct ocfs2_net_wait_ctxt *ocfs2_new_net_wait_ctxt(unsigned int response_id)
 308 {
 309         struct ocfs2_net_wait_ctxt *w;
 310
 311         w = kzalloc(sizeof(*w), GFP_NOFS);
 312         if (!w) {
 313                 mlog_errno(-ENOMEM);
 314                 goto bail;
 315         }
 316
 317         INIT_LIST_HEAD(&w->n_list);
 318         init_waitqueue_head(&w->n_event);
 319         ocfs2_node_map_init(&w->n_node_map);
 320         w->n_response_id = response_id;
 321         w->n_callback = NULL;
 322 bail:
 323         return w;
 324 }
 325
 326 static unsigned int ocfs2_new_response_id(struct ocfs2_super *osb)
 327 {
 328         unsigned int ret;
 329
 330         spin_lock(&osb->net_response_lock);
 331         ret = ++osb->net_response_ids;
 332         spin_unlock(&osb->net_response_lock);
 333
 334         return ret;
 335 }
 336
 337 static void ocfs2_dequeue_net_wait_ctxt(struct ocfs2_super *osb,
 338                                         struct ocfs2_net_wait_ctxt *w)
 339 {
 340         spin_lock(&osb->net_response_lock);
 341         list_del(&w->n_list);
 342         spin_unlock(&osb->net_response_lock);
 343 }
 344
 345 static void ocfs2_queue_net_wait_ctxt(struct ocfs2_super *osb,
 346                                       struct ocfs2_net_wait_ctxt *w)
 347 {
 348         spin_lock(&osb->net_response_lock);
 349         list_add_tail(&w->n_list,
 350                       &osb->net_response_list);
 351         spin_unlock(&osb->net_response_lock);
 352 }
 353
 354 static void __ocfs2_mark_node_responded(struct ocfs2_super *osb,
 355                                         struct ocfs2_net_wait_ctxt *w,
 356                                         int node_num)
 357 {
 358         assert_spin_locked(&osb->net_response_lock);
 359
 360         ocfs2_node_map_clear_bit(osb, &w->n_node_map, node_num);
 361         if (ocfs2_node_map_is_empty(osb, &w->n_node_map))
 362                 wake_up(&w->n_event);
 363 }
 364
 365 /* Intended to be called from the node down callback, we fake remove
 366  * the node from all our response contexts */
 367 void ocfs2_remove_node_from_vote_queues(struct ocfs2_super *osb,
 368                                         int node_num)
 369 {
 370         struct list_head *p;
 371         struct ocfs2_net_wait_ctxt *w = NULL;
 372
 373         spin_lock(&osb->net_response_lock);
 374
 375         list_for_each(p, &osb->net_response_list) {
 376                 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
 377
 378                 __ocfs2_mark_node_responded(osb, w, node_num);
 379         }
 380
 381         spin_unlock(&osb->net_response_lock);
 382 }
 383
 384 static int ocfs2_broadcast_vote(struct ocfs2_super *osb,
 385                                 struct ocfs2_vote_msg *request,
 386                                 unsigned int response_id,
 387                                 int *response,
 388                                 struct ocfs2_net_response_cb *callback)
 389 {
 390         int status, i, remote_err;
 391         struct ocfs2_net_wait_ctxt *w = NULL;
 392         int dequeued = 0;
 393
 394         mlog_entry_void();
 395
 396         w = ocfs2_new_net_wait_ctxt(response_id);
 397         if (!w) {
 398                 status = -ENOMEM;
 399                 mlog_errno(status);
 400                 goto bail;
 401         }
 402         w->n_callback = callback;
 403
 404         /* we're pretty much ready to go at this point, and this fills
 405          * in n_response which we need anyway... */
 406         ocfs2_queue_net_wait_ctxt(osb, w);
 407
 408         i = ocfs2_node_map_iterate(osb, &osb->mounted_map, 0);
 409
 410         while (i != O2NM_INVALID_NODE_NUM) {
 411                 if (i != osb->node_num) {
 412                         mlog(0, "trying to send request to node %i\n", i);
 413                         ocfs2_node_map_set_bit(osb, &w->n_node_map, i);
 414
 415                         remote_err = 0;
 416                         status = o2net_send_message(OCFS2_MESSAGE_TYPE_VOTE,
 417                                                     osb->net_key,
 418                                                     request,
 419                                                     sizeof(*request),
 420                                                     i,
 421                                                     &remote_err);
 422                         if (status == -ETIMEDOUT) {
 423                                 mlog(0, "remote node %d timed out!\n", i);
 424                                 status = -EAGAIN;
 425                                 goto bail;
 426                         }
 427                         if (remote_err < 0) {
 428                                 status = remote_err;
 429                                 mlog(0, "remote error %d on node %d!\n",
 430                                      remote_err, i);
 431                                 mlog_errno(status);
 432                                 goto bail;
 433                         }
 434                         if (status < 0) {
 435                                 mlog_errno(status);
 436                                 goto bail;
 437                         }
 438                 }
 439                 i++;
 440                 i = ocfs2_node_map_iterate(osb, &osb->mounted_map, i);
 441                 mlog(0, "next is %d, i am %d\n", i, osb->node_num);
 442         }
 443         mlog(0, "done sending, now waiting on responses...\n");
 444
 445         wait_event(w->n_event, ocfs2_node_map_is_empty(osb, &w->n_node_map));
 446
 447         ocfs2_dequeue_net_wait_ctxt(osb, w);
 448         dequeued = 1;
 449
 450         *response = w->n_response;
 451         status = 0;
 452 bail:
 453         if (w) {
 454                 if (!dequeued)
 455                         ocfs2_dequeue_net_wait_ctxt(osb, w);
 456                 kfree(w);
 457         }
 458
 459         mlog_exit(status);
 460         return status;
 461 }
 462
 463 static struct ocfs2_vote_msg * ocfs2_new_vote_request(struct ocfs2_super *osb,
 464                                                       u64 blkno,
 465                                                       unsigned int generation,
 466                                                       enum ocfs2_vote_request type)
 467 {
 468         struct ocfs2_vote_msg *request;
 469         struct ocfs2_msg_hdr *hdr;
 470
 471         BUG_ON(!ocfs2_is_valid_vote_request(type));
 472
 473         request = kzalloc(sizeof(*request), GFP_NOFS);
 474         if (!request) {
 475                 mlog_errno(-ENOMEM);
 476         } else {
 477                 hdr = &request->v_hdr;
 478                 hdr->h_node_num = cpu_to_be32(osb->node_num);
 479                 hdr->h_request = cpu_to_be32(type);
 480                 hdr->h_blkno = cpu_to_be64(blkno);
 481                 hdr->h_generation = cpu_to_be32(generation);
 482         }
 483
 484         return request;
 485 }
 486
 487 /* Complete the buildup of a new vote request and process the
 488  * broadcast return value. */
 489 static int ocfs2_do_request_vote(struct ocfs2_super *osb,
 490                                  struct ocfs2_vote_msg *request,
 491                                  struct ocfs2_net_response_cb *callback)
 492 {
 493         int status, response = -EBUSY;
 494         unsigned int response_id;
 495         struct ocfs2_msg_hdr *hdr;
 496
 497         response_id = ocfs2_new_response_id(osb);
 498
 499         hdr = &request->v_hdr;
 500         hdr->h_response_id = cpu_to_be32(response_id);
 501
 502         status = ocfs2_broadcast_vote(osb, request, response_id, &response,
 503                                       callback);
 504         if (status < 0) {
 505                 mlog_errno(status);
 506                 goto bail;
 507         }
 508
 509         status = response;
 510 bail:
 511
 512         return status;
 513 }
 514
 515 int ocfs2_request_mount_vote(struct ocfs2_super *osb)
 516 {
 517         int status;
 518         struct ocfs2_vote_msg *request = NULL;
 519
 520         request = ocfs2_new_vote_request(osb, 0ULL, 0, OCFS2_VOTE_REQ_MOUNT);
 521         if (!request) {
 522                 status = -ENOMEM;
 523                 goto bail;
 524         }
 525
 526         status = -EAGAIN;
 527         while (status == -EAGAIN) {
 528                 if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
 529                     signal_pending(current)) {
 530                         status = -ERESTARTSYS;
 531                         goto bail;
 532                 }
 533
 534                 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
 535                                            osb->node_num)) {
 536                         status = 0;
 537                         goto bail;
 538                 }
 539
 540                 status = ocfs2_do_request_vote(osb, request, NULL);
 541         }
 542
 543 bail:
 544         kfree(request);
 545         return status;
 546 }
 547
 548 int ocfs2_request_umount_vote(struct ocfs2_super *osb)
 549 {
 550         int status;
 551         struct ocfs2_vote_msg *request = NULL;
 552
 553         request = ocfs2_new_vote_request(osb, 0ULL, 0, OCFS2_VOTE_REQ_UMOUNT);
 554         if (!request) {
 555                 status = -ENOMEM;
 556                 goto bail;
 557         }
 558
 559         status = -EAGAIN;
 560         while (status == -EAGAIN) {
 561                 /* Do not check signals on this vote... We really want
 562                  * this one to go all the way through. */
 563
 564                 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
 565                                            osb->node_num)) {
 566                         status = 0;
 567                         goto bail;
 568                 }
 569
 570                 status = ocfs2_do_request_vote(osb, request, NULL);
 571         }
 572
 573 bail:
 574         kfree(request);
 575         return status;
 576 }
 577
 578 /* TODO: This should eventually be a hash table! */
 579 static struct ocfs2_net_wait_ctxt * __ocfs2_find_net_wait_ctxt(struct ocfs2_super *osb,
 580                                                                u32 response_id)
 581 {
 582         struct list_head *p;
 583         struct ocfs2_net_wait_ctxt *w = NULL;
 584
 585         list_for_each(p, &osb->net_response_list) {
 586                 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
 587                 if (response_id == w->n_response_id)
 588                         break;
 589                 w = NULL;
 590         }
 591
 592         return w;
 593 }
 594
 595 /* Translate response codes into local node errno values */
 596 static inline int ocfs2_translate_response(int response)
 597 {
 598         int ret;
 599
 600         switch (response) {
 601         case OCFS2_RESPONSE_OK:
 602                 ret = 0;
 603                 break;
 604
 605         case OCFS2_RESPONSE_BUSY:
 606                 ret = -EBUSY;
 607                 break;
 608
 609         default:
 610                 ret = -EINVAL;
 611         }
 612
 613         return ret;
 614 }
 615
 616 static int ocfs2_handle_response_message(struct o2net_msg *msg,
 617                                          u32 len,
 618                                          void *data, void **ret_data)
 619 {
 620         unsigned int response_id, node_num;
 621         int response_status;
 622         struct ocfs2_super *osb = data;
 623         struct ocfs2_response_msg *resp;
 624         struct ocfs2_net_wait_ctxt * w;
 625         struct ocfs2_net_response_cb *resp_cb;
 626
 627         resp = (struct ocfs2_response_msg *) msg->buf;
 628
 629         response_id = be32_to_cpu(resp->r_hdr.h_response_id);
 630         node_num = be32_to_cpu(resp->r_hdr.h_node_num);
 631         response_status =
 632                 ocfs2_translate_response(be32_to_cpu(resp->r_response));
 633
 634         mlog(0, "received response message:\n");
 635         mlog(0, "h_response_id = %u\n", response_id);
 636         mlog(0, "h_request = %u\n", be32_to_cpu(resp->r_hdr.h_request));
 637         mlog(0, "h_blkno = %llu\n",
 638              (unsigned long long)be64_to_cpu(resp->r_hdr.h_blkno));
 639         mlog(0, "h_generation = %u\n", be32_to_cpu(resp->r_hdr.h_generation));
 640         mlog(0, "h_node_num = %u\n", node_num);
 641         mlog(0, "r_response = %d\n", response_status);
 642
 643         spin_lock(&osb->net_response_lock);
 644         w = __ocfs2_find_net_wait_ctxt(osb, response_id);
 645         if (!w) {
 646                 mlog(0, "request not found!\n");
 647                 goto bail;
 648         }
 649         resp_cb = w->n_callback;
 650
 651         if (response_status && (!w->n_response)) {
 652                 /* we only really need one negative response so don't
 653                  * set it twice. */
 654                 w->n_response = response_status;
 655         }
 656
 657         if (resp_cb) {
 658                 spin_unlock(&osb->net_response_lock);
 659
 660                 resp_cb->rc_cb(resp_cb->rc_priv, resp);
 661
 662                 spin_lock(&osb->net_response_lock);
 663         }
 664
 665         __ocfs2_mark_node_responded(osb, w, node_num);
 666 bail:
 667         spin_unlock(&osb->net_response_lock);
 668
 669         return 0;
 670 }
 671
 672 static int ocfs2_handle_vote_message(struct o2net_msg *msg,
 673                                      u32 len,
 674                                      void *data, void **ret_data)
 675 {
 676         int status;
 677         struct ocfs2_super *osb = data;
 678         struct ocfs2_vote_work *work;
 679
 680         work = kmalloc(sizeof(struct ocfs2_vote_work), GFP_NOFS);
 681         if (!work) {
 682                 status = -ENOMEM;
 683                 mlog_errno(status);
 684                 goto bail;
 685         }
 686
 687         INIT_LIST_HEAD(&work->w_list);
 688         memcpy(&work->w_msg, msg->buf, sizeof(struct ocfs2_vote_msg));
 689
 690         mlog(0, "scheduling vote request:\n");
 691         mlog(0, "h_response_id = %u\n",
 692              be32_to_cpu(work->w_msg.v_hdr.h_response_id));
 693         mlog(0, "h_request = %u\n", be32_to_cpu(work->w_msg.v_hdr.h_request));
 694         mlog(0, "h_blkno = %llu\n",
 695              (unsigned long long)be64_to_cpu(work->w_msg.v_hdr.h_blkno));
 696         mlog(0, "h_generation = %u\n",
 697              be32_to_cpu(work->w_msg.v_hdr.h_generation));
 698         mlog(0, "h_node_num = %u\n",
 699              be32_to_cpu(work->w_msg.v_hdr.h_node_num));
 700
 701         spin_lock(&osb->vote_task_lock);
 702         list_add_tail(&work->w_list, &osb->vote_list);
 703         osb->vote_count++;
 704         spin_unlock(&osb->vote_task_lock);
 705
 706         ocfs2_kick_vote_thread(osb);
 707
 708         status = 0;
 709 bail:
 710         return status;
 711 }
 712
 713 void ocfs2_unregister_net_handlers(struct ocfs2_super *osb)
 714 {
 715         if (!osb->net_key)
 716                 return;
 717
 718         o2net_unregister_handler_list(&osb->osb_net_handlers);
 719
 720         if (!list_empty(&osb->net_response_list))
 721                 mlog(ML_ERROR, "net response list not empty!\n");
 722
 723         osb->net_key = 0;
 724 }
 725
 726 int ocfs2_register_net_handlers(struct ocfs2_super *osb)
 727 {
 728         int status = 0;
 729
 730         if (ocfs2_mount_local(osb))
 731                 return 0;
 732
 733         status = o2net_register_handler(OCFS2_MESSAGE_TYPE_RESPONSE,
 734                                         osb->net_key,
 735                                         sizeof(struct ocfs2_response_msg),
 736                                         ocfs2_handle_response_message,
 737                                         osb, NULL, &osb->osb_net_handlers);
 738         if (status) {
 739                 mlog_errno(status);
 740                 goto bail;
 741         }
 742
 743         status = o2net_register_handler(OCFS2_MESSAGE_TYPE_VOTE,
 744                                         osb->net_key,
 745                                         sizeof(struct ocfs2_vote_msg),
 746                                         ocfs2_handle_vote_message,
 747                                         osb, NULL, &osb->osb_net_handlers);
 748         if (status) {
 749                 mlog_errno(status);
 750                 goto bail;
 751         }
 752 bail:
 753         if (status < 0)
 754                 ocfs2_unregister_net_handlers(osb);
 755
 756         return status;
 757 }