blob: 53049a20419768b3c00eea23fb6acf4ff4c71c0a [file] [log] [blame]
Mark Fashehccd979b2005-12-15 14:31:24 -08001/* -*- 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)
56struct 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
69struct 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 __be32 v_unlink_namelen;
78 __be64 v_unlink_parent;
79 u8 v_unlink_dirent[OCFS2_VOTE_FILENAME_LEN];
80};
81
82/* Responses are given these values to maintain backwards
83 * compatibility with older ocfs2 versions */
84#define OCFS2_RESPONSE_OK (0)
85#define OCFS2_RESPONSE_BUSY (-16)
86#define OCFS2_RESPONSE_BAD_MSG (-22)
87
88struct ocfs2_response_msg
89{
90 struct ocfs2_msg_hdr r_hdr;
91 __be32 r_response;
92 __be32 r_orphaned_slot;
93};
94
95struct ocfs2_vote_work {
96 struct list_head w_list;
97 struct ocfs2_vote_msg w_msg;
98};
99
100enum ocfs2_vote_request {
101 OCFS2_VOTE_REQ_INVALID = 0,
102 OCFS2_VOTE_REQ_DELETE,
103 OCFS2_VOTE_REQ_UNLINK,
104 OCFS2_VOTE_REQ_RENAME,
105 OCFS2_VOTE_REQ_MOUNT,
106 OCFS2_VOTE_REQ_UMOUNT,
107 OCFS2_VOTE_REQ_LAST
108};
109
110static inline int ocfs2_is_valid_vote_request(int request)
111{
112 return OCFS2_VOTE_REQ_INVALID < request &&
113 request < OCFS2_VOTE_REQ_LAST;
114}
115
116typedef void (*ocfs2_net_response_callback)(void *priv,
117 struct ocfs2_response_msg *resp);
118struct ocfs2_net_response_cb {
119 ocfs2_net_response_callback rc_cb;
120 void *rc_priv;
121};
122
123struct ocfs2_net_wait_ctxt {
124 struct list_head n_list;
125 u32 n_response_id;
126 wait_queue_head_t n_event;
127 struct ocfs2_node_map n_node_map;
128 int n_response; /* an agreggate response. 0 if
129 * all nodes are go, < 0 on any
130 * negative response from any
131 * node or network error. */
132 struct ocfs2_net_response_cb *n_callback;
133};
134
135static void ocfs2_process_mount_request(struct ocfs2_super *osb,
136 unsigned int node_num)
137{
138 mlog(0, "MOUNT vote from node %u\n", node_num);
139 /* The other node only sends us this message when he has an EX
140 * on the superblock, so our recovery threads (if having been
141 * launched) are waiting on it.*/
142 ocfs2_recovery_map_clear(osb, node_num);
143 ocfs2_node_map_set_bit(osb, &osb->mounted_map, node_num);
144
145 /* We clear the umount map here because a node may have been
146 * previously mounted, safely unmounted but never stopped
147 * heartbeating - in which case we'd have a stale entry. */
148 ocfs2_node_map_clear_bit(osb, &osb->umount_map, node_num);
149}
150
151static void ocfs2_process_umount_request(struct ocfs2_super *osb,
152 unsigned int node_num)
153{
154 mlog(0, "UMOUNT vote from node %u\n", node_num);
155 ocfs2_node_map_clear_bit(osb, &osb->mounted_map, node_num);
156 ocfs2_node_map_set_bit(osb, &osb->umount_map, node_num);
157}
158
159void ocfs2_mark_inode_remotely_deleted(struct inode *inode)
160{
161 struct ocfs2_inode_info *oi = OCFS2_I(inode);
162
163 assert_spin_locked(&oi->ip_lock);
164 /* We set the SKIP_DELETE flag on the inode so we don't try to
165 * delete it in delete_inode ourselves, thus avoiding
166 * unecessary lock pinging. If the other node failed to wipe
167 * the inode as a result of a crash, then recovery will pick
168 * up the slack. */
169 oi->ip_flags |= OCFS2_INODE_DELETED|OCFS2_INODE_SKIP_DELETE;
170}
171
172static int ocfs2_process_delete_request(struct inode *inode,
173 int *orphaned_slot)
174{
175 int response = OCFS2_RESPONSE_BUSY;
176
177 mlog(0, "DELETE vote on inode %lu, read lnk_cnt = %u, slot = %d\n",
178 inode->i_ino, inode->i_nlink, *orphaned_slot);
179
180 spin_lock(&OCFS2_I(inode)->ip_lock);
181
182 /* Whatever our vote response is, we want to make sure that
183 * the orphaned slot is recorded properly on this node *and*
184 * on the requesting node. Technically, if the requesting node
185 * did not know which slot the inode is orphaned in but we
186 * respond with BUSY he doesn't actually need the orphaned
187 * slot, but it doesn't hurt to do it here anyway. */
188 if ((*orphaned_slot) != OCFS2_INVALID_SLOT) {
189 mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot !=
190 OCFS2_INVALID_SLOT &&
191 OCFS2_I(inode)->ip_orphaned_slot !=
192 (*orphaned_slot),
Mark Fashehb06970532006-03-03 10:24:33 -0800193 "Inode %llu: This node thinks it's "
Mark Fashehccd979b2005-12-15 14:31:24 -0800194 "orphaned in slot %d, messaged it's in %d\n",
Mark Fashehb06970532006-03-03 10:24:33 -0800195 (unsigned long long)OCFS2_I(inode)->ip_blkno,
Mark Fashehccd979b2005-12-15 14:31:24 -0800196 OCFS2_I(inode)->ip_orphaned_slot,
197 *orphaned_slot);
198
Mark Fashehb06970532006-03-03 10:24:33 -0800199 mlog(0, "Setting orphaned slot for inode %llu to %d\n",
200 (unsigned long long)OCFS2_I(inode)->ip_blkno,
201 *orphaned_slot);
Mark Fashehccd979b2005-12-15 14:31:24 -0800202
203 OCFS2_I(inode)->ip_orphaned_slot = *orphaned_slot;
204 } else {
Mark Fashehb06970532006-03-03 10:24:33 -0800205 mlog(0, "Sending back orphaned slot %d for inode %llu\n",
Mark Fashehccd979b2005-12-15 14:31:24 -0800206 OCFS2_I(inode)->ip_orphaned_slot,
Mark Fashehb06970532006-03-03 10:24:33 -0800207 (unsigned long long)OCFS2_I(inode)->ip_blkno);
Mark Fashehccd979b2005-12-15 14:31:24 -0800208
209 *orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
210 }
211
212 /* vote no if the file is still open. */
213 if (OCFS2_I(inode)->ip_open_count) {
214 mlog(0, "open count = %u\n",
215 OCFS2_I(inode)->ip_open_count);
216 spin_unlock(&OCFS2_I(inode)->ip_lock);
217 goto done;
218 }
219 spin_unlock(&OCFS2_I(inode)->ip_lock);
220
221 /* directories are a bit ugly... What if someone is sitting in
222 * it? We want to make sure the inode is removed completely as
223 * a result of the iput in process_vote. */
224 if (S_ISDIR(inode->i_mode) && (atomic_read(&inode->i_count) != 1)) {
225 mlog(0, "i_count = %u\n", atomic_read(&inode->i_count));
226 goto done;
227 }
228
229 if (filemap_fdatawrite(inode->i_mapping)) {
Mark Fashehb06970532006-03-03 10:24:33 -0800230 mlog(ML_ERROR, "Could not sync inode %llu for delete!\n",
231 (unsigned long long)OCFS2_I(inode)->ip_blkno);
Mark Fashehccd979b2005-12-15 14:31:24 -0800232 goto done;
233 }
234 sync_mapping_buffers(inode->i_mapping);
235 truncate_inode_pages(inode->i_mapping, 0);
236 ocfs2_extent_map_trunc(inode, 0);
237
238 spin_lock(&OCFS2_I(inode)->ip_lock);
239 /* double check open count - someone might have raced this
240 * thread into ocfs2_file_open while we were writing out
241 * data. If we're to allow a wipe of this inode now, we *must*
242 * hold the spinlock until we've marked it. */
243 if (OCFS2_I(inode)->ip_open_count) {
244 mlog(0, "Raced to wipe! open count = %u\n",
245 OCFS2_I(inode)->ip_open_count);
246 spin_unlock(&OCFS2_I(inode)->ip_lock);
247 goto done;
248 }
249
250 /* Mark the inode as being wiped from disk. */
251 ocfs2_mark_inode_remotely_deleted(inode);
252 spin_unlock(&OCFS2_I(inode)->ip_lock);
253
254 /* Not sure this is necessary anymore. */
255 d_prune_aliases(inode);
256
257 /* If we get here, then we're voting 'yes', so commit the
258 * delete on our side. */
259 response = OCFS2_RESPONSE_OK;
260done:
261 return response;
262}
263
264static int ocfs2_match_dentry(struct dentry *dentry,
265 u64 parent_blkno,
266 unsigned int namelen,
267 const char *name)
268{
269 struct inode *parent;
270
271 if (!dentry->d_parent) {
272 mlog(0, "Detached from parent.\n");
273 return 0;
274 }
275
276 parent = dentry->d_parent->d_inode;
277 /* Negative parent dentry? */
278 if (!parent)
279 return 0;
280
281 /* Name is in a different directory. */
282 if (OCFS2_I(parent)->ip_blkno != parent_blkno)
283 return 0;
284
285 if (dentry->d_name.len != namelen)
286 return 0;
287
288 /* comparison above guarantees this is safe. */
289 if (memcmp(dentry->d_name.name, name, namelen))
290 return 0;
291
292 return 1;
293}
294
295static void ocfs2_process_dentry_request(struct inode *inode,
296 int rename,
297 unsigned int new_nlink,
298 u64 parent_blkno,
299 unsigned int namelen,
300 const char *name)
301{
302 struct dentry *dentry = NULL;
303 struct list_head *p;
304 struct ocfs2_inode_info *oi = OCFS2_I(inode);
305
Mark Fashehb06970532006-03-03 10:24:33 -0800306 mlog(0, "parent %llu, namelen = %u, name = %.*s\n",
307 (unsigned long long)parent_blkno, namelen, namelen, name);
Mark Fashehccd979b2005-12-15 14:31:24 -0800308
309 spin_lock(&dcache_lock);
310
311 /* Another node is removing this name from the system. It is
312 * up to us to find the corresponding dentry and if it exists,
313 * unhash it from the dcache. */
314 list_for_each(p, &inode->i_dentry) {
315 dentry = list_entry(p, struct dentry, d_alias);
316
317 if (ocfs2_match_dentry(dentry, parent_blkno, namelen, name)) {
318 mlog(0, "dentry found: %.*s\n",
319 dentry->d_name.len, dentry->d_name.name);
320
321 dget_locked(dentry);
322 break;
323 }
324
325 dentry = NULL;
326 }
327
328 spin_unlock(&dcache_lock);
329
330 if (dentry) {
331 d_delete(dentry);
332 dput(dentry);
333 }
334
335 /* rename votes don't send link counts */
336 if (!rename) {
337 mlog(0, "new_nlink = %u\n", new_nlink);
338
339 /* We don't have the proper locks here to directly
340 * change i_nlink and besides, the vote is sent
341 * *before* the operation so it may have failed on the
342 * other node. This passes a hint to ocfs2_drop_inode
343 * to force ocfs2_delete_inode, who will take the
344 * proper cluster locks to sort things out. */
345 if (new_nlink == 0) {
346 spin_lock(&oi->ip_lock);
347 oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
348 spin_unlock(&OCFS2_I(inode)->ip_lock);
349 }
350 }
351}
352
353static void ocfs2_process_vote(struct ocfs2_super *osb,
354 struct ocfs2_vote_msg *msg)
355{
356 int net_status, vote_response;
357 int orphaned_slot = 0;
358 int rename = 0;
359 unsigned int node_num, generation, new_nlink, namelen;
360 u64 blkno, parent_blkno;
361 enum ocfs2_vote_request request;
362 struct inode *inode = NULL;
363 struct ocfs2_msg_hdr *hdr = &msg->v_hdr;
364 struct ocfs2_response_msg response;
365
366 /* decode the network mumbo jumbo into local variables. */
367 request = be32_to_cpu(hdr->h_request);
368 blkno = be64_to_cpu(hdr->h_blkno);
369 generation = be32_to_cpu(hdr->h_generation);
370 node_num = be32_to_cpu(hdr->h_node_num);
371 if (request == OCFS2_VOTE_REQ_DELETE)
372 orphaned_slot = be32_to_cpu(msg->md1.v_orphaned_slot);
373
Mark Fashehb06970532006-03-03 10:24:33 -0800374 mlog(0, "processing vote: request = %u, blkno = %llu, "
Mark Fashehccd979b2005-12-15 14:31:24 -0800375 "generation = %u, node_num = %u, priv1 = %u\n", request,
Mark Fashehb06970532006-03-03 10:24:33 -0800376 (unsigned long long)blkno, generation, node_num,
377 be32_to_cpu(msg->md1.v_generic1));
Mark Fashehccd979b2005-12-15 14:31:24 -0800378
379 if (!ocfs2_is_valid_vote_request(request)) {
380 mlog(ML_ERROR, "Invalid vote request %d from node %u\n",
381 request, node_num);
382 vote_response = OCFS2_RESPONSE_BAD_MSG;
383 goto respond;
384 }
385
386 vote_response = OCFS2_RESPONSE_OK;
387
388 switch (request) {
389 case OCFS2_VOTE_REQ_UMOUNT:
390 ocfs2_process_umount_request(osb, node_num);
391 goto respond;
392 case OCFS2_VOTE_REQ_MOUNT:
393 ocfs2_process_mount_request(osb, node_num);
394 goto respond;
395 default:
396 /* avoids a gcc warning */
397 break;
398 }
399
400 /* We cannot process the remaining message types before we're
401 * fully mounted. It's perfectly safe however to send a 'yes'
402 * response as we can't possibly have any of the state they're
403 * asking us to modify yet. */
404 if (atomic_read(&osb->vol_state) == VOLUME_INIT)
405 goto respond;
406
407 /* If we get here, then the request is against an inode. */
408 inode = ocfs2_ilookup_for_vote(osb, blkno,
409 request == OCFS2_VOTE_REQ_DELETE);
410
411 /* Not finding the inode is perfectly valid - it means we're
412 * not interested in what the other node is about to do to it
413 * so in those cases we automatically respond with an
414 * affirmative. Cluster locking ensures that we won't race
415 * interest in the inode with this vote request. */
416 if (!inode)
417 goto respond;
418
419 /* Check generation values. It's possible for us to get a
420 * request against a stale inode. If so then we proceed as if
421 * we had not found an inode in the first place. */
422 if (inode->i_generation != generation) {
423 mlog(0, "generation passed %u != inode generation = %u, "
Mark Fashehb06970532006-03-03 10:24:33 -0800424 "ip_flags = %x, ip_blkno = %llu, msg %llu, i_count = %u, "
425 "message type = %u\n", generation, inode->i_generation,
426 OCFS2_I(inode)->ip_flags,
427 (unsigned long long)OCFS2_I(inode)->ip_blkno,
428 (unsigned long long)blkno, atomic_read(&inode->i_count),
429 request);
Mark Fashehccd979b2005-12-15 14:31:24 -0800430 iput(inode);
431 inode = NULL;
432 goto respond;
433 }
434
435 switch (request) {
436 case OCFS2_VOTE_REQ_DELETE:
437 vote_response = ocfs2_process_delete_request(inode,
438 &orphaned_slot);
439 break;
440 case OCFS2_VOTE_REQ_RENAME:
441 rename = 1;
442 /* fall through */
443 case OCFS2_VOTE_REQ_UNLINK:
444 parent_blkno = be64_to_cpu(msg->v_unlink_parent);
445 namelen = be32_to_cpu(msg->v_unlink_namelen);
446 /* new_nlink will be ignored in case of a rename vote */
447 new_nlink = be32_to_cpu(msg->md1.v_nlink);
448 ocfs2_process_dentry_request(inode, rename, new_nlink,
449 parent_blkno, namelen,
450 msg->v_unlink_dirent);
451 break;
452 default:
453 mlog(ML_ERROR, "node %u, invalid request: %u\n",
454 node_num, request);
455 vote_response = OCFS2_RESPONSE_BAD_MSG;
456 }
457
458respond:
459 /* Response struture is small so we just put it on the stack
460 * and stuff it inline. */
461 memset(&response, 0, sizeof(struct ocfs2_response_msg));
462 response.r_hdr.h_response_id = hdr->h_response_id;
463 response.r_hdr.h_blkno = hdr->h_blkno;
464 response.r_hdr.h_generation = hdr->h_generation;
465 response.r_hdr.h_node_num = cpu_to_be32(osb->node_num);
466 response.r_response = cpu_to_be32(vote_response);
467 response.r_orphaned_slot = cpu_to_be32(orphaned_slot);
468
469 net_status = o2net_send_message(OCFS2_MESSAGE_TYPE_RESPONSE,
470 osb->net_key,
471 &response,
472 sizeof(struct ocfs2_response_msg),
473 node_num,
474 NULL);
475 /* We still want to error print for ENOPROTOOPT here. The
476 * sending node shouldn't have unregistered his net handler
477 * without sending an unmount vote 1st */
478 if (net_status < 0
479 && net_status != -ETIMEDOUT
480 && net_status != -ENOTCONN)
481 mlog(ML_ERROR, "message to node %u fails with error %d!\n",
482 node_num, net_status);
483
484 if (inode)
485 iput(inode);
486}
487
488static void ocfs2_vote_thread_do_work(struct ocfs2_super *osb)
489{
490 unsigned long processed;
491 struct ocfs2_lock_res *lockres;
492 struct ocfs2_vote_work *work;
493
494 mlog_entry_void();
495
496 spin_lock(&osb->vote_task_lock);
497 /* grab this early so we know to try again if a state change and
498 * wake happens part-way through our work */
499 osb->vote_work_sequence = osb->vote_wake_sequence;
500
501 processed = osb->blocked_lock_count;
502 while (processed) {
503 BUG_ON(list_empty(&osb->blocked_lock_list));
504
505 lockres = list_entry(osb->blocked_lock_list.next,
506 struct ocfs2_lock_res, l_blocked_list);
507 list_del_init(&lockres->l_blocked_list);
508 osb->blocked_lock_count--;
509 spin_unlock(&osb->vote_task_lock);
510
511 BUG_ON(!processed);
512 processed--;
513
514 ocfs2_process_blocked_lock(osb, lockres);
515
516 spin_lock(&osb->vote_task_lock);
517 }
518
519 while (osb->vote_count) {
520 BUG_ON(list_empty(&osb->vote_list));
521 work = list_entry(osb->vote_list.next,
522 struct ocfs2_vote_work, w_list);
523 list_del(&work->w_list);
524 osb->vote_count--;
525 spin_unlock(&osb->vote_task_lock);
526
527 ocfs2_process_vote(osb, &work->w_msg);
528 kfree(work);
529
530 spin_lock(&osb->vote_task_lock);
531 }
532 spin_unlock(&osb->vote_task_lock);
533
534 mlog_exit_void();
535}
536
537static int ocfs2_vote_thread_lists_empty(struct ocfs2_super *osb)
538{
539 int empty = 0;
540
541 spin_lock(&osb->vote_task_lock);
542 if (list_empty(&osb->blocked_lock_list) &&
543 list_empty(&osb->vote_list))
544 empty = 1;
545
546 spin_unlock(&osb->vote_task_lock);
547 return empty;
548}
549
550static int ocfs2_vote_thread_should_wake(struct ocfs2_super *osb)
551{
552 int should_wake = 0;
553
554 spin_lock(&osb->vote_task_lock);
555 if (osb->vote_work_sequence != osb->vote_wake_sequence)
556 should_wake = 1;
557 spin_unlock(&osb->vote_task_lock);
558
559 return should_wake;
560}
561
562int ocfs2_vote_thread(void *arg)
563{
564 int status = 0;
565 struct ocfs2_super *osb = arg;
566
567 /* only quit once we've been asked to stop and there is no more
568 * work available */
569 while (!(kthread_should_stop() &&
570 ocfs2_vote_thread_lists_empty(osb))) {
571
572 wait_event_interruptible(osb->vote_event,
573 ocfs2_vote_thread_should_wake(osb) ||
574 kthread_should_stop());
575
576 mlog(0, "vote_thread: awoken\n");
577
578 ocfs2_vote_thread_do_work(osb);
579 }
580
581 osb->vote_task = NULL;
582 return status;
583}
584
585static struct ocfs2_net_wait_ctxt *ocfs2_new_net_wait_ctxt(unsigned int response_id)
586{
587 struct ocfs2_net_wait_ctxt *w;
588
589 w = kcalloc(1, sizeof(*w), GFP_KERNEL);
590 if (!w) {
591 mlog_errno(-ENOMEM);
592 goto bail;
593 }
594
595 INIT_LIST_HEAD(&w->n_list);
596 init_waitqueue_head(&w->n_event);
597 ocfs2_node_map_init(&w->n_node_map);
598 w->n_response_id = response_id;
599 w->n_callback = NULL;
600bail:
601 return w;
602}
603
604static unsigned int ocfs2_new_response_id(struct ocfs2_super *osb)
605{
606 unsigned int ret;
607
608 spin_lock(&osb->net_response_lock);
609 ret = ++osb->net_response_ids;
610 spin_unlock(&osb->net_response_lock);
611
612 return ret;
613}
614
615static void ocfs2_dequeue_net_wait_ctxt(struct ocfs2_super *osb,
616 struct ocfs2_net_wait_ctxt *w)
617{
618 spin_lock(&osb->net_response_lock);
619 list_del(&w->n_list);
620 spin_unlock(&osb->net_response_lock);
621}
622
623static void ocfs2_queue_net_wait_ctxt(struct ocfs2_super *osb,
624 struct ocfs2_net_wait_ctxt *w)
625{
626 spin_lock(&osb->net_response_lock);
627 list_add_tail(&w->n_list,
628 &osb->net_response_list);
629 spin_unlock(&osb->net_response_lock);
630}
631
632static void __ocfs2_mark_node_responded(struct ocfs2_super *osb,
633 struct ocfs2_net_wait_ctxt *w,
634 int node_num)
635{
636 assert_spin_locked(&osb->net_response_lock);
637
638 ocfs2_node_map_clear_bit(osb, &w->n_node_map, node_num);
639 if (ocfs2_node_map_is_empty(osb, &w->n_node_map))
640 wake_up(&w->n_event);
641}
642
643/* Intended to be called from the node down callback, we fake remove
644 * the node from all our response contexts */
645void ocfs2_remove_node_from_vote_queues(struct ocfs2_super *osb,
646 int node_num)
647{
648 struct list_head *p;
649 struct ocfs2_net_wait_ctxt *w = NULL;
650
651 spin_lock(&osb->net_response_lock);
652
653 list_for_each(p, &osb->net_response_list) {
654 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
655
656 __ocfs2_mark_node_responded(osb, w, node_num);
657 }
658
659 spin_unlock(&osb->net_response_lock);
660}
661
662static int ocfs2_broadcast_vote(struct ocfs2_super *osb,
663 struct ocfs2_vote_msg *request,
664 unsigned int response_id,
665 int *response,
666 struct ocfs2_net_response_cb *callback)
667{
668 int status, i, remote_err;
669 struct ocfs2_net_wait_ctxt *w = NULL;
670 int dequeued = 0;
671
672 mlog_entry_void();
673
674 w = ocfs2_new_net_wait_ctxt(response_id);
675 if (!w) {
676 status = -ENOMEM;
677 mlog_errno(status);
678 goto bail;
679 }
680 w->n_callback = callback;
681
682 /* we're pretty much ready to go at this point, and this fills
683 * in n_response which we need anyway... */
684 ocfs2_queue_net_wait_ctxt(osb, w);
685
686 i = ocfs2_node_map_iterate(osb, &osb->mounted_map, 0);
687
688 while (i != O2NM_INVALID_NODE_NUM) {
689 if (i != osb->node_num) {
690 mlog(0, "trying to send request to node %i\n", i);
691 ocfs2_node_map_set_bit(osb, &w->n_node_map, i);
692
693 remote_err = 0;
694 status = o2net_send_message(OCFS2_MESSAGE_TYPE_VOTE,
695 osb->net_key,
696 request,
697 sizeof(*request),
698 i,
699 &remote_err);
700 if (status == -ETIMEDOUT) {
701 mlog(0, "remote node %d timed out!\n", i);
702 status = -EAGAIN;
703 goto bail;
704 }
705 if (remote_err < 0) {
706 status = remote_err;
707 mlog(0, "remote error %d on node %d!\n",
708 remote_err, i);
709 mlog_errno(status);
710 goto bail;
711 }
712 if (status < 0) {
713 mlog_errno(status);
714 goto bail;
715 }
716 }
717 i++;
718 i = ocfs2_node_map_iterate(osb, &osb->mounted_map, i);
719 mlog(0, "next is %d, i am %d\n", i, osb->node_num);
720 }
721 mlog(0, "done sending, now waiting on responses...\n");
722
723 wait_event(w->n_event, ocfs2_node_map_is_empty(osb, &w->n_node_map));
724
725 ocfs2_dequeue_net_wait_ctxt(osb, w);
726 dequeued = 1;
727
728 *response = w->n_response;
729 status = 0;
730bail:
731 if (w) {
732 if (!dequeued)
733 ocfs2_dequeue_net_wait_ctxt(osb, w);
734 kfree(w);
735 }
736
737 mlog_exit(status);
738 return status;
739}
740
741static struct ocfs2_vote_msg * ocfs2_new_vote_request(struct ocfs2_super *osb,
742 u64 blkno,
743 unsigned int generation,
744 enum ocfs2_vote_request type,
745 u32 priv)
746{
747 struct ocfs2_vote_msg *request;
748 struct ocfs2_msg_hdr *hdr;
749
750 BUG_ON(!ocfs2_is_valid_vote_request(type));
751
752 request = kcalloc(1, sizeof(*request), GFP_KERNEL);
753 if (!request) {
754 mlog_errno(-ENOMEM);
755 } else {
756 hdr = &request->v_hdr;
757 hdr->h_node_num = cpu_to_be32(osb->node_num);
758 hdr->h_request = cpu_to_be32(type);
759 hdr->h_blkno = cpu_to_be64(blkno);
760 hdr->h_generation = cpu_to_be32(generation);
761
762 request->md1.v_generic1 = cpu_to_be32(priv);
763 }
764
765 return request;
766}
767
768/* Complete the buildup of a new vote request and process the
769 * broadcast return value. */
770static int ocfs2_do_request_vote(struct ocfs2_super *osb,
771 struct ocfs2_vote_msg *request,
772 struct ocfs2_net_response_cb *callback)
773{
774 int status, response;
775 unsigned int response_id;
776 struct ocfs2_msg_hdr *hdr;
777
778 response_id = ocfs2_new_response_id(osb);
779
780 hdr = &request->v_hdr;
781 hdr->h_response_id = cpu_to_be32(response_id);
782
783 status = ocfs2_broadcast_vote(osb, request, response_id, &response,
784 callback);
785 if (status < 0) {
786 mlog_errno(status);
787 goto bail;
788 }
789
790 status = response;
791bail:
792
793 return status;
794}
795
796static int ocfs2_request_vote(struct inode *inode,
797 struct ocfs2_vote_msg *request,
798 struct ocfs2_net_response_cb *callback)
799{
800 int status;
801 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
802
803 if (ocfs2_inode_is_new(inode))
804 return 0;
805
806 status = -EAGAIN;
807 while (status == -EAGAIN) {
808 if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
809 signal_pending(current))
810 return -ERESTARTSYS;
811
812 status = ocfs2_super_lock(osb, 0);
813 if (status < 0) {
814 mlog_errno(status);
815 break;
816 }
817
818 status = 0;
819 if (!ocfs2_node_map_is_only(osb, &osb->mounted_map,
820 osb->node_num))
821 status = ocfs2_do_request_vote(osb, request, callback);
822
823 ocfs2_super_unlock(osb, 0);
824 }
825 return status;
826}
827
828static void ocfs2_delete_response_cb(void *priv,
829 struct ocfs2_response_msg *resp)
830{
831 int orphaned_slot, node;
832 struct inode *inode = priv;
833
834 orphaned_slot = be32_to_cpu(resp->r_orphaned_slot);
835 node = be32_to_cpu(resp->r_hdr.h_node_num);
Mark Fashehb06970532006-03-03 10:24:33 -0800836 mlog(0, "node %d tells us that inode %llu is orphaned in slot %d\n",
837 node, (unsigned long long)OCFS2_I(inode)->ip_blkno,
838 orphaned_slot);
Mark Fashehccd979b2005-12-15 14:31:24 -0800839
840 /* The other node may not actually know which slot the inode
841 * is orphaned in. */
842 if (orphaned_slot == OCFS2_INVALID_SLOT)
843 return;
844
845 /* Ok, the responding node knows which slot this inode is
846 * orphaned in. We verify that the information is correct and
847 * then record this in the inode. ocfs2_delete_inode will use
848 * this information to determine which lock to take. */
849 spin_lock(&OCFS2_I(inode)->ip_lock);
850 mlog_bug_on_msg(OCFS2_I(inode)->ip_orphaned_slot != orphaned_slot &&
851 OCFS2_I(inode)->ip_orphaned_slot
Mark Fashehb06970532006-03-03 10:24:33 -0800852 != OCFS2_INVALID_SLOT, "Inode %llu: Node %d says it's "
853 "orphaned in slot %d, we think it's in %d\n",
854 (unsigned long long)OCFS2_I(inode)->ip_blkno,
Mark Fashehccd979b2005-12-15 14:31:24 -0800855 be32_to_cpu(resp->r_hdr.h_node_num),
856 orphaned_slot, OCFS2_I(inode)->ip_orphaned_slot);
857
858 OCFS2_I(inode)->ip_orphaned_slot = orphaned_slot;
859 spin_unlock(&OCFS2_I(inode)->ip_lock);
860}
861
862int ocfs2_request_delete_vote(struct inode *inode)
863{
864 int orphaned_slot, status;
865 struct ocfs2_net_response_cb delete_cb;
866 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
867 struct ocfs2_vote_msg *request;
868
869 spin_lock(&OCFS2_I(inode)->ip_lock);
870 orphaned_slot = OCFS2_I(inode)->ip_orphaned_slot;
871 spin_unlock(&OCFS2_I(inode)->ip_lock);
872
873 delete_cb.rc_cb = ocfs2_delete_response_cb;
874 delete_cb.rc_priv = inode;
875
Mark Fashehb06970532006-03-03 10:24:33 -0800876 mlog(0, "Inode %llu, we start thinking orphaned slot is %d\n",
877 (unsigned long long)OCFS2_I(inode)->ip_blkno, orphaned_slot);
Mark Fashehccd979b2005-12-15 14:31:24 -0800878
879 status = -ENOMEM;
880 request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
881 inode->i_generation,
882 OCFS2_VOTE_REQ_DELETE, orphaned_slot);
883 if (request) {
884 status = ocfs2_request_vote(inode, request, &delete_cb);
885
886 kfree(request);
887 }
888
889 return status;
890}
891
892static void ocfs2_setup_unlink_vote(struct ocfs2_vote_msg *request,
893 struct dentry *dentry)
894{
895 struct inode *parent = dentry->d_parent->d_inode;
896
897 /* We need some values which will uniquely identify a dentry
898 * on the other nodes so that they can find it and run
899 * d_delete against it. Parent directory block and full name
900 * should suffice. */
901
Mark Fashehb06970532006-03-03 10:24:33 -0800902 mlog(0, "unlink/rename request: parent: %llu name: %.*s\n",
903 (unsigned long long)OCFS2_I(parent)->ip_blkno, dentry->d_name.len,
Mark Fashehccd979b2005-12-15 14:31:24 -0800904 dentry->d_name.name);
905
906 request->v_unlink_parent = cpu_to_be64(OCFS2_I(parent)->ip_blkno);
907 request->v_unlink_namelen = cpu_to_be32(dentry->d_name.len);
908 memcpy(request->v_unlink_dirent, dentry->d_name.name,
909 dentry->d_name.len);
910}
911
912int ocfs2_request_unlink_vote(struct inode *inode,
913 struct dentry *dentry,
914 unsigned int nlink)
915{
916 int status;
917 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
918 struct ocfs2_vote_msg *request;
919
920 if (dentry->d_name.len > OCFS2_VOTE_FILENAME_LEN)
921 return -ENAMETOOLONG;
922
923 status = -ENOMEM;
924 request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
925 inode->i_generation,
926 OCFS2_VOTE_REQ_UNLINK, nlink);
927 if (request) {
928 ocfs2_setup_unlink_vote(request, dentry);
929
930 status = ocfs2_request_vote(inode, request, NULL);
931
932 kfree(request);
933 }
934 return status;
935}
936
937int ocfs2_request_rename_vote(struct inode *inode,
938 struct dentry *dentry)
939{
940 int status;
941 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
942 struct ocfs2_vote_msg *request;
943
944 if (dentry->d_name.len > OCFS2_VOTE_FILENAME_LEN)
945 return -ENAMETOOLONG;
946
947 status = -ENOMEM;
948 request = ocfs2_new_vote_request(osb, OCFS2_I(inode)->ip_blkno,
949 inode->i_generation,
950 OCFS2_VOTE_REQ_RENAME, 0);
951 if (request) {
952 ocfs2_setup_unlink_vote(request, dentry);
953
954 status = ocfs2_request_vote(inode, request, NULL);
955
956 kfree(request);
957 }
958 return status;
959}
960
961int ocfs2_request_mount_vote(struct ocfs2_super *osb)
962{
963 int status;
964 struct ocfs2_vote_msg *request = NULL;
965
966 request = ocfs2_new_vote_request(osb, 0ULL, 0,
967 OCFS2_VOTE_REQ_MOUNT, 0);
968 if (!request) {
969 status = -ENOMEM;
970 goto bail;
971 }
972
973 status = -EAGAIN;
974 while (status == -EAGAIN) {
975 if (!(osb->s_mount_opt & OCFS2_MOUNT_NOINTR) &&
976 signal_pending(current)) {
977 status = -ERESTARTSYS;
978 goto bail;
979 }
980
981 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
982 osb->node_num)) {
983 status = 0;
984 goto bail;
985 }
986
987 status = ocfs2_do_request_vote(osb, request, NULL);
988 }
989
990bail:
991 if (request)
992 kfree(request);
993
994 return status;
995}
996
997int ocfs2_request_umount_vote(struct ocfs2_super *osb)
998{
999 int status;
1000 struct ocfs2_vote_msg *request = NULL;
1001
1002 request = ocfs2_new_vote_request(osb, 0ULL, 0,
1003 OCFS2_VOTE_REQ_UMOUNT, 0);
1004 if (!request) {
1005 status = -ENOMEM;
1006 goto bail;
1007 }
1008
1009 status = -EAGAIN;
1010 while (status == -EAGAIN) {
1011 /* Do not check signals on this vote... We really want
1012 * this one to go all the way through. */
1013
1014 if (ocfs2_node_map_is_only(osb, &osb->mounted_map,
1015 osb->node_num)) {
1016 status = 0;
1017 goto bail;
1018 }
1019
1020 status = ocfs2_do_request_vote(osb, request, NULL);
1021 }
1022
1023bail:
1024 if (request)
1025 kfree(request);
1026
1027 return status;
1028}
1029
1030/* TODO: This should eventually be a hash table! */
1031static struct ocfs2_net_wait_ctxt * __ocfs2_find_net_wait_ctxt(struct ocfs2_super *osb,
1032 u32 response_id)
1033{
1034 struct list_head *p;
1035 struct ocfs2_net_wait_ctxt *w = NULL;
1036
1037 list_for_each(p, &osb->net_response_list) {
1038 w = list_entry(p, struct ocfs2_net_wait_ctxt, n_list);
1039 if (response_id == w->n_response_id)
1040 break;
1041 w = NULL;
1042 }
1043
1044 return w;
1045}
1046
1047/* Translate response codes into local node errno values */
1048static inline int ocfs2_translate_response(int response)
1049{
1050 int ret;
1051
1052 switch (response) {
1053 case OCFS2_RESPONSE_OK:
1054 ret = 0;
1055 break;
1056
1057 case OCFS2_RESPONSE_BUSY:
1058 ret = -EBUSY;
1059 break;
1060
1061 default:
1062 ret = -EINVAL;
1063 }
1064
1065 return ret;
1066}
1067
1068static int ocfs2_handle_response_message(struct o2net_msg *msg,
1069 u32 len,
1070 void *data)
1071{
1072 unsigned int response_id, node_num;
1073 int response_status;
1074 struct ocfs2_super *osb = data;
1075 struct ocfs2_response_msg *resp;
1076 struct ocfs2_net_wait_ctxt * w;
1077 struct ocfs2_net_response_cb *resp_cb;
1078
1079 resp = (struct ocfs2_response_msg *) msg->buf;
1080
1081 response_id = be32_to_cpu(resp->r_hdr.h_response_id);
1082 node_num = be32_to_cpu(resp->r_hdr.h_node_num);
1083 response_status =
1084 ocfs2_translate_response(be32_to_cpu(resp->r_response));
1085
1086 mlog(0, "received response message:\n");
1087 mlog(0, "h_response_id = %u\n", response_id);
1088 mlog(0, "h_request = %u\n", be32_to_cpu(resp->r_hdr.h_request));
Mark Fashehb06970532006-03-03 10:24:33 -08001089 mlog(0, "h_blkno = %llu\n",
1090 (unsigned long long)be64_to_cpu(resp->r_hdr.h_blkno));
Mark Fashehccd979b2005-12-15 14:31:24 -08001091 mlog(0, "h_generation = %u\n", be32_to_cpu(resp->r_hdr.h_generation));
1092 mlog(0, "h_node_num = %u\n", node_num);
1093 mlog(0, "r_response = %d\n", response_status);
1094
1095 spin_lock(&osb->net_response_lock);
1096 w = __ocfs2_find_net_wait_ctxt(osb, response_id);
1097 if (!w) {
1098 mlog(0, "request not found!\n");
1099 goto bail;
1100 }
1101 resp_cb = w->n_callback;
1102
1103 if (response_status && (!w->n_response)) {
1104 /* we only really need one negative response so don't
1105 * set it twice. */
1106 w->n_response = response_status;
1107 }
1108
1109 if (resp_cb) {
1110 spin_unlock(&osb->net_response_lock);
1111
1112 resp_cb->rc_cb(resp_cb->rc_priv, resp);
1113
1114 spin_lock(&osb->net_response_lock);
1115 }
1116
1117 __ocfs2_mark_node_responded(osb, w, node_num);
1118bail:
1119 spin_unlock(&osb->net_response_lock);
1120
1121 return 0;
1122}
1123
1124static int ocfs2_handle_vote_message(struct o2net_msg *msg,
1125 u32 len,
1126 void *data)
1127{
1128 int status;
1129 struct ocfs2_super *osb = data;
1130 struct ocfs2_vote_work *work;
1131
1132 work = kmalloc(sizeof(struct ocfs2_vote_work), GFP_KERNEL);
1133 if (!work) {
1134 status = -ENOMEM;
1135 mlog_errno(status);
1136 goto bail;
1137 }
1138
1139 INIT_LIST_HEAD(&work->w_list);
1140 memcpy(&work->w_msg, msg->buf, sizeof(struct ocfs2_vote_msg));
1141
1142 mlog(0, "scheduling vote request:\n");
1143 mlog(0, "h_response_id = %u\n",
1144 be32_to_cpu(work->w_msg.v_hdr.h_response_id));
1145 mlog(0, "h_request = %u\n", be32_to_cpu(work->w_msg.v_hdr.h_request));
Mark Fashehb06970532006-03-03 10:24:33 -08001146 mlog(0, "h_blkno = %llu\n",
1147 (unsigned long long)be64_to_cpu(work->w_msg.v_hdr.h_blkno));
Mark Fashehccd979b2005-12-15 14:31:24 -08001148 mlog(0, "h_generation = %u\n",
1149 be32_to_cpu(work->w_msg.v_hdr.h_generation));
1150 mlog(0, "h_node_num = %u\n",
1151 be32_to_cpu(work->w_msg.v_hdr.h_node_num));
1152 mlog(0, "v_generic1 = %u\n", be32_to_cpu(work->w_msg.md1.v_generic1));
1153
1154 spin_lock(&osb->vote_task_lock);
1155 list_add_tail(&work->w_list, &osb->vote_list);
1156 osb->vote_count++;
1157 spin_unlock(&osb->vote_task_lock);
1158
1159 ocfs2_kick_vote_thread(osb);
1160
1161 status = 0;
1162bail:
1163 return status;
1164}
1165
1166void ocfs2_unregister_net_handlers(struct ocfs2_super *osb)
1167{
1168 if (!osb->net_key)
1169 return;
1170
1171 o2net_unregister_handler_list(&osb->osb_net_handlers);
1172
1173 if (!list_empty(&osb->net_response_list))
1174 mlog(ML_ERROR, "net response list not empty!\n");
1175
1176 osb->net_key = 0;
1177}
1178
1179int ocfs2_register_net_handlers(struct ocfs2_super *osb)
1180{
1181 int status = 0;
1182
1183 status = o2net_register_handler(OCFS2_MESSAGE_TYPE_RESPONSE,
1184 osb->net_key,
1185 sizeof(struct ocfs2_response_msg),
1186 ocfs2_handle_response_message,
1187 osb, &osb->osb_net_handlers);
1188 if (status) {
1189 mlog_errno(status);
1190 goto bail;
1191 }
1192
1193 status = o2net_register_handler(OCFS2_MESSAGE_TYPE_VOTE,
1194 osb->net_key,
1195 sizeof(struct ocfs2_vote_msg),
1196 ocfs2_handle_vote_message,
1197 osb, &osb->osb_net_handlers);
1198 if (status) {
1199 mlog_errno(status);
1200 goto bail;
1201 }
1202bail:
1203 if (status < 0)
1204 ocfs2_unregister_net_handlers(osb);
1205
1206 return status;
1207}