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