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