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