?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
WIP FPC-III support
[linux/fpc-iii.git] / fs / ocfs2 / dlm / dlmmaster.c
blobf105746063ed32a1cca08538889328f228043bd1
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* -*- mode: c; c-basic-offset: 8; -*-
3 * vim: noexpandtab sw=8 ts=8 sts=0:
4 *
5 * dlmmod.c
6 *
7 * standalone DLM module
8 *
9 * Copyright (C) 2004 Oracle. All rights reserved.
10 */
11
12
13 #include <linux/module.h>
14 #include <linux/fs.h>
15 #include <linux/types.h>
16 #include <linux/slab.h>
17 #include <linux/highmem.h>
18 #include <linux/init.h>
19 #include <linux/sysctl.h>
20 #include <linux/random.h>
21 #include <linux/blkdev.h>
22 #include <linux/socket.h>
23 #include <linux/inet.h>
24 #include <linux/spinlock.h>
25 #include <linux/delay.h>
26
27
28 #include "../cluster/heartbeat.h"
29 #include "../cluster/nodemanager.h"
30 #include "../cluster/tcp.h"
31
32 #include "dlmapi.h"
33 #include "dlmcommon.h"
34 #include "dlmdomain.h"
35 #include "dlmdebug.h"
36
37 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_MASTER)
38 #include "../cluster/masklog.h"
39
40 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
41 struct dlm_master_list_entry *mle,
42 struct o2nm_node *node,
43 int idx);
44 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
45 struct dlm_master_list_entry *mle,
46 struct o2nm_node *node,
47 int idx);
48
49 static void dlm_assert_master_worker(struct dlm_work_item *item, void *data);
50 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
51 struct dlm_lock_resource *res,
52 void *nodemap, u32 flags);
53 static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data);
54
55 static inline int dlm_mle_equal(struct dlm_ctxt *dlm,
56 struct dlm_master_list_entry *mle,
57 const char *name,
58 unsigned int namelen)
59 {
60 if (dlm != mle->dlm)
61 return 0;
62
63 if (namelen != mle->mnamelen ||
64 memcmp(name, mle->mname, namelen) != 0)
65 return 0;
66
67 return 1;
68 }
69
70 static struct kmem_cache *dlm_lockres_cache;
71 static struct kmem_cache *dlm_lockname_cache;
72 static struct kmem_cache *dlm_mle_cache;
73
74 static void dlm_mle_release(struct kref *kref);
75 static void dlm_init_mle(struct dlm_master_list_entry *mle,
76 enum dlm_mle_type type,
77 struct dlm_ctxt *dlm,
78 struct dlm_lock_resource *res,
79 const char *name,
80 unsigned int namelen);
81 static void dlm_put_mle(struct dlm_master_list_entry *mle);
82 static void __dlm_put_mle(struct dlm_master_list_entry *mle);
83 static int dlm_find_mle(struct dlm_ctxt *dlm,
84 struct dlm_master_list_entry **mle,
85 char *name, unsigned int namelen);
86
87 static int dlm_do_master_request(struct dlm_lock_resource *res,
88 struct dlm_master_list_entry *mle, int to);
89
90
91 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
92 struct dlm_lock_resource *res,
93 struct dlm_master_list_entry *mle,
94 int *blocked);
95 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
96 struct dlm_lock_resource *res,
97 struct dlm_master_list_entry *mle,
98 int blocked);
99 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
100 struct dlm_lock_resource *res,
101 struct dlm_master_list_entry *mle,
102 struct dlm_master_list_entry **oldmle,
103 const char *name, unsigned int namelen,
104 u8 new_master, u8 master);
105
106 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
107 struct dlm_lock_resource *res);
108 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
109 struct dlm_lock_resource *res);
110 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
111 struct dlm_lock_resource *res,
112 u8 target);
113 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
114 struct dlm_lock_resource *res);
115
116
117 int dlm_is_host_down(int errno)
118 {
119 switch (errno) {
120 case -EBADF:
121 case -ECONNREFUSED:
122 case -ENOTCONN:
123 case -ECONNRESET:
124 case -EPIPE:
125 case -EHOSTDOWN:
126 case -EHOSTUNREACH:
127 case -ETIMEDOUT:
128 case -ECONNABORTED:
129 case -ENETDOWN:
130 case -ENETUNREACH:
131 case -ENETRESET:
132 case -ESHUTDOWN:
133 case -ENOPROTOOPT:
134 case -EINVAL: /* if returned from our tcp code,
135 this means there is no socket */
136 return 1;
137 }
138 return 0;
139 }
140
141
142 /*
143 * MASTER LIST FUNCTIONS
144 */
145
146
147 /*
148 * regarding master list entries and heartbeat callbacks:
149 *
150 * in order to avoid sleeping and allocation that occurs in
151 * heartbeat, master list entries are simply attached to the
152 * dlm's established heartbeat callbacks. the mle is attached
153 * when it is created, and since the dlm->spinlock is held at
154 * that time, any heartbeat event will be properly discovered
155 * by the mle. the mle needs to be detached from the
156 * dlm->mle_hb_events list as soon as heartbeat events are no
157 * longer useful to the mle, and before the mle is freed.
158 *
159 * as a general rule, heartbeat events are no longer needed by
160 * the mle once an "answer" regarding the lock master has been
161 * received.
162 */
163 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt *dlm,
164 struct dlm_master_list_entry *mle)
165 {
166 assert_spin_locked(&dlm->spinlock);
167
168 list_add_tail(&mle->hb_events, &dlm->mle_hb_events);
169 }
170
171
172 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,
173 struct dlm_master_list_entry *mle)
174 {
175 if (!list_empty(&mle->hb_events))
176 list_del_init(&mle->hb_events);
177 }
178
179
180 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,
181 struct dlm_master_list_entry *mle)
182 {
183 spin_lock(&dlm->spinlock);
184 __dlm_mle_detach_hb_events(dlm, mle);
185 spin_unlock(&dlm->spinlock);
186 }
187
188 static void dlm_get_mle_inuse(struct dlm_master_list_entry *mle)
189 {
190 struct dlm_ctxt *dlm;
191 dlm = mle->dlm;
192
193 assert_spin_locked(&dlm->spinlock);
194 assert_spin_locked(&dlm->master_lock);
195 mle->inuse++;
196 kref_get(&mle->mle_refs);
197 }
198
199 static void dlm_put_mle_inuse(struct dlm_master_list_entry *mle)
200 {
201 struct dlm_ctxt *dlm;
202 dlm = mle->dlm;
203
204 spin_lock(&dlm->spinlock);
205 spin_lock(&dlm->master_lock);
206 mle->inuse--;
207 __dlm_put_mle(mle);
208 spin_unlock(&dlm->master_lock);
209 spin_unlock(&dlm->spinlock);
210
211 }
212
213 /* remove from list and free */
214 static void __dlm_put_mle(struct dlm_master_list_entry *mle)
215 {
216 struct dlm_ctxt *dlm;
217 dlm = mle->dlm;
218
219 assert_spin_locked(&dlm->spinlock);
220 assert_spin_locked(&dlm->master_lock);
221 if (!kref_read(&mle->mle_refs)) {
222 /* this may or may not crash, but who cares.
223 * it's a BUG. */
224 mlog(ML_ERROR, "bad mle: %p\n", mle);
225 dlm_print_one_mle(mle);
226 BUG();
227 } else
228 kref_put(&mle->mle_refs, dlm_mle_release);
229 }
230
231
232 /* must not have any spinlocks coming in */
233 static void dlm_put_mle(struct dlm_master_list_entry *mle)
234 {
235 struct dlm_ctxt *dlm;
236 dlm = mle->dlm;
237
238 spin_lock(&dlm->spinlock);
239 spin_lock(&dlm->master_lock);
240 __dlm_put_mle(mle);
241 spin_unlock(&dlm->master_lock);
242 spin_unlock(&dlm->spinlock);
243 }
244
245 static inline void dlm_get_mle(struct dlm_master_list_entry *mle)
246 {
247 kref_get(&mle->mle_refs);
248 }
249
250 static void dlm_init_mle(struct dlm_master_list_entry *mle,
251 enum dlm_mle_type type,
252 struct dlm_ctxt *dlm,
253 struct dlm_lock_resource *res,
254 const char *name,
255 unsigned int namelen)
256 {
257 assert_spin_locked(&dlm->spinlock);
258
259 mle->dlm = dlm;
260 mle->type = type;
261 INIT_HLIST_NODE(&mle->master_hash_node);
262 INIT_LIST_HEAD(&mle->hb_events);
263 memset(mle->maybe_map, 0, sizeof(mle->maybe_map));
264 spin_lock_init(&mle->spinlock);
265 init_waitqueue_head(&mle->wq);
266 atomic_set(&mle->woken, 0);
267 kref_init(&mle->mle_refs);
268 memset(mle->response_map, 0, sizeof(mle->response_map));
269 mle->master = O2NM_MAX_NODES;
270 mle->new_master = O2NM_MAX_NODES;
271 mle->inuse = 0;
272
273 BUG_ON(mle->type != DLM_MLE_BLOCK &&
274 mle->type != DLM_MLE_MASTER &&
275 mle->type != DLM_MLE_MIGRATION);
276
277 if (mle->type == DLM_MLE_MASTER) {
278 BUG_ON(!res);
279 mle->mleres = res;
280 memcpy(mle->mname, res->lockname.name, res->lockname.len);
281 mle->mnamelen = res->lockname.len;
282 mle->mnamehash = res->lockname.hash;
283 } else {
284 BUG_ON(!name);
285 mle->mleres = NULL;
286 memcpy(mle->mname, name, namelen);
287 mle->mnamelen = namelen;
288 mle->mnamehash = dlm_lockid_hash(name, namelen);
289 }
290
291 atomic_inc(&dlm->mle_tot_count[mle->type]);
292 atomic_inc(&dlm->mle_cur_count[mle->type]);
293
294 /* copy off the node_map and register hb callbacks on our copy */
295 memcpy(mle->node_map, dlm->domain_map, sizeof(mle->node_map));
296 memcpy(mle->vote_map, dlm->domain_map, sizeof(mle->vote_map));
297 clear_bit(dlm->node_num, mle->vote_map);
298 clear_bit(dlm->node_num, mle->node_map);
299
300 /* attach the mle to the domain node up/down events */
301 __dlm_mle_attach_hb_events(dlm, mle);
302 }
303
304 void __dlm_unlink_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)
305 {
306 assert_spin_locked(&dlm->spinlock);
307 assert_spin_locked(&dlm->master_lock);
308
309 if (!hlist_unhashed(&mle->master_hash_node))
310 hlist_del_init(&mle->master_hash_node);
311 }
312
313 void __dlm_insert_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)
314 {
315 struct hlist_head *bucket;
316
317 assert_spin_locked(&dlm->master_lock);
318
319 bucket = dlm_master_hash(dlm, mle->mnamehash);
320 hlist_add_head(&mle->master_hash_node, bucket);
321 }
322
323 /* returns 1 if found, 0 if not */
324 static int dlm_find_mle(struct dlm_ctxt *dlm,
325 struct dlm_master_list_entry **mle,
326 char *name, unsigned int namelen)
327 {
328 struct dlm_master_list_entry *tmpmle;
329 struct hlist_head *bucket;
330 unsigned int hash;
331
332 assert_spin_locked(&dlm->master_lock);
333
334 hash = dlm_lockid_hash(name, namelen);
335 bucket = dlm_master_hash(dlm, hash);
336 hlist_for_each_entry(tmpmle, bucket, master_hash_node) {
337 if (!dlm_mle_equal(dlm, tmpmle, name, namelen))
338 continue;
339 dlm_get_mle(tmpmle);
340 *mle = tmpmle;
341 return 1;
342 }
343 return 0;
344 }
345
346 void dlm_hb_event_notify_attached(struct dlm_ctxt *dlm, int idx, int node_up)
347 {
348 struct dlm_master_list_entry *mle;
349
350 assert_spin_locked(&dlm->spinlock);
351
352 list_for_each_entry(mle, &dlm->mle_hb_events, hb_events) {
353 if (node_up)
354 dlm_mle_node_up(dlm, mle, NULL, idx);
355 else
356 dlm_mle_node_down(dlm, mle, NULL, idx);
357 }
358 }
359
360 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
361 struct dlm_master_list_entry *mle,
362 struct o2nm_node *node, int idx)
363 {
364 spin_lock(&mle->spinlock);
365
366 if (!test_bit(idx, mle->node_map))
367 mlog(0, "node %u already removed from nodemap!\n", idx);
368 else
369 clear_bit(idx, mle->node_map);
370
371 spin_unlock(&mle->spinlock);
372 }
373
374 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
375 struct dlm_master_list_entry *mle,
376 struct o2nm_node *node, int idx)
377 {
378 spin_lock(&mle->spinlock);
379
380 if (test_bit(idx, mle->node_map))
381 mlog(0, "node %u already in node map!\n", idx);
382 else
383 set_bit(idx, mle->node_map);
384
385 spin_unlock(&mle->spinlock);
386 }
387
388
389 int dlm_init_mle_cache(void)
390 {
391 dlm_mle_cache = kmem_cache_create("o2dlm_mle",
392 sizeof(struct dlm_master_list_entry),
393 0, SLAB_HWCACHE_ALIGN,
394 NULL);
395 if (dlm_mle_cache == NULL)
396 return -ENOMEM;
397 return 0;
398 }
399
400 void dlm_destroy_mle_cache(void)
401 {
402 kmem_cache_destroy(dlm_mle_cache);
403 }
404
405 static void dlm_mle_release(struct kref *kref)
406 {
407 struct dlm_master_list_entry *mle;
408 struct dlm_ctxt *dlm;
409
410 mle = container_of(kref, struct dlm_master_list_entry, mle_refs);
411 dlm = mle->dlm;
412
413 assert_spin_locked(&dlm->spinlock);
414 assert_spin_locked(&dlm->master_lock);
415
416 mlog(0, "Releasing mle for %.*s, type %d\n", mle->mnamelen, mle->mname,
417 mle->type);
418
419 /* remove from list if not already */
420 __dlm_unlink_mle(dlm, mle);
421
422 /* detach the mle from the domain node up/down events */
423 __dlm_mle_detach_hb_events(dlm, mle);
424
425 atomic_dec(&dlm->mle_cur_count[mle->type]);
426
427 /* NOTE: kfree under spinlock here.
428 * if this is bad, we can move this to a freelist. */
429 kmem_cache_free(dlm_mle_cache, mle);
430 }
431
432
433 /*
434 * LOCK RESOURCE FUNCTIONS
435 */
436
437 int dlm_init_master_caches(void)
438 {
439 dlm_lockres_cache = kmem_cache_create("o2dlm_lockres",
440 sizeof(struct dlm_lock_resource),
441 0, SLAB_HWCACHE_ALIGN, NULL);
442 if (!dlm_lockres_cache)
443 goto bail;
444
445 dlm_lockname_cache = kmem_cache_create("o2dlm_lockname",
446 DLM_LOCKID_NAME_MAX, 0,
447 SLAB_HWCACHE_ALIGN, NULL);
448 if (!dlm_lockname_cache)
449 goto bail;
450
451 return 0;
452 bail:
453 dlm_destroy_master_caches();
454 return -ENOMEM;
455 }
456
457 void dlm_destroy_master_caches(void)
458 {
459 kmem_cache_destroy(dlm_lockname_cache);
460 dlm_lockname_cache = NULL;
461
462 kmem_cache_destroy(dlm_lockres_cache);
463 dlm_lockres_cache = NULL;
464 }
465
466 static void dlm_lockres_release(struct kref *kref)
467 {
468 struct dlm_lock_resource *res;
469 struct dlm_ctxt *dlm;
470
471 res = container_of(kref, struct dlm_lock_resource, refs);
472 dlm = res->dlm;
473
474 /* This should not happen -- all lockres' have a name
475 * associated with them at init time. */
476 BUG_ON(!res->lockname.name);
477
478 mlog(0, "destroying lockres %.*s\n", res->lockname.len,
479 res->lockname.name);
480
481 atomic_dec(&dlm->res_cur_count);
482
483 if (!hlist_unhashed(&res->hash_node) ||
484 !list_empty(&res->granted) ||
485 !list_empty(&res->converting) ||
486 !list_empty(&res->blocked) ||
487 !list_empty(&res->dirty) ||
488 !list_empty(&res->recovering) ||
489 !list_empty(&res->purge)) {
490 mlog(ML_ERROR,
491 "Going to BUG for resource %.*s."
492 " We're on a list! [%c%c%c%c%c%c%c]\n",
493 res->lockname.len, res->lockname.name,
494 !hlist_unhashed(&res->hash_node) ? 'H' : ' ',
495 !list_empty(&res->granted) ? 'G' : ' ',
496 !list_empty(&res->converting) ? 'C' : ' ',
497 !list_empty(&res->blocked) ? 'B' : ' ',
498 !list_empty(&res->dirty) ? 'D' : ' ',
499 !list_empty(&res->recovering) ? 'R' : ' ',
500 !list_empty(&res->purge) ? 'P' : ' ');
501
502 dlm_print_one_lock_resource(res);
503 }
504
505 /* By the time we're ready to blow this guy away, we shouldn't
506 * be on any lists. */
507 BUG_ON(!hlist_unhashed(&res->hash_node));
508 BUG_ON(!list_empty(&res->granted));
509 BUG_ON(!list_empty(&res->converting));
510 BUG_ON(!list_empty(&res->blocked));
511 BUG_ON(!list_empty(&res->dirty));
512 BUG_ON(!list_empty(&res->recovering));
513 BUG_ON(!list_empty(&res->purge));
514
515 kmem_cache_free(dlm_lockname_cache, (void *)res->lockname.name);
516
517 kmem_cache_free(dlm_lockres_cache, res);
518 }
519
520 void dlm_lockres_put(struct dlm_lock_resource *res)
521 {
522 kref_put(&res->refs, dlm_lockres_release);
523 }
524
525 static void dlm_init_lockres(struct dlm_ctxt *dlm,
526 struct dlm_lock_resource *res,
527 const char *name, unsigned int namelen)
528 {
529 char *qname;
530
531 /* If we memset here, we lose our reference to the kmalloc'd
532 * res->lockname.name, so be sure to init every field
533 * correctly! */
534
535 qname = (char *) res->lockname.name;
536 memcpy(qname, name, namelen);
537
538 res->lockname.len = namelen;
539 res->lockname.hash = dlm_lockid_hash(name, namelen);
540
541 init_waitqueue_head(&res->wq);
542 spin_lock_init(&res->spinlock);
543 INIT_HLIST_NODE(&res->hash_node);
544 INIT_LIST_HEAD(&res->granted);
545 INIT_LIST_HEAD(&res->converting);
546 INIT_LIST_HEAD(&res->blocked);
547 INIT_LIST_HEAD(&res->dirty);
548 INIT_LIST_HEAD(&res->recovering);
549 INIT_LIST_HEAD(&res->purge);
550 INIT_LIST_HEAD(&res->tracking);
551 atomic_set(&res->asts_reserved, 0);
552 res->migration_pending = 0;
553 res->inflight_locks = 0;
554 res->inflight_assert_workers = 0;
555
556 res->dlm = dlm;
557
558 kref_init(&res->refs);
559
560 atomic_inc(&dlm->res_tot_count);
561 atomic_inc(&dlm->res_cur_count);
562
563 /* just for consistency */
564 spin_lock(&res->spinlock);
565 dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);
566 spin_unlock(&res->spinlock);
567
568 res->state = DLM_LOCK_RES_IN_PROGRESS;
569
570 res->last_used = 0;
571
572 spin_lock(&dlm->track_lock);
573 list_add_tail(&res->tracking, &dlm->tracking_list);
574 spin_unlock(&dlm->track_lock);
575
576 memset(res->lvb, 0, DLM_LVB_LEN);
577 memset(res->refmap, 0, sizeof(res->refmap));
578 }
579
580 struct dlm_lock_resource *dlm_new_lockres(struct dlm_ctxt *dlm,
581 const char *name,
582 unsigned int namelen)
583 {
584 struct dlm_lock_resource *res = NULL;
585
586 res = kmem_cache_zalloc(dlm_lockres_cache, GFP_NOFS);
587 if (!res)
588 goto error;
589
590 res->lockname.name = kmem_cache_zalloc(dlm_lockname_cache, GFP_NOFS);
591 if (!res->lockname.name)
592 goto error;
593
594 dlm_init_lockres(dlm, res, name, namelen);
595 return res;
596
597 error:
598 if (res)
599 kmem_cache_free(dlm_lockres_cache, res);
600 return NULL;
601 }
602
603 void dlm_lockres_set_refmap_bit(struct dlm_ctxt *dlm,
604 struct dlm_lock_resource *res, int bit)
605 {
606 assert_spin_locked(&res->spinlock);
607
608 mlog(0, "res %.*s, set node %u, %ps()\n", res->lockname.len,
609 res->lockname.name, bit, __builtin_return_address(0));
610
611 set_bit(bit, res->refmap);
612 }
613
614 void dlm_lockres_clear_refmap_bit(struct dlm_ctxt *dlm,
615 struct dlm_lock_resource *res, int bit)
616 {
617 assert_spin_locked(&res->spinlock);
618
619 mlog(0, "res %.*s, clr node %u, %ps()\n", res->lockname.len,
620 res->lockname.name, bit, __builtin_return_address(0));
621
622 clear_bit(bit, res->refmap);
623 }
624
625 static void __dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,
626 struct dlm_lock_resource *res)
627 {
628 res->inflight_locks++;
629
630 mlog(0, "%s: res %.*s, inflight++: now %u, %ps()\n", dlm->name,
631 res->lockname.len, res->lockname.name, res->inflight_locks,
632 __builtin_return_address(0));
633 }
634
635 void dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,
636 struct dlm_lock_resource *res)
637 {
638 assert_spin_locked(&res->spinlock);
639 __dlm_lockres_grab_inflight_ref(dlm, res);
640 }
641
642 void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,
643 struct dlm_lock_resource *res)
644 {
645 assert_spin_locked(&res->spinlock);
646
647 BUG_ON(res->inflight_locks == 0);
648
649 res->inflight_locks--;
650
651 mlog(0, "%s: res %.*s, inflight--: now %u, %ps()\n", dlm->name,
652 res->lockname.len, res->lockname.name, res->inflight_locks,
653 __builtin_return_address(0));
654
655 wake_up(&res->wq);
656 }
657
658 void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,
659 struct dlm_lock_resource *res)
660 {
661 assert_spin_locked(&res->spinlock);
662 res->inflight_assert_workers++;
663 mlog(0, "%s:%.*s: inflight assert worker++: now %u\n",
664 dlm->name, res->lockname.len, res->lockname.name,
665 res->inflight_assert_workers);
666 }
667
668 static void __dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
669 struct dlm_lock_resource *res)
670 {
671 assert_spin_locked(&res->spinlock);
672 BUG_ON(res->inflight_assert_workers == 0);
673 res->inflight_assert_workers--;
674 mlog(0, "%s:%.*s: inflight assert worker--: now %u\n",
675 dlm->name, res->lockname.len, res->lockname.name,
676 res->inflight_assert_workers);
677 }
678
679 static void dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
680 struct dlm_lock_resource *res)
681 {
682 spin_lock(&res->spinlock);
683 __dlm_lockres_drop_inflight_worker(dlm, res);
684 spin_unlock(&res->spinlock);
685 }
686
687 /*
688 * lookup a lock resource by name.
689 * may already exist in the hashtable.
690 * lockid is null terminated
691 *
692 * if not, allocate enough for the lockres and for
693 * the temporary structure used in doing the mastering.
694 *
695 * also, do a lookup in the dlm->master_list to see
696 * if another node has begun mastering the same lock.
697 * if so, there should be a block entry in there
698 * for this name, and we should *not* attempt to master
699 * the lock here. need to wait around for that node
700 * to assert_master (or die).
701 *
702 */
703 struct dlm_lock_resource * dlm_get_lock_resource(struct dlm_ctxt *dlm,
704 const char *lockid,
705 int namelen,
706 int flags)
707 {
708 struct dlm_lock_resource *tmpres=NULL, *res=NULL;
709 struct dlm_master_list_entry *mle = NULL;
710 struct dlm_master_list_entry *alloc_mle = NULL;
711 int blocked = 0;
712 int ret, nodenum;
713 struct dlm_node_iter iter;
714 unsigned int hash;
715 int tries = 0;
716 int bit, wait_on_recovery = 0;
717
718 BUG_ON(!lockid);
719
720 hash = dlm_lockid_hash(lockid, namelen);
721
722 mlog(0, "get lockres %s (len %d)\n", lockid, namelen);
723
724 lookup:
725 spin_lock(&dlm->spinlock);
726 tmpres = __dlm_lookup_lockres_full(dlm, lockid, namelen, hash);
727 if (tmpres) {
728 spin_unlock(&dlm->spinlock);
729 spin_lock(&tmpres->spinlock);
730
731 /*
732 * Right after dlm spinlock was released, dlm_thread could have
733 * purged the lockres. Check if lockres got unhashed. If so
734 * start over.
735 */
736 if (hlist_unhashed(&tmpres->hash_node)) {
737 spin_unlock(&tmpres->spinlock);
738 dlm_lockres_put(tmpres);
739 tmpres = NULL;
740 goto lookup;
741 }
742
743 /* Wait on the thread that is mastering the resource */
744 if (tmpres->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
745 __dlm_wait_on_lockres(tmpres);
746 BUG_ON(tmpres->owner == DLM_LOCK_RES_OWNER_UNKNOWN);
747 spin_unlock(&tmpres->spinlock);
748 dlm_lockres_put(tmpres);
749 tmpres = NULL;
750 goto lookup;
751 }
752
753 /* Wait on the resource purge to complete before continuing */
754 if (tmpres->state & DLM_LOCK_RES_DROPPING_REF) {
755 BUG_ON(tmpres->owner == dlm->node_num);
756 __dlm_wait_on_lockres_flags(tmpres,
757 DLM_LOCK_RES_DROPPING_REF);
758 spin_unlock(&tmpres->spinlock);
759 dlm_lockres_put(tmpres);
760 tmpres = NULL;
761 goto lookup;
762 }
763
764 /* Grab inflight ref to pin the resource */
765 dlm_lockres_grab_inflight_ref(dlm, tmpres);
766
767 spin_unlock(&tmpres->spinlock);
768 if (res) {
769 spin_lock(&dlm->track_lock);
770 if (!list_empty(&res->tracking))
771 list_del_init(&res->tracking);
772 else
773 mlog(ML_ERROR, "Resource %.*s not "
774 "on the Tracking list\n",
775 res->lockname.len,
776 res->lockname.name);
777 spin_unlock(&dlm->track_lock);
778 dlm_lockres_put(res);
779 }
780 res = tmpres;
781 goto leave;
782 }
783
784 if (!res) {
785 spin_unlock(&dlm->spinlock);
786 mlog(0, "allocating a new resource\n");
787 /* nothing found and we need to allocate one. */
788 alloc_mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
789 if (!alloc_mle)
790 goto leave;
791 res = dlm_new_lockres(dlm, lockid, namelen);
792 if (!res)
793 goto leave;
794 goto lookup;
795 }
796
797 mlog(0, "no lockres found, allocated our own: %p\n", res);
798
799 if (flags & LKM_LOCAL) {
800 /* caller knows it's safe to assume it's not mastered elsewhere
801 * DONE! return right away */
802 spin_lock(&res->spinlock);
803 dlm_change_lockres_owner(dlm, res, dlm->node_num);
804 __dlm_insert_lockres(dlm, res);
805 dlm_lockres_grab_inflight_ref(dlm, res);
806 spin_unlock(&res->spinlock);
807 spin_unlock(&dlm->spinlock);
808 /* lockres still marked IN_PROGRESS */
809 goto wake_waiters;
810 }
811
812 /* check master list to see if another node has started mastering it */
813 spin_lock(&dlm->master_lock);
814
815 /* if we found a block, wait for lock to be mastered by another node */
816 blocked = dlm_find_mle(dlm, &mle, (char *)lockid, namelen);
817 if (blocked) {
818 int mig;
819 if (mle->type == DLM_MLE_MASTER) {
820 mlog(ML_ERROR, "master entry for nonexistent lock!\n");
821 BUG();
822 }
823 mig = (mle->type == DLM_MLE_MIGRATION);
824 /* if there is a migration in progress, let the migration
825 * finish before continuing. we can wait for the absence
826 * of the MIGRATION mle: either the migrate finished or
827 * one of the nodes died and the mle was cleaned up.
828 * if there is a BLOCK here, but it already has a master
829 * set, we are too late. the master does not have a ref
830 * for us in the refmap. detach the mle and drop it.
831 * either way, go back to the top and start over. */
832 if (mig || mle->master != O2NM_MAX_NODES) {
833 BUG_ON(mig && mle->master == dlm->node_num);
834 /* we arrived too late. the master does not
835 * have a ref for us. retry. */
836 mlog(0, "%s:%.*s: late on %s\n",
837 dlm->name, namelen, lockid,
838 mig ? "MIGRATION" : "BLOCK");
839 spin_unlock(&dlm->master_lock);
840 spin_unlock(&dlm->spinlock);
841
842 /* master is known, detach */
843 if (!mig)
844 dlm_mle_detach_hb_events(dlm, mle);
845 dlm_put_mle(mle);
846 mle = NULL;
847 /* this is lame, but we can't wait on either
848 * the mle or lockres waitqueue here */
849 if (mig)
850 msleep(100);
851 goto lookup;
852 }
853 } else {
854 /* go ahead and try to master lock on this node */
855 mle = alloc_mle;
856 /* make sure this does not get freed below */
857 alloc_mle = NULL;
858 dlm_init_mle(mle, DLM_MLE_MASTER, dlm, res, NULL, 0);
859 set_bit(dlm->node_num, mle->maybe_map);
860 __dlm_insert_mle(dlm, mle);
861
862 /* still holding the dlm spinlock, check the recovery map
863 * to see if there are any nodes that still need to be
864 * considered. these will not appear in the mle nodemap
865 * but they might own this lockres. wait on them. */
866 bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);
867 if (bit < O2NM_MAX_NODES) {
868 mlog(0, "%s: res %.*s, At least one node (%d) "
869 "to recover before lock mastery can begin\n",
870 dlm->name, namelen, (char *)lockid, bit);
871 wait_on_recovery = 1;
872 }
873 }
874
875 /* at this point there is either a DLM_MLE_BLOCK or a
876 * DLM_MLE_MASTER on the master list, so it's safe to add the
877 * lockres to the hashtable. anyone who finds the lock will
878 * still have to wait on the IN_PROGRESS. */
879
880 /* finally add the lockres to its hash bucket */
881 __dlm_insert_lockres(dlm, res);
882
883 /* since this lockres is new it doesn't not require the spinlock */
884 __dlm_lockres_grab_inflight_ref(dlm, res);
885
886 /* get an extra ref on the mle in case this is a BLOCK
887 * if so, the creator of the BLOCK may try to put the last
888 * ref at this time in the assert master handler, so we
889 * need an extra one to keep from a bad ptr deref. */
890 dlm_get_mle_inuse(mle);
891 spin_unlock(&dlm->master_lock);
892 spin_unlock(&dlm->spinlock);
893
894 redo_request:
895 while (wait_on_recovery) {
896 /* any cluster changes that occurred after dropping the
897 * dlm spinlock would be detectable be a change on the mle,
898 * so we only need to clear out the recovery map once. */
899 if (dlm_is_recovery_lock(lockid, namelen)) {
900 mlog(0, "%s: Recovery map is not empty, but must "
901 "master $RECOVERY lock now\n", dlm->name);
902 if (!dlm_pre_master_reco_lockres(dlm, res))
903 wait_on_recovery = 0;
904 else {
905 mlog(0, "%s: waiting 500ms for heartbeat state "
906 "change\n", dlm->name);
907 msleep(500);
908 }
909 continue;
910 }
911
912 dlm_kick_recovery_thread(dlm);
913 msleep(1000);
914 dlm_wait_for_recovery(dlm);
915
916 spin_lock(&dlm->spinlock);
917 bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);
918 if (bit < O2NM_MAX_NODES) {
919 mlog(0, "%s: res %.*s, At least one node (%d) "
920 "to recover before lock mastery can begin\n",
921 dlm->name, namelen, (char *)lockid, bit);
922 wait_on_recovery = 1;
923 } else
924 wait_on_recovery = 0;
925 spin_unlock(&dlm->spinlock);
926
927 if (wait_on_recovery)
928 dlm_wait_for_node_recovery(dlm, bit, 10000);
929 }
930
931 /* must wait for lock to be mastered elsewhere */
932 if (blocked)
933 goto wait;
934
935 ret = -EINVAL;
936 dlm_node_iter_init(mle->vote_map, &iter);
937 while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
938 ret = dlm_do_master_request(res, mle, nodenum);
939 if (ret < 0)
940 mlog_errno(ret);
941 if (mle->master != O2NM_MAX_NODES) {
942 /* found a master ! */
943 if (mle->master <= nodenum)
944 break;
945 /* if our master request has not reached the master
946 * yet, keep going until it does. this is how the
947 * master will know that asserts are needed back to
948 * the lower nodes. */
949 mlog(0, "%s: res %.*s, Requests only up to %u but "
950 "master is %u, keep going\n", dlm->name, namelen,
951 lockid, nodenum, mle->master);
952 }
953 }
954
955 wait:
956 /* keep going until the response map includes all nodes */
957 ret = dlm_wait_for_lock_mastery(dlm, res, mle, &blocked);
958 if (ret < 0) {
959 wait_on_recovery = 1;
960 mlog(0, "%s: res %.*s, Node map changed, redo the master "
961 "request now, blocked=%d\n", dlm->name, res->lockname.len,
962 res->lockname.name, blocked);
963 if (++tries > 20) {
964 mlog(ML_ERROR, "%s: res %.*s, Spinning on "
965 "dlm_wait_for_lock_mastery, blocked = %d\n",
966 dlm->name, res->lockname.len,
967 res->lockname.name, blocked);
968 dlm_print_one_lock_resource(res);
969 dlm_print_one_mle(mle);
970 tries = 0;
971 }
972 goto redo_request;
973 }
974
975 mlog(0, "%s: res %.*s, Mastered by %u\n", dlm->name, res->lockname.len,
976 res->lockname.name, res->owner);
977 /* make sure we never continue without this */
978 BUG_ON(res->owner == O2NM_MAX_NODES);
979
980 /* master is known, detach if not already detached */
981 dlm_mle_detach_hb_events(dlm, mle);
982 dlm_put_mle(mle);
983 /* put the extra ref */
984 dlm_put_mle_inuse(mle);
985
986 wake_waiters:
987 spin_lock(&res->spinlock);
988 res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
989 spin_unlock(&res->spinlock);
990 wake_up(&res->wq);
991
992 leave:
993 /* need to free the unused mle */
994 if (alloc_mle)
995 kmem_cache_free(dlm_mle_cache, alloc_mle);
996
997 return res;
998 }
999
1000
1001 #define DLM_MASTERY_TIMEOUT_MS 5000
1002
1003 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
1004 struct dlm_lock_resource *res,
1005 struct dlm_master_list_entry *mle,
1006 int *blocked)
1007 {
1008 u8 m;
1009 int ret, bit;
1010 int map_changed, voting_done;
1011 int assert, sleep;
1012
1013 recheck:
1014 ret = 0;
1015 assert = 0;
1016
1017 /* check if another node has already become the owner */
1018 spin_lock(&res->spinlock);
1019 if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1020 mlog(0, "%s:%.*s: owner is suddenly %u\n", dlm->name,
1021 res->lockname.len, res->lockname.name, res->owner);
1022 spin_unlock(&res->spinlock);
1023 /* this will cause the master to re-assert across
1024 * the whole cluster, freeing up mles */
1025 if (res->owner != dlm->node_num) {
1026 ret = dlm_do_master_request(res, mle, res->owner);
1027 if (ret < 0) {
1028 /* give recovery a chance to run */
1029 mlog(ML_ERROR, "link to %u went down?: %d\n", res->owner, ret);
1030 msleep(500);
1031 goto recheck;
1032 }
1033 }
1034 ret = 0;
1035 goto leave;
1036 }
1037 spin_unlock(&res->spinlock);
1038
1039 spin_lock(&mle->spinlock);
1040 m = mle->master;
1041 map_changed = (memcmp(mle->vote_map, mle->node_map,
1042 sizeof(mle->vote_map)) != 0);
1043 voting_done = (memcmp(mle->vote_map, mle->response_map,
1044 sizeof(mle->vote_map)) == 0);
1045
1046 /* restart if we hit any errors */
1047 if (map_changed) {
1048 int b;
1049 mlog(0, "%s: %.*s: node map changed, restarting\n",
1050 dlm->name, res->lockname.len, res->lockname.name);
1051 ret = dlm_restart_lock_mastery(dlm, res, mle, *blocked);
1052 b = (mle->type == DLM_MLE_BLOCK);
1053 if ((*blocked && !b) || (!*blocked && b)) {
1054 mlog(0, "%s:%.*s: status change: old=%d new=%d\n",
1055 dlm->name, res->lockname.len, res->lockname.name,
1056 *blocked, b);
1057 *blocked = b;
1058 }
1059 spin_unlock(&mle->spinlock);
1060 if (ret < 0) {
1061 mlog_errno(ret);
1062 goto leave;
1063 }
1064 mlog(0, "%s:%.*s: restart lock mastery succeeded, "
1065 "rechecking now\n", dlm->name, res->lockname.len,
1066 res->lockname.name);
1067 goto recheck;
1068 } else {
1069 if (!voting_done) {
1070 mlog(0, "map not changed and voting not done "
1071 "for %s:%.*s\n", dlm->name, res->lockname.len,
1072 res->lockname.name);
1073 }
1074 }
1075
1076 if (m != O2NM_MAX_NODES) {
1077 /* another node has done an assert!
1078 * all done! */
1079 sleep = 0;
1080 } else {
1081 sleep = 1;
1082 /* have all nodes responded? */
1083 if (voting_done && !*blocked) {
1084 bit = find_next_bit(mle->maybe_map, O2NM_MAX_NODES, 0);
1085 if (dlm->node_num <= bit) {
1086 /* my node number is lowest.
1087 * now tell other nodes that I am
1088 * mastering this. */
1089 mle->master = dlm->node_num;
1090 /* ref was grabbed in get_lock_resource
1091 * will be dropped in dlmlock_master */
1092 assert = 1;
1093 sleep = 0;
1094 }
1095 /* if voting is done, but we have not received
1096 * an assert master yet, we must sleep */
1097 }
1098 }
1099
1100 spin_unlock(&mle->spinlock);
1101
1102 /* sleep if we haven't finished voting yet */
1103 if (sleep) {
1104 unsigned long timeo = msecs_to_jiffies(DLM_MASTERY_TIMEOUT_MS);
1105 atomic_set(&mle->woken, 0);
1106 (void)wait_event_timeout(mle->wq,
1107 (atomic_read(&mle->woken) == 1),
1108 timeo);
1109 if (res->owner == O2NM_MAX_NODES) {
1110 mlog(0, "%s:%.*s: waiting again\n", dlm->name,
1111 res->lockname.len, res->lockname.name);
1112 goto recheck;
1113 }
1114 mlog(0, "done waiting, master is %u\n", res->owner);
1115 ret = 0;
1116 goto leave;
1117 }
1118
1119 ret = 0; /* done */
1120 if (assert) {
1121 m = dlm->node_num;
1122 mlog(0, "about to master %.*s here, this=%u\n",
1123 res->lockname.len, res->lockname.name, m);
1124 ret = dlm_do_assert_master(dlm, res, mle->vote_map, 0);
1125 if (ret) {
1126 /* This is a failure in the network path,
1127 * not in the response to the assert_master
1128 * (any nonzero response is a BUG on this node).
1129 * Most likely a socket just got disconnected
1130 * due to node death. */
1131 mlog_errno(ret);
1132 }
1133 /* no longer need to restart lock mastery.
1134 * all living nodes have been contacted. */
1135 ret = 0;
1136 }
1137
1138 /* set the lockres owner */
1139 spin_lock(&res->spinlock);
1140 /* mastery reference obtained either during
1141 * assert_master_handler or in get_lock_resource */
1142 dlm_change_lockres_owner(dlm, res, m);
1143 spin_unlock(&res->spinlock);
1144
1145 leave:
1146 return ret;
1147 }
1148
1149 struct dlm_bitmap_diff_iter
1150 {
1151 int curnode;
1152 unsigned long *orig_bm;
1153 unsigned long *cur_bm;
1154 unsigned long diff_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
1155 };
1156
1157 enum dlm_node_state_change
1158 {
1159 NODE_DOWN = -1,
1160 NODE_NO_CHANGE = 0,
1161 NODE_UP
1162 };
1163
1164 static void dlm_bitmap_diff_iter_init(struct dlm_bitmap_diff_iter *iter,
1165 unsigned long *orig_bm,
1166 unsigned long *cur_bm)
1167 {
1168 unsigned long p1, p2;
1169 int i;
1170
1171 iter->curnode = -1;
1172 iter->orig_bm = orig_bm;
1173 iter->cur_bm = cur_bm;
1174
1175 for (i = 0; i < BITS_TO_LONGS(O2NM_MAX_NODES); i++) {
1176 p1 = *(iter->orig_bm + i);
1177 p2 = *(iter->cur_bm + i);
1178 iter->diff_bm[i] = (p1 & ~p2) | (p2 & ~p1);
1179 }
1180 }
1181
1182 static int dlm_bitmap_diff_iter_next(struct dlm_bitmap_diff_iter *iter,
1183 enum dlm_node_state_change *state)
1184 {
1185 int bit;
1186
1187 if (iter->curnode >= O2NM_MAX_NODES)
1188 return -ENOENT;
1189
1190 bit = find_next_bit(iter->diff_bm, O2NM_MAX_NODES,
1191 iter->curnode+1);
1192 if (bit >= O2NM_MAX_NODES) {
1193 iter->curnode = O2NM_MAX_NODES;
1194 return -ENOENT;
1195 }
1196
1197 /* if it was there in the original then this node died */
1198 if (test_bit(bit, iter->orig_bm))
1199 *state = NODE_DOWN;
1200 else
1201 *state = NODE_UP;
1202
1203 iter->curnode = bit;
1204 return bit;
1205 }
1206
1207
1208 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
1209 struct dlm_lock_resource *res,
1210 struct dlm_master_list_entry *mle,
1211 int blocked)
1212 {
1213 struct dlm_bitmap_diff_iter bdi;
1214 enum dlm_node_state_change sc;
1215 int node;
1216 int ret = 0;
1217
1218 mlog(0, "something happened such that the "
1219 "master process may need to be restarted!\n");
1220
1221 assert_spin_locked(&mle->spinlock);
1222
1223 dlm_bitmap_diff_iter_init(&bdi, mle->vote_map, mle->node_map);
1224 node = dlm_bitmap_diff_iter_next(&bdi, &sc);
1225 while (node >= 0) {
1226 if (sc == NODE_UP) {
1227 /* a node came up. clear any old vote from
1228 * the response map and set it in the vote map
1229 * then restart the mastery. */
1230 mlog(ML_NOTICE, "node %d up while restarting\n", node);
1231
1232 /* redo the master request, but only for the new node */
1233 mlog(0, "sending request to new node\n");
1234 clear_bit(node, mle->response_map);
1235 set_bit(node, mle->vote_map);
1236 } else {
1237 mlog(ML_ERROR, "node down! %d\n", node);
1238 if (blocked) {
1239 int lowest = find_next_bit(mle->maybe_map,
1240 O2NM_MAX_NODES, 0);
1241
1242 /* act like it was never there */
1243 clear_bit(node, mle->maybe_map);
1244
1245 if (node == lowest) {
1246 mlog(0, "expected master %u died"
1247 " while this node was blocked "
1248 "waiting on it!\n", node);
1249 lowest = find_next_bit(mle->maybe_map,
1250 O2NM_MAX_NODES,
1251 lowest+1);
1252 if (lowest < O2NM_MAX_NODES) {
1253 mlog(0, "%s:%.*s:still "
1254 "blocked. waiting on %u "
1255 "now\n", dlm->name,
1256 res->lockname.len,
1257 res->lockname.name,
1258 lowest);
1259 } else {
1260 /* mle is an MLE_BLOCK, but
1261 * there is now nothing left to
1262 * block on. we need to return
1263 * all the way back out and try
1264 * again with an MLE_MASTER.
1265 * dlm_do_local_recovery_cleanup
1266 * has already run, so the mle
1267 * refcount is ok */
1268 mlog(0, "%s:%.*s: no "
1269 "longer blocking. try to "
1270 "master this here\n",
1271 dlm->name,
1272 res->lockname.len,
1273 res->lockname.name);
1274 mle->type = DLM_MLE_MASTER;
1275 mle->mleres = res;
1276 }
1277 }
1278 }
1279
1280 /* now blank out everything, as if we had never
1281 * contacted anyone */
1282 memset(mle->maybe_map, 0, sizeof(mle->maybe_map));
1283 memset(mle->response_map, 0, sizeof(mle->response_map));
1284 /* reset the vote_map to the current node_map */
1285 memcpy(mle->vote_map, mle->node_map,
1286 sizeof(mle->node_map));
1287 /* put myself into the maybe map */
1288 if (mle->type != DLM_MLE_BLOCK)
1289 set_bit(dlm->node_num, mle->maybe_map);
1290 }
1291 ret = -EAGAIN;
1292 node = dlm_bitmap_diff_iter_next(&bdi, &sc);
1293 }
1294 return ret;
1295 }
1296
1297
1298 /*
1299 * DLM_MASTER_REQUEST_MSG
1300 *
1301 * returns: 0 on success,
1302 * -errno on a network error
1303 *
1304 * on error, the caller should assume the target node is "dead"
1305 *
1306 */
1307
1308 static int dlm_do_master_request(struct dlm_lock_resource *res,
1309 struct dlm_master_list_entry *mle, int to)
1310 {
1311 struct dlm_ctxt *dlm = mle->dlm;
1312 struct dlm_master_request request;
1313 int ret, response=0, resend;
1314
1315 memset(&request, 0, sizeof(request));
1316 request.node_idx = dlm->node_num;
1317
1318 BUG_ON(mle->type == DLM_MLE_MIGRATION);
1319
1320 request.namelen = (u8)mle->mnamelen;
1321 memcpy(request.name, mle->mname, request.namelen);
1322
1323 again:
1324 ret = o2net_send_message(DLM_MASTER_REQUEST_MSG, dlm->key, &request,
1325 sizeof(request), to, &response);
1326 if (ret < 0) {
1327 if (ret == -ESRCH) {
1328 /* should never happen */
1329 mlog(ML_ERROR, "TCP stack not ready!\n");
1330 BUG();
1331 } else if (ret == -EINVAL) {
1332 mlog(ML_ERROR, "bad args passed to o2net!\n");
1333 BUG();
1334 } else if (ret == -ENOMEM) {
1335 mlog(ML_ERROR, "out of memory while trying to send "
1336 "network message! retrying\n");
1337 /* this is totally crude */
1338 msleep(50);
1339 goto again;
1340 } else if (!dlm_is_host_down(ret)) {
1341 /* not a network error. bad. */
1342 mlog_errno(ret);
1343 mlog(ML_ERROR, "unhandled error!");
1344 BUG();
1345 }
1346 /* all other errors should be network errors,
1347 * and likely indicate node death */
1348 mlog(ML_ERROR, "link to %d went down!\n", to);
1349 goto out;
1350 }
1351
1352 ret = 0;
1353 resend = 0;
1354 spin_lock(&mle->spinlock);
1355 switch (response) {
1356 case DLM_MASTER_RESP_YES:
1357 set_bit(to, mle->response_map);
1358 mlog(0, "node %u is the master, response=YES\n", to);
1359 mlog(0, "%s:%.*s: master node %u now knows I have a "
1360 "reference\n", dlm->name, res->lockname.len,
1361 res->lockname.name, to);
1362 mle->master = to;
1363 break;
1364 case DLM_MASTER_RESP_NO:
1365 mlog(0, "node %u not master, response=NO\n", to);
1366 set_bit(to, mle->response_map);
1367 break;
1368 case DLM_MASTER_RESP_MAYBE:
1369 mlog(0, "node %u not master, response=MAYBE\n", to);
1370 set_bit(to, mle->response_map);
1371 set_bit(to, mle->maybe_map);
1372 break;
1373 case DLM_MASTER_RESP_ERROR:
1374 mlog(0, "node %u hit an error, resending\n", to);
1375 resend = 1;
1376 response = 0;
1377 break;
1378 default:
1379 mlog(ML_ERROR, "bad response! %u\n", response);
1380 BUG();
1381 }
1382 spin_unlock(&mle->spinlock);
1383 if (resend) {
1384 /* this is also totally crude */
1385 msleep(50);
1386 goto again;
1387 }
1388
1389 out:
1390 return ret;
1391 }
1392
1393 /*
1394 * locks that can be taken here:
1395 * dlm->spinlock
1396 * res->spinlock
1397 * mle->spinlock
1398 * dlm->master_list
1399 *
1400 * if possible, TRIM THIS DOWN!!!
1401 */
1402 int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data,
1403 void **ret_data)
1404 {
1405 u8 response = DLM_MASTER_RESP_MAYBE;
1406 struct dlm_ctxt *dlm = data;
1407 struct dlm_lock_resource *res = NULL;
1408 struct dlm_master_request *request = (struct dlm_master_request *) msg->buf;
1409 struct dlm_master_list_entry *mle = NULL, *tmpmle = NULL;
1410 char *name;
1411 unsigned int namelen, hash;
1412 int found, ret;
1413 int set_maybe;
1414 int dispatch_assert = 0;
1415 int dispatched = 0;
1416
1417 if (!dlm_grab(dlm))
1418 return DLM_MASTER_RESP_NO;
1419
1420 if (!dlm_domain_fully_joined(dlm)) {
1421 response = DLM_MASTER_RESP_NO;
1422 goto send_response;
1423 }
1424
1425 name = request->name;
1426 namelen = request->namelen;
1427 hash = dlm_lockid_hash(name, namelen);
1428
1429 if (namelen > DLM_LOCKID_NAME_MAX) {
1430 response = DLM_IVBUFLEN;
1431 goto send_response;
1432 }
1433
1434 way_up_top:
1435 spin_lock(&dlm->spinlock);
1436 res = __dlm_lookup_lockres(dlm, name, namelen, hash);
1437 if (res) {
1438 spin_unlock(&dlm->spinlock);
1439
1440 /* take care of the easy cases up front */
1441 spin_lock(&res->spinlock);
1442
1443 /*
1444 * Right after dlm spinlock was released, dlm_thread could have
1445 * purged the lockres. Check if lockres got unhashed. If so
1446 * start over.
1447 */
1448 if (hlist_unhashed(&res->hash_node)) {
1449 spin_unlock(&res->spinlock);
1450 dlm_lockres_put(res);
1451 goto way_up_top;
1452 }
1453
1454 if (res->state & (DLM_LOCK_RES_RECOVERING|
1455 DLM_LOCK_RES_MIGRATING)) {
1456 spin_unlock(&res->spinlock);
1457 mlog(0, "returning DLM_MASTER_RESP_ERROR since res is "
1458 "being recovered/migrated\n");
1459 response = DLM_MASTER_RESP_ERROR;
1460 if (mle)
1461 kmem_cache_free(dlm_mle_cache, mle);
1462 goto send_response;
1463 }
1464
1465 if (res->owner == dlm->node_num) {
1466 dlm_lockres_set_refmap_bit(dlm, res, request->node_idx);
1467 spin_unlock(&res->spinlock);
1468 response = DLM_MASTER_RESP_YES;
1469 if (mle)
1470 kmem_cache_free(dlm_mle_cache, mle);
1471
1472 /* this node is the owner.
1473 * there is some extra work that needs to
1474 * happen now. the requesting node has
1475 * caused all nodes up to this one to
1476 * create mles. this node now needs to
1477 * go back and clean those up. */
1478 dispatch_assert = 1;
1479 goto send_response;
1480 } else if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1481 spin_unlock(&res->spinlock);
1482 // mlog(0, "node %u is the master\n", res->owner);
1483 response = DLM_MASTER_RESP_NO;
1484 if (mle)
1485 kmem_cache_free(dlm_mle_cache, mle);
1486 goto send_response;
1487 }
1488
1489 /* ok, there is no owner. either this node is
1490 * being blocked, or it is actively trying to
1491 * master this lock. */
1492 if (!(res->state & DLM_LOCK_RES_IN_PROGRESS)) {
1493 mlog(ML_ERROR, "lock with no owner should be "
1494 "in-progress!\n");
1495 BUG();
1496 }
1497
1498 // mlog(0, "lockres is in progress...\n");
1499 spin_lock(&dlm->master_lock);
1500 found = dlm_find_mle(dlm, &tmpmle, name, namelen);
1501 if (!found) {
1502 mlog(ML_ERROR, "no mle found for this lock!\n");
1503 BUG();
1504 }
1505 set_maybe = 1;
1506 spin_lock(&tmpmle->spinlock);
1507 if (tmpmle->type == DLM_MLE_BLOCK) {
1508 // mlog(0, "this node is waiting for "
1509 // "lockres to be mastered\n");
1510 response = DLM_MASTER_RESP_NO;
1511 } else if (tmpmle->type == DLM_MLE_MIGRATION) {
1512 mlog(0, "node %u is master, but trying to migrate to "
1513 "node %u.\n", tmpmle->master, tmpmle->new_master);
1514 if (tmpmle->master == dlm->node_num) {
1515 mlog(ML_ERROR, "no owner on lockres, but this "
1516 "node is trying to migrate it to %u?!\n",
1517 tmpmle->new_master);
1518 BUG();
1519 } else {
1520 /* the real master can respond on its own */
1521 response = DLM_MASTER_RESP_NO;
1522 }
1523 } else if (tmpmle->master != DLM_LOCK_RES_OWNER_UNKNOWN) {
1524 set_maybe = 0;
1525 if (tmpmle->master == dlm->node_num) {
1526 response = DLM_MASTER_RESP_YES;
1527 /* this node will be the owner.
1528 * go back and clean the mles on any
1529 * other nodes */
1530 dispatch_assert = 1;
1531 dlm_lockres_set_refmap_bit(dlm, res,
1532 request->node_idx);
1533 } else
1534 response = DLM_MASTER_RESP_NO;
1535 } else {
1536 // mlog(0, "this node is attempting to "
1537 // "master lockres\n");
1538 response = DLM_MASTER_RESP_MAYBE;
1539 }
1540 if (set_maybe)
1541 set_bit(request->node_idx, tmpmle->maybe_map);
1542 spin_unlock(&tmpmle->spinlock);
1543
1544 spin_unlock(&dlm->master_lock);
1545 spin_unlock(&res->spinlock);
1546
1547 /* keep the mle attached to heartbeat events */
1548 dlm_put_mle(tmpmle);
1549 if (mle)
1550 kmem_cache_free(dlm_mle_cache, mle);
1551 goto send_response;
1552 }
1553
1554 /*
1555 * lockres doesn't exist on this node
1556 * if there is an MLE_BLOCK, return NO
1557 * if there is an MLE_MASTER, return MAYBE
1558 * otherwise, add an MLE_BLOCK, return NO
1559 */
1560 spin_lock(&dlm->master_lock);
1561 found = dlm_find_mle(dlm, &tmpmle, name, namelen);
1562 if (!found) {
1563 /* this lockid has never been seen on this node yet */
1564 // mlog(0, "no mle found\n");
1565 if (!mle) {
1566 spin_unlock(&dlm->master_lock);
1567 spin_unlock(&dlm->spinlock);
1568
1569 mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
1570 if (!mle) {
1571 response = DLM_MASTER_RESP_ERROR;
1572 mlog_errno(-ENOMEM);
1573 goto send_response;
1574 }
1575 goto way_up_top;
1576 }
1577
1578 // mlog(0, "this is second time thru, already allocated, "
1579 // "add the block.\n");
1580 dlm_init_mle(mle, DLM_MLE_BLOCK, dlm, NULL, name, namelen);
1581 set_bit(request->node_idx, mle->maybe_map);
1582 __dlm_insert_mle(dlm, mle);
1583 response = DLM_MASTER_RESP_NO;
1584 } else {
1585 spin_lock(&tmpmle->spinlock);
1586 if (tmpmle->master == dlm->node_num) {
1587 mlog(ML_ERROR, "no lockres, but an mle with this node as master!\n");
1588 BUG();
1589 }
1590 if (tmpmle->type == DLM_MLE_BLOCK)
1591 response = DLM_MASTER_RESP_NO;
1592 else if (tmpmle->type == DLM_MLE_MIGRATION) {
1593 mlog(0, "migration mle was found (%u->%u)\n",
1594 tmpmle->master, tmpmle->new_master);
1595 /* real master can respond on its own */
1596 response = DLM_MASTER_RESP_NO;
1597 } else
1598 response = DLM_MASTER_RESP_MAYBE;
1599 set_bit(request->node_idx, tmpmle->maybe_map);
1600 spin_unlock(&tmpmle->spinlock);
1601 }
1602 spin_unlock(&dlm->master_lock);
1603 spin_unlock(&dlm->spinlock);
1604
1605 if (found) {
1606 /* keep the mle attached to heartbeat events */
1607 dlm_put_mle(tmpmle);
1608 }
1609 send_response:
1610 /*
1611 * __dlm_lookup_lockres() grabbed a reference to this lockres.
1612 * The reference is released by dlm_assert_master_worker() under
1613 * the call to dlm_dispatch_assert_master(). If
1614 * dlm_assert_master_worker() isn't called, we drop it here.
1615 */
1616 if (dispatch_assert) {
1617 mlog(0, "%u is the owner of %.*s, cleaning everyone else\n",
1618 dlm->node_num, res->lockname.len, res->lockname.name);
1619 spin_lock(&res->spinlock);
1620 ret = dlm_dispatch_assert_master(dlm, res, 0, request->node_idx,
1621 DLM_ASSERT_MASTER_MLE_CLEANUP);
1622 if (ret < 0) {
1623 mlog(ML_ERROR, "failed to dispatch assert master work\n");
1624 response = DLM_MASTER_RESP_ERROR;
1625 spin_unlock(&res->spinlock);
1626 dlm_lockres_put(res);
1627 } else {
1628 dispatched = 1;
1629 __dlm_lockres_grab_inflight_worker(dlm, res);
1630 spin_unlock(&res->spinlock);
1631 }
1632 } else {
1633 if (res)
1634 dlm_lockres_put(res);
1635 }
1636
1637 if (!dispatched)
1638 dlm_put(dlm);
1639 return response;
1640 }
1641
1642 /*
1643 * DLM_ASSERT_MASTER_MSG
1644 */
1645
1646
1647 /*
1648 * NOTE: this can be used for debugging
1649 * can periodically run all locks owned by this node
1650 * and re-assert across the cluster...
1651 */
1652 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
1653 struct dlm_lock_resource *res,
1654 void *nodemap, u32 flags)
1655 {
1656 struct dlm_assert_master assert;
1657 int to, tmpret;
1658 struct dlm_node_iter iter;
1659 int ret = 0;
1660 int reassert;
1661 const char *lockname = res->lockname.name;
1662 unsigned int namelen = res->lockname.len;
1663
1664 BUG_ON(namelen > O2NM_MAX_NAME_LEN);
1665
1666 spin_lock(&res->spinlock);
1667 res->state |= DLM_LOCK_RES_SETREF_INPROG;
1668 spin_unlock(&res->spinlock);
1669
1670 again:
1671 reassert = 0;
1672
1673 /* note that if this nodemap is empty, it returns 0 */
1674 dlm_node_iter_init(nodemap, &iter);
1675 while ((to = dlm_node_iter_next(&iter)) >= 0) {
1676 int r = 0;
1677 struct dlm_master_list_entry *mle = NULL;
1678
1679 mlog(0, "sending assert master to %d (%.*s)\n", to,
1680 namelen, lockname);
1681 memset(&assert, 0, sizeof(assert));
1682 assert.node_idx = dlm->node_num;
1683 assert.namelen = namelen;
1684 memcpy(assert.name, lockname, namelen);
1685 assert.flags = cpu_to_be32(flags);
1686
1687 tmpret = o2net_send_message(DLM_ASSERT_MASTER_MSG, dlm->key,
1688 &assert, sizeof(assert), to, &r);
1689 if (tmpret < 0) {
1690 mlog(ML_ERROR, "Error %d when sending message %u (key "
1691 "0x%x) to node %u\n", tmpret,
1692 DLM_ASSERT_MASTER_MSG, dlm->key, to);
1693 if (!dlm_is_host_down(tmpret)) {
1694 mlog(ML_ERROR, "unhandled error=%d!\n", tmpret);
1695 BUG();
1696 }
1697 /* a node died. finish out the rest of the nodes. */
1698 mlog(0, "link to %d went down!\n", to);
1699 /* any nonzero status return will do */
1700 ret = tmpret;
1701 r = 0;
1702 } else if (r < 0) {
1703 /* ok, something horribly messed. kill thyself. */
1704 mlog(ML_ERROR,"during assert master of %.*s to %u, "
1705 "got %d.\n", namelen, lockname, to, r);
1706 spin_lock(&dlm->spinlock);
1707 spin_lock(&dlm->master_lock);
1708 if (dlm_find_mle(dlm, &mle, (char *)lockname,
1709 namelen)) {
1710 dlm_print_one_mle(mle);
1711 __dlm_put_mle(mle);
1712 }
1713 spin_unlock(&dlm->master_lock);
1714 spin_unlock(&dlm->spinlock);
1715 BUG();
1716 }
1717
1718 if (r & DLM_ASSERT_RESPONSE_REASSERT &&
1719 !(r & DLM_ASSERT_RESPONSE_MASTERY_REF)) {
1720 mlog(ML_ERROR, "%.*s: very strange, "
1721 "master MLE but no lockres on %u\n",
1722 namelen, lockname, to);
1723 }
1724
1725 if (r & DLM_ASSERT_RESPONSE_REASSERT) {
1726 mlog(0, "%.*s: node %u create mles on other "
1727 "nodes and requests a re-assert\n",
1728 namelen, lockname, to);
1729 reassert = 1;
1730 }
1731 if (r & DLM_ASSERT_RESPONSE_MASTERY_REF) {
1732 mlog(0, "%.*s: node %u has a reference to this "
1733 "lockres, set the bit in the refmap\n",
1734 namelen, lockname, to);
1735 spin_lock(&res->spinlock);
1736 dlm_lockres_set_refmap_bit(dlm, res, to);
1737 spin_unlock(&res->spinlock);
1738 }
1739 }
1740
1741 if (reassert)
1742 goto again;
1743
1744 spin_lock(&res->spinlock);
1745 res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
1746 spin_unlock(&res->spinlock);
1747 wake_up(&res->wq);
1748
1749 return ret;
1750 }
1751
1752 /*
1753 * locks that can be taken here:
1754 * dlm->spinlock
1755 * res->spinlock
1756 * mle->spinlock
1757 * dlm->master_list
1758 *
1759 * if possible, TRIM THIS DOWN!!!
1760 */
1761 int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data,
1762 void **ret_data)
1763 {
1764 struct dlm_ctxt *dlm = data;
1765 struct dlm_master_list_entry *mle = NULL;
1766 struct dlm_assert_master *assert = (struct dlm_assert_master *)msg->buf;
1767 struct dlm_lock_resource *res = NULL;
1768 char *name;
1769 unsigned int namelen, hash;
1770 u32 flags;
1771 int master_request = 0, have_lockres_ref = 0;
1772 int ret = 0;
1773
1774 if (!dlm_grab(dlm))
1775 return 0;
1776
1777 name = assert->name;
1778 namelen = assert->namelen;
1779 hash = dlm_lockid_hash(name, namelen);
1780 flags = be32_to_cpu(assert->flags);
1781
1782 if (namelen > DLM_LOCKID_NAME_MAX) {
1783 mlog(ML_ERROR, "Invalid name length!");
1784 goto done;
1785 }
1786
1787 spin_lock(&dlm->spinlock);
1788
1789 if (flags)
1790 mlog(0, "assert_master with flags: %u\n", flags);
1791
1792 /* find the MLE */
1793 spin_lock(&dlm->master_lock);
1794 if (!dlm_find_mle(dlm, &mle, name, namelen)) {
1795 /* not an error, could be master just re-asserting */
1796 mlog(0, "just got an assert_master from %u, but no "
1797 "MLE for it! (%.*s)\n", assert->node_idx,
1798 namelen, name);
1799 } else {
1800 int bit = find_next_bit (mle->maybe_map, O2NM_MAX_NODES, 0);
1801 if (bit >= O2NM_MAX_NODES) {
1802 /* not necessarily an error, though less likely.
1803 * could be master just re-asserting. */
1804 mlog(0, "no bits set in the maybe_map, but %u "
1805 "is asserting! (%.*s)\n", assert->node_idx,
1806 namelen, name);
1807 } else if (bit != assert->node_idx) {
1808 if (flags & DLM_ASSERT_MASTER_MLE_CLEANUP) {
1809 mlog(0, "master %u was found, %u should "
1810 "back off\n", assert->node_idx, bit);
1811 } else {
1812 /* with the fix for bug 569, a higher node
1813 * number winning the mastery will respond
1814 * YES to mastery requests, but this node
1815 * had no way of knowing. let it pass. */
1816 mlog(0, "%u is the lowest node, "
1817 "%u is asserting. (%.*s) %u must "
1818 "have begun after %u won.\n", bit,
1819 assert->node_idx, namelen, name, bit,
1820 assert->node_idx);
1821 }
1822 }
1823 if (mle->type == DLM_MLE_MIGRATION) {
1824 if (flags & DLM_ASSERT_MASTER_MLE_CLEANUP) {
1825 mlog(0, "%s:%.*s: got cleanup assert"
1826 " from %u for migration\n",
1827 dlm->name, namelen, name,
1828 assert->node_idx);
1829 } else if (!(flags & DLM_ASSERT_MASTER_FINISH_MIGRATION)) {
1830 mlog(0, "%s:%.*s: got unrelated assert"
1831 " from %u for migration, ignoring\n",
1832 dlm->name, namelen, name,
1833 assert->node_idx);
1834 __dlm_put_mle(mle);
1835 spin_unlock(&dlm->master_lock);
1836 spin_unlock(&dlm->spinlock);
1837 goto done;
1838 }
1839 }
1840 }
1841 spin_unlock(&dlm->master_lock);
1842
1843 /* ok everything checks out with the MLE
1844 * now check to see if there is a lockres */
1845 res = __dlm_lookup_lockres(dlm, name, namelen, hash);
1846 if (res) {
1847 spin_lock(&res->spinlock);
1848 if (res->state & DLM_LOCK_RES_RECOVERING) {
1849 mlog(ML_ERROR, "%u asserting but %.*s is "
1850 "RECOVERING!\n", assert->node_idx, namelen, name);
1851 goto kill;
1852 }
1853 if (!mle) {
1854 if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN &&
1855 res->owner != assert->node_idx) {
1856 mlog(ML_ERROR, "DIE! Mastery assert from %u, "
1857 "but current owner is %u! (%.*s)\n",
1858 assert->node_idx, res->owner, namelen,
1859 name);
1860 __dlm_print_one_lock_resource(res);
1861 BUG();
1862 }
1863 } else if (mle->type != DLM_MLE_MIGRATION) {
1864 if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1865 /* owner is just re-asserting */
1866 if (res->owner == assert->node_idx) {
1867 mlog(0, "owner %u re-asserting on "
1868 "lock %.*s\n", assert->node_idx,
1869 namelen, name);
1870 goto ok;
1871 }
1872 mlog(ML_ERROR, "got assert_master from "
1873 "node %u, but %u is the owner! "
1874 "(%.*s)\n", assert->node_idx,
1875 res->owner, namelen, name);
1876 goto kill;
1877 }
1878 if (!(res->state & DLM_LOCK_RES_IN_PROGRESS)) {
1879 mlog(ML_ERROR, "got assert from %u, but lock "
1880 "with no owner should be "
1881 "in-progress! (%.*s)\n",
1882 assert->node_idx,
1883 namelen, name);
1884 goto kill;
1885 }
1886 } else /* mle->type == DLM_MLE_MIGRATION */ {
1887 /* should only be getting an assert from new master */
1888 if (assert->node_idx != mle->new_master) {
1889 mlog(ML_ERROR, "got assert from %u, but "
1890 "new master is %u, and old master "
1891 "was %u (%.*s)\n",
1892 assert->node_idx, mle->new_master,
1893 mle->master, namelen, name);
1894 goto kill;
1895 }
1896
1897 }
1898 ok:
1899 spin_unlock(&res->spinlock);
1900 }
1901
1902 // mlog(0, "woo! got an assert_master from node %u!\n",
1903 // assert->node_idx);
1904 if (mle) {
1905 int extra_ref = 0;
1906 int nn = -1;
1907 int rr, err = 0;
1908
1909 spin_lock(&mle->spinlock);
1910 if (mle->type == DLM_MLE_BLOCK || mle->type == DLM_MLE_MIGRATION)
1911 extra_ref = 1;
1912 else {
1913 /* MASTER mle: if any bits set in the response map
1914 * then the calling node needs to re-assert to clear
1915 * up nodes that this node contacted */
1916 while ((nn = find_next_bit (mle->response_map, O2NM_MAX_NODES,
1917 nn+1)) < O2NM_MAX_NODES) {
1918 if (nn != dlm->node_num && nn != assert->node_idx) {
1919 master_request = 1;
1920 break;
1921 }
1922 }
1923 }
1924 mle->master = assert->node_idx;
1925 atomic_set(&mle->woken, 1);
1926 wake_up(&mle->wq);
1927 spin_unlock(&mle->spinlock);
1928
1929 if (res) {
1930 int wake = 0;
1931 spin_lock(&res->spinlock);
1932 if (mle->type == DLM_MLE_MIGRATION) {
1933 mlog(0, "finishing off migration of lockres %.*s, "
1934 "from %u to %u\n",
1935 res->lockname.len, res->lockname.name,
1936 dlm->node_num, mle->new_master);
1937 res->state &= ~DLM_LOCK_RES_MIGRATING;
1938 wake = 1;
1939 dlm_change_lockres_owner(dlm, res, mle->new_master);
1940 BUG_ON(res->state & DLM_LOCK_RES_DIRTY);
1941 } else {
1942 dlm_change_lockres_owner(dlm, res, mle->master);
1943 }
1944 spin_unlock(&res->spinlock);
1945 have_lockres_ref = 1;
1946 if (wake)
1947 wake_up(&res->wq);
1948 }
1949
1950 /* master is known, detach if not already detached.
1951 * ensures that only one assert_master call will happen
1952 * on this mle. */
1953 spin_lock(&dlm->master_lock);
1954
1955 rr = kref_read(&mle->mle_refs);
1956 if (mle->inuse > 0) {
1957 if (extra_ref && rr < 3)
1958 err = 1;
1959 else if (!extra_ref && rr < 2)
1960 err = 1;
1961 } else {
1962 if (extra_ref && rr < 2)
1963 err = 1;
1964 else if (!extra_ref && rr < 1)
1965 err = 1;
1966 }
1967 if (err) {
1968 mlog(ML_ERROR, "%s:%.*s: got assert master from %u "
1969 "that will mess up this node, refs=%d, extra=%d, "
1970 "inuse=%d\n", dlm->name, namelen, name,
1971 assert->node_idx, rr, extra_ref, mle->inuse);
1972 dlm_print_one_mle(mle);
1973 }
1974 __dlm_unlink_mle(dlm, mle);
1975 __dlm_mle_detach_hb_events(dlm, mle);
1976 __dlm_put_mle(mle);
1977 if (extra_ref) {
1978 /* the assert master message now balances the extra
1979 * ref given by the master / migration request message.
1980 * if this is the last put, it will be removed
1981 * from the list. */
1982 __dlm_put_mle(mle);
1983 }
1984 spin_unlock(&dlm->master_lock);
1985 } else if (res) {
1986 if (res->owner != assert->node_idx) {
1987 mlog(0, "assert_master from %u, but current "
1988 "owner is %u (%.*s), no mle\n", assert->node_idx,
1989 res->owner, namelen, name);
1990 }
1991 }
1992 spin_unlock(&dlm->spinlock);
1993
1994 done:
1995 ret = 0;
1996 if (res) {
1997 spin_lock(&res->spinlock);
1998 res->state |= DLM_LOCK_RES_SETREF_INPROG;
1999 spin_unlock(&res->spinlock);
2000 *ret_data = (void *)res;
2001 }
2002 dlm_put(dlm);
2003 if (master_request) {
2004 mlog(0, "need to tell master to reassert\n");
2005 /* positive. negative would shoot down the node. */
2006 ret |= DLM_ASSERT_RESPONSE_REASSERT;
2007 if (!have_lockres_ref) {
2008 mlog(ML_ERROR, "strange, got assert from %u, MASTER "
2009 "mle present here for %s:%.*s, but no lockres!\n",
2010 assert->node_idx, dlm->name, namelen, name);
2011 }
2012 }
2013 if (have_lockres_ref) {
2014 /* let the master know we have a reference to the lockres */
2015 ret |= DLM_ASSERT_RESPONSE_MASTERY_REF;
2016 mlog(0, "%s:%.*s: got assert from %u, need a ref\n",
2017 dlm->name, namelen, name, assert->node_idx);
2018 }
2019 return ret;
2020
2021 kill:
2022 /* kill the caller! */
2023 mlog(ML_ERROR, "Bad message received from another node. Dumping state "
2024 "and killing the other node now! This node is OK and can continue.\n");
2025 __dlm_print_one_lock_resource(res);
2026 spin_unlock(&res->spinlock);
2027 spin_lock(&dlm->master_lock);
2028 if (mle)
2029 __dlm_put_mle(mle);
2030 spin_unlock(&dlm->master_lock);
2031 spin_unlock(&dlm->spinlock);
2032 *ret_data = (void *)res;
2033 dlm_put(dlm);
2034 return -EINVAL;
2035 }
2036
2037 void dlm_assert_master_post_handler(int status, void *data, void *ret_data)
2038 {
2039 struct dlm_lock_resource *res = (struct dlm_lock_resource *)ret_data;
2040
2041 if (ret_data) {
2042 spin_lock(&res->spinlock);
2043 res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
2044 spin_unlock(&res->spinlock);
2045 wake_up(&res->wq);
2046 dlm_lockres_put(res);
2047 }
2048 return;
2049 }
2050
2051 int dlm_dispatch_assert_master(struct dlm_ctxt *dlm,
2052 struct dlm_lock_resource *res,
2053 int ignore_higher, u8 request_from, u32 flags)
2054 {
2055 struct dlm_work_item *item;
2056 item = kzalloc(sizeof(*item), GFP_ATOMIC);
2057 if (!item)
2058 return -ENOMEM;
2059
2060
2061 /* queue up work for dlm_assert_master_worker */
2062 dlm_init_work_item(dlm, item, dlm_assert_master_worker, NULL);
2063 item->u.am.lockres = res; /* already have a ref */
2064 /* can optionally ignore node numbers higher than this node */
2065 item->u.am.ignore_higher = ignore_higher;
2066 item->u.am.request_from = request_from;
2067 item->u.am.flags = flags;
2068
2069 if (ignore_higher)
2070 mlog(0, "IGNORE HIGHER: %.*s\n", res->lockname.len,
2071 res->lockname.name);
2072
2073 spin_lock(&dlm->work_lock);
2074 list_add_tail(&item->list, &dlm->work_list);
2075 spin_unlock(&dlm->work_lock);
2076
2077 queue_work(dlm->dlm_worker, &dlm->dispatched_work);
2078 return 0;
2079 }
2080
2081 static void dlm_assert_master_worker(struct dlm_work_item *item, void *data)
2082 {
2083 struct dlm_ctxt *dlm = data;
2084 int ret = 0;
2085 struct dlm_lock_resource *res;
2086 unsigned long nodemap[BITS_TO_LONGS(O2NM_MAX_NODES)];
2087 int ignore_higher;
2088 int bit;
2089 u8 request_from;
2090 u32 flags;
2091
2092 dlm = item->dlm;
2093 res = item->u.am.lockres;
2094 ignore_higher = item->u.am.ignore_higher;
2095 request_from = item->u.am.request_from;
2096 flags = item->u.am.flags;
2097
2098 spin_lock(&dlm->spinlock);
2099 memcpy(nodemap, dlm->domain_map, sizeof(nodemap));
2100 spin_unlock(&dlm->spinlock);
2101
2102 clear_bit(dlm->node_num, nodemap);
2103 if (ignore_higher) {
2104 /* if is this just to clear up mles for nodes below
2105 * this node, do not send the message to the original
2106 * caller or any node number higher than this */
2107 clear_bit(request_from, nodemap);
2108 bit = dlm->node_num;
2109 while (1) {
2110 bit = find_next_bit(nodemap, O2NM_MAX_NODES,
2111 bit+1);
2112 if (bit >= O2NM_MAX_NODES)
2113 break;
2114 clear_bit(bit, nodemap);
2115 }
2116 }
2117
2118 /*
2119 * If we're migrating this lock to someone else, we are no
2120 * longer allowed to assert out own mastery. OTOH, we need to
2121 * prevent migration from starting while we're still asserting
2122 * our dominance. The reserved ast delays migration.
2123 */
2124 spin_lock(&res->spinlock);
2125 if (res->state & DLM_LOCK_RES_MIGRATING) {
2126 mlog(0, "Someone asked us to assert mastery, but we're "
2127 "in the middle of migration. Skipping assert, "
2128 "the new master will handle that.\n");
2129 spin_unlock(&res->spinlock);
2130 goto put;
2131 } else
2132 __dlm_lockres_reserve_ast(res);
2133 spin_unlock(&res->spinlock);
2134
2135 /* this call now finishes out the nodemap
2136 * even if one or more nodes die */
2137 mlog(0, "worker about to master %.*s here, this=%u\n",
2138 res->lockname.len, res->lockname.name, dlm->node_num);
2139 ret = dlm_do_assert_master(dlm, res, nodemap, flags);
2140 if (ret < 0) {
2141 /* no need to restart, we are done */
2142 if (!dlm_is_host_down(ret))
2143 mlog_errno(ret);
2144 }
2145
2146 /* Ok, we've asserted ourselves. Let's let migration start. */
2147 dlm_lockres_release_ast(dlm, res);
2148
2149 put:
2150 dlm_lockres_drop_inflight_worker(dlm, res);
2151
2152 dlm_lockres_put(res);
2153
2154 mlog(0, "finished with dlm_assert_master_worker\n");
2155 }
2156
2157 /* SPECIAL CASE for the $RECOVERY lock used by the recovery thread.
2158 * We cannot wait for node recovery to complete to begin mastering this
2159 * lockres because this lockres is used to kick off recovery! ;-)
2160 * So, do a pre-check on all living nodes to see if any of those nodes
2161 * think that $RECOVERY is currently mastered by a dead node. If so,
2162 * we wait a short time to allow that node to get notified by its own
2163 * heartbeat stack, then check again. All $RECOVERY lock resources
2164 * mastered by dead nodes are purged when the heartbeat callback is
2165 * fired, so we can know for sure that it is safe to continue once
2166 * the node returns a live node or no node. */
2167 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
2168 struct dlm_lock_resource *res)
2169 {
2170 struct dlm_node_iter iter;
2171 int nodenum;
2172 int ret = 0;
2173 u8 master = DLM_LOCK_RES_OWNER_UNKNOWN;
2174
2175 spin_lock(&dlm->spinlock);
2176 dlm_node_iter_init(dlm->domain_map, &iter);
2177 spin_unlock(&dlm->spinlock);
2178
2179 while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
2180 /* do not send to self */
2181 if (nodenum == dlm->node_num)
2182 continue;
2183 ret = dlm_do_master_requery(dlm, res, nodenum, &master);
2184 if (ret < 0) {
2185 mlog_errno(ret);
2186 if (!dlm_is_host_down(ret))
2187 BUG();
2188 /* host is down, so answer for that node would be
2189 * DLM_LOCK_RES_OWNER_UNKNOWN. continue. */
2190 ret = 0;
2191 }
2192
2193 if (master != DLM_LOCK_RES_OWNER_UNKNOWN) {
2194 /* check to see if this master is in the recovery map */
2195 spin_lock(&dlm->spinlock);
2196 if (test_bit(master, dlm->recovery_map)) {
2197 mlog(ML_NOTICE, "%s: node %u has not seen "
2198 "node %u go down yet, and thinks the "
2199 "dead node is mastering the recovery "
2200 "lock. must wait.\n", dlm->name,
2201 nodenum, master);
2202 ret = -EAGAIN;
2203 }
2204 spin_unlock(&dlm->spinlock);
2205 mlog(0, "%s: reco lock master is %u\n", dlm->name,
2206 master);
2207 break;
2208 }
2209 }
2210 return ret;
2211 }
2212
2213 /*
2214 * DLM_DEREF_LOCKRES_MSG
2215 */
2216
2217 int dlm_drop_lockres_ref(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
2218 {
2219 struct dlm_deref_lockres deref;
2220 int ret = 0, r;
2221 const char *lockname;
2222 unsigned int namelen;
2223
2224 lockname = res->lockname.name;
2225 namelen = res->lockname.len;
2226 BUG_ON(namelen > O2NM_MAX_NAME_LEN);
2227
2228 memset(&deref, 0, sizeof(deref));
2229 deref.node_idx = dlm->node_num;
2230 deref.namelen = namelen;
2231 memcpy(deref.name, lockname, namelen);
2232
2233 ret = o2net_send_message(DLM_DEREF_LOCKRES_MSG, dlm->key,
2234 &deref, sizeof(deref), res->owner, &r);
2235 if (ret < 0)
2236 mlog(ML_ERROR, "%s: res %.*s, error %d send DEREF to node %u\n",
2237 dlm->name, namelen, lockname, ret, res->owner);
2238 else if (r < 0) {
2239 /* BAD. other node says I did not have a ref. */
2240 mlog(ML_ERROR, "%s: res %.*s, DEREF to node %u got %d\n",
2241 dlm->name, namelen, lockname, res->owner, r);
2242 dlm_print_one_lock_resource(res);
2243 if (r == -ENOMEM)
2244 BUG();
2245 } else
2246 ret = r;
2247
2248 return ret;
2249 }
2250
2251 int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
2252 void **ret_data)
2253 {
2254 struct dlm_ctxt *dlm = data;
2255 struct dlm_deref_lockres *deref = (struct dlm_deref_lockres *)msg->buf;
2256 struct dlm_lock_resource *res = NULL;
2257 char *name;
2258 unsigned int namelen;
2259 int ret = -EINVAL;
2260 u8 node;
2261 unsigned int hash;
2262 struct dlm_work_item *item;
2263 int cleared = 0;
2264 int dispatch = 0;
2265
2266 if (!dlm_grab(dlm))
2267 return 0;
2268
2269 name = deref->name;
2270 namelen = deref->namelen;
2271 node = deref->node_idx;
2272
2273 if (namelen > DLM_LOCKID_NAME_MAX) {
2274 mlog(ML_ERROR, "Invalid name length!");
2275 goto done;
2276 }
2277 if (deref->node_idx >= O2NM_MAX_NODES) {
2278 mlog(ML_ERROR, "Invalid node number: %u\n", node);
2279 goto done;
2280 }
2281
2282 hash = dlm_lockid_hash(name, namelen);
2283
2284 spin_lock(&dlm->spinlock);
2285 res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);
2286 if (!res) {
2287 spin_unlock(&dlm->spinlock);
2288 mlog(ML_ERROR, "%s:%.*s: bad lockres name\n",
2289 dlm->name, namelen, name);
2290 goto done;
2291 }
2292 spin_unlock(&dlm->spinlock);
2293
2294 spin_lock(&res->spinlock);
2295 if (res->state & DLM_LOCK_RES_SETREF_INPROG)
2296 dispatch = 1;
2297 else {
2298 BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2299 if (test_bit(node, res->refmap)) {
2300 dlm_lockres_clear_refmap_bit(dlm, res, node);
2301 cleared = 1;
2302 }
2303 }
2304 spin_unlock(&res->spinlock);
2305
2306 if (!dispatch) {
2307 if (cleared)
2308 dlm_lockres_calc_usage(dlm, res);
2309 else {
2310 mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2311 "but it is already dropped!\n", dlm->name,
2312 res->lockname.len, res->lockname.name, node);
2313 dlm_print_one_lock_resource(res);
2314 }
2315 ret = DLM_DEREF_RESPONSE_DONE;
2316 goto done;
2317 }
2318
2319 item = kzalloc(sizeof(*item), GFP_NOFS);
2320 if (!item) {
2321 ret = -ENOMEM;
2322 mlog_errno(ret);
2323 goto done;
2324 }
2325
2326 dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);
2327 item->u.dl.deref_res = res;
2328 item->u.dl.deref_node = node;
2329
2330 spin_lock(&dlm->work_lock);
2331 list_add_tail(&item->list, &dlm->work_list);
2332 spin_unlock(&dlm->work_lock);
2333
2334 queue_work(dlm->dlm_worker, &dlm->dispatched_work);
2335 return DLM_DEREF_RESPONSE_INPROG;
2336
2337 done:
2338 if (res)
2339 dlm_lockres_put(res);
2340 dlm_put(dlm);
2341
2342 return ret;
2343 }
2344
2345 int dlm_deref_lockres_done_handler(struct o2net_msg *msg, u32 len, void *data,
2346 void **ret_data)
2347 {
2348 struct dlm_ctxt *dlm = data;
2349 struct dlm_deref_lockres_done *deref
2350 = (struct dlm_deref_lockres_done *)msg->buf;
2351 struct dlm_lock_resource *res = NULL;
2352 char *name;
2353 unsigned int namelen;
2354 int ret = -EINVAL;
2355 u8 node;
2356 unsigned int hash;
2357
2358 if (!dlm_grab(dlm))
2359 return 0;
2360
2361 name = deref->name;
2362 namelen = deref->namelen;
2363 node = deref->node_idx;
2364
2365 if (namelen > DLM_LOCKID_NAME_MAX) {
2366 mlog(ML_ERROR, "Invalid name length!");
2367 goto done;
2368 }
2369 if (deref->node_idx >= O2NM_MAX_NODES) {
2370 mlog(ML_ERROR, "Invalid node number: %u\n", node);
2371 goto done;
2372 }
2373
2374 hash = dlm_lockid_hash(name, namelen);
2375
2376 spin_lock(&dlm->spinlock);
2377 res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);
2378 if (!res) {
2379 spin_unlock(&dlm->spinlock);
2380 mlog(ML_ERROR, "%s:%.*s: bad lockres name\n",
2381 dlm->name, namelen, name);
2382 goto done;
2383 }
2384
2385 spin_lock(&res->spinlock);
2386 if (!(res->state & DLM_LOCK_RES_DROPPING_REF)) {
2387 spin_unlock(&res->spinlock);
2388 spin_unlock(&dlm->spinlock);
2389 mlog(ML_NOTICE, "%s:%.*s: node %u sends deref done "
2390 "but it is already derefed!\n", dlm->name,
2391 res->lockname.len, res->lockname.name, node);
2392 ret = 0;
2393 goto done;
2394 }
2395
2396 __dlm_do_purge_lockres(dlm, res);
2397 spin_unlock(&res->spinlock);
2398 wake_up(&res->wq);
2399
2400 spin_unlock(&dlm->spinlock);
2401
2402 ret = 0;
2403 done:
2404 if (res)
2405 dlm_lockres_put(res);
2406 dlm_put(dlm);
2407 return ret;
2408 }
2409
2410 static void dlm_drop_lockres_ref_done(struct dlm_ctxt *dlm,
2411 struct dlm_lock_resource *res, u8 node)
2412 {
2413 struct dlm_deref_lockres_done deref;
2414 int ret = 0, r;
2415 const char *lockname;
2416 unsigned int namelen;
2417
2418 lockname = res->lockname.name;
2419 namelen = res->lockname.len;
2420 BUG_ON(namelen > O2NM_MAX_NAME_LEN);
2421
2422 memset(&deref, 0, sizeof(deref));
2423 deref.node_idx = dlm->node_num;
2424 deref.namelen = namelen;
2425 memcpy(deref.name, lockname, namelen);
2426
2427 ret = o2net_send_message(DLM_DEREF_LOCKRES_DONE, dlm->key,
2428 &deref, sizeof(deref), node, &r);
2429 if (ret < 0) {
2430 mlog(ML_ERROR, "%s: res %.*s, error %d send DEREF DONE "
2431 " to node %u\n", dlm->name, namelen,
2432 lockname, ret, node);
2433 } else if (r < 0) {
2434 /* ignore the error */
2435 mlog(ML_ERROR, "%s: res %.*s, DEREF to node %u got %d\n",
2436 dlm->name, namelen, lockname, node, r);
2437 dlm_print_one_lock_resource(res);
2438 }
2439 }
2440
2441 static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
2442 {
2443 struct dlm_ctxt *dlm;
2444 struct dlm_lock_resource *res;
2445 u8 node;
2446 u8 cleared = 0;
2447
2448 dlm = item->dlm;
2449 res = item->u.dl.deref_res;
2450 node = item->u.dl.deref_node;
2451
2452 spin_lock(&res->spinlock);
2453 BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2454 __dlm_wait_on_lockres_flags(res, DLM_LOCK_RES_SETREF_INPROG);
2455 if (test_bit(node, res->refmap)) {
2456 dlm_lockres_clear_refmap_bit(dlm, res, node);
2457 cleared = 1;
2458 }
2459 spin_unlock(&res->spinlock);
2460
2461 dlm_drop_lockres_ref_done(dlm, res, node);
2462
2463 if (cleared) {
2464 mlog(0, "%s:%.*s node %u ref dropped in dispatch\n",
2465 dlm->name, res->lockname.len, res->lockname.name, node);
2466 dlm_lockres_calc_usage(dlm, res);
2467 } else {
2468 mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2469 "but it is already dropped!\n", dlm->name,
2470 res->lockname.len, res->lockname.name, node);
2471 dlm_print_one_lock_resource(res);
2472 }
2473
2474 dlm_lockres_put(res);
2475 }
2476
2477 /*
2478 * A migratable resource is one that is :
2479 * 1. locally mastered, and,
2480 * 2. zero local locks, and,
2481 * 3. one or more non-local locks, or, one or more references
2482 * Returns 1 if yes, 0 if not.
2483 */
2484 static int dlm_is_lockres_migratable(struct dlm_ctxt *dlm,
2485 struct dlm_lock_resource *res)
2486 {
2487 enum dlm_lockres_list idx;
2488 int nonlocal = 0, node_ref;
2489 struct list_head *queue;
2490 struct dlm_lock *lock;
2491 u64 cookie;
2492
2493 assert_spin_locked(&res->spinlock);
2494
2495 /* delay migration when the lockres is in MIGRATING state */
2496 if (res->state & DLM_LOCK_RES_MIGRATING)
2497 return 0;
2498
2499 /* delay migration when the lockres is in RECOCERING state */
2500 if (res->state & (DLM_LOCK_RES_RECOVERING|
2501 DLM_LOCK_RES_RECOVERY_WAITING))
2502 return 0;
2503
2504 if (res->owner != dlm->node_num)
2505 return 0;
2506
2507 for (idx = DLM_GRANTED_LIST; idx <= DLM_BLOCKED_LIST; idx++) {
2508 queue = dlm_list_idx_to_ptr(res, idx);
2509 list_for_each_entry(lock, queue, list) {
2510 if (lock->ml.node != dlm->node_num) {
2511 nonlocal++;
2512 continue;
2513 }
2514 cookie = be64_to_cpu(lock->ml.cookie);
2515 mlog(0, "%s: Not migratable res %.*s, lock %u:%llu on "
2516 "%s list\n", dlm->name, res->lockname.len,
2517 res->lockname.name,
2518 dlm_get_lock_cookie_node(cookie),
2519 dlm_get_lock_cookie_seq(cookie),
2520 dlm_list_in_text(idx));
2521 return 0;
2522 }
2523 }
2524
2525 if (!nonlocal) {
2526 node_ref = find_next_bit(res->refmap, O2NM_MAX_NODES, 0);
2527 if (node_ref >= O2NM_MAX_NODES)
2528 return 0;
2529 }
2530
2531 mlog(0, "%s: res %.*s, Migratable\n", dlm->name, res->lockname.len,
2532 res->lockname.name);
2533
2534 return 1;
2535 }
2536
2537 /*
2538 * DLM_MIGRATE_LOCKRES
2539 */
2540
2541
2542 static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
2543 struct dlm_lock_resource *res, u8 target)
2544 {
2545 struct dlm_master_list_entry *mle = NULL;
2546 struct dlm_master_list_entry *oldmle = NULL;
2547 struct dlm_migratable_lockres *mres = NULL;
2548 int ret = 0;
2549 const char *name;
2550 unsigned int namelen;
2551 int mle_added = 0;
2552 int wake = 0;
2553
2554 if (!dlm_grab(dlm))
2555 return -EINVAL;
2556
2557 name = res->lockname.name;
2558 namelen = res->lockname.len;
2559
2560 mlog(0, "%s: Migrating %.*s to node %u\n", dlm->name, namelen, name,
2561 target);
2562
2563 /* preallocate up front. if this fails, abort */
2564 ret = -ENOMEM;
2565 mres = (struct dlm_migratable_lockres *) __get_free_page(GFP_NOFS);
2566 if (!mres) {
2567 mlog_errno(ret);
2568 goto leave;
2569 }
2570
2571 mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
2572 if (!mle) {
2573 mlog_errno(ret);
2574 goto leave;
2575 }
2576 ret = 0;
2577
2578 /*
2579 * clear any existing master requests and
2580 * add the migration mle to the list
2581 */
2582 spin_lock(&dlm->spinlock);
2583 spin_lock(&dlm->master_lock);
2584 ret = dlm_add_migration_mle(dlm, res, mle, &oldmle, name,
2585 namelen, target, dlm->node_num);
2586 /* get an extra reference on the mle.
2587 * otherwise the assert_master from the new
2588 * master will destroy this.
2589 */
2590 if (ret != -EEXIST)
2591 dlm_get_mle_inuse(mle);
2592
2593 spin_unlock(&dlm->master_lock);
2594 spin_unlock(&dlm->spinlock);
2595
2596 if (ret == -EEXIST) {
2597 mlog(0, "another process is already migrating it\n");
2598 goto fail;
2599 }
2600 mle_added = 1;
2601
2602 /*
2603 * set the MIGRATING flag and flush asts
2604 * if we fail after this we need to re-dirty the lockres
2605 */
2606 if (dlm_mark_lockres_migrating(dlm, res, target) < 0) {
2607 mlog(ML_ERROR, "tried to migrate %.*s to %u, but "
2608 "the target went down.\n", res->lockname.len,
2609 res->lockname.name, target);
2610 spin_lock(&res->spinlock);
2611 res->state &= ~DLM_LOCK_RES_MIGRATING;
2612 wake = 1;
2613 spin_unlock(&res->spinlock);
2614 ret = -EINVAL;
2615 }
2616
2617 fail:
2618 if (ret != -EEXIST && oldmle) {
2619 /* master is known, detach if not already detached */
2620 dlm_mle_detach_hb_events(dlm, oldmle);
2621 dlm_put_mle(oldmle);
2622 }
2623
2624 if (ret < 0) {
2625 if (mle_added) {
2626 dlm_mle_detach_hb_events(dlm, mle);
2627 dlm_put_mle(mle);
2628 dlm_put_mle_inuse(mle);
2629 } else if (mle) {
2630 kmem_cache_free(dlm_mle_cache, mle);
2631 mle = NULL;
2632 }
2633 goto leave;
2634 }
2635
2636 /*
2637 * at this point, we have a migration target, an mle
2638 * in the master list, and the MIGRATING flag set on
2639 * the lockres
2640 */
2641
2642 /* now that remote nodes are spinning on the MIGRATING flag,
2643 * ensure that all assert_master work is flushed. */
2644 flush_workqueue(dlm->dlm_worker);
2645
2646 /* notify new node and send all lock state */
2647 /* call send_one_lockres with migration flag.
2648 * this serves as notice to the target node that a
2649 * migration is starting. */
2650 ret = dlm_send_one_lockres(dlm, res, mres, target,
2651 DLM_MRES_MIGRATION);
2652
2653 if (ret < 0) {
2654 mlog(0, "migration to node %u failed with %d\n",
2655 target, ret);
2656 /* migration failed, detach and clean up mle */
2657 dlm_mle_detach_hb_events(dlm, mle);
2658 dlm_put_mle(mle);
2659 dlm_put_mle_inuse(mle);
2660 spin_lock(&res->spinlock);
2661 res->state &= ~DLM_LOCK_RES_MIGRATING;
2662 wake = 1;
2663 spin_unlock(&res->spinlock);
2664 if (dlm_is_host_down(ret))
2665 dlm_wait_for_node_death(dlm, target,
2666 DLM_NODE_DEATH_WAIT_MAX);
2667 goto leave;
2668 }
2669
2670 /* at this point, the target sends a message to all nodes,
2671 * (using dlm_do_migrate_request). this node is skipped since
2672 * we had to put an mle in the list to begin the process. this
2673 * node now waits for target to do an assert master. this node
2674 * will be the last one notified, ensuring that the migration
2675 * is complete everywhere. if the target dies while this is
2676 * going on, some nodes could potentially see the target as the
2677 * master, so it is important that my recovery finds the migration
2678 * mle and sets the master to UNKNOWN. */
2679
2680
2681 /* wait for new node to assert master */
2682 while (1) {
2683 ret = wait_event_interruptible_timeout(mle->wq,
2684 (atomic_read(&mle->woken) == 1),
2685 msecs_to_jiffies(5000));
2686
2687 if (ret >= 0) {
2688 if (atomic_read(&mle->woken) == 1 ||
2689 res->owner == target)
2690 break;
2691
2692 mlog(0, "%s:%.*s: timed out during migration\n",
2693 dlm->name, res->lockname.len, res->lockname.name);
2694 /* avoid hang during shutdown when migrating lockres
2695 * to a node which also goes down */
2696 if (dlm_is_node_dead(dlm, target)) {
2697 mlog(0, "%s:%.*s: expected migration "
2698 "target %u is no longer up, restarting\n",
2699 dlm->name, res->lockname.len,
2700 res->lockname.name, target);
2701 ret = -EINVAL;
2702 /* migration failed, detach and clean up mle */
2703 dlm_mle_detach_hb_events(dlm, mle);
2704 dlm_put_mle(mle);
2705 dlm_put_mle_inuse(mle);
2706 spin_lock(&res->spinlock);
2707 res->state &= ~DLM_LOCK_RES_MIGRATING;
2708 wake = 1;
2709 spin_unlock(&res->spinlock);
2710 goto leave;
2711 }
2712 } else
2713 mlog(0, "%s:%.*s: caught signal during migration\n",
2714 dlm->name, res->lockname.len, res->lockname.name);
2715 }
2716
2717 /* all done, set the owner, clear the flag */
2718 spin_lock(&res->spinlock);
2719 dlm_set_lockres_owner(dlm, res, target);
2720 res->state &= ~DLM_LOCK_RES_MIGRATING;
2721 dlm_remove_nonlocal_locks(dlm, res);
2722 spin_unlock(&res->spinlock);
2723 wake_up(&res->wq);
2724
2725 /* master is known, detach if not already detached */
2726 dlm_mle_detach_hb_events(dlm, mle);
2727 dlm_put_mle_inuse(mle);
2728 ret = 0;
2729
2730 dlm_lockres_calc_usage(dlm, res);
2731
2732 leave:
2733 /* re-dirty the lockres if we failed */
2734 if (ret < 0)
2735 dlm_kick_thread(dlm, res);
2736
2737 /* wake up waiters if the MIGRATING flag got set
2738 * but migration failed */
2739 if (wake)
2740 wake_up(&res->wq);
2741
2742 if (mres)
2743 free_page((unsigned long)mres);
2744
2745 dlm_put(dlm);
2746
2747 mlog(0, "%s: Migrating %.*s to %u, returns %d\n", dlm->name, namelen,
2748 name, target, ret);
2749 return ret;
2750 }
2751
2752 /*
2753 * Should be called only after beginning the domain leave process.
2754 * There should not be any remaining locks on nonlocal lock resources,
2755 * and there should be no local locks left on locally mastered resources.
2756 *
2757 * Called with the dlm spinlock held, may drop it to do migration, but
2758 * will re-acquire before exit.
2759 *
2760 * Returns: 1 if dlm->spinlock was dropped/retaken, 0 if never dropped
2761 */
2762 int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
2763 __must_hold(&dlm->spinlock)
2764 {
2765 int ret;
2766 int lock_dropped = 0;
2767 u8 target = O2NM_MAX_NODES;
2768
2769 assert_spin_locked(&dlm->spinlock);
2770
2771 spin_lock(&res->spinlock);
2772 if (dlm_is_lockres_migratable(dlm, res))
2773 target = dlm_pick_migration_target(dlm, res);
2774 spin_unlock(&res->spinlock);
2775
2776 if (target == O2NM_MAX_NODES)
2777 goto leave;
2778
2779 /* Wheee! Migrate lockres here! Will sleep so drop spinlock. */
2780 spin_unlock(&dlm->spinlock);
2781 lock_dropped = 1;
2782 ret = dlm_migrate_lockres(dlm, res, target);
2783 if (ret)
2784 mlog(0, "%s: res %.*s, Migrate to node %u failed with %d\n",
2785 dlm->name, res->lockname.len, res->lockname.name,
2786 target, ret);
2787 spin_lock(&dlm->spinlock);
2788 leave:
2789 return lock_dropped;
2790 }
2791
2792 int dlm_lock_basts_flushed(struct dlm_ctxt *dlm, struct dlm_lock *lock)
2793 {
2794 int ret;
2795 spin_lock(&dlm->ast_lock);
2796 spin_lock(&lock->spinlock);
2797 ret = (list_empty(&lock->bast_list) && !lock->bast_pending);
2798 spin_unlock(&lock->spinlock);
2799 spin_unlock(&dlm->ast_lock);
2800 return ret;
2801 }
2802
2803 static int dlm_migration_can_proceed(struct dlm_ctxt *dlm,
2804 struct dlm_lock_resource *res,
2805 u8 mig_target)
2806 {
2807 int can_proceed;
2808 spin_lock(&res->spinlock);
2809 can_proceed = !!(res->state & DLM_LOCK_RES_MIGRATING);
2810 spin_unlock(&res->spinlock);
2811
2812 /* target has died, so make the caller break out of the
2813 * wait_event, but caller must recheck the domain_map */
2814 spin_lock(&dlm->spinlock);
2815 if (!test_bit(mig_target, dlm->domain_map))
2816 can_proceed = 1;
2817 spin_unlock(&dlm->spinlock);
2818 return can_proceed;
2819 }
2820
2821 static int dlm_lockres_is_dirty(struct dlm_ctxt *dlm,
2822 struct dlm_lock_resource *res)
2823 {
2824 int ret;
2825 spin_lock(&res->spinlock);
2826 ret = !!(res->state & DLM_LOCK_RES_DIRTY);
2827 spin_unlock(&res->spinlock);
2828 return ret;
2829 }
2830
2831
2832 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
2833 struct dlm_lock_resource *res,
2834 u8 target)
2835 {
2836 int ret = 0;
2837
2838 mlog(0, "dlm_mark_lockres_migrating: %.*s, from %u to %u\n",
2839 res->lockname.len, res->lockname.name, dlm->node_num,
2840 target);
2841 /* need to set MIGRATING flag on lockres. this is done by
2842 * ensuring that all asts have been flushed for this lockres. */
2843 spin_lock(&res->spinlock);
2844 BUG_ON(res->migration_pending);
2845 res->migration_pending = 1;
2846 /* strategy is to reserve an extra ast then release
2847 * it below, letting the release do all of the work */
2848 __dlm_lockres_reserve_ast(res);
2849 spin_unlock(&res->spinlock);
2850
2851 /* now flush all the pending asts */
2852 dlm_kick_thread(dlm, res);
2853 /* before waiting on DIRTY, block processes which may
2854 * try to dirty the lockres before MIGRATING is set */
2855 spin_lock(&res->spinlock);
2856 BUG_ON(res->state & DLM_LOCK_RES_BLOCK_DIRTY);
2857 res->state |= DLM_LOCK_RES_BLOCK_DIRTY;
2858 spin_unlock(&res->spinlock);
2859 /* now wait on any pending asts and the DIRTY state */
2860 wait_event(dlm->ast_wq, !dlm_lockres_is_dirty(dlm, res));
2861 dlm_lockres_release_ast(dlm, res);
2862
2863 mlog(0, "about to wait on migration_wq, dirty=%s\n",
2864 res->state & DLM_LOCK_RES_DIRTY ? "yes" : "no");
2865 /* if the extra ref we just put was the final one, this
2866 * will pass thru immediately. otherwise, we need to wait
2867 * for the last ast to finish. */
2868 again:
2869 ret = wait_event_interruptible_timeout(dlm->migration_wq,
2870 dlm_migration_can_proceed(dlm, res, target),
2871 msecs_to_jiffies(1000));
2872 if (ret < 0) {
2873 mlog(0, "woken again: migrating? %s, dead? %s\n",
2874 res->state & DLM_LOCK_RES_MIGRATING ? "yes":"no",
2875 test_bit(target, dlm->domain_map) ? "no":"yes");
2876 } else {
2877 mlog(0, "all is well: migrating? %s, dead? %s\n",
2878 res->state & DLM_LOCK_RES_MIGRATING ? "yes":"no",
2879 test_bit(target, dlm->domain_map) ? "no":"yes");
2880 }
2881 if (!dlm_migration_can_proceed(dlm, res, target)) {
2882 mlog(0, "trying again...\n");
2883 goto again;
2884 }
2885
2886 ret = 0;
2887 /* did the target go down or die? */
2888 spin_lock(&dlm->spinlock);
2889 if (!test_bit(target, dlm->domain_map)) {
2890 mlog(ML_ERROR, "aha. migration target %u just went down\n",
2891 target);
2892 ret = -EHOSTDOWN;
2893 }
2894 spin_unlock(&dlm->spinlock);
2895
2896 /*
2897 * if target is down, we need to clear DLM_LOCK_RES_BLOCK_DIRTY for
2898 * another try; otherwise, we are sure the MIGRATING state is there,
2899 * drop the unneeded state which blocked threads trying to DIRTY
2900 */
2901 spin_lock(&res->spinlock);
2902 BUG_ON(!(res->state & DLM_LOCK_RES_BLOCK_DIRTY));
2903 res->state &= ~DLM_LOCK_RES_BLOCK_DIRTY;
2904 if (!ret)
2905 BUG_ON(!(res->state & DLM_LOCK_RES_MIGRATING));
2906 else
2907 res->migration_pending = 0;
2908 spin_unlock(&res->spinlock);
2909
2910 /*
2911 * at this point:
2912 *
2913 * o the DLM_LOCK_RES_MIGRATING flag is set if target not down
2914 * o there are no pending asts on this lockres
2915 * o all processes trying to reserve an ast on this
2916 * lockres must wait for the MIGRATING flag to clear
2917 */
2918 return ret;
2919 }
2920
2921 /* last step in the migration process.
2922 * original master calls this to free all of the dlm_lock
2923 * structures that used to be for other nodes. */
2924 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
2925 struct dlm_lock_resource *res)
2926 {
2927 struct list_head *queue = &res->granted;
2928 int i, bit;
2929 struct dlm_lock *lock, *next;
2930
2931 assert_spin_locked(&res->spinlock);
2932
2933 BUG_ON(res->owner == dlm->node_num);
2934
2935 for (i=0; i<3; i++) {
2936 list_for_each_entry_safe(lock, next, queue, list) {
2937 if (lock->ml.node != dlm->node_num) {
2938 mlog(0, "putting lock for node %u\n",
2939 lock->ml.node);
2940 /* be extra careful */
2941 BUG_ON(!list_empty(&lock->ast_list));
2942 BUG_ON(!list_empty(&lock->bast_list));
2943 BUG_ON(lock->ast_pending);
2944 BUG_ON(lock->bast_pending);
2945 dlm_lockres_clear_refmap_bit(dlm, res,
2946 lock->ml.node);
2947 list_del_init(&lock->list);
2948 dlm_lock_put(lock);
2949 /* In a normal unlock, we would have added a
2950 * DLM_UNLOCK_FREE_LOCK action. Force it. */
2951 dlm_lock_put(lock);
2952 }
2953 }
2954 queue++;
2955 }
2956 bit = 0;
2957 while (1) {
2958 bit = find_next_bit(res->refmap, O2NM_MAX_NODES, bit);
2959 if (bit >= O2NM_MAX_NODES)
2960 break;
2961 /* do not clear the local node reference, if there is a
2962 * process holding this, let it drop the ref itself */
2963 if (bit != dlm->node_num) {
2964 mlog(0, "%s:%.*s: node %u had a ref to this "
2965 "migrating lockres, clearing\n", dlm->name,
2966 res->lockname.len, res->lockname.name, bit);
2967 dlm_lockres_clear_refmap_bit(dlm, res, bit);
2968 }
2969 bit++;
2970 }
2971 }
2972
2973 /*
2974 * Pick a node to migrate the lock resource to. This function selects a
2975 * potential target based first on the locks and then on refmap. It skips
2976 * nodes that are in the process of exiting the domain.
2977 */
2978 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
2979 struct dlm_lock_resource *res)
2980 {
2981 enum dlm_lockres_list idx;
2982 struct list_head *queue = &res->granted;
2983 struct dlm_lock *lock;
2984 int noderef;
2985 u8 nodenum = O2NM_MAX_NODES;
2986
2987 assert_spin_locked(&dlm->spinlock);
2988 assert_spin_locked(&res->spinlock);
2989
2990 /* Go through all the locks */
2991 for (idx = DLM_GRANTED_LIST; idx <= DLM_BLOCKED_LIST; idx++) {
2992 queue = dlm_list_idx_to_ptr(res, idx);
2993 list_for_each_entry(lock, queue, list) {
2994 if (lock->ml.node == dlm->node_num)
2995 continue;
2996 if (test_bit(lock->ml.node, dlm->exit_domain_map))
2997 continue;
2998 nodenum = lock->ml.node;
2999 goto bail;
3000 }
3001 }
3002
3003 /* Go thru the refmap */
3004 noderef = -1;
3005 while (1) {
3006 noderef = find_next_bit(res->refmap, O2NM_MAX_NODES,
3007 noderef + 1);
3008 if (noderef >= O2NM_MAX_NODES)
3009 break;
3010 if (noderef == dlm->node_num)
3011 continue;
3012 if (test_bit(noderef, dlm->exit_domain_map))
3013 continue;
3014 nodenum = noderef;
3015 goto bail;
3016 }
3017
3018 bail:
3019 return nodenum;
3020 }
3021
3022 /* this is called by the new master once all lockres
3023 * data has been received */
3024 static int dlm_do_migrate_request(struct dlm_ctxt *dlm,
3025 struct dlm_lock_resource *res,
3026 u8 master, u8 new_master,
3027 struct dlm_node_iter *iter)
3028 {
3029 struct dlm_migrate_request migrate;
3030 int ret, skip, status = 0;
3031 int nodenum;
3032
3033 memset(&migrate, 0, sizeof(migrate));
3034 migrate.namelen = res->lockname.len;
3035 memcpy(migrate.name, res->lockname.name, migrate.namelen);
3036 migrate.new_master = new_master;
3037 migrate.master = master;
3038
3039 ret = 0;
3040
3041 /* send message to all nodes, except the master and myself */
3042 while ((nodenum = dlm_node_iter_next(iter)) >= 0) {
3043 if (nodenum == master ||
3044 nodenum == new_master)
3045 continue;
3046
3047 /* We could race exit domain. If exited, skip. */
3048 spin_lock(&dlm->spinlock);
3049 skip = (!test_bit(nodenum, dlm->domain_map));
3050 spin_unlock(&dlm->spinlock);
3051 if (skip) {
3052 clear_bit(nodenum, iter->node_map);
3053 continue;
3054 }
3055
3056 ret = o2net_send_message(DLM_MIGRATE_REQUEST_MSG, dlm->key,
3057 &migrate, sizeof(migrate), nodenum,
3058 &status);
3059 if (ret < 0) {
3060 mlog(ML_ERROR, "%s: res %.*s, Error %d send "
3061 "MIGRATE_REQUEST to node %u\n", dlm->name,
3062 migrate.namelen, migrate.name, ret, nodenum);
3063 if (!dlm_is_host_down(ret)) {
3064 mlog(ML_ERROR, "unhandled error=%d!\n", ret);
3065 BUG();
3066 }
3067 clear_bit(nodenum, iter->node_map);
3068 ret = 0;
3069 } else if (status < 0) {
3070 mlog(0, "migrate request (node %u) returned %d!\n",
3071 nodenum, status);
3072 ret = status;
3073 } else if (status == DLM_MIGRATE_RESPONSE_MASTERY_REF) {
3074 /* during the migration request we short-circuited
3075 * the mastery of the lockres. make sure we have
3076 * a mastery ref for nodenum */
3077 mlog(0, "%s:%.*s: need ref for node %u\n",
3078 dlm->name, res->lockname.len, res->lockname.name,
3079 nodenum);
3080 spin_lock(&res->spinlock);
3081 dlm_lockres_set_refmap_bit(dlm, res, nodenum);
3082 spin_unlock(&res->spinlock);
3083 }
3084 }
3085
3086 if (ret < 0)
3087 mlog_errno(ret);
3088
3089 mlog(0, "returning ret=%d\n", ret);
3090 return ret;
3091 }
3092
3093
3094 /* if there is an existing mle for this lockres, we now know who the master is.
3095 * (the one who sent us *this* message) we can clear it up right away.
3096 * since the process that put the mle on the list still has a reference to it,
3097 * we can unhash it now, set the master and wake the process. as a result,
3098 * we will have no mle in the list to start with. now we can add an mle for
3099 * the migration and this should be the only one found for those scanning the
3100 * list. */
3101 int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
3102 void **ret_data)
3103 {
3104 struct dlm_ctxt *dlm = data;
3105 struct dlm_lock_resource *res = NULL;
3106 struct dlm_migrate_request *migrate = (struct dlm_migrate_request *) msg->buf;
3107 struct dlm_master_list_entry *mle = NULL, *oldmle = NULL;
3108 const char *name;
3109 unsigned int namelen, hash;
3110 int ret = 0;
3111
3112 if (!dlm_grab(dlm))
3113 return 0;
3114
3115 name = migrate->name;
3116 namelen = migrate->namelen;
3117 hash = dlm_lockid_hash(name, namelen);
3118
3119 /* preallocate.. if this fails, abort */
3120 mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
3121
3122 if (!mle) {
3123 ret = -ENOMEM;
3124 goto leave;
3125 }
3126
3127 /* check for pre-existing lock */
3128 spin_lock(&dlm->spinlock);
3129 res = __dlm_lookup_lockres(dlm, name, namelen, hash);
3130 if (res) {
3131 spin_lock(&res->spinlock);
3132 if (res->state & DLM_LOCK_RES_RECOVERING) {
3133 /* if all is working ok, this can only mean that we got
3134 * a migrate request from a node that we now see as
3135 * dead. what can we do here? drop it to the floor? */
3136 spin_unlock(&res->spinlock);
3137 mlog(ML_ERROR, "Got a migrate request, but the "
3138 "lockres is marked as recovering!");
3139 kmem_cache_free(dlm_mle_cache, mle);
3140 ret = -EINVAL; /* need a better solution */
3141 goto unlock;
3142 }
3143 res->state |= DLM_LOCK_RES_MIGRATING;
3144 spin_unlock(&res->spinlock);
3145 }
3146
3147 spin_lock(&dlm->master_lock);
3148 /* ignore status. only nonzero status would BUG. */
3149 ret = dlm_add_migration_mle(dlm, res, mle, &oldmle,
3150 name, namelen,
3151 migrate->new_master,
3152 migrate->master);
3153
3154 if (ret < 0)
3155 kmem_cache_free(dlm_mle_cache, mle);
3156
3157 spin_unlock(&dlm->master_lock);
3158 unlock:
3159 spin_unlock(&dlm->spinlock);
3160
3161 if (oldmle) {
3162 /* master is known, detach if not already detached */
3163 dlm_mle_detach_hb_events(dlm, oldmle);
3164 dlm_put_mle(oldmle);
3165 }
3166
3167 if (res)
3168 dlm_lockres_put(res);
3169 leave:
3170 dlm_put(dlm);
3171 return ret;
3172 }
3173
3174 /* must be holding dlm->spinlock and dlm->master_lock
3175 * when adding a migration mle, we can clear any other mles
3176 * in the master list because we know with certainty that
3177 * the master is "master". so we remove any old mle from
3178 * the list after setting it's master field, and then add
3179 * the new migration mle. this way we can hold with the rule
3180 * of having only one mle for a given lock name at all times. */
3181 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
3182 struct dlm_lock_resource *res,
3183 struct dlm_master_list_entry *mle,
3184 struct dlm_master_list_entry **oldmle,
3185 const char *name, unsigned int namelen,
3186 u8 new_master, u8 master)
3187 {
3188 int found;
3189 int ret = 0;
3190
3191 *oldmle = NULL;
3192
3193 assert_spin_locked(&dlm->spinlock);
3194 assert_spin_locked(&dlm->master_lock);
3195
3196 /* caller is responsible for any ref taken here on oldmle */
3197 found = dlm_find_mle(dlm, oldmle, (char *)name, namelen);
3198 if (found) {
3199 struct dlm_master_list_entry *tmp = *oldmle;
3200 spin_lock(&tmp->spinlock);
3201 if (tmp->type == DLM_MLE_MIGRATION) {
3202 if (master == dlm->node_num) {
3203 /* ah another process raced me to it */
3204 mlog(0, "tried to migrate %.*s, but some "
3205 "process beat me to it\n",
3206 namelen, name);
3207 spin_unlock(&tmp->spinlock);
3208 return -EEXIST;
3209 } else {
3210 /* bad. 2 NODES are trying to migrate! */
3211 mlog(ML_ERROR, "migration error mle: "
3212 "master=%u new_master=%u // request: "
3213 "master=%u new_master=%u // "
3214 "lockres=%.*s\n",
3215 tmp->master, tmp->new_master,
3216 master, new_master,
3217 namelen, name);
3218 BUG();
3219 }
3220 } else {
3221 /* this is essentially what assert_master does */
3222 tmp->master = master;
3223 atomic_set(&tmp->woken, 1);
3224 wake_up(&tmp->wq);
3225 /* remove it so that only one mle will be found */
3226 __dlm_unlink_mle(dlm, tmp);
3227 __dlm_mle_detach_hb_events(dlm, tmp);
3228 if (tmp->type == DLM_MLE_MASTER) {
3229 ret = DLM_MIGRATE_RESPONSE_MASTERY_REF;
3230 mlog(0, "%s:%.*s: master=%u, newmaster=%u, "
3231 "telling master to get ref "
3232 "for cleared out mle during "
3233 "migration\n", dlm->name,
3234 namelen, name, master,
3235 new_master);
3236 }
3237 }
3238 spin_unlock(&tmp->spinlock);
3239 }
3240
3241 /* now add a migration mle to the tail of the list */
3242 dlm_init_mle(mle, DLM_MLE_MIGRATION, dlm, res, name, namelen);
3243 mle->new_master = new_master;
3244 /* the new master will be sending an assert master for this.
3245 * at that point we will get the refmap reference */
3246 mle->master = master;
3247 /* do this for consistency with other mle types */
3248 set_bit(new_master, mle->maybe_map);
3249 __dlm_insert_mle(dlm, mle);
3250
3251 return ret;
3252 }
3253
3254 /*
3255 * Sets the owner of the lockres, associated to the mle, to UNKNOWN
3256 */
3257 static struct dlm_lock_resource *dlm_reset_mleres_owner(struct dlm_ctxt *dlm,
3258 struct dlm_master_list_entry *mle)
3259 {
3260 struct dlm_lock_resource *res;
3261
3262 /* Find the lockres associated to the mle and set its owner to UNK */
3263 res = __dlm_lookup_lockres(dlm, mle->mname, mle->mnamelen,
3264 mle->mnamehash);
3265 if (res) {
3266 spin_unlock(&dlm->master_lock);
3267
3268 /* move lockres onto recovery list */
3269 spin_lock(&res->spinlock);
3270 dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);
3271 dlm_move_lockres_to_recovery_list(dlm, res);
3272 spin_unlock(&res->spinlock);
3273 dlm_lockres_put(res);
3274
3275 /* about to get rid of mle, detach from heartbeat */
3276 __dlm_mle_detach_hb_events(dlm, mle);
3277
3278 /* dump the mle */
3279 spin_lock(&dlm->master_lock);
3280 __dlm_put_mle(mle);
3281 spin_unlock(&dlm->master_lock);
3282 }
3283
3284 return res;
3285 }
3286
3287 static void dlm_clean_migration_mle(struct dlm_ctxt *dlm,
3288 struct dlm_master_list_entry *mle)
3289 {
3290 __dlm_mle_detach_hb_events(dlm, mle);
3291
3292 spin_lock(&mle->spinlock);
3293 __dlm_unlink_mle(dlm, mle);
3294 atomic_set(&mle->woken, 1);
3295 spin_unlock(&mle->spinlock);
3296
3297 wake_up(&mle->wq);
3298 }
3299
3300 static void dlm_clean_block_mle(struct dlm_ctxt *dlm,
3301 struct dlm_master_list_entry *mle, u8 dead_node)
3302 {
3303 int bit;
3304
3305 BUG_ON(mle->type != DLM_MLE_BLOCK);
3306
3307 spin_lock(&mle->spinlock);
3308 bit = find_next_bit(mle->maybe_map, O2NM_MAX_NODES, 0);
3309 if (bit != dead_node) {
3310 mlog(0, "mle found, but dead node %u would not have been "
3311 "master\n", dead_node);
3312 spin_unlock(&mle->spinlock);
3313 } else {
3314 /* Must drop the refcount by one since the assert_master will
3315 * never arrive. This may result in the mle being unlinked and
3316 * freed, but there may still be a process waiting in the
3317 * dlmlock path which is fine. */
3318 mlog(0, "node %u was expected master\n", dead_node);
3319 atomic_set(&mle->woken, 1);
3320 spin_unlock(&mle->spinlock);
3321 wake_up(&mle->wq);
3322
3323 /* Do not need events any longer, so detach from heartbeat */
3324 __dlm_mle_detach_hb_events(dlm, mle);
3325 __dlm_put_mle(mle);
3326 }
3327 }
3328
3329 void dlm_clean_master_list(struct dlm_ctxt *dlm, u8 dead_node)
3330 {
3331 struct dlm_master_list_entry *mle;
3332 struct dlm_lock_resource *res;
3333 struct hlist_head *bucket;
3334 struct hlist_node *tmp;
3335 unsigned int i;
3336
3337 mlog(0, "dlm=%s, dead node=%u\n", dlm->name, dead_node);
3338 top:
3339 assert_spin_locked(&dlm->spinlock);
3340
3341 /* clean the master list */
3342 spin_lock(&dlm->master_lock);
3343 for (i = 0; i < DLM_HASH_BUCKETS; i++) {
3344 bucket = dlm_master_hash(dlm, i);
3345 hlist_for_each_entry_safe(mle, tmp, bucket, master_hash_node) {
3346 BUG_ON(mle->type != DLM_MLE_BLOCK &&
3347 mle->type != DLM_MLE_MASTER &&
3348 mle->type != DLM_MLE_MIGRATION);
3349
3350 /* MASTER mles are initiated locally. The waiting
3351 * process will notice the node map change shortly.
3352 * Let that happen as normal. */
3353 if (mle->type == DLM_MLE_MASTER)
3354 continue;
3355
3356 /* BLOCK mles are initiated by other nodes. Need to
3357 * clean up if the dead node would have been the
3358 * master. */
3359 if (mle->type == DLM_MLE_BLOCK) {
3360 dlm_clean_block_mle(dlm, mle, dead_node);
3361 continue;
3362 }
3363
3364 /* Everything else is a MIGRATION mle */
3365
3366 /* The rule for MIGRATION mles is that the master
3367 * becomes UNKNOWN if *either* the original or the new
3368 * master dies. All UNKNOWN lockres' are sent to
3369 * whichever node becomes the recovery master. The new
3370 * master is responsible for determining if there is
3371 * still a master for this lockres, or if he needs to
3372 * take over mastery. Either way, this node should
3373 * expect another message to resolve this. */
3374
3375 if (mle->master != dead_node &&
3376 mle->new_master != dead_node)
3377 continue;
3378
3379 if (mle->new_master == dead_node && mle->inuse) {
3380 mlog(ML_NOTICE, "%s: target %u died during "
3381 "migration from %u, the MLE is "
3382 "still keep used, ignore it!\n",
3383 dlm->name, dead_node,
3384 mle->master);
3385 continue;
3386 }
3387
3388 /* If we have reached this point, this mle needs to be
3389 * removed from the list and freed. */
3390 dlm_clean_migration_mle(dlm, mle);
3391
3392 mlog(0, "%s: node %u died during migration from "
3393 "%u to %u!\n", dlm->name, dead_node, mle->master,
3394 mle->new_master);
3395
3396 /* If we find a lockres associated with the mle, we've
3397 * hit this rare case that messes up our lock ordering.
3398 * If so, we need to drop the master lock so that we can
3399 * take the lockres lock, meaning that we will have to
3400 * restart from the head of list. */
3401 res = dlm_reset_mleres_owner(dlm, mle);
3402 if (res)
3403 /* restart */
3404 goto top;
3405
3406 /* This may be the last reference */
3407 __dlm_put_mle(mle);
3408 }
3409 }
3410 spin_unlock(&dlm->master_lock);
3411 }
3412
3413 int dlm_finish_migration(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
3414 u8 old_master)
3415 {
3416 struct dlm_node_iter iter;
3417 int ret = 0;
3418
3419 spin_lock(&dlm->spinlock);
3420 dlm_node_iter_init(dlm->domain_map, &iter);
3421 clear_bit(old_master, iter.node_map);
3422 clear_bit(dlm->node_num, iter.node_map);
3423 spin_unlock(&dlm->spinlock);
3424
3425 /* ownership of the lockres is changing. account for the
3426 * mastery reference here since old_master will briefly have
3427 * a reference after the migration completes */
3428 spin_lock(&res->spinlock);
3429 dlm_lockres_set_refmap_bit(dlm, res, old_master);
3430 spin_unlock(&res->spinlock);
3431
3432 mlog(0, "now time to do a migrate request to other nodes\n");
3433 ret = dlm_do_migrate_request(dlm, res, old_master,
3434 dlm->node_num, &iter);
3435 if (ret < 0) {
3436 mlog_errno(ret);
3437 goto leave;
3438 }
3439
3440 mlog(0, "doing assert master of %.*s to all except the original node\n",
3441 res->lockname.len, res->lockname.name);
3442 /* this call now finishes out the nodemap
3443 * even if one or more nodes die */
3444 ret = dlm_do_assert_master(dlm, res, iter.node_map,
3445 DLM_ASSERT_MASTER_FINISH_MIGRATION);
3446 if (ret < 0) {
3447 /* no longer need to retry. all living nodes contacted. */
3448 mlog_errno(ret);
3449 ret = 0;
3450 }
3451
3452 memset(iter.node_map, 0, sizeof(iter.node_map));
3453 set_bit(old_master, iter.node_map);
3454 mlog(0, "doing assert master of %.*s back to %u\n",
3455 res->lockname.len, res->lockname.name, old_master);
3456 ret = dlm_do_assert_master(dlm, res, iter.node_map,
3457 DLM_ASSERT_MASTER_FINISH_MIGRATION);
3458 if (ret < 0) {
3459 mlog(0, "assert master to original master failed "
3460 "with %d.\n", ret);
3461 /* the only nonzero status here would be because of
3462 * a dead original node. we're done. */
3463 ret = 0;
3464 }
3465
3466 /* all done, set the owner, clear the flag */
3467 spin_lock(&res->spinlock);
3468 dlm_set_lockres_owner(dlm, res, dlm->node_num);
3469 res->state &= ~DLM_LOCK_RES_MIGRATING;
3470 spin_unlock(&res->spinlock);
3471 /* re-dirty it on the new master */
3472 dlm_kick_thread(dlm, res);
3473 wake_up(&res->wq);
3474 leave:
3475 return ret;
3476 }
3477
3478 /*
3479 * LOCKRES AST REFCOUNT
3480 * this is integral to migration
3481 */
3482
3483 /* for future intent to call an ast, reserve one ahead of time.
3484 * this should be called only after waiting on the lockres
3485 * with dlm_wait_on_lockres, and while still holding the
3486 * spinlock after the call. */
3487 void __dlm_lockres_reserve_ast(struct dlm_lock_resource *res)
3488 {
3489 assert_spin_locked(&res->spinlock);
3490 if (res->state & DLM_LOCK_RES_MIGRATING) {
3491 __dlm_print_one_lock_resource(res);
3492 }
3493 BUG_ON(res->state & DLM_LOCK_RES_MIGRATING);
3494
3495 atomic_inc(&res->asts_reserved);
3496 }
3497
3498 /*
3499 * used to drop the reserved ast, either because it went unused,
3500 * or because the ast/bast was actually called.
3501 *
3502 * also, if there is a pending migration on this lockres,
3503 * and this was the last pending ast on the lockres,
3504 * atomically set the MIGRATING flag before we drop the lock.
3505 * this is how we ensure that migration can proceed with no
3506 * asts in progress. note that it is ok if the state of the
3507 * queues is such that a lock should be granted in the future
3508 * or that a bast should be fired, because the new master will
3509 * shuffle the lists on this lockres as soon as it is migrated.
3510 */
3511 void dlm_lockres_release_ast(struct dlm_ctxt *dlm,
3512 struct dlm_lock_resource *res)
3513 {
3514 if (!atomic_dec_and_lock(&res->asts_reserved, &res->spinlock))
3515 return;
3516
3517 if (!res->migration_pending) {
3518 spin_unlock(&res->spinlock);
3519 return;
3520 }
3521
3522 BUG_ON(res->state & DLM_LOCK_RES_MIGRATING);
3523 res->migration_pending = 0;
3524 res->state |= DLM_LOCK_RES_MIGRATING;
3525 spin_unlock(&res->spinlock);
3526 wake_up(&res->wq);
3527 wake_up(&dlm->migration_wq);
3528 }
3529
3530 void dlm_force_free_mles(struct dlm_ctxt *dlm)
3531 {
3532 int i;
3533 struct hlist_head *bucket;
3534 struct dlm_master_list_entry *mle;
3535 struct hlist_node *tmp;
3536
3537 /*
3538 * We notified all other nodes that we are exiting the domain and
3539 * marked the dlm state to DLM_CTXT_LEAVING. If any mles are still
3540 * around we force free them and wake any processes that are waiting
3541 * on the mles
3542 */
3543 spin_lock(&dlm->spinlock);
3544 spin_lock(&dlm->master_lock);
3545
3546 BUG_ON(dlm->dlm_state != DLM_CTXT_LEAVING);
3547 BUG_ON((find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 0) < O2NM_MAX_NODES));
3548
3549 for (i = 0; i < DLM_HASH_BUCKETS; i++) {
3550 bucket = dlm_master_hash(dlm, i);
3551 hlist_for_each_entry_safe(mle, tmp, bucket, master_hash_node) {
3552 if (mle->type != DLM_MLE_BLOCK) {
3553 mlog(ML_ERROR, "bad mle: %p\n", mle);
3554 dlm_print_one_mle(mle);
3555 }
3556 atomic_set(&mle->woken, 1);
3557 wake_up(&mle->wq);
3558
3559 __dlm_unlink_mle(dlm, mle);
3560 __dlm_mle_detach_hb_events(dlm, mle);
3561 __dlm_put_mle(mle);
3562 }
3563 }
3564 spin_unlock(&dlm->master_lock);
3565 spin_unlock(&dlm->spinlock);
3566 }
Linux with added FPC-III support