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