?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Merge git://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux-2.6
[wrt350n-kernel.git] / fs / ocfs2 / dlm / dlmrecovery.c
blobb45b6d64c9a802e57c73ad5eeba74216e2f3451b
1 /* -*- mode: c; c-basic-offset: 8; -*-
2 * vim: noexpandtab sw=8 ts=8 sts=0:
3 *
4 * dlmrecovery.c
5 *
6 * recovery stuff
7 *
8 * Copyright (C) 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
27
28 #include <linux/module.h>
29 #include <linux/fs.h>
30 #include <linux/types.h>
31 #include <linux/slab.h>
32 #include <linux/highmem.h>
33 #include <linux/utsname.h>
34 #include <linux/init.h>
35 #include <linux/sysctl.h>
36 #include <linux/random.h>
37 #include <linux/blkdev.h>
38 #include <linux/socket.h>
39 #include <linux/inet.h>
40 #include <linux/timer.h>
41 #include <linux/kthread.h>
42 #include <linux/delay.h>
43
44
45 #include "cluster/heartbeat.h"
46 #include "cluster/nodemanager.h"
47 #include "cluster/tcp.h"
48
49 #include "dlmapi.h"
50 #include "dlmcommon.h"
51 #include "dlmdomain.h"
52
53 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_RECOVERY)
54 #include "cluster/masklog.h"
55
56 static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node);
57
58 static int dlm_recovery_thread(void *data);
59 void dlm_complete_recovery_thread(struct dlm_ctxt *dlm);
60 int dlm_launch_recovery_thread(struct dlm_ctxt *dlm);
61 void dlm_kick_recovery_thread(struct dlm_ctxt *dlm);
62 static int dlm_do_recovery(struct dlm_ctxt *dlm);
63
64 static int dlm_pick_recovery_master(struct dlm_ctxt *dlm);
65 static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node);
66 static int dlm_init_recovery_area(struct dlm_ctxt *dlm, u8 dead_node);
67 static int dlm_request_all_locks(struct dlm_ctxt *dlm,
68 u8 request_from, u8 dead_node);
69 static void dlm_destroy_recovery_area(struct dlm_ctxt *dlm, u8 dead_node);
70
71 static inline int dlm_num_locks_in_lockres(struct dlm_lock_resource *res);
72 static void dlm_init_migratable_lockres(struct dlm_migratable_lockres *mres,
73 const char *lockname, int namelen,
74 int total_locks, u64 cookie,
75 u8 flags, u8 master);
76 static int dlm_send_mig_lockres_msg(struct dlm_ctxt *dlm,
77 struct dlm_migratable_lockres *mres,
78 u8 send_to,
79 struct dlm_lock_resource *res,
80 int total_locks);
81 static int dlm_process_recovery_data(struct dlm_ctxt *dlm,
82 struct dlm_lock_resource *res,
83 struct dlm_migratable_lockres *mres);
84 static int dlm_send_finalize_reco_message(struct dlm_ctxt *dlm);
85 static int dlm_send_all_done_msg(struct dlm_ctxt *dlm,
86 u8 dead_node, u8 send_to);
87 static int dlm_send_begin_reco_message(struct dlm_ctxt *dlm, u8 dead_node);
88 static void dlm_move_reco_locks_to_list(struct dlm_ctxt *dlm,
89 struct list_head *list, u8 dead_node);
90 static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
91 u8 dead_node, u8 new_master);
92 static void dlm_reco_ast(void *astdata);
93 static void dlm_reco_bast(void *astdata, int blocked_type);
94 static void dlm_reco_unlock_ast(void *astdata, enum dlm_status st);
95 static void dlm_request_all_locks_worker(struct dlm_work_item *item,
96 void *data);
97 static void dlm_mig_lockres_worker(struct dlm_work_item *item, void *data);
98 static int dlm_lockres_master_requery(struct dlm_ctxt *dlm,
99 struct dlm_lock_resource *res,
100 u8 *real_master);
101
102 static u64 dlm_get_next_mig_cookie(void);
103
104 static DEFINE_SPINLOCK(dlm_reco_state_lock);
105 static DEFINE_SPINLOCK(dlm_mig_cookie_lock);
106 static u64 dlm_mig_cookie = 1;
107
108 static u64 dlm_get_next_mig_cookie(void)
109 {
110 u64 c;
111 spin_lock(&dlm_mig_cookie_lock);
112 c = dlm_mig_cookie;
113 if (dlm_mig_cookie == (~0ULL))
114 dlm_mig_cookie = 1;
115 else
116 dlm_mig_cookie++;
117 spin_unlock(&dlm_mig_cookie_lock);
118 return c;
119 }
120
121 static inline void dlm_set_reco_dead_node(struct dlm_ctxt *dlm,
122 u8 dead_node)
123 {
124 assert_spin_locked(&dlm->spinlock);
125 if (dlm->reco.dead_node != dead_node)
126 mlog(0, "%s: changing dead_node from %u to %u\n",
127 dlm->name, dlm->reco.dead_node, dead_node);
128 dlm->reco.dead_node = dead_node;
129 }
130
131 static inline void dlm_set_reco_master(struct dlm_ctxt *dlm,
132 u8 master)
133 {
134 assert_spin_locked(&dlm->spinlock);
135 mlog(0, "%s: changing new_master from %u to %u\n",
136 dlm->name, dlm->reco.new_master, master);
137 dlm->reco.new_master = master;
138 }
139
140 static inline void __dlm_reset_recovery(struct dlm_ctxt *dlm)
141 {
142 assert_spin_locked(&dlm->spinlock);
143 clear_bit(dlm->reco.dead_node, dlm->recovery_map);
144 dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
145 dlm_set_reco_master(dlm, O2NM_INVALID_NODE_NUM);
146 }
147
148 static inline void dlm_reset_recovery(struct dlm_ctxt *dlm)
149 {
150 spin_lock(&dlm->spinlock);
151 __dlm_reset_recovery(dlm);
152 spin_unlock(&dlm->spinlock);
153 }
154
155 /* Worker function used during recovery. */
156 void dlm_dispatch_work(struct work_struct *work)
157 {
158 struct dlm_ctxt *dlm =
159 container_of(work, struct dlm_ctxt, dispatched_work);
160 LIST_HEAD(tmp_list);
161 struct dlm_work_item *item, *next;
162 dlm_workfunc_t *workfunc;
163 int tot=0;
164
165 spin_lock(&dlm->work_lock);
166 list_splice_init(&dlm->work_list, &tmp_list);
167 spin_unlock(&dlm->work_lock);
168
169 list_for_each_entry(item, &tmp_list, list) {
170 tot++;
171 }
172 mlog(0, "%s: work thread has %d work items\n", dlm->name, tot);
173
174 list_for_each_entry_safe(item, next, &tmp_list, list) {
175 workfunc = item->func;
176 list_del_init(&item->list);
177
178 /* already have ref on dlm to avoid having
179 * it disappear. just double-check. */
180 BUG_ON(item->dlm != dlm);
181
182 /* this is allowed to sleep and
183 * call network stuff */
184 workfunc(item, item->data);
185
186 dlm_put(dlm);
187 kfree(item);
188 }
189 }
190
191 /*
192 * RECOVERY THREAD
193 */
194
195 void dlm_kick_recovery_thread(struct dlm_ctxt *dlm)
196 {
197 /* wake the recovery thread
198 * this will wake the reco thread in one of three places
199 * 1) sleeping with no recovery happening
200 * 2) sleeping with recovery mastered elsewhere
201 * 3) recovery mastered here, waiting on reco data */
202
203 wake_up(&dlm->dlm_reco_thread_wq);
204 }
205
206 /* Launch the recovery thread */
207 int dlm_launch_recovery_thread(struct dlm_ctxt *dlm)
208 {
209 mlog(0, "starting dlm recovery thread...\n");
210
211 dlm->dlm_reco_thread_task = kthread_run(dlm_recovery_thread, dlm,
212 "dlm_reco_thread");
213 if (IS_ERR(dlm->dlm_reco_thread_task)) {
214 mlog_errno(PTR_ERR(dlm->dlm_reco_thread_task));
215 dlm->dlm_reco_thread_task = NULL;
216 return -EINVAL;
217 }
218
219 return 0;
220 }
221
222 void dlm_complete_recovery_thread(struct dlm_ctxt *dlm)
223 {
224 if (dlm->dlm_reco_thread_task) {
225 mlog(0, "waiting for dlm recovery thread to exit\n");
226 kthread_stop(dlm->dlm_reco_thread_task);
227 dlm->dlm_reco_thread_task = NULL;
228 }
229 }
230
231
232
233 /*
234 * this is lame, but here's how recovery works...
235 * 1) all recovery threads cluster wide will work on recovering
236 * ONE node at a time
237 * 2) negotiate who will take over all the locks for the dead node.
238 * thats right... ALL the locks.
239 * 3) once a new master is chosen, everyone scans all locks
240 * and moves aside those mastered by the dead guy
241 * 4) each of these locks should be locked until recovery is done
242 * 5) the new master collects up all of secondary lock queue info
243 * one lock at a time, forcing each node to communicate back
244 * before continuing
245 * 6) each secondary lock queue responds with the full known lock info
246 * 7) once the new master has run all its locks, it sends a ALLDONE!
247 * message to everyone
248 * 8) upon receiving this message, the secondary queue node unlocks
249 * and responds to the ALLDONE
250 * 9) once the new master gets responses from everyone, he unlocks
251 * everything and recovery for this dead node is done
252 *10) go back to 2) while there are still dead nodes
253 *
254 */
255
256 static void dlm_print_reco_node_status(struct dlm_ctxt *dlm)
257 {
258 struct dlm_reco_node_data *ndata;
259 struct dlm_lock_resource *res;
260
261 mlog(ML_NOTICE, "%s(%d): recovery info, state=%s, dead=%u, master=%u\n",
262 dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
263 dlm->reco.state & DLM_RECO_STATE_ACTIVE ? "ACTIVE" : "inactive",
264 dlm->reco.dead_node, dlm->reco.new_master);
265
266 list_for_each_entry(ndata, &dlm->reco.node_data, list) {
267 char *st = "unknown";
268 switch (ndata->state) {
269 case DLM_RECO_NODE_DATA_INIT:
270 st = "init";
271 break;
272 case DLM_RECO_NODE_DATA_REQUESTING:
273 st = "requesting";
274 break;
275 case DLM_RECO_NODE_DATA_DEAD:
276 st = "dead";
277 break;
278 case DLM_RECO_NODE_DATA_RECEIVING:
279 st = "receiving";
280 break;
281 case DLM_RECO_NODE_DATA_REQUESTED:
282 st = "requested";
283 break;
284 case DLM_RECO_NODE_DATA_DONE:
285 st = "done";
286 break;
287 case DLM_RECO_NODE_DATA_FINALIZE_SENT:
288 st = "finalize-sent";
289 break;
290 default:
291 st = "bad";
292 break;
293 }
294 mlog(ML_NOTICE, "%s: reco state, node %u, state=%s\n",
295 dlm->name, ndata->node_num, st);
296 }
297 list_for_each_entry(res, &dlm->reco.resources, recovering) {
298 mlog(ML_NOTICE, "%s: lockres %.*s on recovering list\n",
299 dlm->name, res->lockname.len, res->lockname.name);
300 }
301 }
302
303 #define DLM_RECO_THREAD_TIMEOUT_MS (5 * 1000)
304
305 static int dlm_recovery_thread(void *data)
306 {
307 int status;
308 struct dlm_ctxt *dlm = data;
309 unsigned long timeout = msecs_to_jiffies(DLM_RECO_THREAD_TIMEOUT_MS);
310
311 mlog(0, "dlm thread running for %s...\n", dlm->name);
312
313 while (!kthread_should_stop()) {
314 if (dlm_joined(dlm)) {
315 status = dlm_do_recovery(dlm);
316 if (status == -EAGAIN) {
317 /* do not sleep, recheck immediately. */
318 continue;
319 }
320 if (status < 0)
321 mlog_errno(status);
322 }
323
324 wait_event_interruptible_timeout(dlm->dlm_reco_thread_wq,
325 kthread_should_stop(),
326 timeout);
327 }
328
329 mlog(0, "quitting DLM recovery thread\n");
330 return 0;
331 }
332
333 /* returns true when the recovery master has contacted us */
334 static int dlm_reco_master_ready(struct dlm_ctxt *dlm)
335 {
336 int ready;
337 spin_lock(&dlm->spinlock);
338 ready = (dlm->reco.new_master != O2NM_INVALID_NODE_NUM);
339 spin_unlock(&dlm->spinlock);
340 return ready;
341 }
342
343 /* returns true if node is no longer in the domain
344 * could be dead or just not joined */
345 int dlm_is_node_dead(struct dlm_ctxt *dlm, u8 node)
346 {
347 int dead;
348 spin_lock(&dlm->spinlock);
349 dead = !test_bit(node, dlm->domain_map);
350 spin_unlock(&dlm->spinlock);
351 return dead;
352 }
353
354 /* returns true if node is no longer in the domain
355 * could be dead or just not joined */
356 static int dlm_is_node_recovered(struct dlm_ctxt *dlm, u8 node)
357 {
358 int recovered;
359 spin_lock(&dlm->spinlock);
360 recovered = !test_bit(node, dlm->recovery_map);
361 spin_unlock(&dlm->spinlock);
362 return recovered;
363 }
364
365
366 int dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
367 {
368 if (timeout) {
369 mlog(ML_NOTICE, "%s: waiting %dms for notification of "
370 "death of node %u\n", dlm->name, timeout, node);
371 wait_event_timeout(dlm->dlm_reco_thread_wq,
372 dlm_is_node_dead(dlm, node),
373 msecs_to_jiffies(timeout));
374 } else {
375 mlog(ML_NOTICE, "%s: waiting indefinitely for notification "
376 "of death of node %u\n", dlm->name, node);
377 wait_event(dlm->dlm_reco_thread_wq,
378 dlm_is_node_dead(dlm, node));
379 }
380 /* for now, return 0 */
381 return 0;
382 }
383
384 int dlm_wait_for_node_recovery(struct dlm_ctxt *dlm, u8 node, int timeout)
385 {
386 if (timeout) {
387 mlog(0, "%s: waiting %dms for notification of "
388 "recovery of node %u\n", dlm->name, timeout, node);
389 wait_event_timeout(dlm->dlm_reco_thread_wq,
390 dlm_is_node_recovered(dlm, node),
391 msecs_to_jiffies(timeout));
392 } else {
393 mlog(0, "%s: waiting indefinitely for notification "
394 "of recovery of node %u\n", dlm->name, node);
395 wait_event(dlm->dlm_reco_thread_wq,
396 dlm_is_node_recovered(dlm, node));
397 }
398 /* for now, return 0 */
399 return 0;
400 }
401
402 /* callers of the top-level api calls (dlmlock/dlmunlock) should
403 * block on the dlm->reco.event when recovery is in progress.
404 * the dlm recovery thread will set this state when it begins
405 * recovering a dead node (as the new master or not) and clear
406 * the state and wake as soon as all affected lock resources have
407 * been marked with the RECOVERY flag */
408 static int dlm_in_recovery(struct dlm_ctxt *dlm)
409 {
410 int in_recovery;
411 spin_lock(&dlm->spinlock);
412 in_recovery = !!(dlm->reco.state & DLM_RECO_STATE_ACTIVE);
413 spin_unlock(&dlm->spinlock);
414 return in_recovery;
415 }
416
417
418 void dlm_wait_for_recovery(struct dlm_ctxt *dlm)
419 {
420 if (dlm_in_recovery(dlm)) {
421 mlog(0, "%s: reco thread %d in recovery: "
422 "state=%d, master=%u, dead=%u\n",
423 dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
424 dlm->reco.state, dlm->reco.new_master,
425 dlm->reco.dead_node);
426 }
427 wait_event(dlm->reco.event, !dlm_in_recovery(dlm));
428 }
429
430 static void dlm_begin_recovery(struct dlm_ctxt *dlm)
431 {
432 spin_lock(&dlm->spinlock);
433 BUG_ON(dlm->reco.state & DLM_RECO_STATE_ACTIVE);
434 dlm->reco.state |= DLM_RECO_STATE_ACTIVE;
435 spin_unlock(&dlm->spinlock);
436 }
437
438 static void dlm_end_recovery(struct dlm_ctxt *dlm)
439 {
440 spin_lock(&dlm->spinlock);
441 BUG_ON(!(dlm->reco.state & DLM_RECO_STATE_ACTIVE));
442 dlm->reco.state &= ~DLM_RECO_STATE_ACTIVE;
443 spin_unlock(&dlm->spinlock);
444 wake_up(&dlm->reco.event);
445 }
446
447 static int dlm_do_recovery(struct dlm_ctxt *dlm)
448 {
449 int status = 0;
450 int ret;
451
452 spin_lock(&dlm->spinlock);
453
454 /* check to see if the new master has died */
455 if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM &&
456 test_bit(dlm->reco.new_master, dlm->recovery_map)) {
457 mlog(0, "new master %u died while recovering %u!\n",
458 dlm->reco.new_master, dlm->reco.dead_node);
459 /* unset the new_master, leave dead_node */
460 dlm_set_reco_master(dlm, O2NM_INVALID_NODE_NUM);
461 }
462
463 /* select a target to recover */
464 if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
465 int bit;
466
467 bit = find_next_bit (dlm->recovery_map, O2NM_MAX_NODES+1, 0);
468 if (bit >= O2NM_MAX_NODES || bit < 0)
469 dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
470 else
471 dlm_set_reco_dead_node(dlm, bit);
472 } else if (!test_bit(dlm->reco.dead_node, dlm->recovery_map)) {
473 /* BUG? */
474 mlog(ML_ERROR, "dead_node %u no longer in recovery map!\n",
475 dlm->reco.dead_node);
476 dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
477 }
478
479 if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
480 // mlog(0, "nothing to recover! sleeping now!\n");
481 spin_unlock(&dlm->spinlock);
482 /* return to main thread loop and sleep. */
483 return 0;
484 }
485 mlog(0, "%s(%d):recovery thread found node %u in the recovery map!\n",
486 dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
487 dlm->reco.dead_node);
488 spin_unlock(&dlm->spinlock);
489
490 /* take write barrier */
491 /* (stops the list reshuffling thread, proxy ast handling) */
492 dlm_begin_recovery(dlm);
493
494 if (dlm->reco.new_master == dlm->node_num)
495 goto master_here;
496
497 if (dlm->reco.new_master == O2NM_INVALID_NODE_NUM) {
498 /* choose a new master, returns 0 if this node
499 * is the master, -EEXIST if it's another node.
500 * this does not return until a new master is chosen
501 * or recovery completes entirely. */
502 ret = dlm_pick_recovery_master(dlm);
503 if (!ret) {
504 /* already notified everyone. go. */
505 goto master_here;
506 }
507 mlog(0, "another node will master this recovery session.\n");
508 }
509 mlog(0, "dlm=%s (%d), new_master=%u, this node=%u, dead_node=%u\n",
510 dlm->name, task_pid_nr(dlm->dlm_reco_thread_task), dlm->reco.new_master,
511 dlm->node_num, dlm->reco.dead_node);
512
513 /* it is safe to start everything back up here
514 * because all of the dead node's lock resources
515 * have been marked as in-recovery */
516 dlm_end_recovery(dlm);
517
518 /* sleep out in main dlm_recovery_thread loop. */
519 return 0;
520
521 master_here:
522 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
523 mlog(0, "(%d) mastering recovery of %s:%u here(this=%u)!\n",
524 task_pid_nr(dlm->dlm_reco_thread_task),
525 dlm->name, dlm->reco.dead_node, dlm->node_num);
526 =======
527 mlog(ML_NOTICE, "(%d) Node %u is the Recovery Master for the Dead Node "
528 "%u for Domain %s\n", task_pid_nr(dlm->dlm_reco_thread_task),
529 dlm->node_num, dlm->reco.dead_node, dlm->name);
530 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
531
532 status = dlm_remaster_locks(dlm, dlm->reco.dead_node);
533 if (status < 0) {
534 /* we should never hit this anymore */
535 mlog(ML_ERROR, "error %d remastering locks for node %u, "
536 "retrying.\n", status, dlm->reco.dead_node);
537 /* yield a bit to allow any final network messages
538 * to get handled on remaining nodes */
539 msleep(100);
540 } else {
541 /* success! see if any other nodes need recovery */
542 mlog(0, "DONE mastering recovery of %s:%u here(this=%u)!\n",
543 dlm->name, dlm->reco.dead_node, dlm->node_num);
544 dlm_reset_recovery(dlm);
545 }
546 dlm_end_recovery(dlm);
547
548 /* continue and look for another dead node */
549 return -EAGAIN;
550 }
551
552 static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node)
553 {
554 int status = 0;
555 struct dlm_reco_node_data *ndata;
556 int all_nodes_done;
557 int destroy = 0;
558 int pass = 0;
559
560 do {
561 /* we have become recovery master. there is no escaping
562 * this, so just keep trying until we get it. */
563 status = dlm_init_recovery_area(dlm, dead_node);
564 if (status < 0) {
565 mlog(ML_ERROR, "%s: failed to alloc recovery area, "
566 "retrying\n", dlm->name);
567 msleep(1000);
568 }
569 } while (status != 0);
570
571 /* safe to access the node data list without a lock, since this
572 * process is the only one to change the list */
573 list_for_each_entry(ndata, &dlm->reco.node_data, list) {
574 BUG_ON(ndata->state != DLM_RECO_NODE_DATA_INIT);
575 ndata->state = DLM_RECO_NODE_DATA_REQUESTING;
576
577 mlog(0, "requesting lock info from node %u\n",
578 ndata->node_num);
579
580 if (ndata->node_num == dlm->node_num) {
581 ndata->state = DLM_RECO_NODE_DATA_DONE;
582 continue;
583 }
584
585 do {
586 status = dlm_request_all_locks(dlm, ndata->node_num,
587 dead_node);
588 if (status < 0) {
589 mlog_errno(status);
590 if (dlm_is_host_down(status)) {
591 /* node died, ignore it for recovery */
592 status = 0;
593 ndata->state = DLM_RECO_NODE_DATA_DEAD;
594 /* wait for the domain map to catch up
595 * with the network state. */
596 wait_event_timeout(dlm->dlm_reco_thread_wq,
597 dlm_is_node_dead(dlm,
598 ndata->node_num),
599 msecs_to_jiffies(1000));
600 mlog(0, "waited 1 sec for %u, "
601 "dead? %s\n", ndata->node_num,
602 dlm_is_node_dead(dlm, ndata->node_num) ?
603 "yes" : "no");
604 } else {
605 /* -ENOMEM on the other node */
606 mlog(0, "%s: node %u returned "
607 "%d during recovery, retrying "
608 "after a short wait\n",
609 dlm->name, ndata->node_num,
610 status);
611 msleep(100);
612 }
613 }
614 } while (status != 0);
615
616 spin_lock(&dlm_reco_state_lock);
617 switch (ndata->state) {
618 case DLM_RECO_NODE_DATA_INIT:
619 case DLM_RECO_NODE_DATA_FINALIZE_SENT:
620 case DLM_RECO_NODE_DATA_REQUESTED:
621 BUG();
622 break;
623 case DLM_RECO_NODE_DATA_DEAD:
624 mlog(0, "node %u died after requesting "
625 "recovery info for node %u\n",
626 ndata->node_num, dead_node);
627 /* fine. don't need this node's info.
628 * continue without it. */
629 break;
630 case DLM_RECO_NODE_DATA_REQUESTING:
631 ndata->state = DLM_RECO_NODE_DATA_REQUESTED;
632 mlog(0, "now receiving recovery data from "
633 "node %u for dead node %u\n",
634 ndata->node_num, dead_node);
635 break;
636 case DLM_RECO_NODE_DATA_RECEIVING:
637 mlog(0, "already receiving recovery data from "
638 "node %u for dead node %u\n",
639 ndata->node_num, dead_node);
640 break;
641 case DLM_RECO_NODE_DATA_DONE:
642 mlog(0, "already DONE receiving recovery data "
643 "from node %u for dead node %u\n",
644 ndata->node_num, dead_node);
645 break;
646 }
647 spin_unlock(&dlm_reco_state_lock);
648 }
649
650 mlog(0, "done requesting all lock info\n");
651
652 /* nodes should be sending reco data now
653 * just need to wait */
654
655 while (1) {
656 /* check all the nodes now to see if we are
657 * done, or if anyone died */
658 all_nodes_done = 1;
659 spin_lock(&dlm_reco_state_lock);
660 list_for_each_entry(ndata, &dlm->reco.node_data, list) {
661 mlog(0, "checking recovery state of node %u\n",
662 ndata->node_num);
663 switch (ndata->state) {
664 case DLM_RECO_NODE_DATA_INIT:
665 case DLM_RECO_NODE_DATA_REQUESTING:
666 mlog(ML_ERROR, "bad ndata state for "
667 "node %u: state=%d\n",
668 ndata->node_num, ndata->state);
669 BUG();
670 break;
671 case DLM_RECO_NODE_DATA_DEAD:
672 mlog(0, "node %u died after "
673 "requesting recovery info for "
674 "node %u\n", ndata->node_num,
675 dead_node);
676 break;
677 case DLM_RECO_NODE_DATA_RECEIVING:
678 case DLM_RECO_NODE_DATA_REQUESTED:
679 mlog(0, "%s: node %u still in state %s\n",
680 dlm->name, ndata->node_num,
681 ndata->state==DLM_RECO_NODE_DATA_RECEIVING ?
682 "receiving" : "requested");
683 all_nodes_done = 0;
684 break;
685 case DLM_RECO_NODE_DATA_DONE:
686 mlog(0, "%s: node %u state is done\n",
687 dlm->name, ndata->node_num);
688 break;
689 case DLM_RECO_NODE_DATA_FINALIZE_SENT:
690 mlog(0, "%s: node %u state is finalize\n",
691 dlm->name, ndata->node_num);
692 break;
693 }
694 }
695 spin_unlock(&dlm_reco_state_lock);
696
697 mlog(0, "pass #%d, all_nodes_done?: %s\n", ++pass,
698 all_nodes_done?"yes":"no");
699 if (all_nodes_done) {
700 int ret;
701
702 /* all nodes are now in DLM_RECO_NODE_DATA_DONE state
703 * just send a finalize message to everyone and
704 * clean up */
705 mlog(0, "all nodes are done! send finalize\n");
706 ret = dlm_send_finalize_reco_message(dlm);
707 if (ret < 0)
708 mlog_errno(ret);
709
710 spin_lock(&dlm->spinlock);
711 dlm_finish_local_lockres_recovery(dlm, dead_node,
712 dlm->node_num);
713 spin_unlock(&dlm->spinlock);
714 mlog(0, "should be done with recovery!\n");
715
716 mlog(0, "finishing recovery of %s at %lu, "
717 "dead=%u, this=%u, new=%u\n", dlm->name,
718 jiffies, dlm->reco.dead_node,
719 dlm->node_num, dlm->reco.new_master);
720 destroy = 1;
721 status = 0;
722 /* rescan everything marked dirty along the way */
723 dlm_kick_thread(dlm, NULL);
724 break;
725 }
726 /* wait to be signalled, with periodic timeout
727 * to check for node death */
728 wait_event_interruptible_timeout(dlm->dlm_reco_thread_wq,
729 kthread_should_stop(),
730 msecs_to_jiffies(DLM_RECO_THREAD_TIMEOUT_MS));
731
732 }
733
734 if (destroy)
735 dlm_destroy_recovery_area(dlm, dead_node);
736
737 mlog_exit(status);
738 return status;
739 }
740
741 static int dlm_init_recovery_area(struct dlm_ctxt *dlm, u8 dead_node)
742 {
743 int num=0;
744 struct dlm_reco_node_data *ndata;
745
746 spin_lock(&dlm->spinlock);
747 memcpy(dlm->reco.node_map, dlm->domain_map, sizeof(dlm->domain_map));
748 /* nodes can only be removed (by dying) after dropping
749 * this lock, and death will be trapped later, so this should do */
750 spin_unlock(&dlm->spinlock);
751
752 while (1) {
753 num = find_next_bit (dlm->reco.node_map, O2NM_MAX_NODES, num);
754 if (num >= O2NM_MAX_NODES) {
755 break;
756 }
757 BUG_ON(num == dead_node);
758
759 ndata = kzalloc(sizeof(*ndata), GFP_NOFS);
760 if (!ndata) {
761 dlm_destroy_recovery_area(dlm, dead_node);
762 return -ENOMEM;
763 }
764 ndata->node_num = num;
765 ndata->state = DLM_RECO_NODE_DATA_INIT;
766 spin_lock(&dlm_reco_state_lock);
767 list_add_tail(&ndata->list, &dlm->reco.node_data);
768 spin_unlock(&dlm_reco_state_lock);
769 num++;
770 }
771
772 return 0;
773 }
774
775 static void dlm_destroy_recovery_area(struct dlm_ctxt *dlm, u8 dead_node)
776 {
777 struct dlm_reco_node_data *ndata, *next;
778 LIST_HEAD(tmplist);
779
780 spin_lock(&dlm_reco_state_lock);
781 list_splice_init(&dlm->reco.node_data, &tmplist);
782 spin_unlock(&dlm_reco_state_lock);
783
784 list_for_each_entry_safe(ndata, next, &tmplist, list) {
785 list_del_init(&ndata->list);
786 kfree(ndata);
787 }
788 }
789
790 static int dlm_request_all_locks(struct dlm_ctxt *dlm, u8 request_from,
791 u8 dead_node)
792 {
793 struct dlm_lock_request lr;
794 enum dlm_status ret;
795
796 mlog(0, "\n");
797
798
799 mlog(0, "dlm_request_all_locks: dead node is %u, sending request "
800 "to %u\n", dead_node, request_from);
801
802 memset(&lr, 0, sizeof(lr));
803 lr.node_idx = dlm->node_num;
804 lr.dead_node = dead_node;
805
806 // send message
807 ret = DLM_NOLOCKMGR;
808 ret = o2net_send_message(DLM_LOCK_REQUEST_MSG, dlm->key,
809 &lr, sizeof(lr), request_from, NULL);
810
811 /* negative status is handled by caller */
812 if (ret < 0)
813 mlog_errno(ret);
814
815 // return from here, then
816 // sleep until all received or error
817 return ret;
818
819 }
820
821 int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data,
822 void **ret_data)
823 {
824 struct dlm_ctxt *dlm = data;
825 struct dlm_lock_request *lr = (struct dlm_lock_request *)msg->buf;
826 char *buf = NULL;
827 struct dlm_work_item *item = NULL;
828
829 if (!dlm_grab(dlm))
830 return -EINVAL;
831
832 if (lr->dead_node != dlm->reco.dead_node) {
833 mlog(ML_ERROR, "%s: node %u sent dead_node=%u, but local "
834 "dead_node is %u\n", dlm->name, lr->node_idx,
835 lr->dead_node, dlm->reco.dead_node);
836 dlm_print_reco_node_status(dlm);
837 /* this is a hack */
838 dlm_put(dlm);
839 return -ENOMEM;
840 }
841 BUG_ON(lr->dead_node != dlm->reco.dead_node);
842
843 item = kzalloc(sizeof(*item), GFP_NOFS);
844 if (!item) {
845 dlm_put(dlm);
846 return -ENOMEM;
847 }
848
849 /* this will get freed by dlm_request_all_locks_worker */
850 buf = (char *) __get_free_page(GFP_NOFS);
851 if (!buf) {
852 kfree(item);
853 dlm_put(dlm);
854 return -ENOMEM;
855 }
856
857 /* queue up work for dlm_request_all_locks_worker */
858 dlm_grab(dlm); /* get an extra ref for the work item */
859 dlm_init_work_item(dlm, item, dlm_request_all_locks_worker, buf);
860 item->u.ral.reco_master = lr->node_idx;
861 item->u.ral.dead_node = lr->dead_node;
862 spin_lock(&dlm->work_lock);
863 list_add_tail(&item->list, &dlm->work_list);
864 spin_unlock(&dlm->work_lock);
865 queue_work(dlm->dlm_worker, &dlm->dispatched_work);
866
867 dlm_put(dlm);
868 return 0;
869 }
870
871 static void dlm_request_all_locks_worker(struct dlm_work_item *item, void *data)
872 {
873 struct dlm_migratable_lockres *mres;
874 struct dlm_lock_resource *res;
875 struct dlm_ctxt *dlm;
876 LIST_HEAD(resources);
877 int ret;
878 u8 dead_node, reco_master;
879 int skip_all_done = 0;
880
881 dlm = item->dlm;
882 dead_node = item->u.ral.dead_node;
883 reco_master = item->u.ral.reco_master;
884 mres = (struct dlm_migratable_lockres *)data;
885
886 mlog(0, "%s: recovery worker started, dead=%u, master=%u\n",
887 dlm->name, dead_node, reco_master);
888
889 if (dead_node != dlm->reco.dead_node ||
890 reco_master != dlm->reco.new_master) {
891 /* worker could have been created before the recovery master
892 * died. if so, do not continue, but do not error. */
893 if (dlm->reco.new_master == O2NM_INVALID_NODE_NUM) {
894 mlog(ML_NOTICE, "%s: will not send recovery state, "
895 "recovery master %u died, thread=(dead=%u,mas=%u)"
896 " current=(dead=%u,mas=%u)\n", dlm->name,
897 reco_master, dead_node, reco_master,
898 dlm->reco.dead_node, dlm->reco.new_master);
899 } else {
900 mlog(ML_NOTICE, "%s: reco state invalid: reco(dead=%u, "
901 "master=%u), request(dead=%u, master=%u)\n",
902 dlm->name, dlm->reco.dead_node,
903 dlm->reco.new_master, dead_node, reco_master);
904 }
905 goto leave;
906 }
907
908 /* lock resources should have already been moved to the
909 * dlm->reco.resources list. now move items from that list
910 * to a temp list if the dead owner matches. note that the
911 * whole cluster recovers only one node at a time, so we
912 * can safely move UNKNOWN lock resources for each recovery
913 * session. */
914 dlm_move_reco_locks_to_list(dlm, &resources, dead_node);
915
916 /* now we can begin blasting lockreses without the dlm lock */
917
918 /* any errors returned will be due to the new_master dying,
919 * the dlm_reco_thread should detect this */
920 list_for_each_entry(res, &resources, recovering) {
921 ret = dlm_send_one_lockres(dlm, res, mres, reco_master,
922 DLM_MRES_RECOVERY);
923 if (ret < 0) {
924 mlog(ML_ERROR, "%s: node %u went down while sending "
925 "recovery state for dead node %u, ret=%d\n", dlm->name,
926 reco_master, dead_node, ret);
927 skip_all_done = 1;
928 break;
929 }
930 }
931
932 /* move the resources back to the list */
933 spin_lock(&dlm->spinlock);
934 list_splice_init(&resources, &dlm->reco.resources);
935 spin_unlock(&dlm->spinlock);
936
937 if (!skip_all_done) {
938 ret = dlm_send_all_done_msg(dlm, dead_node, reco_master);
939 if (ret < 0) {
940 mlog(ML_ERROR, "%s: node %u went down while sending "
941 "recovery all-done for dead node %u, ret=%d\n",
942 dlm->name, reco_master, dead_node, ret);
943 }
944 }
945 leave:
946 free_page((unsigned long)data);
947 }
948
949
950 static int dlm_send_all_done_msg(struct dlm_ctxt *dlm, u8 dead_node, u8 send_to)
951 {
952 int ret, tmpret;
953 struct dlm_reco_data_done done_msg;
954
955 memset(&done_msg, 0, sizeof(done_msg));
956 done_msg.node_idx = dlm->node_num;
957 done_msg.dead_node = dead_node;
958 mlog(0, "sending DATA DONE message to %u, "
959 "my node=%u, dead node=%u\n", send_to, done_msg.node_idx,
960 done_msg.dead_node);
961
962 ret = o2net_send_message(DLM_RECO_DATA_DONE_MSG, dlm->key, &done_msg,
963 sizeof(done_msg), send_to, &tmpret);
964 if (ret < 0) {
965 if (!dlm_is_host_down(ret)) {
966 mlog_errno(ret);
967 mlog(ML_ERROR, "%s: unknown error sending data-done "
968 "to %u\n", dlm->name, send_to);
969 BUG();
970 }
971 } else
972 ret = tmpret;
973 return ret;
974 }
975
976
977 int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data,
978 void **ret_data)
979 {
980 struct dlm_ctxt *dlm = data;
981 struct dlm_reco_data_done *done = (struct dlm_reco_data_done *)msg->buf;
982 struct dlm_reco_node_data *ndata = NULL;
983 int ret = -EINVAL;
984
985 if (!dlm_grab(dlm))
986 return -EINVAL;
987
988 mlog(0, "got DATA DONE: dead_node=%u, reco.dead_node=%u, "
989 "node_idx=%u, this node=%u\n", done->dead_node,
990 dlm->reco.dead_node, done->node_idx, dlm->node_num);
991
992 mlog_bug_on_msg((done->dead_node != dlm->reco.dead_node),
993 "Got DATA DONE: dead_node=%u, reco.dead_node=%u, "
994 "node_idx=%u, this node=%u\n", done->dead_node,
995 dlm->reco.dead_node, done->node_idx, dlm->node_num);
996
997 spin_lock(&dlm_reco_state_lock);
998 list_for_each_entry(ndata, &dlm->reco.node_data, list) {
999 if (ndata->node_num != done->node_idx)
1000 continue;
1001
1002 switch (ndata->state) {
1003 /* should have moved beyond INIT but not to FINALIZE yet */
1004 case DLM_RECO_NODE_DATA_INIT:
1005 case DLM_RECO_NODE_DATA_DEAD:
1006 case DLM_RECO_NODE_DATA_FINALIZE_SENT:
1007 mlog(ML_ERROR, "bad ndata state for node %u:"
1008 " state=%d\n", ndata->node_num,
1009 ndata->state);
1010 BUG();
1011 break;
1012 /* these states are possible at this point, anywhere along
1013 * the line of recovery */
1014 case DLM_RECO_NODE_DATA_DONE:
1015 case DLM_RECO_NODE_DATA_RECEIVING:
1016 case DLM_RECO_NODE_DATA_REQUESTED:
1017 case DLM_RECO_NODE_DATA_REQUESTING:
1018 mlog(0, "node %u is DONE sending "
1019 "recovery data!\n",
1020 ndata->node_num);
1021
1022 ndata->state = DLM_RECO_NODE_DATA_DONE;
1023 ret = 0;
1024 break;
1025 }
1026 }
1027 spin_unlock(&dlm_reco_state_lock);
1028
1029 /* wake the recovery thread, some node is done */
1030 if (!ret)
1031 dlm_kick_recovery_thread(dlm);
1032
1033 if (ret < 0)
1034 mlog(ML_ERROR, "failed to find recovery node data for node "
1035 "%u\n", done->node_idx);
1036 dlm_put(dlm);
1037
1038 mlog(0, "leaving reco data done handler, ret=%d\n", ret);
1039 return ret;
1040 }
1041
1042 static void dlm_move_reco_locks_to_list(struct dlm_ctxt *dlm,
1043 struct list_head *list,
1044 u8 dead_node)
1045 {
1046 struct dlm_lock_resource *res, *next;
1047 struct dlm_lock *lock;
1048
1049 spin_lock(&dlm->spinlock);
1050 list_for_each_entry_safe(res, next, &dlm->reco.resources, recovering) {
1051 /* always prune any $RECOVERY entries for dead nodes,
1052 * otherwise hangs can occur during later recovery */
1053 if (dlm_is_recovery_lock(res->lockname.name,
1054 res->lockname.len)) {
1055 spin_lock(&res->spinlock);
1056 list_for_each_entry(lock, &res->granted, list) {
1057 if (lock->ml.node == dead_node) {
1058 mlog(0, "AHA! there was "
1059 "a $RECOVERY lock for dead "
1060 "node %u (%s)!\n",
1061 dead_node, dlm->name);
1062 list_del_init(&lock->list);
1063 dlm_lock_put(lock);
1064 break;
1065 }
1066 }
1067 spin_unlock(&res->spinlock);
1068 continue;
1069 }
1070
1071 if (res->owner == dead_node) {
1072 mlog(0, "found lockres owned by dead node while "
1073 "doing recovery for node %u. sending it.\n",
1074 dead_node);
1075 list_move_tail(&res->recovering, list);
1076 } else if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
1077 mlog(0, "found UNKNOWN owner while doing recovery "
1078 "for node %u. sending it.\n", dead_node);
1079 list_move_tail(&res->recovering, list);
1080 }
1081 }
1082 spin_unlock(&dlm->spinlock);
1083 }
1084
1085 static inline int dlm_num_locks_in_lockres(struct dlm_lock_resource *res)
1086 {
1087 int total_locks = 0;
1088 struct list_head *iter, *queue = &res->granted;
1089 int i;
1090
1091 for (i=0; i<3; i++) {
1092 list_for_each(iter, queue)
1093 total_locks++;
1094 queue++;
1095 }
1096 return total_locks;
1097 }
1098
1099
1100 static int dlm_send_mig_lockres_msg(struct dlm_ctxt *dlm,
1101 struct dlm_migratable_lockres *mres,
1102 u8 send_to,
1103 struct dlm_lock_resource *res,
1104 int total_locks)
1105 {
1106 u64 mig_cookie = be64_to_cpu(mres->mig_cookie);
1107 int mres_total_locks = be32_to_cpu(mres->total_locks);
1108 int sz, ret = 0, status = 0;
1109 u8 orig_flags = mres->flags,
1110 orig_master = mres->master;
1111
1112 BUG_ON(mres->num_locks > DLM_MAX_MIGRATABLE_LOCKS);
1113 if (!mres->num_locks)
1114 return 0;
1115
1116 sz = sizeof(struct dlm_migratable_lockres) +
1117 (mres->num_locks * sizeof(struct dlm_migratable_lock));
1118
1119 /* add an all-done flag if we reached the last lock */
1120 orig_flags = mres->flags;
1121 BUG_ON(total_locks > mres_total_locks);
1122 if (total_locks == mres_total_locks)
1123 mres->flags |= DLM_MRES_ALL_DONE;
1124
1125 mlog(0, "%s:%.*s: sending mig lockres (%s) to %u\n",
1126 dlm->name, res->lockname.len, res->lockname.name,
1127 orig_flags & DLM_MRES_MIGRATION ? "migrate" : "recovery",
1128 send_to);
1129
1130 /* send it */
1131 ret = o2net_send_message(DLM_MIG_LOCKRES_MSG, dlm->key, mres,
1132 sz, send_to, &status);
1133 if (ret < 0) {
1134 /* XXX: negative status is not handled.
1135 * this will end up killing this node. */
1136 mlog_errno(ret);
1137 } else {
1138 /* might get an -ENOMEM back here */
1139 ret = status;
1140 if (ret < 0) {
1141 mlog_errno(ret);
1142
1143 if (ret == -EFAULT) {
1144 mlog(ML_ERROR, "node %u told me to kill "
1145 "myself!\n", send_to);
1146 BUG();
1147 }
1148 }
1149 }
1150
1151 /* zero and reinit the message buffer */
1152 dlm_init_migratable_lockres(mres, res->lockname.name,
1153 res->lockname.len, mres_total_locks,
1154 mig_cookie, orig_flags, orig_master);
1155 return ret;
1156 }
1157
1158 static void dlm_init_migratable_lockres(struct dlm_migratable_lockres *mres,
1159 const char *lockname, int namelen,
1160 int total_locks, u64 cookie,
1161 u8 flags, u8 master)
1162 {
1163 /* mres here is one full page */
1164 clear_page(mres);
1165 mres->lockname_len = namelen;
1166 memcpy(mres->lockname, lockname, namelen);
1167 mres->num_locks = 0;
1168 mres->total_locks = cpu_to_be32(total_locks);
1169 mres->mig_cookie = cpu_to_be64(cookie);
1170 mres->flags = flags;
1171 mres->master = master;
1172 }
1173
1174
1175 /* returns 1 if this lock fills the network structure,
1176 * 0 otherwise */
1177 static int dlm_add_lock_to_array(struct dlm_lock *lock,
1178 struct dlm_migratable_lockres *mres, int queue)
1179 {
1180 struct dlm_migratable_lock *ml;
1181 int lock_num = mres->num_locks;
1182
1183 ml = &(mres->ml[lock_num]);
1184 ml->cookie = lock->ml.cookie;
1185 ml->type = lock->ml.type;
1186 ml->convert_type = lock->ml.convert_type;
1187 ml->highest_blocked = lock->ml.highest_blocked;
1188 ml->list = queue;
1189 if (lock->lksb) {
1190 ml->flags = lock->lksb->flags;
1191 /* send our current lvb */
1192 if (ml->type == LKM_EXMODE ||
1193 ml->type == LKM_PRMODE) {
1194 /* if it is already set, this had better be a PR
1195 * and it has to match */
1196 if (!dlm_lvb_is_empty(mres->lvb) &&
1197 (ml->type == LKM_EXMODE ||
1198 memcmp(mres->lvb, lock->lksb->lvb, DLM_LVB_LEN))) {
1199 mlog(ML_ERROR, "mismatched lvbs!\n");
1200 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1201 __dlm_print_one_lock_resource(lock->lockres);
1202 =======
1203 dlm_print_one_lock_resource(lock->lockres);
1204 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1205 BUG();
1206 }
1207 memcpy(mres->lvb, lock->lksb->lvb, DLM_LVB_LEN);
1208 }
1209 }
1210 ml->node = lock->ml.node;
1211 mres->num_locks++;
1212 /* we reached the max, send this network message */
1213 if (mres->num_locks == DLM_MAX_MIGRATABLE_LOCKS)
1214 return 1;
1215 return 0;
1216 }
1217
1218 static void dlm_add_dummy_lock(struct dlm_ctxt *dlm,
1219 struct dlm_migratable_lockres *mres)
1220 {
1221 struct dlm_lock dummy;
1222 memset(&dummy, 0, sizeof(dummy));
1223 dummy.ml.cookie = 0;
1224 dummy.ml.type = LKM_IVMODE;
1225 dummy.ml.convert_type = LKM_IVMODE;
1226 dummy.ml.highest_blocked = LKM_IVMODE;
1227 dummy.lksb = NULL;
1228 dummy.ml.node = dlm->node_num;
1229 dlm_add_lock_to_array(&dummy, mres, DLM_BLOCKED_LIST);
1230 }
1231
1232 static inline int dlm_is_dummy_lock(struct dlm_ctxt *dlm,
1233 struct dlm_migratable_lock *ml,
1234 u8 *nodenum)
1235 {
1236 if (unlikely(ml->cookie == 0 &&
1237 ml->type == LKM_IVMODE &&
1238 ml->convert_type == LKM_IVMODE &&
1239 ml->highest_blocked == LKM_IVMODE &&
1240 ml->list == DLM_BLOCKED_LIST)) {
1241 *nodenum = ml->node;
1242 return 1;
1243 }
1244 return 0;
1245 }
1246
1247 int dlm_send_one_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
1248 struct dlm_migratable_lockres *mres,
1249 u8 send_to, u8 flags)
1250 {
1251 struct list_head *queue;
1252 int total_locks, i;
1253 u64 mig_cookie = 0;
1254 struct dlm_lock *lock;
1255 int ret = 0;
1256
1257 BUG_ON(!(flags & (DLM_MRES_RECOVERY|DLM_MRES_MIGRATION)));
1258
1259 mlog(0, "sending to %u\n", send_to);
1260
1261 total_locks = dlm_num_locks_in_lockres(res);
1262 if (total_locks > DLM_MAX_MIGRATABLE_LOCKS) {
1263 /* rare, but possible */
1264 mlog(0, "argh. lockres has %d locks. this will "
1265 "require more than one network packet to "
1266 "migrate\n", total_locks);
1267 mig_cookie = dlm_get_next_mig_cookie();
1268 }
1269
1270 dlm_init_migratable_lockres(mres, res->lockname.name,
1271 res->lockname.len, total_locks,
1272 mig_cookie, flags, res->owner);
1273
1274 total_locks = 0;
1275 for (i=DLM_GRANTED_LIST; i<=DLM_BLOCKED_LIST; i++) {
1276 queue = dlm_list_idx_to_ptr(res, i);
1277 list_for_each_entry(lock, queue, list) {
1278 /* add another lock. */
1279 total_locks++;
1280 if (!dlm_add_lock_to_array(lock, mres, i))
1281 continue;
1282
1283 /* this filled the lock message,
1284 * we must send it immediately. */
1285 ret = dlm_send_mig_lockres_msg(dlm, mres, send_to,
1286 res, total_locks);
1287 if (ret < 0)
1288 goto error;
1289 }
1290 }
1291 if (total_locks == 0) {
1292 /* send a dummy lock to indicate a mastery reference only */
1293 mlog(0, "%s:%.*s: sending dummy lock to %u, %s\n",
1294 dlm->name, res->lockname.len, res->lockname.name,
1295 send_to, flags & DLM_MRES_RECOVERY ? "recovery" :
1296 "migration");
1297 dlm_add_dummy_lock(dlm, mres);
1298 }
1299 /* flush any remaining locks */
1300 ret = dlm_send_mig_lockres_msg(dlm, mres, send_to, res, total_locks);
1301 if (ret < 0)
1302 goto error;
1303 return ret;
1304
1305 error:
1306 mlog(ML_ERROR, "%s: dlm_send_mig_lockres_msg returned %d\n",
1307 dlm->name, ret);
1308 if (!dlm_is_host_down(ret))
1309 BUG();
1310 mlog(0, "%s: node %u went down while sending %s "
1311 "lockres %.*s\n", dlm->name, send_to,
1312 flags & DLM_MRES_RECOVERY ? "recovery" : "migration",
1313 res->lockname.len, res->lockname.name);
1314 return ret;
1315 }
1316
1317
1318
1319 /*
1320 * this message will contain no more than one page worth of
1321 * recovery data, and it will work on only one lockres.
1322 * there may be many locks in this page, and we may need to wait
1323 * for additional packets to complete all the locks (rare, but
1324 * possible).
1325 */
1326 /*
1327 * NOTE: the allocation error cases here are scary
1328 * we really cannot afford to fail an alloc in recovery
1329 * do we spin? returning an error only delays the problem really
1330 */
1331
1332 int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
1333 void **ret_data)
1334 {
1335 struct dlm_ctxt *dlm = data;
1336 struct dlm_migratable_lockres *mres =
1337 (struct dlm_migratable_lockres *)msg->buf;
1338 int ret = 0;
1339 u8 real_master;
1340 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1341 =======
1342 u8 extra_refs = 0;
1343 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1344 char *buf = NULL;
1345 struct dlm_work_item *item = NULL;
1346 struct dlm_lock_resource *res = NULL;
1347
1348 if (!dlm_grab(dlm))
1349 return -EINVAL;
1350
1351 BUG_ON(!(mres->flags & (DLM_MRES_RECOVERY|DLM_MRES_MIGRATION)));
1352
1353 real_master = mres->master;
1354 if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
1355 /* cannot migrate a lockres with no master */
1356 BUG_ON(!(mres->flags & DLM_MRES_RECOVERY));
1357 }
1358
1359 mlog(0, "%s message received from node %u\n",
1360 (mres->flags & DLM_MRES_RECOVERY) ?
1361 "recovery" : "migration", mres->master);
1362 if (mres->flags & DLM_MRES_ALL_DONE)
1363 mlog(0, "all done flag. all lockres data received!\n");
1364
1365 ret = -ENOMEM;
1366 buf = kmalloc(be16_to_cpu(msg->data_len), GFP_NOFS);
1367 item = kzalloc(sizeof(*item), GFP_NOFS);
1368 if (!buf || !item)
1369 goto leave;
1370
1371 /* lookup the lock to see if we have a secondary queue for this
1372 * already... just add the locks in and this will have its owner
1373 * and RECOVERY flag changed when it completes. */
1374 res = dlm_lookup_lockres(dlm, mres->lockname, mres->lockname_len);
1375 if (res) {
1376 /* this will get a ref on res */
1377 /* mark it as recovering/migrating and hash it */
1378 spin_lock(&res->spinlock);
1379 if (mres->flags & DLM_MRES_RECOVERY) {
1380 res->state |= DLM_LOCK_RES_RECOVERING;
1381 } else {
1382 if (res->state & DLM_LOCK_RES_MIGRATING) {
1383 /* this is at least the second
1384 * lockres message */
1385 mlog(0, "lock %.*s is already migrating\n",
1386 mres->lockname_len,
1387 mres->lockname);
1388 } else if (res->state & DLM_LOCK_RES_RECOVERING) {
1389 /* caller should BUG */
1390 mlog(ML_ERROR, "node is attempting to migrate "
1391 "lock %.*s, but marked as recovering!\n",
1392 mres->lockname_len, mres->lockname);
1393 ret = -EFAULT;
1394 spin_unlock(&res->spinlock);
1395 goto leave;
1396 }
1397 res->state |= DLM_LOCK_RES_MIGRATING;
1398 }
1399 spin_unlock(&res->spinlock);
1400 } else {
1401 /* need to allocate, just like if it was
1402 * mastered here normally */
1403 res = dlm_new_lockres(dlm, mres->lockname, mres->lockname_len);
1404 if (!res)
1405 goto leave;
1406
1407 /* to match the ref that we would have gotten if
1408 * dlm_lookup_lockres had succeeded */
1409 dlm_lockres_get(res);
1410
1411 /* mark it as recovering/migrating and hash it */
1412 if (mres->flags & DLM_MRES_RECOVERY)
1413 res->state |= DLM_LOCK_RES_RECOVERING;
1414 else
1415 res->state |= DLM_LOCK_RES_MIGRATING;
1416
1417 spin_lock(&dlm->spinlock);
1418 __dlm_insert_lockres(dlm, res);
1419 spin_unlock(&dlm->spinlock);
1420
1421 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1422 =======
1423 /* Add an extra ref for this lock-less lockres lest the
1424 * dlm_thread purges it before we get the chance to add
1425 * locks to it */
1426 dlm_lockres_get(res);
1427
1428 /* There are three refs that need to be put.
1429 * 1. Taken above.
1430 * 2. kref_init in dlm_new_lockres()->dlm_init_lockres().
1431 * 3. dlm_lookup_lockres()
1432 * The first one is handled at the end of this function. The
1433 * other two are handled in the worker thread after locks have
1434 * been attached. Yes, we don't wait for purge time to match
1435 * kref_init. The lockres will still have atleast one ref
1436 * added because it is in the hash __dlm_insert_lockres() */
1437 extra_refs++;
1438
1439 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1440 /* now that the new lockres is inserted,
1441 * make it usable by other processes */
1442 spin_lock(&res->spinlock);
1443 res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
1444 spin_unlock(&res->spinlock);
1445 wake_up(&res->wq);
1446 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1447
1448 /* add an extra ref for just-allocated lockres
1449 * otherwise the lockres will be purged immediately */
1450 dlm_lockres_get(res);
1451 =======
1452 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1453 }
1454
1455 /* at this point we have allocated everything we need,
1456 * and we have a hashed lockres with an extra ref and
1457 * the proper res->state flags. */
1458 ret = 0;
1459 spin_lock(&res->spinlock);
1460 /* drop this either when master requery finds a different master
1461 * or when a lock is added by the recovery worker */
1462 dlm_lockres_grab_inflight_ref(dlm, res);
1463 if (mres->master == DLM_LOCK_RES_OWNER_UNKNOWN) {
1464 /* migration cannot have an unknown master */
1465 BUG_ON(!(mres->flags & DLM_MRES_RECOVERY));
1466 mlog(0, "recovery has passed me a lockres with an "
1467 "unknown owner.. will need to requery: "
1468 "%.*s\n", mres->lockname_len, mres->lockname);
1469 } else {
1470 /* take a reference now to pin the lockres, drop it
1471 * when locks are added in the worker */
1472 dlm_change_lockres_owner(dlm, res, dlm->node_num);
1473 }
1474 spin_unlock(&res->spinlock);
1475
1476 /* queue up work for dlm_mig_lockres_worker */
1477 dlm_grab(dlm); /* get an extra ref for the work item */
1478 memcpy(buf, msg->buf, be16_to_cpu(msg->data_len)); /* copy the whole message */
1479 dlm_init_work_item(dlm, item, dlm_mig_lockres_worker, buf);
1480 item->u.ml.lockres = res; /* already have a ref */
1481 item->u.ml.real_master = real_master;
1482 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1483 =======
1484 item->u.ml.extra_ref = extra_refs;
1485 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1486 spin_lock(&dlm->work_lock);
1487 list_add_tail(&item->list, &dlm->work_list);
1488 spin_unlock(&dlm->work_lock);
1489 queue_work(dlm->dlm_worker, &dlm->dispatched_work);
1490
1491 leave:
1492 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1493 =======
1494 /* One extra ref taken needs to be put here */
1495 if (extra_refs)
1496 dlm_lockres_put(res);
1497
1498 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1499 dlm_put(dlm);
1500 if (ret < 0) {
1501 if (buf)
1502 kfree(buf);
1503 if (item)
1504 kfree(item);
1505 }
1506
1507 mlog_exit(ret);
1508 return ret;
1509 }
1510
1511
1512 static void dlm_mig_lockres_worker(struct dlm_work_item *item, void *data)
1513 {
1514 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1515 struct dlm_ctxt *dlm = data;
1516 =======
1517 struct dlm_ctxt *dlm;
1518 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1519 struct dlm_migratable_lockres *mres;
1520 int ret = 0;
1521 struct dlm_lock_resource *res;
1522 u8 real_master;
1523 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1524 =======
1525 u8 extra_ref;
1526 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1527
1528 dlm = item->dlm;
1529 mres = (struct dlm_migratable_lockres *)data;
1530
1531 res = item->u.ml.lockres;
1532 real_master = item->u.ml.real_master;
1533 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1534 =======
1535 extra_ref = item->u.ml.extra_ref;
1536 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1537
1538 if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
1539 /* this case is super-rare. only occurs if
1540 * node death happens during migration. */
1541 again:
1542 ret = dlm_lockres_master_requery(dlm, res, &real_master);
1543 if (ret < 0) {
1544 mlog(0, "dlm_lockres_master_requery ret=%d\n",
1545 ret);
1546 goto again;
1547 }
1548 if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
1549 mlog(0, "lockres %.*s not claimed. "
1550 "this node will take it.\n",
1551 res->lockname.len, res->lockname.name);
1552 } else {
1553 spin_lock(&res->spinlock);
1554 dlm_lockres_drop_inflight_ref(dlm, res);
1555 spin_unlock(&res->spinlock);
1556 mlog(0, "master needs to respond to sender "
1557 "that node %u still owns %.*s\n",
1558 real_master, res->lockname.len,
1559 res->lockname.name);
1560 /* cannot touch this lockres */
1561 goto leave;
1562 }
1563 }
1564
1565 ret = dlm_process_recovery_data(dlm, res, mres);
1566 if (ret < 0)
1567 mlog(0, "dlm_process_recovery_data returned %d\n", ret);
1568 else
1569 mlog(0, "dlm_process_recovery_data succeeded\n");
1570
1571 if ((mres->flags & (DLM_MRES_MIGRATION|DLM_MRES_ALL_DONE)) ==
1572 (DLM_MRES_MIGRATION|DLM_MRES_ALL_DONE)) {
1573 ret = dlm_finish_migration(dlm, res, mres->master);
1574 if (ret < 0)
1575 mlog_errno(ret);
1576 }
1577
1578 leave:
1579 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1580 =======
1581 /* See comment in dlm_mig_lockres_handler() */
1582 if (res) {
1583 if (extra_ref)
1584 dlm_lockres_put(res);
1585 dlm_lockres_put(res);
1586 }
1587 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1588 kfree(data);
1589 mlog_exit(ret);
1590 }
1591
1592
1593
1594 static int dlm_lockres_master_requery(struct dlm_ctxt *dlm,
1595 struct dlm_lock_resource *res,
1596 u8 *real_master)
1597 {
1598 struct dlm_node_iter iter;
1599 int nodenum;
1600 int ret = 0;
1601
1602 *real_master = DLM_LOCK_RES_OWNER_UNKNOWN;
1603
1604 /* we only reach here if one of the two nodes in a
1605 * migration died while the migration was in progress.
1606 * at this point we need to requery the master. we
1607 * know that the new_master got as far as creating
1608 * an mle on at least one node, but we do not know
1609 * if any nodes had actually cleared the mle and set
1610 * the master to the new_master. the old master
1611 * is supposed to set the owner to UNKNOWN in the
1612 * event of a new_master death, so the only possible
1613 * responses that we can get from nodes here are
1614 * that the master is new_master, or that the master
1615 * is UNKNOWN.
1616 * if all nodes come back with UNKNOWN then we know
1617 * the lock needs remastering here.
1618 * if any node comes back with a valid master, check
1619 * to see if that master is the one that we are
1620 * recovering. if so, then the new_master died and
1621 * we need to remaster this lock. if not, then the
1622 * new_master survived and that node will respond to
1623 * other nodes about the owner.
1624 * if there is an owner, this node needs to dump this
1625 * lockres and alert the sender that this lockres
1626 * was rejected. */
1627 spin_lock(&dlm->spinlock);
1628 dlm_node_iter_init(dlm->domain_map, &iter);
1629 spin_unlock(&dlm->spinlock);
1630
1631 while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
1632 /* do not send to self */
1633 if (nodenum == dlm->node_num)
1634 continue;
1635 ret = dlm_do_master_requery(dlm, res, nodenum, real_master);
1636 if (ret < 0) {
1637 mlog_errno(ret);
1638 if (!dlm_is_host_down(ret))
1639 BUG();
1640 /* host is down, so answer for that node would be
1641 * DLM_LOCK_RES_OWNER_UNKNOWN. continue. */
1642 }
1643 if (*real_master != DLM_LOCK_RES_OWNER_UNKNOWN) {
1644 mlog(0, "lock master is %u\n", *real_master);
1645 break;
1646 }
1647 }
1648 return ret;
1649 }
1650
1651
1652 int dlm_do_master_requery(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
1653 u8 nodenum, u8 *real_master)
1654 {
1655 int ret = -EINVAL;
1656 struct dlm_master_requery req;
1657 int status = DLM_LOCK_RES_OWNER_UNKNOWN;
1658
1659 memset(&req, 0, sizeof(req));
1660 req.node_idx = dlm->node_num;
1661 req.namelen = res->lockname.len;
1662 memcpy(req.name, res->lockname.name, res->lockname.len);
1663
1664 ret = o2net_send_message(DLM_MASTER_REQUERY_MSG, dlm->key,
1665 &req, sizeof(req), nodenum, &status);
1666 /* XXX: negative status not handled properly here. */
1667 if (ret < 0)
1668 mlog_errno(ret);
1669 else {
1670 BUG_ON(status < 0);
1671 BUG_ON(status > DLM_LOCK_RES_OWNER_UNKNOWN);
1672 *real_master = (u8) (status & 0xff);
1673 mlog(0, "node %u responded to master requery with %u\n",
1674 nodenum, *real_master);
1675 ret = 0;
1676 }
1677 return ret;
1678 }
1679
1680
1681 /* this function cannot error, so unless the sending
1682 * or receiving of the message failed, the owner can
1683 * be trusted */
1684 int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data,
1685 void **ret_data)
1686 {
1687 struct dlm_ctxt *dlm = data;
1688 struct dlm_master_requery *req = (struct dlm_master_requery *)msg->buf;
1689 struct dlm_lock_resource *res = NULL;
1690 unsigned int hash;
1691 int master = DLM_LOCK_RES_OWNER_UNKNOWN;
1692 u32 flags = DLM_ASSERT_MASTER_REQUERY;
1693
1694 if (!dlm_grab(dlm)) {
1695 /* since the domain has gone away on this
1696 * node, the proper response is UNKNOWN */
1697 return master;
1698 }
1699
1700 hash = dlm_lockid_hash(req->name, req->namelen);
1701
1702 spin_lock(&dlm->spinlock);
1703 res = __dlm_lookup_lockres(dlm, req->name, req->namelen, hash);
1704 if (res) {
1705 spin_lock(&res->spinlock);
1706 master = res->owner;
1707 if (master == dlm->node_num) {
1708 int ret = dlm_dispatch_assert_master(dlm, res,
1709 0, 0, flags);
1710 if (ret < 0) {
1711 mlog_errno(-ENOMEM);
1712 /* retry!? */
1713 BUG();
1714 }
1715 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1716 }
1717 =======
1718 } else /* put.. incase we are not the master */
1719 dlm_lockres_put(res);
1720 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
1721 spin_unlock(&res->spinlock);
1722 }
1723 spin_unlock(&dlm->spinlock);
1724
1725 dlm_put(dlm);
1726 return master;
1727 }
1728
1729 static inline struct list_head *
1730 dlm_list_num_to_pointer(struct dlm_lock_resource *res, int list_num)
1731 {
1732 struct list_head *ret;
1733 BUG_ON(list_num < 0);
1734 BUG_ON(list_num > 2);
1735 ret = &(res->granted);
1736 ret += list_num;
1737 return ret;
1738 }
1739 /* TODO: do ast flush business
1740 * TODO: do MIGRATING and RECOVERING spinning
1741 */
1742
1743 /*
1744 * NOTE about in-flight requests during migration:
1745 *
1746 * Before attempting the migrate, the master has marked the lockres as
1747 * MIGRATING and then flushed all of its pending ASTS. So any in-flight
1748 * requests either got queued before the MIGRATING flag got set, in which
1749 * case the lock data will reflect the change and a return message is on
1750 * the way, or the request failed to get in before MIGRATING got set. In
1751 * this case, the caller will be told to spin and wait for the MIGRATING
1752 * flag to be dropped, then recheck the master.
1753 * This holds true for the convert, cancel and unlock cases, and since lvb
1754 * updates are tied to these same messages, it applies to lvb updates as
1755 * well. For the lock case, there is no way a lock can be on the master
1756 * queue and not be on the secondary queue since the lock is always added
1757 * locally first. This means that the new target node will never be sent
1758 * a lock that he doesn't already have on the list.
1759 * In total, this means that the local lock is correct and should not be
1760 * updated to match the one sent by the master. Any messages sent back
1761 * from the master before the MIGRATING flag will bring the lock properly
1762 * up-to-date, and the change will be ordered properly for the waiter.
1763 * We will *not* attempt to modify the lock underneath the waiter.
1764 */
1765
1766 static int dlm_process_recovery_data(struct dlm_ctxt *dlm,
1767 struct dlm_lock_resource *res,
1768 struct dlm_migratable_lockres *mres)
1769 {
1770 struct dlm_migratable_lock *ml;
1771 struct list_head *queue;
1772 struct list_head *tmpq = NULL;
1773 struct dlm_lock *newlock = NULL;
1774 struct dlm_lockstatus *lksb = NULL;
1775 int ret = 0;
1776 int i, j, bad;
1777 struct dlm_lock *lock = NULL;
1778 u8 from = O2NM_MAX_NODES;
1779 unsigned int added = 0;
1780
1781 mlog(0, "running %d locks for this lockres\n", mres->num_locks);
1782 for (i=0; i<mres->num_locks; i++) {
1783 ml = &(mres->ml[i]);
1784
1785 if (dlm_is_dummy_lock(dlm, ml, &from)) {
1786 /* placeholder, just need to set the refmap bit */
1787 BUG_ON(mres->num_locks != 1);
1788 mlog(0, "%s:%.*s: dummy lock for %u\n",
1789 dlm->name, mres->lockname_len, mres->lockname,
1790 from);
1791 spin_lock(&res->spinlock);
1792 dlm_lockres_set_refmap_bit(from, res);
1793 spin_unlock(&res->spinlock);
1794 added++;
1795 break;
1796 }
1797 BUG_ON(ml->highest_blocked != LKM_IVMODE);
1798 newlock = NULL;
1799 lksb = NULL;
1800
1801 queue = dlm_list_num_to_pointer(res, ml->list);
1802 tmpq = NULL;
1803
1804 /* if the lock is for the local node it needs to
1805 * be moved to the proper location within the queue.
1806 * do not allocate a new lock structure. */
1807 if (ml->node == dlm->node_num) {
1808 /* MIGRATION ONLY! */
1809 BUG_ON(!(mres->flags & DLM_MRES_MIGRATION));
1810
1811 spin_lock(&res->spinlock);
1812 for (j = DLM_GRANTED_LIST; j <= DLM_BLOCKED_LIST; j++) {
1813 tmpq = dlm_list_idx_to_ptr(res, j);
1814 list_for_each_entry(lock, tmpq, list) {
1815 if (lock->ml.cookie != ml->cookie)
1816 lock = NULL;
1817 else
1818 break;
1819 }
1820 if (lock)
1821 break;
1822 }
1823
1824 /* lock is always created locally first, and
1825 * destroyed locally last. it must be on the list */
1826 if (!lock) {
1827 __be64 c = ml->cookie;
1828 mlog(ML_ERROR, "could not find local lock "
1829 "with cookie %u:%llu!\n",
1830 dlm_get_lock_cookie_node(be64_to_cpu(c)),
1831 dlm_get_lock_cookie_seq(be64_to_cpu(c)));
1832 __dlm_print_one_lock_resource(res);
1833 BUG();
1834 }
1835 BUG_ON(lock->ml.node != ml->node);
1836
1837 if (tmpq != queue) {
1838 mlog(0, "lock was on %u instead of %u for %.*s\n",
1839 j, ml->list, res->lockname.len, res->lockname.name);
1840 spin_unlock(&res->spinlock);
1841 continue;
1842 }
1843
1844 /* see NOTE above about why we do not update
1845 * to match the master here */
1846
1847 /* move the lock to its proper place */
1848 /* do not alter lock refcount. switching lists. */
1849 list_move_tail(&lock->list, queue);
1850 spin_unlock(&res->spinlock);
1851 added++;
1852
1853 mlog(0, "just reordered a local lock!\n");
1854 continue;
1855 }
1856
1857 /* lock is for another node. */
1858 newlock = dlm_new_lock(ml->type, ml->node,
1859 be64_to_cpu(ml->cookie), NULL);
1860 if (!newlock) {
1861 ret = -ENOMEM;
1862 goto leave;
1863 }
1864 lksb = newlock->lksb;
1865 dlm_lock_attach_lockres(newlock, res);
1866
1867 if (ml->convert_type != LKM_IVMODE) {
1868 BUG_ON(queue != &res->converting);
1869 newlock->ml.convert_type = ml->convert_type;
1870 }
1871 lksb->flags |= (ml->flags &
1872 (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
1873
1874 if (ml->type == LKM_NLMODE)
1875 goto skip_lvb;
1876
1877 if (!dlm_lvb_is_empty(mres->lvb)) {
1878 if (lksb->flags & DLM_LKSB_PUT_LVB) {
1879 /* other node was trying to update
1880 * lvb when node died. recreate the
1881 * lksb with the updated lvb. */
1882 memcpy(lksb->lvb, mres->lvb, DLM_LVB_LEN);
1883 /* the lock resource lvb update must happen
1884 * NOW, before the spinlock is dropped.
1885 * we no longer wait for the AST to update
1886 * the lvb. */
1887 memcpy(res->lvb, mres->lvb, DLM_LVB_LEN);
1888 } else {
1889 /* otherwise, the node is sending its
1890 * most recent valid lvb info */
1891 BUG_ON(ml->type != LKM_EXMODE &&
1892 ml->type != LKM_PRMODE);
1893 if (!dlm_lvb_is_empty(res->lvb) &&
1894 (ml->type == LKM_EXMODE ||
1895 memcmp(res->lvb, mres->lvb, DLM_LVB_LEN))) {
1896 int i;
1897 mlog(ML_ERROR, "%s:%.*s: received bad "
1898 "lvb! type=%d\n", dlm->name,
1899 res->lockname.len,
1900 res->lockname.name, ml->type);
1901 printk("lockres lvb=[");
1902 for (i=0; i<DLM_LVB_LEN; i++)
1903 printk("%02x", res->lvb[i]);
1904 printk("]\nmigrated lvb=[");
1905 for (i=0; i<DLM_LVB_LEN; i++)
1906 printk("%02x", mres->lvb[i]);
1907 printk("]\n");
1908 dlm_print_one_lock_resource(res);
1909 BUG();
1910 }
1911 memcpy(res->lvb, mres->lvb, DLM_LVB_LEN);
1912 }
1913 }
1914 skip_lvb:
1915
1916 /* NOTE:
1917 * wrt lock queue ordering and recovery:
1918 * 1. order of locks on granted queue is
1919 * meaningless.
1920 * 2. order of locks on converting queue is
1921 * LOST with the node death. sorry charlie.
1922 * 3. order of locks on the blocked queue is
1923 * also LOST.
1924 * order of locks does not affect integrity, it
1925 * just means that a lock request may get pushed
1926 * back in line as a result of the node death.
1927 * also note that for a given node the lock order
1928 * for its secondary queue locks is preserved
1929 * relative to each other, but clearly *not*
1930 * preserved relative to locks from other nodes.
1931 */
1932 bad = 0;
1933 spin_lock(&res->spinlock);
1934 list_for_each_entry(lock, queue, list) {
1935 if (lock->ml.cookie == ml->cookie) {
1936 __be64 c = lock->ml.cookie;
1937 mlog(ML_ERROR, "%s:%.*s: %u:%llu: lock already "
1938 "exists on this lockres!\n", dlm->name,
1939 res->lockname.len, res->lockname.name,
1940 dlm_get_lock_cookie_node(be64_to_cpu(c)),
1941 dlm_get_lock_cookie_seq(be64_to_cpu(c)));
1942
1943 mlog(ML_NOTICE, "sent lock: type=%d, conv=%d, "
1944 "node=%u, cookie=%u:%llu, queue=%d\n",
1945 ml->type, ml->convert_type, ml->node,
1946 dlm_get_lock_cookie_node(be64_to_cpu(ml->cookie)),
1947 dlm_get_lock_cookie_seq(be64_to_cpu(ml->cookie)),
1948 ml->list);
1949
1950 __dlm_print_one_lock_resource(res);
1951 bad = 1;
1952 break;
1953 }
1954 }
1955 if (!bad) {
1956 dlm_lock_get(newlock);
1957 list_add_tail(&newlock->list, queue);
1958 mlog(0, "%s:%.*s: added lock for node %u, "
1959 "setting refmap bit\n", dlm->name,
1960 res->lockname.len, res->lockname.name, ml->node);
1961 dlm_lockres_set_refmap_bit(ml->node, res);
1962 added++;
1963 }
1964 spin_unlock(&res->spinlock);
1965 }
1966 mlog(0, "done running all the locks\n");
1967
1968 leave:
1969 /* balance the ref taken when the work was queued */
1970 spin_lock(&res->spinlock);
1971 dlm_lockres_drop_inflight_ref(dlm, res);
1972 spin_unlock(&res->spinlock);
1973
1974 if (ret < 0) {
1975 mlog_errno(ret);
1976 if (newlock)
1977 dlm_lock_put(newlock);
1978 }
1979
1980 mlog_exit(ret);
1981 return ret;
1982 }
1983
1984 void dlm_move_lockres_to_recovery_list(struct dlm_ctxt *dlm,
1985 struct dlm_lock_resource *res)
1986 {
1987 int i;
1988 struct list_head *queue;
1989 struct dlm_lock *lock, *next;
1990
1991 res->state |= DLM_LOCK_RES_RECOVERING;
1992 if (!list_empty(&res->recovering)) {
1993 mlog(0,
1994 "Recovering res %s:%.*s, is already on recovery list!\n",
1995 dlm->name, res->lockname.len, res->lockname.name);
1996 list_del_init(&res->recovering);
1997 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
1998 =======
1999 dlm_lockres_put(res);
2000 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
2001 }
2002 /* We need to hold a reference while on the recovery list */
2003 dlm_lockres_get(res);
2004 list_add_tail(&res->recovering, &dlm->reco.resources);
2005
2006 /* find any pending locks and put them back on proper list */
2007 for (i=DLM_BLOCKED_LIST; i>=DLM_GRANTED_LIST; i--) {
2008 queue = dlm_list_idx_to_ptr(res, i);
2009 list_for_each_entry_safe(lock, next, queue, list) {
2010 dlm_lock_get(lock);
2011 if (lock->convert_pending) {
2012 /* move converting lock back to granted */
2013 BUG_ON(i != DLM_CONVERTING_LIST);
2014 mlog(0, "node died with convert pending "
2015 "on %.*s. move back to granted list.\n",
2016 res->lockname.len, res->lockname.name);
2017 dlm_revert_pending_convert(res, lock);
2018 lock->convert_pending = 0;
2019 } else if (lock->lock_pending) {
2020 /* remove pending lock requests completely */
2021 BUG_ON(i != DLM_BLOCKED_LIST);
2022 mlog(0, "node died with lock pending "
2023 "on %.*s. remove from blocked list and skip.\n",
2024 res->lockname.len, res->lockname.name);
2025 /* lock will be floating until ref in
2026 * dlmlock_remote is freed after the network
2027 * call returns. ok for it to not be on any
2028 * list since no ast can be called
2029 * (the master is dead). */
2030 dlm_revert_pending_lock(res, lock);
2031 lock->lock_pending = 0;
2032 } else if (lock->unlock_pending) {
2033 /* if an unlock was in progress, treat as
2034 * if this had completed successfully
2035 * before sending this lock state to the
2036 * new master. note that the dlm_unlock
2037 * call is still responsible for calling
2038 * the unlockast. that will happen after
2039 * the network call times out. for now,
2040 * just move lists to prepare the new
2041 * recovery master. */
2042 BUG_ON(i != DLM_GRANTED_LIST);
2043 mlog(0, "node died with unlock pending "
2044 "on %.*s. remove from blocked list and skip.\n",
2045 res->lockname.len, res->lockname.name);
2046 dlm_commit_pending_unlock(res, lock);
2047 lock->unlock_pending = 0;
2048 } else if (lock->cancel_pending) {
2049 /* if a cancel was in progress, treat as
2050 * if this had completed successfully
2051 * before sending this lock state to the
2052 * new master */
2053 BUG_ON(i != DLM_CONVERTING_LIST);
2054 mlog(0, "node died with cancel pending "
2055 "on %.*s. move back to granted list.\n",
2056 res->lockname.len, res->lockname.name);
2057 dlm_commit_pending_cancel(res, lock);
2058 lock->cancel_pending = 0;
2059 }
2060 dlm_lock_put(lock);
2061 }
2062 }
2063 }
2064
2065
2066
2067 /* removes all recovered locks from the recovery list.
2068 * sets the res->owner to the new master.
2069 * unsets the RECOVERY flag and wakes waiters. */
2070 static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
2071 u8 dead_node, u8 new_master)
2072 {
2073 int i;
2074 struct hlist_node *hash_iter;
2075 struct hlist_head *bucket;
2076 struct dlm_lock_resource *res, *next;
2077
2078 mlog_entry_void();
2079
2080 assert_spin_locked(&dlm->spinlock);
2081
2082 list_for_each_entry_safe(res, next, &dlm->reco.resources, recovering) {
2083 if (res->owner == dead_node) {
2084 list_del_init(&res->recovering);
2085 spin_lock(&res->spinlock);
2086 /* new_master has our reference from
2087 * the lock state sent during recovery */
2088 dlm_change_lockres_owner(dlm, res, new_master);
2089 res->state &= ~DLM_LOCK_RES_RECOVERING;
2090 if (__dlm_lockres_has_locks(res))
2091 __dlm_dirty_lockres(dlm, res);
2092 spin_unlock(&res->spinlock);
2093 wake_up(&res->wq);
2094 dlm_lockres_put(res);
2095 }
2096 }
2097
2098 /* this will become unnecessary eventually, but
2099 * for now we need to run the whole hash, clear
2100 * the RECOVERING state and set the owner
2101 * if necessary */
2102 for (i = 0; i < DLM_HASH_BUCKETS; i++) {
2103 bucket = dlm_lockres_hash(dlm, i);
2104 hlist_for_each_entry(res, hash_iter, bucket, hash_node) {
2105 if (res->state & DLM_LOCK_RES_RECOVERING) {
2106 if (res->owner == dead_node) {
2107 mlog(0, "(this=%u) res %.*s owner=%u "
2108 "was not on recovering list, but "
2109 "clearing state anyway\n",
2110 dlm->node_num, res->lockname.len,
2111 res->lockname.name, new_master);
2112 } else if (res->owner == dlm->node_num) {
2113 mlog(0, "(this=%u) res %.*s owner=%u "
2114 "was not on recovering list, "
2115 "owner is THIS node, clearing\n",
2116 dlm->node_num, res->lockname.len,
2117 res->lockname.name, new_master);
2118 } else
2119 continue;
2120
2121 if (!list_empty(&res->recovering)) {
2122 mlog(0, "%s:%.*s: lockres was "
2123 "marked RECOVERING, owner=%u\n",
2124 dlm->name, res->lockname.len,
2125 res->lockname.name, res->owner);
2126 list_del_init(&res->recovering);
2127 dlm_lockres_put(res);
2128 }
2129 spin_lock(&res->spinlock);
2130 /* new_master has our reference from
2131 * the lock state sent during recovery */
2132 dlm_change_lockres_owner(dlm, res, new_master);
2133 res->state &= ~DLM_LOCK_RES_RECOVERING;
2134 if (__dlm_lockres_has_locks(res))
2135 __dlm_dirty_lockres(dlm, res);
2136 spin_unlock(&res->spinlock);
2137 wake_up(&res->wq);
2138 }
2139 }
2140 }
2141 }
2142
2143 static inline int dlm_lvb_needs_invalidation(struct dlm_lock *lock, int local)
2144 {
2145 if (local) {
2146 if (lock->ml.type != LKM_EXMODE &&
2147 lock->ml.type != LKM_PRMODE)
2148 return 1;
2149 } else if (lock->ml.type == LKM_EXMODE)
2150 return 1;
2151 return 0;
2152 }
2153
2154 static void dlm_revalidate_lvb(struct dlm_ctxt *dlm,
2155 struct dlm_lock_resource *res, u8 dead_node)
2156 {
2157 struct list_head *queue;
2158 struct dlm_lock *lock;
2159 int blank_lvb = 0, local = 0;
2160 int i;
2161 u8 search_node;
2162
2163 assert_spin_locked(&dlm->spinlock);
2164 assert_spin_locked(&res->spinlock);
2165
2166 if (res->owner == dlm->node_num)
2167 /* if this node owned the lockres, and if the dead node
2168 * had an EX when he died, blank out the lvb */
2169 search_node = dead_node;
2170 else {
2171 /* if this is a secondary lockres, and we had no EX or PR
2172 * locks granted, we can no longer trust the lvb */
2173 search_node = dlm->node_num;
2174 local = 1; /* check local state for valid lvb */
2175 }
2176
2177 for (i=DLM_GRANTED_LIST; i<=DLM_CONVERTING_LIST; i++) {
2178 queue = dlm_list_idx_to_ptr(res, i);
2179 list_for_each_entry(lock, queue, list) {
2180 if (lock->ml.node == search_node) {
2181 if (dlm_lvb_needs_invalidation(lock, local)) {
2182 /* zero the lksb lvb and lockres lvb */
2183 blank_lvb = 1;
2184 memset(lock->lksb->lvb, 0, DLM_LVB_LEN);
2185 }
2186 }
2187 }
2188 }
2189
2190 if (blank_lvb) {
2191 mlog(0, "clearing %.*s lvb, dead node %u had EX\n",
2192 res->lockname.len, res->lockname.name, dead_node);
2193 memset(res->lvb, 0, DLM_LVB_LEN);
2194 }
2195 }
2196
2197 static void dlm_free_dead_locks(struct dlm_ctxt *dlm,
2198 struct dlm_lock_resource *res, u8 dead_node)
2199 {
2200 struct dlm_lock *lock, *next;
2201 unsigned int freed = 0;
2202
2203 /* this node is the lockres master:
2204 * 1) remove any stale locks for the dead node
2205 * 2) if the dead node had an EX when he died, blank out the lvb
2206 */
2207 assert_spin_locked(&dlm->spinlock);
2208 assert_spin_locked(&res->spinlock);
2209
2210 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
2211 =======
2212 /* We do two dlm_lock_put(). One for removing from list and the other is
2213 * to force the DLM_UNLOCK_FREE_LOCK action so as to free the locks */
2214
2215 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
2216 /* TODO: check pending_asts, pending_basts here */
2217 list_for_each_entry_safe(lock, next, &res->granted, list) {
2218 if (lock->ml.node == dead_node) {
2219 list_del_init(&lock->list);
2220 dlm_lock_put(lock);
2221 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
2222 =======
2223 /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
2224 dlm_lock_put(lock);
2225 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
2226 freed++;
2227 }
2228 }
2229 list_for_each_entry_safe(lock, next, &res->converting, list) {
2230 if (lock->ml.node == dead_node) {
2231 list_del_init(&lock->list);
2232 dlm_lock_put(lock);
2233 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
2234 =======
2235 /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
2236 dlm_lock_put(lock);
2237 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
2238 freed++;
2239 }
2240 }
2241 list_for_each_entry_safe(lock, next, &res->blocked, list) {
2242 if (lock->ml.node == dead_node) {
2243 list_del_init(&lock->list);
2244 dlm_lock_put(lock);
2245 <<<<<<< HEAD:fs/ocfs2/dlm/dlmrecovery.c
2246 =======
2247 /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
2248 dlm_lock_put(lock);
2249 >>>>>>> 264e3e889d86e552b4191d69bb60f4f3b383135a:fs/ocfs2/dlm/dlmrecovery.c
2250 freed++;
2251 }
2252 }
2253
2254 if (freed) {
2255 mlog(0, "%s:%.*s: freed %u locks for dead node %u, "
2256 "dropping ref from lockres\n", dlm->name,
2257 res->lockname.len, res->lockname.name, freed, dead_node);
2258 BUG_ON(!test_bit(dead_node, res->refmap));
2259 dlm_lockres_clear_refmap_bit(dead_node, res);
2260 } else if (test_bit(dead_node, res->refmap)) {
2261 mlog(0, "%s:%.*s: dead node %u had a ref, but had "
2262 "no locks and had not purged before dying\n", dlm->name,
2263 res->lockname.len, res->lockname.name, dead_node);
2264 dlm_lockres_clear_refmap_bit(dead_node, res);
2265 }
2266
2267 /* do not kick thread yet */
2268 __dlm_dirty_lockres(dlm, res);
2269 }
2270
2271 /* if this node is the recovery master, and there are no
2272 * locks for a given lockres owned by this node that are in
2273 * either PR or EX mode, zero out the lvb before requesting.
2274 *
2275 */
2276
2277
2278 static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node)
2279 {
2280 struct hlist_node *iter;
2281 struct dlm_lock_resource *res;
2282 int i;
2283 struct hlist_head *bucket;
2284 struct dlm_lock *lock;
2285
2286
2287 /* purge any stale mles */
2288 dlm_clean_master_list(dlm, dead_node);
2289
2290 /*
2291 * now clean up all lock resources. there are two rules:
2292 *
2293 * 1) if the dead node was the master, move the lockres
2294 * to the recovering list. set the RECOVERING flag.
2295 * this lockres needs to be cleaned up before it can
2296 * be used further.
2297 *
2298 * 2) if this node was the master, remove all locks from
2299 * each of the lockres queues that were owned by the
2300 * dead node. once recovery finishes, the dlm thread
2301 * can be kicked again to see if any ASTs or BASTs
2302 * need to be fired as a result.
2303 */
2304 for (i = 0; i < DLM_HASH_BUCKETS; i++) {
2305 bucket = dlm_lockres_hash(dlm, i);
2306 hlist_for_each_entry(res, iter, bucket, hash_node) {
2307 /* always prune any $RECOVERY entries for dead nodes,
2308 * otherwise hangs can occur during later recovery */
2309 if (dlm_is_recovery_lock(res->lockname.name,
2310 res->lockname.len)) {
2311 spin_lock(&res->spinlock);
2312 list_for_each_entry(lock, &res->granted, list) {
2313 if (lock->ml.node == dead_node) {
2314 mlog(0, "AHA! there was "
2315 "a $RECOVERY lock for dead "
2316 "node %u (%s)!\n",
2317 dead_node, dlm->name);
2318 list_del_init(&lock->list);
2319 dlm_lock_put(lock);
2320 break;
2321 }
2322 }
2323 spin_unlock(&res->spinlock);
2324 continue;
2325 }
2326 spin_lock(&res->spinlock);
2327 /* zero the lvb if necessary */
2328 dlm_revalidate_lvb(dlm, res, dead_node);
2329 if (res->owner == dead_node) {
2330 if (res->state & DLM_LOCK_RES_DROPPING_REF)
2331 mlog(0, "%s:%.*s: owned by "
2332 "dead node %u, this node was "
2333 "dropping its ref when it died. "
2334 "continue, dropping the flag.\n",
2335 dlm->name, res->lockname.len,
2336 res->lockname.name, dead_node);
2337
2338 /* the wake_up for this will happen when the
2339 * RECOVERING flag is dropped later */
2340 res->state &= ~DLM_LOCK_RES_DROPPING_REF;
2341
2342 dlm_move_lockres_to_recovery_list(dlm, res);
2343 } else if (res->owner == dlm->node_num) {
2344 dlm_free_dead_locks(dlm, res, dead_node);
2345 __dlm_lockres_calc_usage(dlm, res);
2346 }
2347 spin_unlock(&res->spinlock);
2348 }
2349 }
2350
2351 }
2352
2353 static void __dlm_hb_node_down(struct dlm_ctxt *dlm, int idx)
2354 {
2355 assert_spin_locked(&dlm->spinlock);
2356
2357 if (dlm->reco.new_master == idx) {
2358 mlog(0, "%s: recovery master %d just died\n",
2359 dlm->name, idx);
2360 if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
2361 /* finalize1 was reached, so it is safe to clear
2362 * the new_master and dead_node. that recovery
2363 * is complete. */
2364 mlog(0, "%s: dead master %d had reached "
2365 "finalize1 state, clearing\n", dlm->name, idx);
2366 dlm->reco.state &= ~DLM_RECO_STATE_FINALIZE;
2367 __dlm_reset_recovery(dlm);
2368 }
2369 }
2370
2371 /* Clean up join state on node death. */
2372 if (dlm->joining_node == idx) {
2373 mlog(0, "Clearing join state for node %u\n", idx);
2374 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
2375 }
2376
2377 /* check to see if the node is already considered dead */
2378 if (!test_bit(idx, dlm->live_nodes_map)) {
2379 mlog(0, "for domain %s, node %d is already dead. "
2380 "another node likely did recovery already.\n",
2381 dlm->name, idx);
2382 return;
2383 }
2384
2385 /* check to see if we do not care about this node */
2386 if (!test_bit(idx, dlm->domain_map)) {
2387 /* This also catches the case that we get a node down
2388 * but haven't joined the domain yet. */
2389 mlog(0, "node %u already removed from domain!\n", idx);
2390 return;
2391 }
2392
2393 clear_bit(idx, dlm->live_nodes_map);
2394
2395 /* make sure local cleanup occurs before the heartbeat events */
2396 if (!test_bit(idx, dlm->recovery_map))
2397 dlm_do_local_recovery_cleanup(dlm, idx);
2398
2399 /* notify anything attached to the heartbeat events */
2400 dlm_hb_event_notify_attached(dlm, idx, 0);
2401
2402 mlog(0, "node %u being removed from domain map!\n", idx);
2403 clear_bit(idx, dlm->domain_map);
2404 /* wake up migration waiters if a node goes down.
2405 * perhaps later we can genericize this for other waiters. */
2406 wake_up(&dlm->migration_wq);
2407
2408 if (test_bit(idx, dlm->recovery_map))
2409 mlog(0, "domain %s, node %u already added "
2410 "to recovery map!\n", dlm->name, idx);
2411 else
2412 set_bit(idx, dlm->recovery_map);
2413 }
2414
2415 void dlm_hb_node_down_cb(struct o2nm_node *node, int idx, void *data)
2416 {
2417 struct dlm_ctxt *dlm = data;
2418
2419 if (!dlm_grab(dlm))
2420 return;
2421
2422 /*
2423 * This will notify any dlm users that a node in our domain
2424 * went away without notifying us first.
2425 */
2426 if (test_bit(idx, dlm->domain_map))
2427 dlm_fire_domain_eviction_callbacks(dlm, idx);
2428
2429 spin_lock(&dlm->spinlock);
2430 __dlm_hb_node_down(dlm, idx);
2431 spin_unlock(&dlm->spinlock);
2432
2433 dlm_put(dlm);
2434 }
2435
2436 void dlm_hb_node_up_cb(struct o2nm_node *node, int idx, void *data)
2437 {
2438 struct dlm_ctxt *dlm = data;
2439
2440 if (!dlm_grab(dlm))
2441 return;
2442
2443 spin_lock(&dlm->spinlock);
2444 set_bit(idx, dlm->live_nodes_map);
2445 /* do NOT notify mle attached to the heartbeat events.
2446 * new nodes are not interesting in mastery until joined. */
2447 spin_unlock(&dlm->spinlock);
2448
2449 dlm_put(dlm);
2450 }
2451
2452 static void dlm_reco_ast(void *astdata)
2453 {
2454 struct dlm_ctxt *dlm = astdata;
2455 mlog(0, "ast for recovery lock fired!, this=%u, dlm=%s\n",
2456 dlm->node_num, dlm->name);
2457 }
2458 static void dlm_reco_bast(void *astdata, int blocked_type)
2459 {
2460 struct dlm_ctxt *dlm = astdata;
2461 mlog(0, "bast for recovery lock fired!, this=%u, dlm=%s\n",
2462 dlm->node_num, dlm->name);
2463 }
2464 static void dlm_reco_unlock_ast(void *astdata, enum dlm_status st)
2465 {
2466 mlog(0, "unlockast for recovery lock fired!\n");
2467 }
2468
2469 /*
2470 * dlm_pick_recovery_master will continually attempt to use
2471 * dlmlock() on the special "$RECOVERY" lockres with the
2472 * LKM_NOQUEUE flag to get an EX. every thread that enters
2473 * this function on each node racing to become the recovery
2474 * master will not stop attempting this until either:
2475 * a) this node gets the EX (and becomes the recovery master),
2476 * or b) dlm->reco.new_master gets set to some nodenum
2477 * != O2NM_INVALID_NODE_NUM (another node will do the reco).
2478 * so each time a recovery master is needed, the entire cluster
2479 * will sync at this point. if the new master dies, that will
2480 * be detected in dlm_do_recovery */
2481 static int dlm_pick_recovery_master(struct dlm_ctxt *dlm)
2482 {
2483 enum dlm_status ret;
2484 struct dlm_lockstatus lksb;
2485 int status = -EINVAL;
2486
2487 mlog(0, "starting recovery of %s at %lu, dead=%u, this=%u\n",
2488 dlm->name, jiffies, dlm->reco.dead_node, dlm->node_num);
2489 again:
2490 memset(&lksb, 0, sizeof(lksb));
2491
2492 ret = dlmlock(dlm, LKM_EXMODE, &lksb, LKM_NOQUEUE|LKM_RECOVERY,
2493 DLM_RECOVERY_LOCK_NAME, DLM_RECOVERY_LOCK_NAME_LEN,
2494 dlm_reco_ast, dlm, dlm_reco_bast);
2495
2496 mlog(0, "%s: dlmlock($RECOVERY) returned %d, lksb=%d\n",
2497 dlm->name, ret, lksb.status);
2498
2499 if (ret == DLM_NORMAL) {
2500 mlog(0, "dlm=%s dlmlock says I got it (this=%u)\n",
2501 dlm->name, dlm->node_num);
2502
2503 /* got the EX lock. check to see if another node
2504 * just became the reco master */
2505 if (dlm_reco_master_ready(dlm)) {
2506 mlog(0, "%s: got reco EX lock, but %u will "
2507 "do the recovery\n", dlm->name,
2508 dlm->reco.new_master);
2509 status = -EEXIST;
2510 } else {
2511 status = 0;
2512
2513 /* see if recovery was already finished elsewhere */
2514 spin_lock(&dlm->spinlock);
2515 if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
2516 status = -EINVAL;
2517 mlog(0, "%s: got reco EX lock, but "
2518 "node got recovered already\n", dlm->name);
2519 if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM) {
2520 mlog(ML_ERROR, "%s: new master is %u "
2521 "but no dead node!\n",
2522 dlm->name, dlm->reco.new_master);
2523 BUG();
2524 }
2525 }
2526 spin_unlock(&dlm->spinlock);
2527 }
2528
2529 /* if this node has actually become the recovery master,
2530 * set the master and send the messages to begin recovery */
2531 if (!status) {
2532 mlog(0, "%s: dead=%u, this=%u, sending "
2533 "begin_reco now\n", dlm->name,
2534 dlm->reco.dead_node, dlm->node_num);
2535 status = dlm_send_begin_reco_message(dlm,
2536 dlm->reco.dead_node);
2537 /* this always succeeds */
2538 BUG_ON(status);
2539
2540 /* set the new_master to this node */
2541 spin_lock(&dlm->spinlock);
2542 dlm_set_reco_master(dlm, dlm->node_num);
2543 spin_unlock(&dlm->spinlock);
2544 }
2545
2546 /* recovery lock is a special case. ast will not get fired,
2547 * so just go ahead and unlock it. */
2548 ret = dlmunlock(dlm, &lksb, 0, dlm_reco_unlock_ast, dlm);
2549 if (ret == DLM_DENIED) {
2550 mlog(0, "got DLM_DENIED, trying LKM_CANCEL\n");
2551 ret = dlmunlock(dlm, &lksb, LKM_CANCEL, dlm_reco_unlock_ast, dlm);
2552 }
2553 if (ret != DLM_NORMAL) {
2554 /* this would really suck. this could only happen
2555 * if there was a network error during the unlock
2556 * because of node death. this means the unlock
2557 * is actually "done" and the lock structure is
2558 * even freed. we can continue, but only
2559 * because this specific lock name is special. */
2560 mlog(ML_ERROR, "dlmunlock returned %d\n", ret);
2561 }
2562 } else if (ret == DLM_NOTQUEUED) {
2563 mlog(0, "dlm=%s dlmlock says another node got it (this=%u)\n",
2564 dlm->name, dlm->node_num);
2565 /* another node is master. wait on
2566 * reco.new_master != O2NM_INVALID_NODE_NUM
2567 * for at most one second */
2568 wait_event_timeout(dlm->dlm_reco_thread_wq,
2569 dlm_reco_master_ready(dlm),
2570 msecs_to_jiffies(1000));
2571 if (!dlm_reco_master_ready(dlm)) {
2572 mlog(0, "%s: reco master taking awhile\n",
2573 dlm->name);
2574 goto again;
2575 }
2576 /* another node has informed this one that it is reco master */
2577 mlog(0, "%s: reco master %u is ready to recover %u\n",
2578 dlm->name, dlm->reco.new_master, dlm->reco.dead_node);
2579 status = -EEXIST;
2580 } else if (ret == DLM_RECOVERING) {
2581 mlog(0, "dlm=%s dlmlock says master node died (this=%u)\n",
2582 dlm->name, dlm->node_num);
2583 goto again;
2584 } else {
2585 struct dlm_lock_resource *res;
2586
2587 /* dlmlock returned something other than NOTQUEUED or NORMAL */
2588 mlog(ML_ERROR, "%s: got %s from dlmlock($RECOVERY), "
2589 "lksb.status=%s\n", dlm->name, dlm_errname(ret),
2590 dlm_errname(lksb.status));
2591 res = dlm_lookup_lockres(dlm, DLM_RECOVERY_LOCK_NAME,
2592 DLM_RECOVERY_LOCK_NAME_LEN);
2593 if (res) {
2594 dlm_print_one_lock_resource(res);
2595 dlm_lockres_put(res);
2596 } else {
2597 mlog(ML_ERROR, "recovery lock not found\n");
2598 }
2599 BUG();
2600 }
2601
2602 return status;
2603 }
2604
2605 static int dlm_send_begin_reco_message(struct dlm_ctxt *dlm, u8 dead_node)
2606 {
2607 struct dlm_begin_reco br;
2608 int ret = 0;
2609 struct dlm_node_iter iter;
2610 int nodenum;
2611 int status;
2612
2613 mlog_entry("%u\n", dead_node);
2614
2615 mlog(0, "%s: dead node is %u\n", dlm->name, dead_node);
2616
2617 spin_lock(&dlm->spinlock);
2618 dlm_node_iter_init(dlm->domain_map, &iter);
2619 spin_unlock(&dlm->spinlock);
2620
2621 clear_bit(dead_node, iter.node_map);
2622
2623 memset(&br, 0, sizeof(br));
2624 br.node_idx = dlm->node_num;
2625 br.dead_node = dead_node;
2626
2627 while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
2628 ret = 0;
2629 if (nodenum == dead_node) {
2630 mlog(0, "not sending begin reco to dead node "
2631 "%u\n", dead_node);
2632 continue;
2633 }
2634 if (nodenum == dlm->node_num) {
2635 mlog(0, "not sending begin reco to self\n");
2636 continue;
2637 }
2638 retry:
2639 ret = -EINVAL;
2640 mlog(0, "attempting to send begin reco msg to %d\n",
2641 nodenum);
2642 ret = o2net_send_message(DLM_BEGIN_RECO_MSG, dlm->key,
2643 &br, sizeof(br), nodenum, &status);
2644 /* negative status is handled ok by caller here */
2645 if (ret >= 0)
2646 ret = status;
2647 if (dlm_is_host_down(ret)) {
2648 /* node is down. not involved in recovery
2649 * so just keep going */
2650 mlog(0, "%s: node %u was down when sending "
2651 "begin reco msg (%d)\n", dlm->name, nodenum, ret);
2652 ret = 0;
2653 }
2654 if (ret < 0) {
2655 struct dlm_lock_resource *res;
2656 /* this is now a serious problem, possibly ENOMEM
2657 * in the network stack. must retry */
2658 mlog_errno(ret);
2659 mlog(ML_ERROR, "begin reco of dlm %s to node %u "
2660 " returned %d\n", dlm->name, nodenum, ret);
2661 res = dlm_lookup_lockres(dlm, DLM_RECOVERY_LOCK_NAME,
2662 DLM_RECOVERY_LOCK_NAME_LEN);
2663 if (res) {
2664 dlm_print_one_lock_resource(res);
2665 dlm_lockres_put(res);
2666 } else {
2667 mlog(ML_ERROR, "recovery lock not found\n");
2668 }
2669 /* sleep for a bit in hopes that we can avoid
2670 * another ENOMEM */
2671 msleep(100);
2672 goto retry;
2673 } else if (ret == EAGAIN) {
2674 mlog(0, "%s: trying to start recovery of node "
2675 "%u, but node %u is waiting for last recovery "
2676 "to complete, backoff for a bit\n", dlm->name,
2677 dead_node, nodenum);
2678 /* TODO Look into replacing msleep with cond_resched() */
2679 msleep(100);
2680 goto retry;
2681 }
2682 }
2683
2684 return ret;
2685 }
2686
2687 int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data,
2688 void **ret_data)
2689 {
2690 struct dlm_ctxt *dlm = data;
2691 struct dlm_begin_reco *br = (struct dlm_begin_reco *)msg->buf;
2692
2693 /* ok to return 0, domain has gone away */
2694 if (!dlm_grab(dlm))
2695 return 0;
2696
2697 spin_lock(&dlm->spinlock);
2698 if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
2699 mlog(0, "%s: node %u wants to recover node %u (%u:%u) "
2700 "but this node is in finalize state, waiting on finalize2\n",
2701 dlm->name, br->node_idx, br->dead_node,
2702 dlm->reco.dead_node, dlm->reco.new_master);
2703 spin_unlock(&dlm->spinlock);
2704 return EAGAIN;
2705 }
2706 spin_unlock(&dlm->spinlock);
2707
2708 mlog(0, "%s: node %u wants to recover node %u (%u:%u)\n",
2709 dlm->name, br->node_idx, br->dead_node,
2710 dlm->reco.dead_node, dlm->reco.new_master);
2711
2712 dlm_fire_domain_eviction_callbacks(dlm, br->dead_node);
2713
2714 spin_lock(&dlm->spinlock);
2715 if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM) {
2716 if (test_bit(dlm->reco.new_master, dlm->recovery_map)) {
2717 mlog(0, "%s: new_master %u died, changing "
2718 "to %u\n", dlm->name, dlm->reco.new_master,
2719 br->node_idx);
2720 } else {
2721 mlog(0, "%s: new_master %u NOT DEAD, changing "
2722 "to %u\n", dlm->name, dlm->reco.new_master,
2723 br->node_idx);
2724 /* may not have seen the new master as dead yet */
2725 }
2726 }
2727 if (dlm->reco.dead_node != O2NM_INVALID_NODE_NUM) {
2728 mlog(ML_NOTICE, "%s: dead_node previously set to %u, "
2729 "node %u changing it to %u\n", dlm->name,
2730 dlm->reco.dead_node, br->node_idx, br->dead_node);
2731 }
2732 dlm_set_reco_master(dlm, br->node_idx);
2733 dlm_set_reco_dead_node(dlm, br->dead_node);
2734 if (!test_bit(br->dead_node, dlm->recovery_map)) {
2735 mlog(0, "recovery master %u sees %u as dead, but this "
2736 "node has not yet. marking %u as dead\n",
2737 br->node_idx, br->dead_node, br->dead_node);
2738 if (!test_bit(br->dead_node, dlm->domain_map) ||
2739 !test_bit(br->dead_node, dlm->live_nodes_map))
2740 mlog(0, "%u not in domain/live_nodes map "
2741 "so setting it in reco map manually\n",
2742 br->dead_node);
2743 /* force the recovery cleanup in __dlm_hb_node_down
2744 * both of these will be cleared in a moment */
2745 set_bit(br->dead_node, dlm->domain_map);
2746 set_bit(br->dead_node, dlm->live_nodes_map);
2747 __dlm_hb_node_down(dlm, br->dead_node);
2748 }
2749 spin_unlock(&dlm->spinlock);
2750
2751 dlm_kick_recovery_thread(dlm);
2752
2753 mlog(0, "%s: recovery started by node %u, for %u (%u:%u)\n",
2754 dlm->name, br->node_idx, br->dead_node,
2755 dlm->reco.dead_node, dlm->reco.new_master);
2756
2757 dlm_put(dlm);
2758 return 0;
2759 }
2760
2761 #define DLM_FINALIZE_STAGE2 0x01
2762 static int dlm_send_finalize_reco_message(struct dlm_ctxt *dlm)
2763 {
2764 int ret = 0;
2765 struct dlm_finalize_reco fr;
2766 struct dlm_node_iter iter;
2767 int nodenum;
2768 int status;
2769 int stage = 1;
2770
2771 mlog(0, "finishing recovery for node %s:%u, "
2772 "stage %d\n", dlm->name, dlm->reco.dead_node, stage);
2773
2774 spin_lock(&dlm->spinlock);
2775 dlm_node_iter_init(dlm->domain_map, &iter);
2776 spin_unlock(&dlm->spinlock);
2777
2778 stage2:
2779 memset(&fr, 0, sizeof(fr));
2780 fr.node_idx = dlm->node_num;
2781 fr.dead_node = dlm->reco.dead_node;
2782 if (stage == 2)
2783 fr.flags |= DLM_FINALIZE_STAGE2;
2784
2785 while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
2786 if (nodenum == dlm->node_num)
2787 continue;
2788 ret = o2net_send_message(DLM_FINALIZE_RECO_MSG, dlm->key,
2789 &fr, sizeof(fr), nodenum, &status);
2790 if (ret >= 0)
2791 ret = status;
2792 if (ret < 0) {
2793 mlog_errno(ret);
2794 if (dlm_is_host_down(ret)) {
2795 /* this has no effect on this recovery
2796 * session, so set the status to zero to
2797 * finish out the last recovery */
2798 mlog(ML_ERROR, "node %u went down after this "
2799 "node finished recovery.\n", nodenum);
2800 ret = 0;
2801 continue;
2802 }
2803 break;
2804 }
2805 }
2806 if (stage == 1) {
2807 /* reset the node_iter back to the top and send finalize2 */
2808 iter.curnode = -1;
2809 stage = 2;
2810 goto stage2;
2811 }
2812
2813 return ret;
2814 }
2815
2816 int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data,
2817 void **ret_data)
2818 {
2819 struct dlm_ctxt *dlm = data;
2820 struct dlm_finalize_reco *fr = (struct dlm_finalize_reco *)msg->buf;
2821 int stage = 1;
2822
2823 /* ok to return 0, domain has gone away */
2824 if (!dlm_grab(dlm))
2825 return 0;
2826
2827 if (fr->flags & DLM_FINALIZE_STAGE2)
2828 stage = 2;
2829
2830 mlog(0, "%s: node %u finalizing recovery stage%d of "
2831 "node %u (%u:%u)\n", dlm->name, fr->node_idx, stage,
2832 fr->dead_node, dlm->reco.dead_node, dlm->reco.new_master);
2833
2834 spin_lock(&dlm->spinlock);
2835
2836 if (dlm->reco.new_master != fr->node_idx) {
2837 mlog(ML_ERROR, "node %u sent recovery finalize msg, but node "
2838 "%u is supposed to be the new master, dead=%u\n",
2839 fr->node_idx, dlm->reco.new_master, fr->dead_node);
2840 BUG();
2841 }
2842 if (dlm->reco.dead_node != fr->dead_node) {
2843 mlog(ML_ERROR, "node %u sent recovery finalize msg for dead "
2844 "node %u, but node %u is supposed to be dead\n",
2845 fr->node_idx, fr->dead_node, dlm->reco.dead_node);
2846 BUG();
2847 }
2848
2849 switch (stage) {
2850 case 1:
2851 dlm_finish_local_lockres_recovery(dlm, fr->dead_node, fr->node_idx);
2852 if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
2853 mlog(ML_ERROR, "%s: received finalize1 from "
2854 "new master %u for dead node %u, but "
2855 "this node has already received it!\n",
2856 dlm->name, fr->node_idx, fr->dead_node);
2857 dlm_print_reco_node_status(dlm);
2858 BUG();
2859 }
2860 dlm->reco.state |= DLM_RECO_STATE_FINALIZE;
2861 spin_unlock(&dlm->spinlock);
2862 break;
2863 case 2:
2864 if (!(dlm->reco.state & DLM_RECO_STATE_FINALIZE)) {
2865 mlog(ML_ERROR, "%s: received finalize2 from "
2866 "new master %u for dead node %u, but "
2867 "this node did not have finalize1!\n",
2868 dlm->name, fr->node_idx, fr->dead_node);
2869 dlm_print_reco_node_status(dlm);
2870 BUG();
2871 }
2872 dlm->reco.state &= ~DLM_RECO_STATE_FINALIZE;
2873 spin_unlock(&dlm->spinlock);
2874 dlm_reset_recovery(dlm);
2875 dlm_kick_recovery_thread(dlm);
2876 break;
2877 default:
2878 BUG();
2879 }
2880
2881 mlog(0, "%s: recovery done, reco master was %u, dead now %u, master now %u\n",
2882 dlm->name, fr->node_idx, dlm->reco.dead_node, dlm->reco.new_master);
2883
2884 dlm_put(dlm);
2885 return 0;
2886 }
WRT350N Kernel Patches