ocfs2: local mounts
[linux-2.6] / 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 }