linux/fs/dlm/lock.c
<<
>>
Prefs
   1// SPDX-License-Identifier: GPL-2.0-only
   2/******************************************************************************
   3*******************************************************************************
   4**
   5**  Copyright (C) 2005-2010 Red Hat, Inc.  All rights reserved.
   6**
   7**
   8*******************************************************************************
   9******************************************************************************/
  10
  11/* Central locking logic has four stages:
  12
  13   dlm_lock()
  14   dlm_unlock()
  15
  16   request_lock(ls, lkb)
  17   convert_lock(ls, lkb)
  18   unlock_lock(ls, lkb)
  19   cancel_lock(ls, lkb)
  20
  21   _request_lock(r, lkb)
  22   _convert_lock(r, lkb)
  23   _unlock_lock(r, lkb)
  24   _cancel_lock(r, lkb)
  25
  26   do_request(r, lkb)
  27   do_convert(r, lkb)
  28   do_unlock(r, lkb)
  29   do_cancel(r, lkb)
  30
  31   Stage 1 (lock, unlock) is mainly about checking input args and
  32   splitting into one of the four main operations:
  33
  34       dlm_lock          = request_lock
  35       dlm_lock+CONVERT  = convert_lock
  36       dlm_unlock        = unlock_lock
  37       dlm_unlock+CANCEL = cancel_lock
  38
  39   Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
  40   provided to the next stage.
  41
  42   Stage 3, _xxxx_lock(), determines if the operation is local or remote.
  43   When remote, it calls send_xxxx(), when local it calls do_xxxx().
  44
  45   Stage 4, do_xxxx(), is the guts of the operation.  It manipulates the
  46   given rsb and lkb and queues callbacks.
  47
  48   For remote operations, send_xxxx() results in the corresponding do_xxxx()
  49   function being executed on the remote node.  The connecting send/receive
  50   calls on local (L) and remote (R) nodes:
  51
  52   L: send_xxxx()              ->  R: receive_xxxx()
  53                                   R: do_xxxx()
  54   L: receive_xxxx_reply()     <-  R: send_xxxx_reply()
  55*/
  56#include <linux/types.h>
  57#include <linux/rbtree.h>
  58#include <linux/slab.h>
  59#include "dlm_internal.h"
  60#include <linux/dlm_device.h>
  61#include "memory.h"
  62#include "midcomms.h"
  63#include "requestqueue.h"
  64#include "util.h"
  65#include "dir.h"
  66#include "member.h"
  67#include "lockspace.h"
  68#include "ast.h"
  69#include "lock.h"
  70#include "rcom.h"
  71#include "recover.h"
  72#include "lvb_table.h"
  73#include "user.h"
  74#include "config.h"
  75
  76static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
  77static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
  78static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  79static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
  80static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
  81static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
  82static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
  83static int send_remove(struct dlm_rsb *r);
  84static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  85static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  86static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
  87                                    struct dlm_message *ms);
  88static int receive_extralen(struct dlm_message *ms);
  89static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
  90static void del_timeout(struct dlm_lkb *lkb);
  91static void toss_rsb(struct kref *kref);
  92
  93/*
  94 * Lock compatibilty matrix - thanks Steve
  95 * UN = Unlocked state. Not really a state, used as a flag
  96 * PD = Padding. Used to make the matrix a nice power of two in size
  97 * Other states are the same as the VMS DLM.
  98 * Usage: matrix[grmode+1][rqmode+1]  (although m[rq+1][gr+1] is the same)
  99 */
 100
 101static const int __dlm_compat_matrix[8][8] = {
 102      /* UN NL CR CW PR PW EX PD */
 103        {1, 1, 1, 1, 1, 1, 1, 0},       /* UN */
 104        {1, 1, 1, 1, 1, 1, 1, 0},       /* NL */
 105        {1, 1, 1, 1, 1, 1, 0, 0},       /* CR */
 106        {1, 1, 1, 1, 0, 0, 0, 0},       /* CW */
 107        {1, 1, 1, 0, 1, 0, 0, 0},       /* PR */
 108        {1, 1, 1, 0, 0, 0, 0, 0},       /* PW */
 109        {1, 1, 0, 0, 0, 0, 0, 0},       /* EX */
 110        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 111};
 112
 113/*
 114 * This defines the direction of transfer of LVB data.
 115 * Granted mode is the row; requested mode is the column.
 116 * Usage: matrix[grmode+1][rqmode+1]
 117 * 1 = LVB is returned to the caller
 118 * 0 = LVB is written to the resource
 119 * -1 = nothing happens to the LVB
 120 */
 121
 122const int dlm_lvb_operations[8][8] = {
 123        /* UN   NL  CR  CW  PR  PW  EX  PD*/
 124        {  -1,  1,  1,  1,  1,  1,  1, -1 }, /* UN */
 125        {  -1,  1,  1,  1,  1,  1,  1,  0 }, /* NL */
 126        {  -1, -1,  1,  1,  1,  1,  1,  0 }, /* CR */
 127        {  -1, -1, -1,  1,  1,  1,  1,  0 }, /* CW */
 128        {  -1, -1, -1, -1,  1,  1,  1,  0 }, /* PR */
 129        {  -1,  0,  0,  0,  0,  0,  1,  0 }, /* PW */
 130        {  -1,  0,  0,  0,  0,  0,  0,  0 }, /* EX */
 131        {  -1,  0,  0,  0,  0,  0,  0,  0 }  /* PD */
 132};
 133
 134#define modes_compat(gr, rq) \
 135        __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
 136
 137int dlm_modes_compat(int mode1, int mode2)
 138{
 139        return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
 140}
 141
 142/*
 143 * Compatibility matrix for conversions with QUECVT set.
 144 * Granted mode is the row; requested mode is the column.
 145 * Usage: matrix[grmode+1][rqmode+1]
 146 */
 147
 148static const int __quecvt_compat_matrix[8][8] = {
 149      /* UN NL CR CW PR PW EX PD */
 150        {0, 0, 0, 0, 0, 0, 0, 0},       /* UN */
 151        {0, 0, 1, 1, 1, 1, 1, 0},       /* NL */
 152        {0, 0, 0, 1, 1, 1, 1, 0},       /* CR */
 153        {0, 0, 0, 0, 1, 1, 1, 0},       /* CW */
 154        {0, 0, 0, 1, 0, 1, 1, 0},       /* PR */
 155        {0, 0, 0, 0, 0, 0, 1, 0},       /* PW */
 156        {0, 0, 0, 0, 0, 0, 0, 0},       /* EX */
 157        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 158};
 159
 160void dlm_print_lkb(struct dlm_lkb *lkb)
 161{
 162        printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
 163               "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
 164               lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
 165               lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
 166               lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
 167               (unsigned long long)lkb->lkb_recover_seq);
 168}
 169
 170static void dlm_print_rsb(struct dlm_rsb *r)
 171{
 172        printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
 173               "rlc %d name %s\n",
 174               r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
 175               r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
 176               r->res_name);
 177}
 178
 179void dlm_dump_rsb(struct dlm_rsb *r)
 180{
 181        struct dlm_lkb *lkb;
 182
 183        dlm_print_rsb(r);
 184
 185        printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
 186               list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
 187        printk(KERN_ERR "rsb lookup list\n");
 188        list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
 189                dlm_print_lkb(lkb);
 190        printk(KERN_ERR "rsb grant queue:\n");
 191        list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
 192                dlm_print_lkb(lkb);
 193        printk(KERN_ERR "rsb convert queue:\n");
 194        list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
 195                dlm_print_lkb(lkb);
 196        printk(KERN_ERR "rsb wait queue:\n");
 197        list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
 198                dlm_print_lkb(lkb);
 199}
 200
 201/* Threads cannot use the lockspace while it's being recovered */
 202
 203static inline void dlm_lock_recovery(struct dlm_ls *ls)
 204{
 205        down_read(&ls->ls_in_recovery);
 206}
 207
 208void dlm_unlock_recovery(struct dlm_ls *ls)
 209{
 210        up_read(&ls->ls_in_recovery);
 211}
 212
 213int dlm_lock_recovery_try(struct dlm_ls *ls)
 214{
 215        return down_read_trylock(&ls->ls_in_recovery);
 216}
 217
 218static inline int can_be_queued(struct dlm_lkb *lkb)
 219{
 220        return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
 221}
 222
 223static inline int force_blocking_asts(struct dlm_lkb *lkb)
 224{
 225        return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
 226}
 227
 228static inline int is_demoted(struct dlm_lkb *lkb)
 229{
 230        return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
 231}
 232
 233static inline int is_altmode(struct dlm_lkb *lkb)
 234{
 235        return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
 236}
 237
 238static inline int is_granted(struct dlm_lkb *lkb)
 239{
 240        return (lkb->lkb_status == DLM_LKSTS_GRANTED);
 241}
 242
 243static inline int is_remote(struct dlm_rsb *r)
 244{
 245        DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
 246        return !!r->res_nodeid;
 247}
 248
 249static inline int is_process_copy(struct dlm_lkb *lkb)
 250{
 251        return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
 252}
 253
 254static inline int is_master_copy(struct dlm_lkb *lkb)
 255{
 256        return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
 257}
 258
 259static inline int middle_conversion(struct dlm_lkb *lkb)
 260{
 261        if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
 262            (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
 263                return 1;
 264        return 0;
 265}
 266
 267static inline int down_conversion(struct dlm_lkb *lkb)
 268{
 269        return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
 270}
 271
 272static inline int is_overlap_unlock(struct dlm_lkb *lkb)
 273{
 274        return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
 275}
 276
 277static inline int is_overlap_cancel(struct dlm_lkb *lkb)
 278{
 279        return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
 280}
 281
 282static inline int is_overlap(struct dlm_lkb *lkb)
 283{
 284        return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
 285                                  DLM_IFL_OVERLAP_CANCEL));
 286}
 287
 288static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
 289{
 290        if (is_master_copy(lkb))
 291                return;
 292
 293        del_timeout(lkb);
 294
 295        DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
 296
 297        /* if the operation was a cancel, then return -DLM_ECANCEL, if a
 298           timeout caused the cancel then return -ETIMEDOUT */
 299        if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
 300                lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
 301                rv = -ETIMEDOUT;
 302        }
 303
 304        if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
 305                lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
 306                rv = -EDEADLK;
 307        }
 308
 309        dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
 310}
 311
 312static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
 313{
 314        queue_cast(r, lkb,
 315                   is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
 316}
 317
 318static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
 319{
 320        if (is_master_copy(lkb)) {
 321                send_bast(r, lkb, rqmode);
 322        } else {
 323                dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
 324        }
 325}
 326
 327/*
 328 * Basic operations on rsb's and lkb's
 329 */
 330
 331/* This is only called to add a reference when the code already holds
 332   a valid reference to the rsb, so there's no need for locking. */
 333
 334static inline void hold_rsb(struct dlm_rsb *r)
 335{
 336        kref_get(&r->res_ref);
 337}
 338
 339void dlm_hold_rsb(struct dlm_rsb *r)
 340{
 341        hold_rsb(r);
 342}
 343
 344/* When all references to the rsb are gone it's transferred to
 345   the tossed list for later disposal. */
 346
 347static void put_rsb(struct dlm_rsb *r)
 348{
 349        struct dlm_ls *ls = r->res_ls;
 350        uint32_t bucket = r->res_bucket;
 351
 352        spin_lock(&ls->ls_rsbtbl[bucket].lock);
 353        kref_put(&r->res_ref, toss_rsb);
 354        spin_unlock(&ls->ls_rsbtbl[bucket].lock);
 355}
 356
 357void dlm_put_rsb(struct dlm_rsb *r)
 358{
 359        put_rsb(r);
 360}
 361
 362static int pre_rsb_struct(struct dlm_ls *ls)
 363{
 364        struct dlm_rsb *r1, *r2;
 365        int count = 0;
 366
 367        spin_lock(&ls->ls_new_rsb_spin);
 368        if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
 369                spin_unlock(&ls->ls_new_rsb_spin);
 370                return 0;
 371        }
 372        spin_unlock(&ls->ls_new_rsb_spin);
 373
 374        r1 = dlm_allocate_rsb(ls);
 375        r2 = dlm_allocate_rsb(ls);
 376
 377        spin_lock(&ls->ls_new_rsb_spin);
 378        if (r1) {
 379                list_add(&r1->res_hashchain, &ls->ls_new_rsb);
 380                ls->ls_new_rsb_count++;
 381        }
 382        if (r2) {
 383                list_add(&r2->res_hashchain, &ls->ls_new_rsb);
 384                ls->ls_new_rsb_count++;
 385        }
 386        count = ls->ls_new_rsb_count;
 387        spin_unlock(&ls->ls_new_rsb_spin);
 388
 389        if (!count)
 390                return -ENOMEM;
 391        return 0;
 392}
 393
 394/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
 395   unlock any spinlocks, go back and call pre_rsb_struct again.
 396   Otherwise, take an rsb off the list and return it. */
 397
 398static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
 399                          struct dlm_rsb **r_ret)
 400{
 401        struct dlm_rsb *r;
 402        int count;
 403
 404        spin_lock(&ls->ls_new_rsb_spin);
 405        if (list_empty(&ls->ls_new_rsb)) {
 406                count = ls->ls_new_rsb_count;
 407                spin_unlock(&ls->ls_new_rsb_spin);
 408                log_debug(ls, "find_rsb retry %d %d %s",
 409                          count, dlm_config.ci_new_rsb_count, name);
 410                return -EAGAIN;
 411        }
 412
 413        r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
 414        list_del(&r->res_hashchain);
 415        /* Convert the empty list_head to a NULL rb_node for tree usage: */
 416        memset(&r->res_hashnode, 0, sizeof(struct rb_node));
 417        ls->ls_new_rsb_count--;
 418        spin_unlock(&ls->ls_new_rsb_spin);
 419
 420        r->res_ls = ls;
 421        r->res_length = len;
 422        memcpy(r->res_name, name, len);
 423        mutex_init(&r->res_mutex);
 424
 425        INIT_LIST_HEAD(&r->res_lookup);
 426        INIT_LIST_HEAD(&r->res_grantqueue);
 427        INIT_LIST_HEAD(&r->res_convertqueue);
 428        INIT_LIST_HEAD(&r->res_waitqueue);
 429        INIT_LIST_HEAD(&r->res_root_list);
 430        INIT_LIST_HEAD(&r->res_recover_list);
 431
 432        *r_ret = r;
 433        return 0;
 434}
 435
 436static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
 437{
 438        char maxname[DLM_RESNAME_MAXLEN];
 439
 440        memset(maxname, 0, DLM_RESNAME_MAXLEN);
 441        memcpy(maxname, name, nlen);
 442        return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
 443}
 444
 445int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
 446                        struct dlm_rsb **r_ret)
 447{
 448        struct rb_node *node = tree->rb_node;
 449        struct dlm_rsb *r;
 450        int rc;
 451
 452        while (node) {
 453                r = rb_entry(node, struct dlm_rsb, res_hashnode);
 454                rc = rsb_cmp(r, name, len);
 455                if (rc < 0)
 456                        node = node->rb_left;
 457                else if (rc > 0)
 458                        node = node->rb_right;
 459                else
 460                        goto found;
 461        }
 462        *r_ret = NULL;
 463        return -EBADR;
 464
 465 found:
 466        *r_ret = r;
 467        return 0;
 468}
 469
 470static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
 471{
 472        struct rb_node **newn = &tree->rb_node;
 473        struct rb_node *parent = NULL;
 474        int rc;
 475
 476        while (*newn) {
 477                struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
 478                                               res_hashnode);
 479
 480                parent = *newn;
 481                rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
 482                if (rc < 0)
 483                        newn = &parent->rb_left;
 484                else if (rc > 0)
 485                        newn = &parent->rb_right;
 486                else {
 487                        log_print("rsb_insert match");
 488                        dlm_dump_rsb(rsb);
 489                        dlm_dump_rsb(cur);
 490                        return -EEXIST;
 491                }
 492        }
 493
 494        rb_link_node(&rsb->res_hashnode, parent, newn);
 495        rb_insert_color(&rsb->res_hashnode, tree);
 496        return 0;
 497}
 498
 499/*
 500 * Find rsb in rsbtbl and potentially create/add one
 501 *
 502 * Delaying the release of rsb's has a similar benefit to applications keeping
 503 * NL locks on an rsb, but without the guarantee that the cached master value
 504 * will still be valid when the rsb is reused.  Apps aren't always smart enough
 505 * to keep NL locks on an rsb that they may lock again shortly; this can lead
 506 * to excessive master lookups and removals if we don't delay the release.
 507 *
 508 * Searching for an rsb means looking through both the normal list and toss
 509 * list.  When found on the toss list the rsb is moved to the normal list with
 510 * ref count of 1; when found on normal list the ref count is incremented.
 511 *
 512 * rsb's on the keep list are being used locally and refcounted.
 513 * rsb's on the toss list are not being used locally, and are not refcounted.
 514 *
 515 * The toss list rsb's were either
 516 * - previously used locally but not any more (were on keep list, then
 517 *   moved to toss list when last refcount dropped)
 518 * - created and put on toss list as a directory record for a lookup
 519 *   (we are the dir node for the res, but are not using the res right now,
 520 *   but some other node is)
 521 *
 522 * The purpose of find_rsb() is to return a refcounted rsb for local use.
 523 * So, if the given rsb is on the toss list, it is moved to the keep list
 524 * before being returned.
 525 *
 526 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
 527 * more refcounts exist, so the rsb is moved from the keep list to the
 528 * toss list.
 529 *
 530 * rsb's on both keep and toss lists are used for doing a name to master
 531 * lookups.  rsb's that are in use locally (and being refcounted) are on
 532 * the keep list, rsb's that are not in use locally (not refcounted) and
 533 * only exist for name/master lookups are on the toss list.
 534 *
 535 * rsb's on the toss list who's dir_nodeid is not local can have stale
 536 * name/master mappings.  So, remote requests on such rsb's can potentially
 537 * return with an error, which means the mapping is stale and needs to
 538 * be updated with a new lookup.  (The idea behind MASTER UNCERTAIN and
 539 * first_lkid is to keep only a single outstanding request on an rsb
 540 * while that rsb has a potentially stale master.)
 541 */
 542
 543static int find_rsb_dir(struct dlm_ls *ls, char *name, int len,
 544                        uint32_t hash, uint32_t b,
 545                        int dir_nodeid, int from_nodeid,
 546                        unsigned int flags, struct dlm_rsb **r_ret)
 547{
 548        struct dlm_rsb *r = NULL;
 549        int our_nodeid = dlm_our_nodeid();
 550        int from_local = 0;
 551        int from_other = 0;
 552        int from_dir = 0;
 553        int create = 0;
 554        int error;
 555
 556        if (flags & R_RECEIVE_REQUEST) {
 557                if (from_nodeid == dir_nodeid)
 558                        from_dir = 1;
 559                else
 560                        from_other = 1;
 561        } else if (flags & R_REQUEST) {
 562                from_local = 1;
 563        }
 564
 565        /*
 566         * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
 567         * from_nodeid has sent us a lock in dlm_recover_locks, believing
 568         * we're the new master.  Our local recovery may not have set
 569         * res_master_nodeid to our_nodeid yet, so allow either.  Don't
 570         * create the rsb; dlm_recover_process_copy() will handle EBADR
 571         * by resending.
 572         *
 573         * If someone sends us a request, we are the dir node, and we do
 574         * not find the rsb anywhere, then recreate it.  This happens if
 575         * someone sends us a request after we have removed/freed an rsb
 576         * from our toss list.  (They sent a request instead of lookup
 577         * because they are using an rsb from their toss list.)
 578         */
 579
 580        if (from_local || from_dir ||
 581            (from_other && (dir_nodeid == our_nodeid))) {
 582                create = 1;
 583        }
 584
 585 retry:
 586        if (create) {
 587                error = pre_rsb_struct(ls);
 588                if (error < 0)
 589                        goto out;
 590        }
 591
 592        spin_lock(&ls->ls_rsbtbl[b].lock);
 593
 594        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 595        if (error)
 596                goto do_toss;
 597        
 598        /*
 599         * rsb is active, so we can't check master_nodeid without lock_rsb.
 600         */
 601
 602        kref_get(&r->res_ref);
 603        error = 0;
 604        goto out_unlock;
 605
 606
 607 do_toss:
 608        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 609        if (error)
 610                goto do_new;
 611
 612        /*
 613         * rsb found inactive (master_nodeid may be out of date unless
 614         * we are the dir_nodeid or were the master)  No other thread
 615         * is using this rsb because it's on the toss list, so we can
 616         * look at or update res_master_nodeid without lock_rsb.
 617         */
 618
 619        if ((r->res_master_nodeid != our_nodeid) && from_other) {
 620                /* our rsb was not master, and another node (not the dir node)
 621                   has sent us a request */
 622                log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
 623                          from_nodeid, r->res_master_nodeid, dir_nodeid,
 624                          r->res_name);
 625                error = -ENOTBLK;
 626                goto out_unlock;
 627        }
 628
 629        if ((r->res_master_nodeid != our_nodeid) && from_dir) {
 630                /* don't think this should ever happen */
 631                log_error(ls, "find_rsb toss from_dir %d master %d",
 632                          from_nodeid, r->res_master_nodeid);
 633                dlm_print_rsb(r);
 634                /* fix it and go on */
 635                r->res_master_nodeid = our_nodeid;
 636                r->res_nodeid = 0;
 637                rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
 638                r->res_first_lkid = 0;
 639        }
 640
 641        if (from_local && (r->res_master_nodeid != our_nodeid)) {
 642                /* Because we have held no locks on this rsb,
 643                   res_master_nodeid could have become stale. */
 644                rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
 645                r->res_first_lkid = 0;
 646        }
 647
 648        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 649        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 650        goto out_unlock;
 651
 652
 653 do_new:
 654        /*
 655         * rsb not found
 656         */
 657
 658        if (error == -EBADR && !create)
 659                goto out_unlock;
 660
 661        error = get_rsb_struct(ls, name, len, &r);
 662        if (error == -EAGAIN) {
 663                spin_unlock(&ls->ls_rsbtbl[b].lock);
 664                goto retry;
 665        }
 666        if (error)
 667                goto out_unlock;
 668
 669        r->res_hash = hash;
 670        r->res_bucket = b;
 671        r->res_dir_nodeid = dir_nodeid;
 672        kref_init(&r->res_ref);
 673
 674        if (from_dir) {
 675                /* want to see how often this happens */
 676                log_debug(ls, "find_rsb new from_dir %d recreate %s",
 677                          from_nodeid, r->res_name);
 678                r->res_master_nodeid = our_nodeid;
 679                r->res_nodeid = 0;
 680                goto out_add;
 681        }
 682
 683        if (from_other && (dir_nodeid != our_nodeid)) {
 684                /* should never happen */
 685                log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
 686                          from_nodeid, dir_nodeid, our_nodeid, r->res_name);
 687                dlm_free_rsb(r);
 688                r = NULL;
 689                error = -ENOTBLK;
 690                goto out_unlock;
 691        }
 692
 693        if (from_other) {
 694                log_debug(ls, "find_rsb new from_other %d dir %d %s",
 695                          from_nodeid, dir_nodeid, r->res_name);
 696        }
 697
 698        if (dir_nodeid == our_nodeid) {
 699                /* When we are the dir nodeid, we can set the master
 700                   node immediately */
 701                r->res_master_nodeid = our_nodeid;
 702                r->res_nodeid = 0;
 703        } else {
 704                /* set_master will send_lookup to dir_nodeid */
 705                r->res_master_nodeid = 0;
 706                r->res_nodeid = -1;
 707        }
 708
 709 out_add:
 710        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 711 out_unlock:
 712        spin_unlock(&ls->ls_rsbtbl[b].lock);
 713 out:
 714        *r_ret = r;
 715        return error;
 716}
 717
 718/* During recovery, other nodes can send us new MSTCPY locks (from
 719   dlm_recover_locks) before we've made ourself master (in
 720   dlm_recover_masters). */
 721
 722static int find_rsb_nodir(struct dlm_ls *ls, char *name, int len,
 723                          uint32_t hash, uint32_t b,
 724                          int dir_nodeid, int from_nodeid,
 725                          unsigned int flags, struct dlm_rsb **r_ret)
 726{
 727        struct dlm_rsb *r = NULL;
 728        int our_nodeid = dlm_our_nodeid();
 729        int recover = (flags & R_RECEIVE_RECOVER);
 730        int error;
 731
 732 retry:
 733        error = pre_rsb_struct(ls);
 734        if (error < 0)
 735                goto out;
 736
 737        spin_lock(&ls->ls_rsbtbl[b].lock);
 738
 739        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 740        if (error)
 741                goto do_toss;
 742
 743        /*
 744         * rsb is active, so we can't check master_nodeid without lock_rsb.
 745         */
 746
 747        kref_get(&r->res_ref);
 748        goto out_unlock;
 749
 750
 751 do_toss:
 752        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 753        if (error)
 754                goto do_new;
 755
 756        /*
 757         * rsb found inactive. No other thread is using this rsb because
 758         * it's on the toss list, so we can look at or update
 759         * res_master_nodeid without lock_rsb.
 760         */
 761
 762        if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
 763                /* our rsb is not master, and another node has sent us a
 764                   request; this should never happen */
 765                log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
 766                          from_nodeid, r->res_master_nodeid, dir_nodeid);
 767                dlm_print_rsb(r);
 768                error = -ENOTBLK;
 769                goto out_unlock;
 770        }
 771
 772        if (!recover && (r->res_master_nodeid != our_nodeid) &&
 773            (dir_nodeid == our_nodeid)) {
 774                /* our rsb is not master, and we are dir; may as well fix it;
 775                   this should never happen */
 776                log_error(ls, "find_rsb toss our %d master %d dir %d",
 777                          our_nodeid, r->res_master_nodeid, dir_nodeid);
 778                dlm_print_rsb(r);
 779                r->res_master_nodeid = our_nodeid;
 780                r->res_nodeid = 0;
 781        }
 782
 783        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 784        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 785        goto out_unlock;
 786
 787
 788 do_new:
 789        /*
 790         * rsb not found
 791         */
 792
 793        error = get_rsb_struct(ls, name, len, &r);
 794        if (error == -EAGAIN) {
 795                spin_unlock(&ls->ls_rsbtbl[b].lock);
 796                goto retry;
 797        }
 798        if (error)
 799                goto out_unlock;
 800
 801        r->res_hash = hash;
 802        r->res_bucket = b;
 803        r->res_dir_nodeid = dir_nodeid;
 804        r->res_master_nodeid = dir_nodeid;
 805        r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
 806        kref_init(&r->res_ref);
 807
 808        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 809 out_unlock:
 810        spin_unlock(&ls->ls_rsbtbl[b].lock);
 811 out:
 812        *r_ret = r;
 813        return error;
 814}
 815
 816static int find_rsb(struct dlm_ls *ls, char *name, int len, int from_nodeid,
 817                    unsigned int flags, struct dlm_rsb **r_ret)
 818{
 819        uint32_t hash, b;
 820        int dir_nodeid;
 821
 822        if (len > DLM_RESNAME_MAXLEN)
 823                return -EINVAL;
 824
 825        hash = jhash(name, len, 0);
 826        b = hash & (ls->ls_rsbtbl_size - 1);
 827
 828        dir_nodeid = dlm_hash2nodeid(ls, hash);
 829
 830        if (dlm_no_directory(ls))
 831                return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
 832                                      from_nodeid, flags, r_ret);
 833        else
 834                return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
 835                                      from_nodeid, flags, r_ret);
 836}
 837
 838/* we have received a request and found that res_master_nodeid != our_nodeid,
 839   so we need to return an error or make ourself the master */
 840
 841static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
 842                                  int from_nodeid)
 843{
 844        if (dlm_no_directory(ls)) {
 845                log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
 846                          from_nodeid, r->res_master_nodeid,
 847                          r->res_dir_nodeid);
 848                dlm_print_rsb(r);
 849                return -ENOTBLK;
 850        }
 851
 852        if (from_nodeid != r->res_dir_nodeid) {
 853                /* our rsb is not master, and another node (not the dir node)
 854                   has sent us a request.  this is much more common when our
 855                   master_nodeid is zero, so limit debug to non-zero.  */
 856
 857                if (r->res_master_nodeid) {
 858                        log_debug(ls, "validate master from_other %d master %d "
 859                                  "dir %d first %x %s", from_nodeid,
 860                                  r->res_master_nodeid, r->res_dir_nodeid,
 861                                  r->res_first_lkid, r->res_name);
 862                }
 863                return -ENOTBLK;
 864        } else {
 865                /* our rsb is not master, but the dir nodeid has sent us a
 866                   request; this could happen with master 0 / res_nodeid -1 */
 867
 868                if (r->res_master_nodeid) {
 869                        log_error(ls, "validate master from_dir %d master %d "
 870                                  "first %x %s",
 871                                  from_nodeid, r->res_master_nodeid,
 872                                  r->res_first_lkid, r->res_name);
 873                }
 874
 875                r->res_master_nodeid = dlm_our_nodeid();
 876                r->res_nodeid = 0;
 877                return 0;
 878        }
 879}
 880
 881/*
 882 * We're the dir node for this res and another node wants to know the
 883 * master nodeid.  During normal operation (non recovery) this is only
 884 * called from receive_lookup(); master lookups when the local node is
 885 * the dir node are done by find_rsb().
 886 *
 887 * normal operation, we are the dir node for a resource
 888 * . _request_lock
 889 * . set_master
 890 * . send_lookup
 891 * . receive_lookup
 892 * . dlm_master_lookup flags 0
 893 *
 894 * recover directory, we are rebuilding dir for all resources
 895 * . dlm_recover_directory
 896 * . dlm_rcom_names
 897 *   remote node sends back the rsb names it is master of and we are dir of
 898 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
 899 *   we either create new rsb setting remote node as master, or find existing
 900 *   rsb and set master to be the remote node.
 901 *
 902 * recover masters, we are finding the new master for resources
 903 * . dlm_recover_masters
 904 * . recover_master
 905 * . dlm_send_rcom_lookup
 906 * . receive_rcom_lookup
 907 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
 908 */
 909
 910int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
 911                      unsigned int flags, int *r_nodeid, int *result)
 912{
 913        struct dlm_rsb *r = NULL;
 914        uint32_t hash, b;
 915        int from_master = (flags & DLM_LU_RECOVER_DIR);
 916        int fix_master = (flags & DLM_LU_RECOVER_MASTER);
 917        int our_nodeid = dlm_our_nodeid();
 918        int dir_nodeid, error, toss_list = 0;
 919
 920        if (len > DLM_RESNAME_MAXLEN)
 921                return -EINVAL;
 922
 923        if (from_nodeid == our_nodeid) {
 924                log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
 925                          our_nodeid, flags);
 926                return -EINVAL;
 927        }
 928
 929        hash = jhash(name, len, 0);
 930        b = hash & (ls->ls_rsbtbl_size - 1);
 931
 932        dir_nodeid = dlm_hash2nodeid(ls, hash);
 933        if (dir_nodeid != our_nodeid) {
 934                log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
 935                          from_nodeid, dir_nodeid, our_nodeid, hash,
 936                          ls->ls_num_nodes);
 937                *r_nodeid = -1;
 938                return -EINVAL;
 939        }
 940
 941 retry:
 942        error = pre_rsb_struct(ls);
 943        if (error < 0)
 944                return error;
 945
 946        spin_lock(&ls->ls_rsbtbl[b].lock);
 947        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 948        if (!error) {
 949                /* because the rsb is active, we need to lock_rsb before
 950                   checking/changing re_master_nodeid */
 951
 952                hold_rsb(r);
 953                spin_unlock(&ls->ls_rsbtbl[b].lock);
 954                lock_rsb(r);
 955                goto found;
 956        }
 957
 958        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 959        if (error)
 960                goto not_found;
 961
 962        /* because the rsb is inactive (on toss list), it's not refcounted
 963           and lock_rsb is not used, but is protected by the rsbtbl lock */
 964
 965        toss_list = 1;
 966 found:
 967        if (r->res_dir_nodeid != our_nodeid) {
 968                /* should not happen, but may as well fix it and carry on */
 969                log_error(ls, "dlm_master_lookup res_dir %d our %d %s",
 970                          r->res_dir_nodeid, our_nodeid, r->res_name);
 971                r->res_dir_nodeid = our_nodeid;
 972        }
 973
 974        if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
 975                /* Recovery uses this function to set a new master when
 976                   the previous master failed.  Setting NEW_MASTER will
 977                   force dlm_recover_masters to call recover_master on this
 978                   rsb even though the res_nodeid is no longer removed. */
 979
 980                r->res_master_nodeid = from_nodeid;
 981                r->res_nodeid = from_nodeid;
 982                rsb_set_flag(r, RSB_NEW_MASTER);
 983
 984                if (toss_list) {
 985                        /* I don't think we should ever find it on toss list. */
 986                        log_error(ls, "dlm_master_lookup fix_master on toss");
 987                        dlm_dump_rsb(r);
 988                }
 989        }
 990
 991        if (from_master && (r->res_master_nodeid != from_nodeid)) {
 992                /* this will happen if from_nodeid became master during
 993                   a previous recovery cycle, and we aborted the previous
 994                   cycle before recovering this master value */
 995
 996                log_limit(ls, "dlm_master_lookup from_master %d "
 997                          "master_nodeid %d res_nodeid %d first %x %s",
 998                          from_nodeid, r->res_master_nodeid, r->res_nodeid,
 999                          r->res_first_lkid, r->res_name);
1000
1001                if (r->res_master_nodeid == our_nodeid) {
1002                        log_error(ls, "from_master %d our_master", from_nodeid);
1003                        dlm_dump_rsb(r);
1004                        goto out_found;
1005                }
1006
1007                r->res_master_nodeid = from_nodeid;
1008                r->res_nodeid = from_nodeid;
1009                rsb_set_flag(r, RSB_NEW_MASTER);
1010        }
1011
1012        if (!r->res_master_nodeid) {
1013                /* this will happen if recovery happens while we're looking
1014                   up the master for this rsb */
1015
1016                log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s",
1017                          from_nodeid, r->res_first_lkid, r->res_name);
1018                r->res_master_nodeid = from_nodeid;
1019                r->res_nodeid = from_nodeid;
1020        }
1021
1022        if (!from_master && !fix_master &&
1023            (r->res_master_nodeid == from_nodeid)) {
1024                /* this can happen when the master sends remove, the dir node
1025                   finds the rsb on the keep list and ignores the remove,
1026                   and the former master sends a lookup */
1027
1028                log_limit(ls, "dlm_master_lookup from master %d flags %x "
1029                          "first %x %s", from_nodeid, flags,
1030                          r->res_first_lkid, r->res_name);
1031        }
1032
1033 out_found:
1034        *r_nodeid = r->res_master_nodeid;
1035        if (result)
1036                *result = DLM_LU_MATCH;
1037
1038        if (toss_list) {
1039                r->res_toss_time = jiffies;
1040                /* the rsb was inactive (on toss list) */
1041                spin_unlock(&ls->ls_rsbtbl[b].lock);
1042        } else {
1043                /* the rsb was active */
1044                unlock_rsb(r);
1045                put_rsb(r);
1046        }
1047        return 0;
1048
1049 not_found:
1050        error = get_rsb_struct(ls, name, len, &r);
1051        if (error == -EAGAIN) {
1052                spin_unlock(&ls->ls_rsbtbl[b].lock);
1053                goto retry;
1054        }
1055        if (error)
1056                goto out_unlock;
1057
1058        r->res_hash = hash;
1059        r->res_bucket = b;
1060        r->res_dir_nodeid = our_nodeid;
1061        r->res_master_nodeid = from_nodeid;
1062        r->res_nodeid = from_nodeid;
1063        kref_init(&r->res_ref);
1064        r->res_toss_time = jiffies;
1065
1066        error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1067        if (error) {
1068                /* should never happen */
1069                dlm_free_rsb(r);
1070                spin_unlock(&ls->ls_rsbtbl[b].lock);
1071                goto retry;
1072        }
1073
1074        if (result)
1075                *result = DLM_LU_ADD;
1076        *r_nodeid = from_nodeid;
1077        error = 0;
1078 out_unlock:
1079        spin_unlock(&ls->ls_rsbtbl[b].lock);
1080        return error;
1081}
1082
1083static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1084{
1085        struct rb_node *n;
1086        struct dlm_rsb *r;
1087        int i;
1088
1089        for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1090                spin_lock(&ls->ls_rsbtbl[i].lock);
1091                for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1092                        r = rb_entry(n, struct dlm_rsb, res_hashnode);
1093                        if (r->res_hash == hash)
1094                                dlm_dump_rsb(r);
1095                }
1096                spin_unlock(&ls->ls_rsbtbl[i].lock);
1097        }
1098}
1099
1100void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1101{
1102        struct dlm_rsb *r = NULL;
1103        uint32_t hash, b;
1104        int error;
1105
1106        hash = jhash(name, len, 0);
1107        b = hash & (ls->ls_rsbtbl_size - 1);
1108
1109        spin_lock(&ls->ls_rsbtbl[b].lock);
1110        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1111        if (!error)
1112                goto out_dump;
1113
1114        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1115        if (error)
1116                goto out;
1117 out_dump:
1118        dlm_dump_rsb(r);
1119 out:
1120        spin_unlock(&ls->ls_rsbtbl[b].lock);
1121}
1122
1123static void toss_rsb(struct kref *kref)
1124{
1125        struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1126        struct dlm_ls *ls = r->res_ls;
1127
1128        DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1129        kref_init(&r->res_ref);
1130        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1131        rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1132        r->res_toss_time = jiffies;
1133        ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1134        if (r->res_lvbptr) {
1135                dlm_free_lvb(r->res_lvbptr);
1136                r->res_lvbptr = NULL;
1137        }
1138}
1139
1140/* See comment for unhold_lkb */
1141
1142static void unhold_rsb(struct dlm_rsb *r)
1143{
1144        int rv;
1145        rv = kref_put(&r->res_ref, toss_rsb);
1146        DLM_ASSERT(!rv, dlm_dump_rsb(r););
1147}
1148
1149static void kill_rsb(struct kref *kref)
1150{
1151        struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1152
1153        /* All work is done after the return from kref_put() so we
1154           can release the write_lock before the remove and free. */
1155
1156        DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1157        DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1158        DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1159        DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1160        DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1161        DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1162}
1163
1164/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1165   The rsb must exist as long as any lkb's for it do. */
1166
1167static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1168{
1169        hold_rsb(r);
1170        lkb->lkb_resource = r;
1171}
1172
1173static void detach_lkb(struct dlm_lkb *lkb)
1174{
1175        if (lkb->lkb_resource) {
1176                put_rsb(lkb->lkb_resource);
1177                lkb->lkb_resource = NULL;
1178        }
1179}
1180
1181static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1182{
1183        struct dlm_lkb *lkb;
1184        int rv;
1185
1186        lkb = dlm_allocate_lkb(ls);
1187        if (!lkb)
1188                return -ENOMEM;
1189
1190        lkb->lkb_nodeid = -1;
1191        lkb->lkb_grmode = DLM_LOCK_IV;
1192        kref_init(&lkb->lkb_ref);
1193        INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1194        INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1195        INIT_LIST_HEAD(&lkb->lkb_time_list);
1196        INIT_LIST_HEAD(&lkb->lkb_cb_list);
1197        mutex_init(&lkb->lkb_cb_mutex);
1198        INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
1199
1200        idr_preload(GFP_NOFS);
1201        spin_lock(&ls->ls_lkbidr_spin);
1202        rv = idr_alloc(&ls->ls_lkbidr, lkb, 1, 0, GFP_NOWAIT);
1203        if (rv >= 0)
1204                lkb->lkb_id = rv;
1205        spin_unlock(&ls->ls_lkbidr_spin);
1206        idr_preload_end();
1207
1208        if (rv < 0) {
1209                log_error(ls, "create_lkb idr error %d", rv);
1210                dlm_free_lkb(lkb);
1211                return rv;
1212        }
1213
1214        *lkb_ret = lkb;
1215        return 0;
1216}
1217
1218static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1219{
1220        struct dlm_lkb *lkb;
1221
1222        spin_lock(&ls->ls_lkbidr_spin);
1223        lkb = idr_find(&ls->ls_lkbidr, lkid);
1224        if (lkb)
1225                kref_get(&lkb->lkb_ref);
1226        spin_unlock(&ls->ls_lkbidr_spin);
1227
1228        *lkb_ret = lkb;
1229        return lkb ? 0 : -ENOENT;
1230}
1231
1232static void kill_lkb(struct kref *kref)
1233{
1234        struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1235
1236        /* All work is done after the return from kref_put() so we
1237           can release the write_lock before the detach_lkb */
1238
1239        DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1240}
1241
1242/* __put_lkb() is used when an lkb may not have an rsb attached to
1243   it so we need to provide the lockspace explicitly */
1244
1245static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1246{
1247        uint32_t lkid = lkb->lkb_id;
1248
1249        spin_lock(&ls->ls_lkbidr_spin);
1250        if (kref_put(&lkb->lkb_ref, kill_lkb)) {
1251                idr_remove(&ls->ls_lkbidr, lkid);
1252                spin_unlock(&ls->ls_lkbidr_spin);
1253
1254                detach_lkb(lkb);
1255
1256                /* for local/process lkbs, lvbptr points to caller's lksb */
1257                if (lkb->lkb_lvbptr && is_master_copy(lkb))
1258                        dlm_free_lvb(lkb->lkb_lvbptr);
1259                dlm_free_lkb(lkb);
1260                return 1;
1261        } else {
1262                spin_unlock(&ls->ls_lkbidr_spin);
1263                return 0;
1264        }
1265}
1266
1267int dlm_put_lkb(struct dlm_lkb *lkb)
1268{
1269        struct dlm_ls *ls;
1270
1271        DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1272        DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1273
1274        ls = lkb->lkb_resource->res_ls;
1275        return __put_lkb(ls, lkb);
1276}
1277
1278/* This is only called to add a reference when the code already holds
1279   a valid reference to the lkb, so there's no need for locking. */
1280
1281static inline void hold_lkb(struct dlm_lkb *lkb)
1282{
1283        kref_get(&lkb->lkb_ref);
1284}
1285
1286/* This is called when we need to remove a reference and are certain
1287   it's not the last ref.  e.g. del_lkb is always called between a
1288   find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1289   put_lkb would work fine, but would involve unnecessary locking */
1290
1291static inline void unhold_lkb(struct dlm_lkb *lkb)
1292{
1293        int rv;
1294        rv = kref_put(&lkb->lkb_ref, kill_lkb);
1295        DLM_ASSERT(!rv, dlm_print_lkb(lkb););
1296}
1297
1298static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1299                            int mode)
1300{
1301        struct dlm_lkb *lkb = NULL;
1302
1303        list_for_each_entry(lkb, head, lkb_statequeue)
1304                if (lkb->lkb_rqmode < mode)
1305                        break;
1306
1307        __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
1308}
1309
1310/* add/remove lkb to rsb's grant/convert/wait queue */
1311
1312static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1313{
1314        kref_get(&lkb->lkb_ref);
1315
1316        DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1317
1318        lkb->lkb_timestamp = ktime_get();
1319
1320        lkb->lkb_status = status;
1321
1322        switch (status) {
1323        case DLM_LKSTS_WAITING:
1324                if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1325                        list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1326                else
1327                        list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1328                break;
1329        case DLM_LKSTS_GRANTED:
1330                /* convention says granted locks kept in order of grmode */
1331                lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1332                                lkb->lkb_grmode);
1333                break;
1334        case DLM_LKSTS_CONVERT:
1335                if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1336                        list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1337                else
1338                        list_add_tail(&lkb->lkb_statequeue,
1339                                      &r->res_convertqueue);
1340                break;
1341        default:
1342                DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1343        }
1344}
1345
1346static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1347{
1348        lkb->lkb_status = 0;
1349        list_del(&lkb->lkb_statequeue);
1350        unhold_lkb(lkb);
1351}
1352
1353static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1354{
1355        hold_lkb(lkb);
1356        del_lkb(r, lkb);
1357        add_lkb(r, lkb, sts);
1358        unhold_lkb(lkb);
1359}
1360
1361static int msg_reply_type(int mstype)
1362{
1363        switch (mstype) {
1364        case DLM_MSG_REQUEST:
1365                return DLM_MSG_REQUEST_REPLY;
1366        case DLM_MSG_CONVERT:
1367                return DLM_MSG_CONVERT_REPLY;
1368        case DLM_MSG_UNLOCK:
1369                return DLM_MSG_UNLOCK_REPLY;
1370        case DLM_MSG_CANCEL:
1371                return DLM_MSG_CANCEL_REPLY;
1372        case DLM_MSG_LOOKUP:
1373                return DLM_MSG_LOOKUP_REPLY;
1374        }
1375        return -1;
1376}
1377
1378static int nodeid_warned(int nodeid, int num_nodes, int *warned)
1379{
1380        int i;
1381
1382        for (i = 0; i < num_nodes; i++) {
1383                if (!warned[i]) {
1384                        warned[i] = nodeid;
1385                        return 0;
1386                }
1387                if (warned[i] == nodeid)
1388                        return 1;
1389        }
1390        return 0;
1391}
1392
1393void dlm_scan_waiters(struct dlm_ls *ls)
1394{
1395        struct dlm_lkb *lkb;
1396        s64 us;
1397        s64 debug_maxus = 0;
1398        u32 debug_scanned = 0;
1399        u32 debug_expired = 0;
1400        int num_nodes = 0;
1401        int *warned = NULL;
1402
1403        if (!dlm_config.ci_waitwarn_us)
1404                return;
1405
1406        mutex_lock(&ls->ls_waiters_mutex);
1407
1408        list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1409                if (!lkb->lkb_wait_time)
1410                        continue;
1411
1412                debug_scanned++;
1413
1414                us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
1415
1416                if (us < dlm_config.ci_waitwarn_us)
1417                        continue;
1418
1419                lkb->lkb_wait_time = 0;
1420
1421                debug_expired++;
1422                if (us > debug_maxus)
1423                        debug_maxus = us;
1424
1425                if (!num_nodes) {
1426                        num_nodes = ls->ls_num_nodes;
1427                        warned = kcalloc(num_nodes, sizeof(int), GFP_KERNEL);
1428                }
1429                if (!warned)
1430                        continue;
1431                if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
1432                        continue;
1433
1434                log_error(ls, "waitwarn %x %lld %d us check connection to "
1435                          "node %d", lkb->lkb_id, (long long)us,
1436                          dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
1437        }
1438        mutex_unlock(&ls->ls_waiters_mutex);
1439        kfree(warned);
1440
1441        if (debug_expired)
1442                log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
1443                          debug_scanned, debug_expired,
1444                          dlm_config.ci_waitwarn_us, (long long)debug_maxus);
1445}
1446
1447/* add/remove lkb from global waiters list of lkb's waiting for
1448   a reply from a remote node */
1449
1450static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1451{
1452        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1453        int error = 0;
1454
1455        mutex_lock(&ls->ls_waiters_mutex);
1456
1457        if (is_overlap_unlock(lkb) ||
1458            (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1459                error = -EINVAL;
1460                goto out;
1461        }
1462
1463        if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1464                switch (mstype) {
1465                case DLM_MSG_UNLOCK:
1466                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1467                        break;
1468                case DLM_MSG_CANCEL:
1469                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1470                        break;
1471                default:
1472                        error = -EBUSY;
1473                        goto out;
1474                }
1475                lkb->lkb_wait_count++;
1476                hold_lkb(lkb);
1477
1478                log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1479                          lkb->lkb_id, lkb->lkb_wait_type, mstype,
1480                          lkb->lkb_wait_count, lkb->lkb_flags);
1481                goto out;
1482        }
1483
1484        DLM_ASSERT(!lkb->lkb_wait_count,
1485                   dlm_print_lkb(lkb);
1486                   printk("wait_count %d\n", lkb->lkb_wait_count););
1487
1488        lkb->lkb_wait_count++;
1489        lkb->lkb_wait_type = mstype;
1490        lkb->lkb_wait_time = ktime_get();
1491        lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1492        hold_lkb(lkb);
1493        list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1494 out:
1495        if (error)
1496                log_error(ls, "addwait error %x %d flags %x %d %d %s",
1497                          lkb->lkb_id, error, lkb->lkb_flags, mstype,
1498                          lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1499        mutex_unlock(&ls->ls_waiters_mutex);
1500        return error;
1501}
1502
1503/* We clear the RESEND flag because we might be taking an lkb off the waiters
1504   list as part of process_requestqueue (e.g. a lookup that has an optimized
1505   request reply on the requestqueue) between dlm_recover_waiters_pre() which
1506   set RESEND and dlm_recover_waiters_post() */
1507
1508static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1509                                struct dlm_message *ms)
1510{
1511        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1512        int overlap_done = 0;
1513
1514        if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1515                log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1516                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1517                overlap_done = 1;
1518                goto out_del;
1519        }
1520
1521        if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1522                log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1523                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1524                overlap_done = 1;
1525                goto out_del;
1526        }
1527
1528        /* Cancel state was preemptively cleared by a successful convert,
1529           see next comment, nothing to do. */
1530
1531        if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1532            (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1533                log_debug(ls, "remwait %x cancel_reply wait_type %d",
1534                          lkb->lkb_id, lkb->lkb_wait_type);
1535                return -1;
1536        }
1537
1538        /* Remove for the convert reply, and premptively remove for the
1539           cancel reply.  A convert has been granted while there's still
1540           an outstanding cancel on it (the cancel is moot and the result
1541           in the cancel reply should be 0).  We preempt the cancel reply
1542           because the app gets the convert result and then can follow up
1543           with another op, like convert.  This subsequent op would see the
1544           lingering state of the cancel and fail with -EBUSY. */
1545
1546        if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1547            (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1548            is_overlap_cancel(lkb) && ms && !ms->m_result) {
1549                log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1550                          lkb->lkb_id);
1551                lkb->lkb_wait_type = 0;
1552                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1553                lkb->lkb_wait_count--;
1554                goto out_del;
1555        }
1556
1557        /* N.B. type of reply may not always correspond to type of original
1558           msg due to lookup->request optimization, verify others? */
1559
1560        if (lkb->lkb_wait_type) {
1561                lkb->lkb_wait_type = 0;
1562                goto out_del;
1563        }
1564
1565        log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1566                  lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid,
1567                  mstype, lkb->lkb_flags);
1568        return -1;
1569
1570 out_del:
1571        /* the force-unlock/cancel has completed and we haven't recvd a reply
1572           to the op that was in progress prior to the unlock/cancel; we
1573           give up on any reply to the earlier op.  FIXME: not sure when/how
1574           this would happen */
1575
1576        if (overlap_done && lkb->lkb_wait_type) {
1577                log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1578                          lkb->lkb_id, mstype, lkb->lkb_wait_type);
1579                lkb->lkb_wait_count--;
1580                lkb->lkb_wait_type = 0;
1581        }
1582
1583        DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1584
1585        lkb->lkb_flags &= ~DLM_IFL_RESEND;
1586        lkb->lkb_wait_count--;
1587        if (!lkb->lkb_wait_count)
1588                list_del_init(&lkb->lkb_wait_reply);
1589        unhold_lkb(lkb);
1590        return 0;
1591}
1592
1593static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1594{
1595        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1596        int error;
1597
1598        mutex_lock(&ls->ls_waiters_mutex);
1599        error = _remove_from_waiters(lkb, mstype, NULL);
1600        mutex_unlock(&ls->ls_waiters_mutex);
1601        return error;
1602}
1603
1604/* Handles situations where we might be processing a "fake" or "stub" reply in
1605   which we can't try to take waiters_mutex again. */
1606
1607static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1608{
1609        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1610        int error;
1611
1612        if (ms->m_flags != DLM_IFL_STUB_MS)
1613                mutex_lock(&ls->ls_waiters_mutex);
1614        error = _remove_from_waiters(lkb, ms->m_type, ms);
1615        if (ms->m_flags != DLM_IFL_STUB_MS)
1616                mutex_unlock(&ls->ls_waiters_mutex);
1617        return error;
1618}
1619
1620/* If there's an rsb for the same resource being removed, ensure
1621   that the remove message is sent before the new lookup message.
1622   It should be rare to need a delay here, but if not, then it may
1623   be worthwhile to add a proper wait mechanism rather than a delay. */
1624
1625static void wait_pending_remove(struct dlm_rsb *r)
1626{
1627        struct dlm_ls *ls = r->res_ls;
1628 restart:
1629        spin_lock(&ls->ls_remove_spin);
1630        if (ls->ls_remove_len &&
1631            !rsb_cmp(r, ls->ls_remove_name, ls->ls_remove_len)) {
1632                log_debug(ls, "delay lookup for remove dir %d %s",
1633                          r->res_dir_nodeid, r->res_name);
1634                spin_unlock(&ls->ls_remove_spin);
1635                msleep(1);
1636                goto restart;
1637        }
1638        spin_unlock(&ls->ls_remove_spin);
1639}
1640
1641/*
1642 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1643 * read by other threads in wait_pending_remove.  ls_remove_names
1644 * and ls_remove_lens are only used by the scan thread, so they do
1645 * not need protection.
1646 */
1647
1648static void shrink_bucket(struct dlm_ls *ls, int b)
1649{
1650        struct rb_node *n, *next;
1651        struct dlm_rsb *r;
1652        char *name;
1653        int our_nodeid = dlm_our_nodeid();
1654        int remote_count = 0;
1655        int need_shrink = 0;
1656        int i, len, rv;
1657
1658        memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1659
1660        spin_lock(&ls->ls_rsbtbl[b].lock);
1661
1662        if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1663                spin_unlock(&ls->ls_rsbtbl[b].lock);
1664                return;
1665        }
1666
1667        for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1668                next = rb_next(n);
1669                r = rb_entry(n, struct dlm_rsb, res_hashnode);
1670
1671                /* If we're the directory record for this rsb, and
1672                   we're not the master of it, then we need to wait
1673                   for the master node to send us a dir remove for
1674                   before removing the dir record. */
1675
1676                if (!dlm_no_directory(ls) &&
1677                    (r->res_master_nodeid != our_nodeid) &&
1678                    (dlm_dir_nodeid(r) == our_nodeid)) {
1679                        continue;
1680                }
1681
1682                need_shrink = 1;
1683
1684                if (!time_after_eq(jiffies, r->res_toss_time +
1685                                   dlm_config.ci_toss_secs * HZ)) {
1686                        continue;
1687                }
1688
1689                if (!dlm_no_directory(ls) &&
1690                    (r->res_master_nodeid == our_nodeid) &&
1691                    (dlm_dir_nodeid(r) != our_nodeid)) {
1692
1693                        /* We're the master of this rsb but we're not
1694                           the directory record, so we need to tell the
1695                           dir node to remove the dir record. */
1696
1697                        ls->ls_remove_lens[remote_count] = r->res_length;
1698                        memcpy(ls->ls_remove_names[remote_count], r->res_name,
1699                               DLM_RESNAME_MAXLEN);
1700                        remote_count++;
1701
1702                        if (remote_count >= DLM_REMOVE_NAMES_MAX)
1703                                break;
1704                        continue;
1705                }
1706
1707                if (!kref_put(&r->res_ref, kill_rsb)) {
1708                        log_error(ls, "tossed rsb in use %s", r->res_name);
1709                        continue;
1710                }
1711
1712                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1713                dlm_free_rsb(r);
1714        }
1715
1716        if (need_shrink)
1717                ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1718        else
1719                ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1720        spin_unlock(&ls->ls_rsbtbl[b].lock);
1721
1722        /*
1723         * While searching for rsb's to free, we found some that require
1724         * remote removal.  We leave them in place and find them again here
1725         * so there is a very small gap between removing them from the toss
1726         * list and sending the removal.  Keeping this gap small is
1727         * important to keep us (the master node) from being out of sync
1728         * with the remote dir node for very long.
1729         *
1730         * From the time the rsb is removed from toss until just after
1731         * send_remove, the rsb name is saved in ls_remove_name.  A new
1732         * lookup checks this to ensure that a new lookup message for the
1733         * same resource name is not sent just before the remove message.
1734         */
1735
1736        for (i = 0; i < remote_count; i++) {
1737                name = ls->ls_remove_names[i];
1738                len = ls->ls_remove_lens[i];
1739
1740                spin_lock(&ls->ls_rsbtbl[b].lock);
1741                rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1742                if (rv) {
1743                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1744                        log_debug(ls, "remove_name not toss %s", name);
1745                        continue;
1746                }
1747
1748                if (r->res_master_nodeid != our_nodeid) {
1749                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1750                        log_debug(ls, "remove_name master %d dir %d our %d %s",
1751                                  r->res_master_nodeid, r->res_dir_nodeid,
1752                                  our_nodeid, name);
1753                        continue;
1754                }
1755
1756                if (r->res_dir_nodeid == our_nodeid) {
1757                        /* should never happen */
1758                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1759                        log_error(ls, "remove_name dir %d master %d our %d %s",
1760                                  r->res_dir_nodeid, r->res_master_nodeid,
1761                                  our_nodeid, name);
1762                        continue;
1763                }
1764
1765                if (!time_after_eq(jiffies, r->res_toss_time +
1766                                   dlm_config.ci_toss_secs * HZ)) {
1767                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1768                        log_debug(ls, "remove_name toss_time %lu now %lu %s",
1769                                  r->res_toss_time, jiffies, name);
1770                        continue;
1771                }
1772
1773                if (!kref_put(&r->res_ref, kill_rsb)) {
1774                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1775                        log_error(ls, "remove_name in use %s", name);
1776                        continue;
1777                }
1778
1779                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1780
1781                /* block lookup of same name until we've sent remove */
1782                spin_lock(&ls->ls_remove_spin);
1783                ls->ls_remove_len = len;
1784                memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1785                spin_unlock(&ls->ls_remove_spin);
1786                spin_unlock(&ls->ls_rsbtbl[b].lock);
1787
1788                send_remove(r);
1789
1790                /* allow lookup of name again */
1791                spin_lock(&ls->ls_remove_spin);
1792                ls->ls_remove_len = 0;
1793                memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1794                spin_unlock(&ls->ls_remove_spin);
1795
1796                dlm_free_rsb(r);
1797        }
1798}
1799
1800void dlm_scan_rsbs(struct dlm_ls *ls)
1801{
1802        int i;
1803
1804        for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1805                shrink_bucket(ls, i);
1806                if (dlm_locking_stopped(ls))
1807                        break;
1808                cond_resched();
1809        }
1810}
1811
1812static void add_timeout(struct dlm_lkb *lkb)
1813{
1814        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1815
1816        if (is_master_copy(lkb))
1817                return;
1818
1819        if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1820            !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1821                lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1822                goto add_it;
1823        }
1824        if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1825                goto add_it;
1826        return;
1827
1828 add_it:
1829        DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1830        mutex_lock(&ls->ls_timeout_mutex);
1831        hold_lkb(lkb);
1832        list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1833        mutex_unlock(&ls->ls_timeout_mutex);
1834}
1835
1836static void del_timeout(struct dlm_lkb *lkb)
1837{
1838        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1839
1840        mutex_lock(&ls->ls_timeout_mutex);
1841        if (!list_empty(&lkb->lkb_time_list)) {
1842                list_del_init(&lkb->lkb_time_list);
1843                unhold_lkb(lkb);
1844        }
1845        mutex_unlock(&ls->ls_timeout_mutex);
1846}
1847
1848/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1849   lkb_lksb_timeout without lock_rsb?  Note: we can't lock timeout_mutex
1850   and then lock rsb because of lock ordering in add_timeout.  We may need
1851   to specify some special timeout-related bits in the lkb that are just to
1852   be accessed under the timeout_mutex. */
1853
1854void dlm_scan_timeout(struct dlm_ls *ls)
1855{
1856        struct dlm_rsb *r;
1857        struct dlm_lkb *lkb;
1858        int do_cancel, do_warn;
1859        s64 wait_us;
1860
1861        for (;;) {
1862                if (dlm_locking_stopped(ls))
1863                        break;
1864
1865                do_cancel = 0;
1866                do_warn = 0;
1867                mutex_lock(&ls->ls_timeout_mutex);
1868                list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1869
1870                        wait_us = ktime_to_us(ktime_sub(ktime_get(),
1871                                                        lkb->lkb_timestamp));
1872
1873                        if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1874                            wait_us >= (lkb->lkb_timeout_cs * 10000))
1875                                do_cancel = 1;
1876
1877                        if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1878                            wait_us >= dlm_config.ci_timewarn_cs * 10000)
1879                                do_warn = 1;
1880
1881                        if (!do_cancel && !do_warn)
1882                                continue;
1883                        hold_lkb(lkb);
1884                        break;
1885                }
1886                mutex_unlock(&ls->ls_timeout_mutex);
1887
1888                if (!do_cancel && !do_warn)
1889                        break;
1890
1891                r = lkb->lkb_resource;
1892                hold_rsb(r);
1893                lock_rsb(r);
1894
1895                if (do_warn) {
1896                        /* clear flag so we only warn once */
1897                        lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1898                        if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1899                                del_timeout(lkb);
1900                        dlm_timeout_warn(lkb);
1901                }
1902
1903                if (do_cancel) {
1904                        log_debug(ls, "timeout cancel %x node %d %s",
1905                                  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1906                        lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1907                        lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1908                        del_timeout(lkb);
1909                        _cancel_lock(r, lkb);
1910                }
1911
1912                unlock_rsb(r);
1913                unhold_rsb(r);
1914                dlm_put_lkb(lkb);
1915        }
1916}
1917
1918/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1919   dlm_recoverd before checking/setting ls_recover_begin. */
1920
1921void dlm_adjust_timeouts(struct dlm_ls *ls)
1922{
1923        struct dlm_lkb *lkb;
1924        u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1925
1926        ls->ls_recover_begin = 0;
1927        mutex_lock(&ls->ls_timeout_mutex);
1928        list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1929                lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1930        mutex_unlock(&ls->ls_timeout_mutex);
1931
1932        if (!dlm_config.ci_waitwarn_us)
1933                return;
1934
1935        mutex_lock(&ls->ls_waiters_mutex);
1936        list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1937                if (ktime_to_us(lkb->lkb_wait_time))
1938                        lkb->lkb_wait_time = ktime_get();
1939        }
1940        mutex_unlock(&ls->ls_waiters_mutex);
1941}
1942
1943/* lkb is master or local copy */
1944
1945static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1946{
1947        int b, len = r->res_ls->ls_lvblen;
1948
1949        /* b=1 lvb returned to caller
1950           b=0 lvb written to rsb or invalidated
1951           b=-1 do nothing */
1952
1953        b =  dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1954
1955        if (b == 1) {
1956                if (!lkb->lkb_lvbptr)
1957                        return;
1958
1959                if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1960                        return;
1961
1962                if (!r->res_lvbptr)
1963                        return;
1964
1965                memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1966                lkb->lkb_lvbseq = r->res_lvbseq;
1967
1968        } else if (b == 0) {
1969                if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1970                        rsb_set_flag(r, RSB_VALNOTVALID);
1971                        return;
1972                }
1973
1974                if (!lkb->lkb_lvbptr)
1975                        return;
1976
1977                if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1978                        return;
1979
1980                if (!r->res_lvbptr)
1981                        r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1982
1983                if (!r->res_lvbptr)
1984                        return;
1985
1986                memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1987                r->res_lvbseq++;
1988                lkb->lkb_lvbseq = r->res_lvbseq;
1989                rsb_clear_flag(r, RSB_VALNOTVALID);
1990        }
1991
1992        if (rsb_flag(r, RSB_VALNOTVALID))
1993                lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1994}
1995
1996static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1997{
1998        if (lkb->lkb_grmode < DLM_LOCK_PW)
1999                return;
2000
2001        if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2002                rsb_set_flag(r, RSB_VALNOTVALID);
2003                return;
2004        }
2005
2006        if (!lkb->lkb_lvbptr)
2007                return;
2008
2009        if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2010                return;
2011
2012        if (!r->res_lvbptr)
2013                r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
2014
2015        if (!r->res_lvbptr)
2016                return;
2017
2018        memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2019        r->res_lvbseq++;
2020        rsb_clear_flag(r, RSB_VALNOTVALID);
2021}
2022
2023/* lkb is process copy (pc) */
2024
2025static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2026                            struct dlm_message *ms)
2027{
2028        int b;
2029
2030        if (!lkb->lkb_lvbptr)
2031                return;
2032
2033        if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2034                return;
2035
2036        b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2037        if (b == 1) {
2038                int len = receive_extralen(ms);
2039                if (len > r->res_ls->ls_lvblen)
2040                        len = r->res_ls->ls_lvblen;
2041                memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2042                lkb->lkb_lvbseq = ms->m_lvbseq;
2043        }
2044}
2045
2046/* Manipulate lkb's on rsb's convert/granted/waiting queues
2047   remove_lock -- used for unlock, removes lkb from granted
2048   revert_lock -- used for cancel, moves lkb from convert to granted
2049   grant_lock  -- used for request and convert, adds lkb to granted or
2050                  moves lkb from convert or waiting to granted
2051
2052   Each of these is used for master or local copy lkb's.  There is
2053   also a _pc() variation used to make the corresponding change on
2054   a process copy (pc) lkb. */
2055
2056static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2057{
2058        del_lkb(r, lkb);
2059        lkb->lkb_grmode = DLM_LOCK_IV;
2060        /* this unhold undoes the original ref from create_lkb()
2061           so this leads to the lkb being freed */
2062        unhold_lkb(lkb);
2063}
2064
2065static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2066{
2067        set_lvb_unlock(r, lkb);
2068        _remove_lock(r, lkb);
2069}
2070
2071static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2072{
2073        _remove_lock(r, lkb);
2074}
2075
2076/* returns: 0 did nothing
2077            1 moved lock to granted
2078           -1 removed lock */
2079
2080static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2081{
2082        int rv = 0;
2083
2084        lkb->lkb_rqmode = DLM_LOCK_IV;
2085
2086        switch (lkb->lkb_status) {
2087        case DLM_LKSTS_GRANTED:
2088                break;
2089        case DLM_LKSTS_CONVERT:
2090                move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2091                rv = 1;
2092                break;
2093        case DLM_LKSTS_WAITING:
2094                del_lkb(r, lkb);
2095                lkb->lkb_grmode = DLM_LOCK_IV;
2096                /* this unhold undoes the original ref from create_lkb()
2097                   so this leads to the lkb being freed */
2098                unhold_lkb(lkb);
2099                rv = -1;
2100                break;
2101        default:
2102                log_print("invalid status for revert %d", lkb->lkb_status);
2103        }
2104        return rv;
2105}
2106
2107static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2108{
2109        return revert_lock(r, lkb);
2110}
2111
2112static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2113{
2114        if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2115                lkb->lkb_grmode = lkb->lkb_rqmode;
2116                if (lkb->lkb_status)
2117                        move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2118                else
2119                        add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2120        }
2121
2122        lkb->lkb_rqmode = DLM_LOCK_IV;
2123        lkb->lkb_highbast = 0;
2124}
2125
2126static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2127{
2128        set_lvb_lock(r, lkb);
2129        _grant_lock(r, lkb);
2130}
2131
2132static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2133                          struct dlm_message *ms)
2134{
2135        set_lvb_lock_pc(r, lkb, ms);
2136        _grant_lock(r, lkb);
2137}
2138
2139/* called by grant_pending_locks() which means an async grant message must
2140   be sent to the requesting node in addition to granting the lock if the
2141   lkb belongs to a remote node. */
2142
2143static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2144{
2145        grant_lock(r, lkb);
2146        if (is_master_copy(lkb))
2147                send_grant(r, lkb);
2148        else
2149                queue_cast(r, lkb, 0);
2150}
2151
2152/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2153   change the granted/requested modes.  We're munging things accordingly in
2154   the process copy.
2155   CONVDEADLK: our grmode may have been forced down to NL to resolve a
2156   conversion deadlock
2157   ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2158   compatible with other granted locks */
2159
2160static void munge_demoted(struct dlm_lkb *lkb)
2161{
2162        if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2163                log_print("munge_demoted %x invalid modes gr %d rq %d",
2164                          lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2165                return;
2166        }
2167
2168        lkb->lkb_grmode = DLM_LOCK_NL;
2169}
2170
2171static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2172{
2173        if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
2174            ms->m_type != DLM_MSG_GRANT) {
2175                log_print("munge_altmode %x invalid reply type %d",
2176                          lkb->lkb_id, ms->m_type);
2177                return;
2178        }
2179
2180        if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2181                lkb->lkb_rqmode = DLM_LOCK_PR;
2182        else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2183                lkb->lkb_rqmode = DLM_LOCK_CW;
2184        else {
2185                log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2186                dlm_print_lkb(lkb);
2187        }
2188}
2189
2190static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2191{
2192        struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2193                                           lkb_statequeue);
2194        if (lkb->lkb_id == first->lkb_id)
2195                return 1;
2196
2197        return 0;
2198}
2199
2200/* Check if the given lkb conflicts with another lkb on the queue. */
2201
2202static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2203{
2204        struct dlm_lkb *this;
2205
2206        list_for_each_entry(this, head, lkb_statequeue) {
2207                if (this == lkb)
2208                        continue;
2209                if (!modes_compat(this, lkb))
2210                        return 1;
2211        }
2212        return 0;
2213}
2214
2215/*
2216 * "A conversion deadlock arises with a pair of lock requests in the converting
2217 * queue for one resource.  The granted mode of each lock blocks the requested
2218 * mode of the other lock."
2219 *
2220 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2221 * convert queue from being granted, then deadlk/demote lkb.
2222 *
2223 * Example:
2224 * Granted Queue: empty
2225 * Convert Queue: NL->EX (first lock)
2226 *                PR->EX (second lock)
2227 *
2228 * The first lock can't be granted because of the granted mode of the second
2229 * lock and the second lock can't be granted because it's not first in the
2230 * list.  We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2231 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2232 * flag set and return DEMOTED in the lksb flags.
2233 *
2234 * Originally, this function detected conv-deadlk in a more limited scope:
2235 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2236 * - if lkb1 was the first entry in the queue (not just earlier), and was
2237 *   blocked by the granted mode of lkb2, and there was nothing on the
2238 *   granted queue preventing lkb1 from being granted immediately, i.e.
2239 *   lkb2 was the only thing preventing lkb1 from being granted.
2240 *
2241 * That second condition meant we'd only say there was conv-deadlk if
2242 * resolving it (by demotion) would lead to the first lock on the convert
2243 * queue being granted right away.  It allowed conversion deadlocks to exist
2244 * between locks on the convert queue while they couldn't be granted anyway.
2245 *
2246 * Now, we detect and take action on conversion deadlocks immediately when
2247 * they're created, even if they may not be immediately consequential.  If
2248 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2249 * mode that would prevent lkb1's conversion from being granted, we do a
2250 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2251 * I think this means that the lkb_is_ahead condition below should always
2252 * be zero, i.e. there will never be conv-deadlk between two locks that are
2253 * both already on the convert queue.
2254 */
2255
2256static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2257{
2258        struct dlm_lkb *lkb1;
2259        int lkb_is_ahead = 0;
2260
2261        list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2262                if (lkb1 == lkb2) {
2263                        lkb_is_ahead = 1;
2264                        continue;
2265                }
2266
2267                if (!lkb_is_ahead) {
2268                        if (!modes_compat(lkb2, lkb1))
2269                                return 1;
2270                } else {
2271                        if (!modes_compat(lkb2, lkb1) &&
2272                            !modes_compat(lkb1, lkb2))
2273                                return 1;
2274                }
2275        }
2276        return 0;
2277}
2278
2279/*
2280 * Return 1 if the lock can be granted, 0 otherwise.
2281 * Also detect and resolve conversion deadlocks.
2282 *
2283 * lkb is the lock to be granted
2284 *
2285 * now is 1 if the function is being called in the context of the
2286 * immediate request, it is 0 if called later, after the lock has been
2287 * queued.
2288 *
2289 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2290 * after recovery.
2291 *
2292 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2293 */
2294
2295static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2296                           int recover)
2297{
2298        int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2299
2300        /*
2301         * 6-10: Version 5.4 introduced an option to address the phenomenon of
2302         * a new request for a NL mode lock being blocked.
2303         *
2304         * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2305         * request, then it would be granted.  In essence, the use of this flag
2306         * tells the Lock Manager to expedite theis request by not considering
2307         * what may be in the CONVERTING or WAITING queues...  As of this
2308         * writing, the EXPEDITE flag can be used only with new requests for NL
2309         * mode locks.  This flag is not valid for conversion requests.
2310         *
2311         * A shortcut.  Earlier checks return an error if EXPEDITE is used in a
2312         * conversion or used with a non-NL requested mode.  We also know an
2313         * EXPEDITE request is always granted immediately, so now must always
2314         * be 1.  The full condition to grant an expedite request: (now &&
2315         * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2316         * therefore be shortened to just checking the flag.
2317         */
2318
2319        if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2320                return 1;
2321
2322        /*
2323         * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2324         * added to the remaining conditions.
2325         */
2326
2327        if (queue_conflict(&r->res_grantqueue, lkb))
2328                return 0;
2329
2330        /*
2331         * 6-3: By default, a conversion request is immediately granted if the
2332         * requested mode is compatible with the modes of all other granted
2333         * locks
2334         */
2335
2336        if (queue_conflict(&r->res_convertqueue, lkb))
2337                return 0;
2338
2339        /*
2340         * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2341         * locks for a recovered rsb, on which lkb's have been rebuilt.
2342         * The lkb's may have been rebuilt on the queues in a different
2343         * order than they were in on the previous master.  So, granting
2344         * queued conversions in order after recovery doesn't make sense
2345         * since the order hasn't been preserved anyway.  The new order
2346         * could also have created a new "in place" conversion deadlock.
2347         * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2348         * After recovery, there would be no granted locks, and possibly
2349         * NL->EX, PR->EX, an in-place conversion deadlock.)  So, after
2350         * recovery, grant conversions without considering order.
2351         */
2352
2353        if (conv && recover)
2354                return 1;
2355
2356        /*
2357         * 6-5: But the default algorithm for deciding whether to grant or
2358         * queue conversion requests does not by itself guarantee that such
2359         * requests are serviced on a "first come first serve" basis.  This, in
2360         * turn, can lead to a phenomenon known as "indefinate postponement".
2361         *
2362         * 6-7: This issue is dealt with by using the optional QUECVT flag with
2363         * the system service employed to request a lock conversion.  This flag
2364         * forces certain conversion requests to be queued, even if they are
2365         * compatible with the granted modes of other locks on the same
2366         * resource.  Thus, the use of this flag results in conversion requests
2367         * being ordered on a "first come first servce" basis.
2368         *
2369         * DCT: This condition is all about new conversions being able to occur
2370         * "in place" while the lock remains on the granted queue (assuming
2371         * nothing else conflicts.)  IOW if QUECVT isn't set, a conversion
2372         * doesn't _have_ to go onto the convert queue where it's processed in
2373         * order.  The "now" variable is necessary to distinguish converts
2374         * being received and processed for the first time now, because once a
2375         * convert is moved to the conversion queue the condition below applies
2376         * requiring fifo granting.
2377         */
2378
2379        if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2380                return 1;
2381
2382        /*
2383         * Even if the convert is compat with all granted locks,
2384         * QUECVT forces it behind other locks on the convert queue.
2385         */
2386
2387        if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2388                if (list_empty(&r->res_convertqueue))
2389                        return 1;
2390                else
2391                        return 0;
2392        }
2393
2394        /*
2395         * The NOORDER flag is set to avoid the standard vms rules on grant
2396         * order.
2397         */
2398
2399        if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2400                return 1;
2401
2402        /*
2403         * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2404         * granted until all other conversion requests ahead of it are granted
2405         * and/or canceled.
2406         */
2407
2408        if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2409                return 1;
2410
2411        /*
2412         * 6-4: By default, a new request is immediately granted only if all
2413         * three of the following conditions are satisfied when the request is
2414         * issued:
2415         * - The queue of ungranted conversion requests for the resource is
2416         *   empty.
2417         * - The queue of ungranted new requests for the resource is empty.
2418         * - The mode of the new request is compatible with the most
2419         *   restrictive mode of all granted locks on the resource.
2420         */
2421
2422        if (now && !conv && list_empty(&r->res_convertqueue) &&
2423            list_empty(&r->res_waitqueue))
2424                return 1;
2425
2426        /*
2427         * 6-4: Once a lock request is in the queue of ungranted new requests,
2428         * it cannot be granted until the queue of ungranted conversion
2429         * requests is empty, all ungranted new requests ahead of it are
2430         * granted and/or canceled, and it is compatible with the granted mode
2431         * of the most restrictive lock granted on the resource.
2432         */
2433
2434        if (!now && !conv && list_empty(&r->res_convertqueue) &&
2435            first_in_list(lkb, &r->res_waitqueue))
2436                return 1;
2437
2438        return 0;
2439}
2440
2441static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2442                          int recover, int *err)
2443{
2444        int rv;
2445        int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2446        int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2447
2448        if (err)
2449                *err = 0;
2450
2451        rv = _can_be_granted(r, lkb, now, recover);
2452        if (rv)
2453                goto out;
2454
2455        /*
2456         * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2457         * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2458         * cancels one of the locks.
2459         */
2460
2461        if (is_convert && can_be_queued(lkb) &&
2462            conversion_deadlock_detect(r, lkb)) {
2463                if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2464                        lkb->lkb_grmode = DLM_LOCK_NL;
2465                        lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2466                } else if (err) {
2467                        *err = -EDEADLK;
2468                } else {
2469                        log_print("can_be_granted deadlock %x now %d",
2470                                  lkb->lkb_id, now);
2471                        dlm_dump_rsb(r);
2472                }
2473                goto out;
2474        }
2475
2476        /*
2477         * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2478         * to grant a request in a mode other than the normal rqmode.  It's a
2479         * simple way to provide a big optimization to applications that can
2480         * use them.
2481         */
2482
2483        if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2484                alt = DLM_LOCK_PR;
2485        else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2486                alt = DLM_LOCK_CW;
2487
2488        if (alt) {
2489                lkb->lkb_rqmode = alt;
2490                rv = _can_be_granted(r, lkb, now, 0);
2491                if (rv)
2492                        lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2493                else
2494                        lkb->lkb_rqmode = rqmode;
2495        }
2496 out:
2497        return rv;
2498}
2499
2500/* Returns the highest requested mode of all blocked conversions; sets
2501   cw if there's a blocked conversion to DLM_LOCK_CW. */
2502
2503static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2504                                 unsigned int *count)
2505{
2506        struct dlm_lkb *lkb, *s;
2507        int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2508        int hi, demoted, quit, grant_restart, demote_restart;
2509        int deadlk;
2510
2511        quit = 0;
2512 restart:
2513        grant_restart = 0;
2514        demote_restart = 0;
2515        hi = DLM_LOCK_IV;
2516
2517        list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2518                demoted = is_demoted(lkb);
2519                deadlk = 0;
2520
2521                if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2522                        grant_lock_pending(r, lkb);
2523                        grant_restart = 1;
2524                        if (count)
2525                                (*count)++;
2526                        continue;
2527                }
2528
2529                if (!demoted && is_demoted(lkb)) {
2530                        log_print("WARN: pending demoted %x node %d %s",
2531                                  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2532                        demote_restart = 1;
2533                        continue;
2534                }
2535
2536                if (deadlk) {
2537                        /*
2538                         * If DLM_LKB_NODLKWT flag is set and conversion
2539                         * deadlock is detected, we request blocking AST and
2540                         * down (or cancel) conversion.
2541                         */
2542                        if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2543                                if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2544                                        queue_bast(r, lkb, lkb->lkb_rqmode);
2545                                        lkb->lkb_highbast = lkb->lkb_rqmode;
2546                                }
2547                        } else {
2548                                log_print("WARN: pending deadlock %x node %d %s",
2549                                          lkb->lkb_id, lkb->lkb_nodeid,
2550                                          r->res_name);
2551                                dlm_dump_rsb(r);
2552                        }
2553                        continue;
2554                }
2555
2556                hi = max_t(int, lkb->lkb_rqmode, hi);
2557
2558                if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2559                        *cw = 1;
2560        }
2561
2562        if (grant_restart)
2563                goto restart;
2564        if (demote_restart && !quit) {
2565                quit = 1;
2566                goto restart;
2567        }
2568
2569        return max_t(int, high, hi);
2570}
2571
2572static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2573                              unsigned int *count)
2574{
2575        struct dlm_lkb *lkb, *s;
2576
2577        list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2578                if (can_be_granted(r, lkb, 0, 0, NULL)) {
2579                        grant_lock_pending(r, lkb);
2580                        if (count)
2581                                (*count)++;
2582                } else {
2583                        high = max_t(int, lkb->lkb_rqmode, high);
2584                        if (lkb->lkb_rqmode == DLM_LOCK_CW)
2585                                *cw = 1;
2586                }
2587        }
2588
2589        return high;
2590}
2591
2592/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2593   on either the convert or waiting queue.
2594   high is the largest rqmode of all locks blocked on the convert or
2595   waiting queue. */
2596
2597static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2598{
2599        if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2600                if (gr->lkb_highbast < DLM_LOCK_EX)
2601                        return 1;
2602                return 0;
2603        }
2604
2605        if (gr->lkb_highbast < high &&
2606            !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2607                return 1;
2608        return 0;
2609}
2610
2611static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2612{
2613        struct dlm_lkb *lkb, *s;
2614        int high = DLM_LOCK_IV;
2615        int cw = 0;
2616
2617        if (!is_master(r)) {
2618                log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2619                dlm_dump_rsb(r);
2620                return;
2621        }
2622
2623        high = grant_pending_convert(r, high, &cw, count);
2624        high = grant_pending_wait(r, high, &cw, count);
2625
2626        if (high == DLM_LOCK_IV)
2627                return;
2628
2629        /*
2630         * If there are locks left on the wait/convert queue then send blocking
2631         * ASTs to granted locks based on the largest requested mode (high)
2632         * found above.
2633         */
2634
2635        list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2636                if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2637                        if (cw && high == DLM_LOCK_PR &&
2638                            lkb->lkb_grmode == DLM_LOCK_PR)
2639                                queue_bast(r, lkb, DLM_LOCK_CW);
2640                        else
2641                                queue_bast(r, lkb, high);
2642                        lkb->lkb_highbast = high;
2643                }
2644        }
2645}
2646
2647static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2648{
2649        if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2650            (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2651                if (gr->lkb_highbast < DLM_LOCK_EX)
2652                        return 1;
2653                return 0;
2654        }
2655
2656        if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2657                return 1;
2658        return 0;
2659}
2660
2661static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2662                            struct dlm_lkb *lkb)
2663{
2664        struct dlm_lkb *gr;
2665
2666        list_for_each_entry(gr, head, lkb_statequeue) {
2667                /* skip self when sending basts to convertqueue */
2668                if (gr == lkb)
2669                        continue;
2670                if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2671                        queue_bast(r, gr, lkb->lkb_rqmode);
2672                        gr->lkb_highbast = lkb->lkb_rqmode;
2673                }
2674        }
2675}
2676
2677static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2678{
2679        send_bast_queue(r, &r->res_grantqueue, lkb);
2680}
2681
2682static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2683{
2684        send_bast_queue(r, &r->res_grantqueue, lkb);
2685        send_bast_queue(r, &r->res_convertqueue, lkb);
2686}
2687
2688/* set_master(r, lkb) -- set the master nodeid of a resource
2689
2690   The purpose of this function is to set the nodeid field in the given
2691   lkb using the nodeid field in the given rsb.  If the rsb's nodeid is
2692   known, it can just be copied to the lkb and the function will return
2693   0.  If the rsb's nodeid is _not_ known, it needs to be looked up
2694   before it can be copied to the lkb.
2695
2696   When the rsb nodeid is being looked up remotely, the initial lkb
2697   causing the lookup is kept on the ls_waiters list waiting for the
2698   lookup reply.  Other lkb's waiting for the same rsb lookup are kept
2699   on the rsb's res_lookup list until the master is verified.
2700
2701   Return values:
2702   0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2703   1: the rsb master is not available and the lkb has been placed on
2704      a wait queue
2705*/
2706
2707static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2708{
2709        int our_nodeid = dlm_our_nodeid();
2710
2711        if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2712                rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2713                r->res_first_lkid = lkb->lkb_id;
2714                lkb->lkb_nodeid = r->res_nodeid;
2715                return 0;
2716        }
2717
2718        if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2719                list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2720                return 1;
2721        }
2722
2723        if (r->res_master_nodeid == our_nodeid) {
2724                lkb->lkb_nodeid = 0;
2725                return 0;
2726        }
2727
2728        if (r->res_master_nodeid) {
2729                lkb->lkb_nodeid = r->res_master_nodeid;
2730                return 0;
2731        }
2732
2733        if (dlm_dir_nodeid(r) == our_nodeid) {
2734                /* This is a somewhat unusual case; find_rsb will usually
2735                   have set res_master_nodeid when dir nodeid is local, but
2736                   there are cases where we become the dir node after we've
2737                   past find_rsb and go through _request_lock again.
2738                   confirm_master() or process_lookup_list() needs to be
2739                   called after this. */
2740                log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2741                          lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2742                          r->res_name);
2743                r->res_master_nodeid = our_nodeid;
2744                r->res_nodeid = 0;
2745                lkb->lkb_nodeid = 0;
2746                return 0;
2747        }
2748
2749        wait_pending_remove(r);
2750
2751        r->res_first_lkid = lkb->lkb_id;
2752        send_lookup(r, lkb);
2753        return 1;
2754}
2755
2756static void process_lookup_list(struct dlm_rsb *r)
2757{
2758        struct dlm_lkb *lkb, *safe;
2759
2760        list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2761                list_del_init(&lkb->lkb_rsb_lookup);
2762                _request_lock(r, lkb);
2763                schedule();
2764        }
2765}
2766
2767/* confirm_master -- confirm (or deny) an rsb's master nodeid */
2768
2769static void confirm_master(struct dlm_rsb *r, int error)
2770{
2771        struct dlm_lkb *lkb;
2772
2773        if (!r->res_first_lkid)
2774                return;
2775
2776        switch (error) {
2777        case 0:
2778        case -EINPROGRESS:
2779                r->res_first_lkid = 0;
2780                process_lookup_list(r);
2781                break;
2782
2783        case -EAGAIN:
2784        case -EBADR:
2785        case -ENOTBLK:
2786                /* the remote request failed and won't be retried (it was
2787                   a NOQUEUE, or has been canceled/unlocked); make a waiting
2788                   lkb the first_lkid */
2789
2790                r->res_first_lkid = 0;
2791
2792                if (!list_empty(&r->res_lookup)) {
2793                        lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2794                                         lkb_rsb_lookup);
2795                        list_del_init(&lkb->lkb_rsb_lookup);
2796                        r->res_first_lkid = lkb->lkb_id;
2797                        _request_lock(r, lkb);
2798                }
2799                break;
2800
2801        default:
2802                log_error(r->res_ls, "confirm_master unknown error %d", error);
2803        }
2804}
2805
2806static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2807                         int namelen, unsigned long timeout_cs,
2808                         void (*ast) (void *astparam),
2809                         void *astparam,
2810                         void (*bast) (void *astparam, int mode),
2811                         struct dlm_args *args)
2812{
2813        int rv = -EINVAL;
2814
2815        /* check for invalid arg usage */
2816
2817        if (mode < 0 || mode > DLM_LOCK_EX)
2818                goto out;
2819
2820        if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2821                goto out;
2822
2823        if (flags & DLM_LKF_CANCEL)
2824                goto out;
2825
2826        if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2827                goto out;
2828
2829        if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2830                goto out;
2831
2832        if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2833                goto out;
2834
2835        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2836                goto out;
2837
2838        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2839                goto out;
2840
2841        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2842                goto out;
2843
2844        if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2845                goto out;
2846
2847        if (!ast || !lksb)
2848                goto out;
2849
2850        if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2851                goto out;
2852
2853        if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2854                goto out;
2855
2856        /* these args will be copied to the lkb in validate_lock_args,
2857           it cannot be done now because when converting locks, fields in
2858           an active lkb cannot be modified before locking the rsb */
2859
2860        args->flags = flags;
2861        args->astfn = ast;
2862        args->astparam = astparam;
2863        args->bastfn = bast;
2864        args->timeout = timeout_cs;
2865        args->mode = mode;
2866        args->lksb = lksb;
2867        rv = 0;
2868 out:
2869        return rv;
2870}
2871
2872static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2873{
2874        if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2875                      DLM_LKF_FORCEUNLOCK))
2876                return -EINVAL;
2877
2878        if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2879                return -EINVAL;
2880
2881        args->flags = flags;
2882        args->astparam = astarg;
2883        return 0;
2884}
2885
2886static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2887                              struct dlm_args *args)
2888{
2889        int rv = -EINVAL;
2890
2891        if (args->flags & DLM_LKF_CONVERT) {
2892                if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2893                        goto out;
2894
2895                if (args->flags & DLM_LKF_QUECVT &&
2896                    !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2897                        goto out;
2898
2899                rv = -EBUSY;
2900                if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2901                        goto out;
2902
2903                if (lkb->lkb_wait_type)
2904                        goto out;
2905
2906                if (is_overlap(lkb))
2907                        goto out;
2908        }
2909
2910        lkb->lkb_exflags = args->flags;
2911        lkb->lkb_sbflags = 0;
2912        lkb->lkb_astfn = args->astfn;
2913        lkb->lkb_astparam = args->astparam;
2914        lkb->lkb_bastfn = args->bastfn;
2915        lkb->lkb_rqmode = args->mode;
2916        lkb->lkb_lksb = args->lksb;
2917        lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2918        lkb->lkb_ownpid = (int) current->pid;
2919        lkb->lkb_timeout_cs = args->timeout;
2920        rv = 0;
2921 out:
2922        if (rv)
2923                log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2924                          rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2925                          lkb->lkb_status, lkb->lkb_wait_type,
2926                          lkb->lkb_resource->res_name);
2927        return rv;
2928}
2929
2930/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2931   for success */
2932
2933/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2934   because there may be a lookup in progress and it's valid to do
2935   cancel/unlockf on it */
2936
2937static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2938{
2939        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2940        int rv = -EINVAL;
2941
2942        if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2943                log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2944                dlm_print_lkb(lkb);
2945                goto out;
2946        }
2947
2948        /* an lkb may still exist even though the lock is EOL'ed due to a
2949           cancel, unlock or failed noqueue request; an app can't use these
2950           locks; return same error as if the lkid had not been found at all */
2951
2952        if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2953                log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2954                rv = -ENOENT;
2955                goto out;
2956        }
2957
2958        /* an lkb may be waiting for an rsb lookup to complete where the
2959           lookup was initiated by another lock */
2960
2961        if (!list_empty(&lkb->lkb_rsb_lookup)) {
2962                if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2963                        log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2964                        list_del_init(&lkb->lkb_rsb_lookup);
2965                        queue_cast(lkb->lkb_resource, lkb,
2966                                   args->flags & DLM_LKF_CANCEL ?
2967                                   -DLM_ECANCEL : -DLM_EUNLOCK);
2968                        unhold_lkb(lkb); /* undoes create_lkb() */
2969                }
2970                /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2971                rv = -EBUSY;
2972                goto out;
2973        }
2974
2975        /* cancel not allowed with another cancel/unlock in progress */
2976
2977        if (args->flags & DLM_LKF_CANCEL) {
2978                if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2979                        goto out;
2980
2981                if (is_overlap(lkb))
2982                        goto out;
2983
2984                /* don't let scand try to do a cancel */
2985                del_timeout(lkb);
2986
2987                if (lkb->lkb_flags & DLM_IFL_RESEND) {
2988                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2989                        rv = -EBUSY;
2990                        goto out;
2991                }
2992
2993                /* there's nothing to cancel */
2994                if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2995                    !lkb->lkb_wait_type) {
2996                        rv = -EBUSY;
2997                        goto out;
2998                }
2999
3000                switch (lkb->lkb_wait_type) {
3001                case DLM_MSG_LOOKUP:
3002                case DLM_MSG_REQUEST:
3003                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3004                        rv = -EBUSY;
3005                        goto out;
3006                case DLM_MSG_UNLOCK:
3007                case DLM_MSG_CANCEL:
3008                        goto out;
3009                }
3010                /* add_to_waiters() will set OVERLAP_CANCEL */
3011                goto out_ok;
3012        }
3013
3014        /* do we need to allow a force-unlock if there's a normal unlock
3015           already in progress?  in what conditions could the normal unlock
3016           fail such that we'd want to send a force-unlock to be sure? */
3017
3018        if (args->flags & DLM_LKF_FORCEUNLOCK) {
3019                if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3020                        goto out;
3021
3022                if (is_overlap_unlock(lkb))
3023                        goto out;
3024
3025                /* don't let scand try to do a cancel */
3026                del_timeout(lkb);
3027
3028                if (lkb->lkb_flags & DLM_IFL_RESEND) {
3029                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3030                        rv = -EBUSY;
3031                        goto out;
3032                }
3033
3034                switch (lkb->lkb_wait_type) {
3035                case DLM_MSG_LOOKUP:
3036                case DLM_MSG_REQUEST:
3037                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3038                        rv = -EBUSY;
3039                        goto out;
3040                case DLM_MSG_UNLOCK:
3041                        goto out;
3042                }
3043                /* add_to_waiters() will set OVERLAP_UNLOCK */
3044                goto out_ok;
3045        }
3046
3047        /* normal unlock not allowed if there's any op in progress */
3048        rv = -EBUSY;
3049        if (lkb->lkb_wait_type || lkb->lkb_wait_count)
3050                goto out;
3051
3052 out_ok:
3053        /* an overlapping op shouldn't blow away exflags from other op */
3054        lkb->lkb_exflags |= args->flags;
3055        lkb->lkb_sbflags = 0;
3056        lkb->lkb_astparam = args->astparam;
3057        rv = 0;
3058 out:
3059        if (rv)
3060                log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3061                          lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3062                          args->flags, lkb->lkb_wait_type,
3063                          lkb->lkb_resource->res_name);
3064        return rv;
3065}
3066
3067/*
3068 * Four stage 4 varieties:
3069 * do_request(), do_convert(), do_unlock(), do_cancel()
3070 * These are called on the master node for the given lock and
3071 * from the central locking logic.
3072 */
3073
3074static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3075{
3076        int error = 0;
3077
3078        if (can_be_granted(r, lkb, 1, 0, NULL)) {
3079                grant_lock(r, lkb);
3080                queue_cast(r, lkb, 0);
3081                goto out;
3082        }
3083
3084        if (can_be_queued(lkb)) {
3085                error = -EINPROGRESS;
3086                add_lkb(r, lkb, DLM_LKSTS_WAITING);
3087                add_timeout(lkb);
3088                goto out;
3089        }
3090
3091        error = -EAGAIN;
3092        queue_cast(r, lkb, -EAGAIN);
3093 out:
3094        return error;
3095}
3096
3097static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3098                               int error)
3099{
3100        switch (error) {
3101        case -EAGAIN:
3102                if (force_blocking_asts(lkb))
3103                        send_blocking_asts_all(r, lkb);
3104                break;
3105        case -EINPROGRESS:
3106                send_blocking_asts(r, lkb);
3107                break;
3108        }
3109}
3110
3111static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3112{
3113        int error = 0;
3114        int deadlk = 0;
3115
3116        /* changing an existing lock may allow others to be granted */
3117
3118        if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3119                grant_lock(r, lkb);
3120                queue_cast(r, lkb, 0);
3121                goto out;
3122        }
3123
3124        /* can_be_granted() detected that this lock would block in a conversion
3125           deadlock, so we leave it on the granted queue and return EDEADLK in
3126           the ast for the convert. */
3127
3128        if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
3129                /* it's left on the granted queue */
3130                revert_lock(r, lkb);
3131                queue_cast(r, lkb, -EDEADLK);
3132                error = -EDEADLK;
3133                goto out;
3134        }
3135
3136        /* is_demoted() means the can_be_granted() above set the grmode
3137           to NL, and left us on the granted queue.  This auto-demotion
3138           (due to CONVDEADLK) might mean other locks, and/or this lock, are
3139           now grantable.  We have to try to grant other converting locks
3140           before we try again to grant this one. */
3141
3142        if (is_demoted(lkb)) {
3143                grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3144                if (_can_be_granted(r, lkb, 1, 0)) {
3145                        grant_lock(r, lkb);
3146                        queue_cast(r, lkb, 0);
3147                        goto out;
3148                }
3149                /* else fall through and move to convert queue */
3150        }
3151
3152        if (can_be_queued(lkb)) {
3153                error = -EINPROGRESS;
3154                del_lkb(r, lkb);
3155                add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3156                add_timeout(lkb);
3157                goto out;
3158        }
3159
3160        error = -EAGAIN;
3161        queue_cast(r, lkb, -EAGAIN);
3162 out:
3163        return error;
3164}
3165
3166static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3167                               int error)
3168{
3169        switch (error) {
3170        case 0:
3171                grant_pending_locks(r, NULL);
3172                /* grant_pending_locks also sends basts */
3173                break;
3174        case -EAGAIN:
3175                if (force_blocking_asts(lkb))
3176                        send_blocking_asts_all(r, lkb);
3177                break;
3178        case -EINPROGRESS:
3179                send_blocking_asts(r, lkb);
3180                break;
3181        }
3182}
3183
3184static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3185{
3186        remove_lock(r, lkb);
3187        queue_cast(r, lkb, -DLM_EUNLOCK);
3188        return -DLM_EUNLOCK;
3189}
3190
3191static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3192                              int error)
3193{
3194        grant_pending_locks(r, NULL);
3195}
3196
3197/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3198
3199static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3200{
3201        int error;
3202
3203        error = revert_lock(r, lkb);
3204        if (error) {
3205                queue_cast(r, lkb, -DLM_ECANCEL);
3206                return -DLM_ECANCEL;
3207        }
3208        return 0;
3209}
3210
3211static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3212                              int error)
3213{
3214        if (error)
3215                grant_pending_locks(r, NULL);
3216}
3217
3218/*
3219 * Four stage 3 varieties:
3220 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3221 */
3222
3223/* add a new lkb to a possibly new rsb, called by requesting process */
3224
3225static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3226{
3227        int error;
3228
3229        /* set_master: sets lkb nodeid from r */
3230
3231        error = set_master(r, lkb);
3232        if (error < 0)
3233                goto out;
3234        if (error) {
3235                error = 0;
3236                goto out;
3237        }
3238
3239        if (is_remote(r)) {
3240                /* receive_request() calls do_request() on remote node */
3241                error = send_request(r, lkb);
3242        } else {
3243                error = do_request(r, lkb);
3244                /* for remote locks the request_reply is sent
3245                   between do_request and do_request_effects */
3246                do_request_effects(r, lkb, error);
3247        }
3248 out:
3249        return error;
3250}
3251
3252/* change some property of an existing lkb, e.g. mode */
3253
3254static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3255{
3256        int error;
3257
3258        if (is_remote(r)) {
3259                /* receive_convert() calls do_convert() on remote node */
3260                error = send_convert(r, lkb);
3261        } else {
3262                error = do_convert(r, lkb);
3263                /* for remote locks the convert_reply is sent
3264                   between do_convert and do_convert_effects */
3265                do_convert_effects(r, lkb, error);
3266        }
3267
3268        return error;
3269}
3270
3271/* remove an existing lkb from the granted queue */
3272
3273static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3274{
3275        int error;
3276
3277        if (is_remote(r)) {
3278                /* receive_unlock() calls do_unlock() on remote node */
3279                error = send_unlock(r, lkb);
3280        } else {
3281                error = do_unlock(r, lkb);
3282                /* for remote locks the unlock_reply is sent
3283                   between do_unlock and do_unlock_effects */
3284                do_unlock_effects(r, lkb, error);
3285        }
3286
3287        return error;
3288}
3289
3290/* remove an existing lkb from the convert or wait queue */
3291
3292static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3293{
3294        int error;
3295
3296        if (is_remote(r)) {
3297                /* receive_cancel() calls do_cancel() on remote node */
3298                error = send_cancel(r, lkb);
3299        } else {
3300                error = do_cancel(r, lkb);
3301                /* for remote locks the cancel_reply is sent
3302                   between do_cancel and do_cancel_effects */
3303                do_cancel_effects(r, lkb, error);
3304        }
3305
3306        return error;
3307}
3308
3309/*
3310 * Four stage 2 varieties:
3311 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3312 */
3313
3314static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3315                        int len, struct dlm_args *args)
3316{
3317        struct dlm_rsb *r;
3318        int error;
3319
3320        error = validate_lock_args(ls, lkb, args);
3321        if (error)
3322                return error;
3323
3324        error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3325        if (error)
3326                return error;
3327
3328        lock_rsb(r);
3329
3330        attach_lkb(r, lkb);
3331        lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3332
3333        error = _request_lock(r, lkb);
3334
3335        unlock_rsb(r);
3336        put_rsb(r);
3337        return error;
3338}
3339
3340static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3341                        struct dlm_args *args)
3342{
3343        struct dlm_rsb *r;
3344        int error;
3345
3346        r = lkb->lkb_resource;
3347
3348        hold_rsb(r);
3349        lock_rsb(r);
3350
3351        error = validate_lock_args(ls, lkb, args);
3352        if (error)
3353                goto out;
3354
3355        error = _convert_lock(r, lkb);
3356 out:
3357        unlock_rsb(r);
3358        put_rsb(r);
3359        return error;
3360}
3361
3362static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3363                       struct dlm_args *args)
3364{
3365        struct dlm_rsb *r;
3366        int error;
3367
3368        r = lkb->lkb_resource;
3369
3370        hold_rsb(r);
3371        lock_rsb(r);
3372
3373        error = validate_unlock_args(lkb, args);
3374        if (error)
3375                goto out;
3376
3377        error = _unlock_lock(r, lkb);
3378 out:
3379        unlock_rsb(r);
3380        put_rsb(r);
3381        return error;
3382}
3383
3384static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3385                       struct dlm_args *args)
3386{
3387        struct dlm_rsb *r;
3388        int error;
3389
3390        r = lkb->lkb_resource;
3391
3392        hold_rsb(r);
3393        lock_rsb(r);
3394
3395        error = validate_unlock_args(lkb, args);
3396        if (error)
3397                goto out;
3398
3399        error = _cancel_lock(r, lkb);
3400 out:
3401        unlock_rsb(r);
3402        put_rsb(r);
3403        return error;
3404}
3405
3406/*
3407 * Two stage 1 varieties:  dlm_lock() and dlm_unlock()
3408 */
3409
3410int dlm_lock(dlm_lockspace_t *lockspace,
3411             int mode,
3412             struct dlm_lksb *lksb,
3413             uint32_t flags,
3414             void *name,
3415             unsigned int namelen,
3416             uint32_t parent_lkid,
3417             void (*ast) (void *astarg),
3418             void *astarg,
3419             void (*bast) (void *astarg, int mode))
3420{
3421        struct dlm_ls *ls;
3422        struct dlm_lkb *lkb;
3423        struct dlm_args args;
3424        int error, convert = flags & DLM_LKF_CONVERT;
3425
3426        ls = dlm_find_lockspace_local(lockspace);
3427        if (!ls)
3428                return -EINVAL;
3429
3430        dlm_lock_recovery(ls);
3431
3432        if (convert)
3433                error = find_lkb(ls, lksb->sb_lkid, &lkb);
3434        else
3435                error = create_lkb(ls, &lkb);
3436
3437        if (error)
3438                goto out;
3439
3440        error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3441                              astarg, bast, &args);
3442        if (error)
3443                goto out_put;
3444
3445        if (convert)
3446                error = convert_lock(ls, lkb, &args);
3447        else
3448                error = request_lock(ls, lkb, name, namelen, &args);
3449
3450        if (error == -EINPROGRESS)
3451                error = 0;
3452 out_put:
3453        if (convert || error)
3454                __put_lkb(ls, lkb);
3455        if (error == -EAGAIN || error == -EDEADLK)
3456                error = 0;
3457 out:
3458        dlm_unlock_recovery(ls);
3459        dlm_put_lockspace(ls);
3460        return error;
3461}
3462
3463int dlm_unlock(dlm_lockspace_t *lockspace,
3464               uint32_t lkid,
3465               uint32_t flags,
3466               struct dlm_lksb *lksb,
3467               void *astarg)
3468{
3469        struct dlm_ls *ls;
3470        struct dlm_lkb *lkb;
3471        struct dlm_args args;
3472        int error;
3473
3474        ls = dlm_find_lockspace_local(lockspace);
3475        if (!ls)
3476                return -EINVAL;
3477
3478        dlm_lock_recovery(ls);
3479
3480        error = find_lkb(ls, lkid, &lkb);
3481        if (error)
3482                goto out;
3483
3484        error = set_unlock_args(flags, astarg, &args);
3485        if (error)
3486                goto out_put;
3487
3488        if (flags & DLM_LKF_CANCEL)
3489                error = cancel_lock(ls, lkb, &args);
3490        else
3491                error = unlock_lock(ls, lkb, &args);
3492
3493        if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3494                error = 0;
3495        if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3496                error = 0;
3497 out_put:
3498        dlm_put_lkb(lkb);
3499 out:
3500        dlm_unlock_recovery(ls);
3501        dlm_put_lockspace(ls);
3502        return error;
3503}
3504
3505/*
3506 * send/receive routines for remote operations and replies
3507 *
3508 * send_args
3509 * send_common
3510 * send_request                 receive_request
3511 * send_convert                 receive_convert
3512 * send_unlock                  receive_unlock
3513 * send_cancel                  receive_cancel
3514 * send_grant                   receive_grant
3515 * send_bast                    receive_bast
3516 * send_lookup                  receive_lookup
3517 * send_remove                  receive_remove
3518 *
3519 *                              send_common_reply
3520 * receive_request_reply        send_request_reply
3521 * receive_convert_reply        send_convert_reply
3522 * receive_unlock_reply         send_unlock_reply
3523 * receive_cancel_reply         send_cancel_reply
3524 * receive_lookup_reply         send_lookup_reply
3525 */
3526
3527static int _create_message(struct dlm_ls *ls, int mb_len,
3528                           int to_nodeid, int mstype,
3529                           struct dlm_message **ms_ret,
3530                           struct dlm_mhandle **mh_ret)
3531{
3532        struct dlm_message *ms;
3533        struct dlm_mhandle *mh;
3534        char *mb;
3535
3536        /* get_buffer gives us a message handle (mh) that we need to
3537           pass into midcomms_commit and a message buffer (mb) that we
3538           write our data into */
3539
3540        mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
3541        if (!mh)
3542                return -ENOBUFS;
3543
3544        ms = (struct dlm_message *) mb;
3545
3546        ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3547        ms->m_header.u.h_lockspace = ls->ls_global_id;
3548        ms->m_header.h_nodeid = dlm_our_nodeid();
3549        ms->m_header.h_length = mb_len;
3550        ms->m_header.h_cmd = DLM_MSG;
3551
3552        ms->m_type = mstype;
3553
3554        *mh_ret = mh;
3555        *ms_ret = ms;
3556        return 0;
3557}
3558
3559static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3560                          int to_nodeid, int mstype,
3561                          struct dlm_message **ms_ret,
3562                          struct dlm_mhandle **mh_ret)
3563{
3564        int mb_len = sizeof(struct dlm_message);
3565
3566        switch (mstype) {
3567        case DLM_MSG_REQUEST:
3568        case DLM_MSG_LOOKUP:
3569        case DLM_MSG_REMOVE:
3570                mb_len += r->res_length;
3571                break;
3572        case DLM_MSG_CONVERT:
3573        case DLM_MSG_UNLOCK:
3574        case DLM_MSG_REQUEST_REPLY:
3575        case DLM_MSG_CONVERT_REPLY:
3576        case DLM_MSG_GRANT:
3577                if (lkb && lkb->lkb_lvbptr)
3578                        mb_len += r->res_ls->ls_lvblen;
3579                break;
3580        }
3581
3582        return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3583                               ms_ret, mh_ret);
3584}
3585
3586/* further lowcomms enhancements or alternate implementations may make
3587   the return value from this function useful at some point */
3588
3589static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3590{
3591        dlm_message_out(ms);
3592        dlm_midcomms_commit_mhandle(mh);
3593        return 0;
3594}
3595
3596static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3597                      struct dlm_message *ms)
3598{
3599        ms->m_nodeid   = lkb->lkb_nodeid;
3600        ms->m_pid      = lkb->lkb_ownpid;
3601        ms->m_lkid     = lkb->lkb_id;
3602        ms->m_remid    = lkb->lkb_remid;
3603        ms->m_exflags  = lkb->lkb_exflags;
3604        ms->m_sbflags  = lkb->lkb_sbflags;
3605        ms->m_flags    = lkb->lkb_flags;
3606        ms->m_lvbseq   = lkb->lkb_lvbseq;
3607        ms->m_status   = lkb->lkb_status;
3608        ms->m_grmode   = lkb->lkb_grmode;
3609        ms->m_rqmode   = lkb->lkb_rqmode;
3610        ms->m_hash     = r->res_hash;
3611
3612        /* m_result and m_bastmode are set from function args,
3613           not from lkb fields */
3614
3615        if (lkb->lkb_bastfn)
3616                ms->m_asts |= DLM_CB_BAST;
3617        if (lkb->lkb_astfn)
3618                ms->m_asts |= DLM_CB_CAST;
3619
3620        /* compare with switch in create_message; send_remove() doesn't
3621           use send_args() */
3622
3623        switch (ms->m_type) {
3624        case DLM_MSG_REQUEST:
3625        case DLM_MSG_LOOKUP:
3626                memcpy(ms->m_extra, r->res_name, r->res_length);
3627                break;
3628        case DLM_MSG_CONVERT:
3629        case DLM_MSG_UNLOCK:
3630        case DLM_MSG_REQUEST_REPLY:
3631        case DLM_MSG_CONVERT_REPLY:
3632        case DLM_MSG_GRANT:
3633                if (!lkb->lkb_lvbptr)
3634                        break;
3635                memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3636                break;
3637        }
3638}
3639
3640static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3641{
3642        struct dlm_message *ms;
3643        struct dlm_mhandle *mh;
3644        int to_nodeid, error;
3645
3646        to_nodeid = r->res_nodeid;
3647
3648        error = add_to_waiters(lkb, mstype, to_nodeid);
3649        if (error)
3650                return error;
3651
3652        error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3653        if (error)
3654                goto fail;
3655
3656        send_args(r, lkb, ms);
3657
3658        error = send_message(mh, ms);
3659        if (error)
3660                goto fail;
3661        return 0;
3662
3663 fail:
3664        remove_from_waiters(lkb, msg_reply_type(mstype));
3665        return error;
3666}
3667
3668static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3669{
3670        return send_common(r, lkb, DLM_MSG_REQUEST);
3671}
3672
3673static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3674{
3675        int error;
3676
3677        error = send_common(r, lkb, DLM_MSG_CONVERT);
3678
3679        /* down conversions go without a reply from the master */
3680        if (!error && down_conversion(lkb)) {
3681                remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3682                r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3683                r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3684                r->res_ls->ls_stub_ms.m_result = 0;
3685                __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3686        }
3687
3688        return error;
3689}
3690
3691/* FIXME: if this lkb is the only lock we hold on the rsb, then set
3692   MASTER_UNCERTAIN to force the next request on the rsb to confirm
3693   that the master is still correct. */
3694
3695static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3696{
3697        return send_common(r, lkb, DLM_MSG_UNLOCK);
3698}
3699
3700static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3701{
3702        return send_common(r, lkb, DLM_MSG_CANCEL);
3703}
3704
3705static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3706{
3707        struct dlm_message *ms;
3708        struct dlm_mhandle *mh;
3709        int to_nodeid, error;
3710
3711        to_nodeid = lkb->lkb_nodeid;
3712
3713        error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3714        if (error)
3715                goto out;
3716
3717        send_args(r, lkb, ms);
3718
3719        ms->m_result = 0;
3720
3721        error = send_message(mh, ms);
3722 out:
3723        return error;
3724}
3725
3726static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3727{
3728        struct dlm_message *ms;
3729        struct dlm_mhandle *mh;
3730        int to_nodeid, error;
3731
3732        to_nodeid = lkb->lkb_nodeid;
3733
3734        error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3735        if (error)
3736                goto out;
3737
3738        send_args(r, lkb, ms);
3739
3740        ms->m_bastmode = mode;
3741
3742        error = send_message(mh, ms);
3743 out:
3744        return error;
3745}
3746
3747static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3748{
3749        struct dlm_message *ms;
3750        struct dlm_mhandle *mh;
3751        int to_nodeid, error;
3752
3753        to_nodeid = dlm_dir_nodeid(r);
3754
3755        error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3756        if (error)
3757                return error;
3758
3759        error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3760        if (error)
3761                goto fail;
3762
3763        send_args(r, lkb, ms);
3764
3765        error = send_message(mh, ms);
3766        if (error)
3767                goto fail;
3768        return 0;
3769
3770 fail:
3771        remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3772        return error;
3773}
3774
3775static int send_remove(struct dlm_rsb *r)
3776{
3777        struct dlm_message *ms;
3778        struct dlm_mhandle *mh;
3779        int to_nodeid, error;
3780
3781        to_nodeid = dlm_dir_nodeid(r);
3782
3783        error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3784        if (error)
3785                goto out;
3786
3787        memcpy(ms->m_extra, r->res_name, r->res_length);
3788        ms->m_hash = r->res_hash;
3789
3790        error = send_message(mh, ms);
3791 out:
3792        return error;
3793}
3794
3795static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3796                             int mstype, int rv)
3797{
3798        struct dlm_message *ms;
3799        struct dlm_mhandle *mh;
3800        int to_nodeid, error;
3801
3802        to_nodeid = lkb->lkb_nodeid;
3803
3804        error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3805        if (error)
3806                goto out;
3807
3808        send_args(r, lkb, ms);
3809
3810        ms->m_result = rv;
3811
3812        error = send_message(mh, ms);
3813 out:
3814        return error;
3815}
3816
3817static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3818{
3819        return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3820}
3821
3822static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3823{
3824        return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3825}
3826
3827static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3828{
3829        return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3830}
3831
3832static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3833{
3834        return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3835}
3836
3837static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3838                             int ret_nodeid, int rv)
3839{
3840        struct dlm_rsb *r = &ls->ls_stub_rsb;
3841        struct dlm_message *ms;
3842        struct dlm_mhandle *mh;
3843        int error, nodeid = ms_in->m_header.h_nodeid;
3844
3845        error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3846        if (error)
3847                goto out;
3848
3849        ms->m_lkid = ms_in->m_lkid;
3850        ms->m_result = rv;
3851        ms->m_nodeid = ret_nodeid;
3852
3853        error = send_message(mh, ms);
3854 out:
3855        return error;
3856}
3857
3858/* which args we save from a received message depends heavily on the type
3859   of message, unlike the send side where we can safely send everything about
3860   the lkb for any type of message */
3861
3862static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3863{
3864        lkb->lkb_exflags = ms->m_exflags;
3865        lkb->lkb_sbflags = ms->m_sbflags;
3866        lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3867                         (ms->m_flags & 0x0000FFFF);
3868}
3869
3870static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3871{
3872        if (ms->m_flags == DLM_IFL_STUB_MS)
3873                return;
3874
3875        lkb->lkb_sbflags = ms->m_sbflags;
3876        lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3877                         (ms->m_flags & 0x0000FFFF);
3878}
3879
3880static int receive_extralen(struct dlm_message *ms)
3881{
3882        return (ms->m_header.h_length - sizeof(struct dlm_message));
3883}
3884
3885static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3886                       struct dlm_message *ms)
3887{
3888        int len;
3889
3890        if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3891                if (!lkb->lkb_lvbptr)
3892                        lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3893                if (!lkb->lkb_lvbptr)
3894                        return -ENOMEM;
3895                len = receive_extralen(ms);
3896                if (len > ls->ls_lvblen)
3897                        len = ls->ls_lvblen;
3898                memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3899        }
3900        return 0;
3901}
3902
3903static void fake_bastfn(void *astparam, int mode)
3904{
3905        log_print("fake_bastfn should not be called");
3906}
3907
3908static void fake_astfn(void *astparam)
3909{
3910        log_print("fake_astfn should not be called");
3911}
3912
3913static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3914                                struct dlm_message *ms)
3915{
3916        lkb->lkb_nodeid = ms->m_header.h_nodeid;
3917        lkb->lkb_ownpid = ms->m_pid;
3918        lkb->lkb_remid = ms->m_lkid;
3919        lkb->lkb_grmode = DLM_LOCK_IV;
3920        lkb->lkb_rqmode = ms->m_rqmode;
3921
3922        lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3923        lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3924
3925        if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3926                /* lkb was just created so there won't be an lvb yet */
3927                lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3928                if (!lkb->lkb_lvbptr)
3929                        return -ENOMEM;
3930        }
3931
3932        return 0;
3933}
3934
3935static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3936                                struct dlm_message *ms)
3937{
3938        if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3939                return -EBUSY;
3940
3941        if (receive_lvb(ls, lkb, ms))
3942                return -ENOMEM;
3943
3944        lkb->lkb_rqmode = ms->m_rqmode;
3945        lkb->lkb_lvbseq = ms->m_lvbseq;
3946
3947        return 0;
3948}
3949
3950static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3951                               struct dlm_message *ms)
3952{
3953        if (receive_lvb(ls, lkb, ms))
3954                return -ENOMEM;
3955        return 0;
3956}
3957
3958/* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3959   uses to send a reply and that the remote end uses to process the reply. */
3960
3961static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3962{
3963        struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3964        lkb->lkb_nodeid = ms->m_header.h_nodeid;
3965        lkb->lkb_remid = ms->m_lkid;
3966}
3967
3968/* This is called after the rsb is locked so that we can safely inspect
3969   fields in the lkb. */
3970
3971static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3972{
3973        int from = ms->m_header.h_nodeid;
3974        int error = 0;
3975
3976        switch (ms->m_type) {
3977        case DLM_MSG_CONVERT:
3978        case DLM_MSG_UNLOCK:
3979        case DLM_MSG_CANCEL:
3980                if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3981                        error = -EINVAL;
3982                break;
3983
3984        case DLM_MSG_CONVERT_REPLY:
3985        case DLM_MSG_UNLOCK_REPLY:
3986        case DLM_MSG_CANCEL_REPLY:
3987        case DLM_MSG_GRANT:
3988        case DLM_MSG_BAST:
3989                if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3990                        error = -EINVAL;
3991                break;
3992
3993        case DLM_MSG_REQUEST_REPLY:
3994                if (!is_process_copy(lkb))
3995                        error = -EINVAL;
3996                else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3997                        error = -EINVAL;
3998                break;
3999
4000        default:
4001                error = -EINVAL;
4002        }
4003
4004        if (error)
4005                log_error(lkb->lkb_resource->res_ls,
4006                          "ignore invalid message %d from %d %x %x %x %d",
4007                          ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
4008                          lkb->lkb_flags, lkb->lkb_nodeid);
4009        return error;
4010}
4011
4012static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4013{
4014        char name[DLM_RESNAME_MAXLEN + 1];
4015        struct dlm_message *ms;
4016        struct dlm_mhandle *mh;
4017        struct dlm_rsb *r;
4018        uint32_t hash, b;
4019        int rv, dir_nodeid;
4020
4021        memset(name, 0, sizeof(name));
4022        memcpy(name, ms_name, len);
4023
4024        hash = jhash(name, len, 0);
4025        b = hash & (ls->ls_rsbtbl_size - 1);
4026
4027        dir_nodeid = dlm_hash2nodeid(ls, hash);
4028
4029        log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4030
4031        spin_lock(&ls->ls_rsbtbl[b].lock);
4032        rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4033        if (!rv) {
4034                spin_unlock(&ls->ls_rsbtbl[b].lock);
4035                log_error(ls, "repeat_remove on keep %s", name);
4036                return;
4037        }
4038
4039        rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4040        if (!rv) {
4041                spin_unlock(&ls->ls_rsbtbl[b].lock);
4042                log_error(ls, "repeat_remove on toss %s", name);
4043                return;
4044        }
4045
4046        /* use ls->remove_name2 to avoid conflict with shrink? */
4047
4048        spin_lock(&ls->ls_remove_spin);
4049        ls->ls_remove_len = len;
4050        memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4051        spin_unlock(&ls->ls_remove_spin);
4052        spin_unlock(&ls->ls_rsbtbl[b].lock);
4053
4054        rv = _create_message(ls, sizeof(struct dlm_message) + len,
4055                             dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4056        if (rv)
4057                return;
4058
4059        memcpy(ms->m_extra, name, len);
4060        ms->m_hash = hash;
4061
4062        send_message(mh, ms);
4063
4064        spin_lock(&ls->ls_remove_spin);
4065        ls->ls_remove_len = 0;
4066        memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4067        spin_unlock(&ls->ls_remove_spin);
4068}
4069
4070static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4071{
4072        struct dlm_lkb *lkb;
4073        struct dlm_rsb *r;
4074        int from_nodeid;
4075        int error, namelen = 0;
4076
4077        from_nodeid = ms->m_header.h_nodeid;
4078
4079        error = create_lkb(ls, &lkb);
4080        if (error)
4081                goto fail;
4082
4083        receive_flags(lkb, ms);
4084        lkb->lkb_flags |= DLM_IFL_MSTCPY;
4085        error = receive_request_args(ls, lkb, ms);
4086        if (error) {
4087                __put_lkb(ls, lkb);
4088                goto fail;
4089        }
4090
4091        /* The dir node is the authority on whether we are the master
4092           for this rsb or not, so if the master sends us a request, we should
4093           recreate the rsb if we've destroyed it.   This race happens when we
4094           send a remove message to the dir node at the same time that the dir
4095           node sends us a request for the rsb. */
4096
4097        namelen = receive_extralen(ms);
4098
4099        error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4100                         R_RECEIVE_REQUEST, &r);
4101        if (error) {
4102                __put_lkb(ls, lkb);
4103                goto fail;
4104        }
4105
4106        lock_rsb(r);
4107
4108        if (r->res_master_nodeid != dlm_our_nodeid()) {
4109                error = validate_master_nodeid(ls, r, from_nodeid);
4110                if (error) {
4111                        unlock_rsb(r);
4112                        put_rsb(r);
4113                        __put_lkb(ls, lkb);
4114                        goto fail;
4115                }
4116        }
4117
4118        attach_lkb(r, lkb);
4119        error = do_request(r, lkb);
4120        send_request_reply(r, lkb, error);
4121        do_request_effects(r, lkb, error);
4122
4123        unlock_rsb(r);
4124        put_rsb(r);
4125
4126        if (error == -EINPROGRESS)
4127                error = 0;
4128        if (error)
4129                dlm_put_lkb(lkb);
4130        return 0;
4131
4132 fail:
4133        /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4134           and do this receive_request again from process_lookup_list once
4135           we get the lookup reply.  This would avoid a many repeated
4136           ENOTBLK request failures when the lookup reply designating us
4137           as master is delayed. */
4138
4139        /* We could repeatedly return -EBADR here if our send_remove() is
4140           delayed in being sent/arriving/being processed on the dir node.
4141           Another node would repeatedly lookup up the master, and the dir
4142           node would continue returning our nodeid until our send_remove
4143           took effect.
4144
4145           We send another remove message in case our previous send_remove
4146           was lost/ignored/missed somehow. */
4147
4148        if (error != -ENOTBLK) {
4149                log_limit(ls, "receive_request %x from %d %d",
4150                          ms->m_lkid, from_nodeid, error);
4151        }
4152
4153        if (namelen && error == -EBADR) {
4154                send_repeat_remove(ls, ms->m_extra, namelen);
4155                msleep(1000);
4156        }
4157
4158        setup_stub_lkb(ls, ms);
4159        send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4160        return error;
4161}
4162
4163static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4164{
4165        struct dlm_lkb *lkb;
4166        struct dlm_rsb *r;
4167        int error, reply = 1;
4168
4169        error = find_lkb(ls, ms->m_remid, &lkb);
4170        if (error)
4171                goto fail;
4172
4173        if (lkb->lkb_remid != ms->m_lkid) {
4174                log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4175                          "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4176                          (unsigned long long)lkb->lkb_recover_seq,
4177                          ms->m_header.h_nodeid, ms->m_lkid);
4178                error = -ENOENT;
4179                dlm_put_lkb(lkb);
4180                goto fail;
4181        }
4182
4183        r = lkb->lkb_resource;
4184
4185        hold_rsb(r);
4186        lock_rsb(r);
4187
4188        error = validate_message(lkb, ms);
4189        if (error)
4190                goto out;
4191
4192        receive_flags(lkb, ms);
4193
4194        error = receive_convert_args(ls, lkb, ms);
4195        if (error) {
4196                send_convert_reply(r, lkb, error);
4197                goto out;
4198        }
4199
4200        reply = !down_conversion(lkb);
4201
4202        error = do_convert(r, lkb);
4203        if (reply)
4204                send_convert_reply(r, lkb, error);
4205        do_convert_effects(r, lkb, error);
4206 out:
4207        unlock_rsb(r);
4208        put_rsb(r);
4209        dlm_put_lkb(lkb);
4210        return 0;
4211
4212 fail:
4213        setup_stub_lkb(ls, ms);
4214        send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4215        return error;
4216}
4217
4218static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4219{
4220        struct dlm_lkb *lkb;
4221        struct dlm_rsb *r;
4222        int error;
4223
4224        error = find_lkb(ls, ms->m_remid, &lkb);
4225        if (error)
4226                goto fail;
4227
4228        if (lkb->lkb_remid != ms->m_lkid) {
4229                log_error(ls, "receive_unlock %x remid %x remote %d %x",
4230                          lkb->lkb_id, lkb->lkb_remid,
4231                          ms->m_header.h_nodeid, ms->m_lkid);
4232                error = -ENOENT;
4233                dlm_put_lkb(lkb);
4234                goto fail;
4235        }
4236
4237        r = lkb->lkb_resource;
4238
4239        hold_rsb(r);
4240        lock_rsb(r);
4241
4242        error = validate_message(lkb, ms);
4243        if (error)
4244                goto out;
4245
4246        receive_flags(lkb, ms);
4247
4248        error = receive_unlock_args(ls, lkb, ms);
4249        if (error) {
4250                send_unlock_reply(r, lkb, error);
4251                goto out;
4252        }
4253
4254        error = do_unlock(r, lkb);
4255        send_unlock_reply(r, lkb, error);
4256        do_unlock_effects(r, lkb, error);
4257 out:
4258        unlock_rsb(r);
4259        put_rsb(r);
4260        dlm_put_lkb(lkb);
4261        return 0;
4262
4263 fail:
4264        setup_stub_lkb(ls, ms);
4265        send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4266        return error;
4267}
4268
4269static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4270{
4271        struct dlm_lkb *lkb;
4272        struct dlm_rsb *r;
4273        int error;
4274
4275        error = find_lkb(ls, ms->m_remid, &lkb);
4276        if (error)
4277                goto fail;
4278
4279        receive_flags(lkb, ms);
4280
4281        r = lkb->lkb_resource;
4282
4283        hold_rsb(r);
4284        lock_rsb(r);
4285
4286        error = validate_message(lkb, ms);
4287        if (error)
4288                goto out;
4289
4290        error = do_cancel(r, lkb);
4291        send_cancel_reply(r, lkb, error);
4292        do_cancel_effects(r, lkb, error);
4293 out:
4294        unlock_rsb(r);
4295        put_rsb(r);
4296        dlm_put_lkb(lkb);
4297        return 0;
4298
4299 fail:
4300        setup_stub_lkb(ls, ms);
4301        send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4302        return error;
4303}
4304
4305static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4306{
4307        struct dlm_lkb *lkb;
4308        struct dlm_rsb *r;
4309        int error;
4310
4311        error = find_lkb(ls, ms->m_remid, &lkb);
4312        if (error)
4313                return error;
4314
4315        r = lkb->lkb_resource;
4316
4317        hold_rsb(r);
4318        lock_rsb(r);
4319
4320        error = validate_message(lkb, ms);
4321        if (error)
4322                goto out;
4323
4324        receive_flags_reply(lkb, ms);
4325        if (is_altmode(lkb))
4326                munge_altmode(lkb, ms);
4327        grant_lock_pc(r, lkb, ms);
4328        queue_cast(r, lkb, 0);
4329 out:
4330        unlock_rsb(r);
4331        put_rsb(r);
4332        dlm_put_lkb(lkb);
4333        return 0;
4334}
4335
4336static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4337{
4338        struct dlm_lkb *lkb;
4339        struct dlm_rsb *r;
4340        int error;
4341
4342        error = find_lkb(ls, ms->m_remid, &lkb);
4343        if (error)
4344                return error;
4345
4346        r = lkb->lkb_resource;
4347
4348        hold_rsb(r);
4349        lock_rsb(r);
4350
4351        error = validate_message(lkb, ms);
4352        if (error)
4353                goto out;
4354
4355        queue_bast(r, lkb, ms->m_bastmode);
4356        lkb->lkb_highbast = ms->m_bastmode;
4357 out:
4358        unlock_rsb(r);
4359        put_rsb(r);
4360        dlm_put_lkb(lkb);
4361        return 0;
4362}
4363
4364static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4365{
4366        int len, error, ret_nodeid, from_nodeid, our_nodeid;
4367
4368        from_nodeid = ms->m_header.h_nodeid;
4369        our_nodeid = dlm_our_nodeid();
4370
4371        len = receive_extralen(ms);
4372
4373        error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4374                                  &ret_nodeid, NULL);
4375
4376        /* Optimization: we're master so treat lookup as a request */
4377        if (!error && ret_nodeid == our_nodeid) {
4378                receive_request(ls, ms);
4379                return;
4380        }
4381        send_lookup_reply(ls, ms, ret_nodeid, error);
4382}
4383
4384static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4385{
4386        char name[DLM_RESNAME_MAXLEN+1];
4387        struct dlm_rsb *r;
4388        uint32_t hash, b;
4389        int rv, len, dir_nodeid, from_nodeid;
4390
4391        from_nodeid = ms->m_header.h_nodeid;
4392
4393        len = receive_extralen(ms);
4394
4395        if (len > DLM_RESNAME_MAXLEN) {
4396                log_error(ls, "receive_remove from %d bad len %d",
4397                          from_nodeid, len);
4398                return;
4399        }
4400
4401        dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4402        if (dir_nodeid != dlm_our_nodeid()) {
4403                log_error(ls, "receive_remove from %d bad nodeid %d",
4404                          from_nodeid, dir_nodeid);
4405                return;
4406        }
4407
4408        /* Look for name on rsbtbl.toss, if it's there, kill it.
4409           If it's on rsbtbl.keep, it's being used, and we should ignore this
4410           message.  This is an expected race between the dir node sending a
4411           request to the master node at the same time as the master node sends
4412           a remove to the dir node.  The resolution to that race is for the
4413           dir node to ignore the remove message, and the master node to
4414           recreate the master rsb when it gets a request from the dir node for
4415           an rsb it doesn't have. */
4416
4417        memset(name, 0, sizeof(name));
4418        memcpy(name, ms->m_extra, len);
4419
4420        hash = jhash(name, len, 0);
4421        b = hash & (ls->ls_rsbtbl_size - 1);
4422
4423        spin_lock(&ls->ls_rsbtbl[b].lock);
4424
4425        rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4426        if (rv) {
4427                /* verify the rsb is on keep list per comment above */
4428                rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4429                if (rv) {
4430                        /* should not happen */
4431                        log_error(ls, "receive_remove from %d not found %s",
4432                                  from_nodeid, name);
4433                        spin_unlock(&ls->ls_rsbtbl[b].lock);
4434                        return;
4435                }
4436                if (r->res_master_nodeid != from_nodeid) {
4437                        /* should not happen */
4438                        log_error(ls, "receive_remove keep from %d master %d",
4439                                  from_nodeid, r->res_master_nodeid);
4440                        dlm_print_rsb(r);
4441                        spin_unlock(&ls->ls_rsbtbl[b].lock);
4442                        return;
4443                }
4444
4445                log_debug(ls, "receive_remove from %d master %d first %x %s",
4446                          from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4447                          name);
4448                spin_unlock(&ls->ls_rsbtbl[b].lock);
4449                return;
4450        }
4451
4452        if (r->res_master_nodeid != from_nodeid) {
4453                log_error(ls, "receive_remove toss from %d master %d",
4454                          from_nodeid, r->res_master_nodeid);
4455                dlm_print_rsb(r);
4456                spin_unlock(&ls->ls_rsbtbl[b].lock);
4457                return;
4458        }
4459
4460        if (kref_put(&r->res_ref, kill_rsb)) {
4461                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4462                spin_unlock(&ls->ls_rsbtbl[b].lock);
4463                dlm_free_rsb(r);
4464        } else {
4465                log_error(ls, "receive_remove from %d rsb ref error",
4466                          from_nodeid);
4467                dlm_print_rsb(r);
4468                spin_unlock(&ls->ls_rsbtbl[b].lock);
4469        }
4470}
4471
4472static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4473{
4474        do_purge(ls, ms->m_nodeid, ms->m_pid);
4475}
4476
4477static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4478{
4479        struct dlm_lkb *lkb;
4480        struct dlm_rsb *r;
4481        int error, mstype, result;
4482        int from_nodeid = ms->m_header.h_nodeid;
4483
4484        error = find_lkb(ls, ms->m_remid, &lkb);
4485        if (error)
4486                return error;
4487
4488        r = lkb->lkb_resource;
4489        hold_rsb(r);
4490        lock_rsb(r);
4491
4492        error = validate_message(lkb, ms);
4493        if (error)
4494                goto out;
4495
4496        mstype = lkb->lkb_wait_type;
4497        error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4498        if (error) {
4499                log_error(ls, "receive_request_reply %x remote %d %x result %d",
4500                          lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4501                dlm_dump_rsb(r);
4502                goto out;
4503        }
4504
4505        /* Optimization: the dir node was also the master, so it took our
4506           lookup as a request and sent request reply instead of lookup reply */
4507        if (mstype == DLM_MSG_LOOKUP) {
4508                r->res_master_nodeid = from_nodeid;
4509                r->res_nodeid = from_nodeid;
4510                lkb->lkb_nodeid = from_nodeid;
4511        }
4512
4513        /* this is the value returned from do_request() on the master */
4514        result = ms->m_result;
4515
4516        switch (result) {
4517        case -EAGAIN:
4518                /* request would block (be queued) on remote master */
4519                queue_cast(r, lkb, -EAGAIN);
4520                confirm_master(r, -EAGAIN);
4521                unhold_lkb(lkb); /* undoes create_lkb() */
4522                break;
4523
4524        case -EINPROGRESS:
4525        case 0:
4526                /* request was queued or granted on remote master */
4527                receive_flags_reply(lkb, ms);
4528                lkb->lkb_remid = ms->m_lkid;
4529                if (is_altmode(lkb))
4530                        munge_altmode(lkb, ms);
4531                if (result) {
4532                        add_lkb(r, lkb, DLM_LKSTS_WAITING);
4533                        add_timeout(lkb);
4534                } else {
4535                        grant_lock_pc(r, lkb, ms);
4536                        queue_cast(r, lkb, 0);
4537                }
4538                confirm_master(r, result);
4539                break;
4540
4541        case -EBADR:
4542        case -ENOTBLK:
4543                /* find_rsb failed to find rsb or rsb wasn't master */
4544                log_limit(ls, "receive_request_reply %x from %d %d "
4545                          "master %d dir %d first %x %s", lkb->lkb_id,
4546                          from_nodeid, result, r->res_master_nodeid,
4547                          r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4548
4549                if (r->res_dir_nodeid != dlm_our_nodeid() &&
4550                    r->res_master_nodeid != dlm_our_nodeid()) {
4551                        /* cause _request_lock->set_master->send_lookup */
4552                        r->res_master_nodeid = 0;
4553                        r->res_nodeid = -1;
4554                        lkb->lkb_nodeid = -1;
4555                }
4556
4557                if (is_overlap(lkb)) {
4558                        /* we'll ignore error in cancel/unlock reply */
4559                        queue_cast_overlap(r, lkb);
4560                        confirm_master(r, result);
4561                        unhold_lkb(lkb); /* undoes create_lkb() */
4562                } else {
4563                        _request_lock(r, lkb);
4564
4565                        if (r->res_master_nodeid == dlm_our_nodeid())
4566                                confirm_master(r, 0);
4567                }
4568                break;
4569
4570        default:
4571                log_error(ls, "receive_request_reply %x error %d",
4572                          lkb->lkb_id, result);
4573        }
4574
4575        if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4576                log_debug(ls, "receive_request_reply %x result %d unlock",
4577                          lkb->lkb_id, result);
4578                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4579                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4580                send_unlock(r, lkb);
4581        } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4582                log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4583                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4584                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4585                send_cancel(r, lkb);
4586        } else {
4587                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4588                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4589        }
4590 out:
4591        unlock_rsb(r);
4592        put_rsb(r);
4593        dlm_put_lkb(lkb);
4594        return 0;
4595}
4596
4597static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4598                                    struct dlm_message *ms)
4599{
4600        /* this is the value returned from do_convert() on the master */
4601        switch (ms->m_result) {
4602        case -EAGAIN:
4603                /* convert would block (be queued) on remote master */
4604                queue_cast(r, lkb, -EAGAIN);
4605                break;
4606
4607        case -EDEADLK:
4608                receive_flags_reply(lkb, ms);
4609                revert_lock_pc(r, lkb);
4610                queue_cast(r, lkb, -EDEADLK);
4611                break;
4612
4613        case -EINPROGRESS:
4614                /* convert was queued on remote master */
4615                receive_flags_reply(lkb, ms);
4616                if (is_demoted(lkb))
4617                        munge_demoted(lkb);
4618                del_lkb(r, lkb);
4619                add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4620                add_timeout(lkb);
4621                break;
4622
4623        case 0:
4624                /* convert was granted on remote master */
4625                receive_flags_reply(lkb, ms);
4626                if (is_demoted(lkb))
4627                        munge_demoted(lkb);
4628                grant_lock_pc(r, lkb, ms);
4629                queue_cast(r, lkb, 0);
4630                break;
4631
4632        default:
4633                log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4634                          lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4635                          ms->m_result);
4636                dlm_print_rsb(r);
4637                dlm_print_lkb(lkb);
4638        }
4639}
4640
4641static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4642{
4643        struct dlm_rsb *r = lkb->lkb_resource;
4644        int error;
4645
4646        hold_rsb(r);
4647        lock_rsb(r);
4648
4649        error = validate_message(lkb, ms);
4650        if (error)
4651                goto out;
4652
4653        /* stub reply can happen with waiters_mutex held */
4654        error = remove_from_waiters_ms(lkb, ms);
4655        if (error)
4656                goto out;
4657
4658        __receive_convert_reply(r, lkb, ms);
4659 out:
4660        unlock_rsb(r);
4661        put_rsb(r);
4662}
4663
4664static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4665{
4666        struct dlm_lkb *lkb;
4667        int error;
4668
4669        error = find_lkb(ls, ms->m_remid, &lkb);
4670        if (error)
4671                return error;
4672
4673        _receive_convert_reply(lkb, ms);
4674        dlm_put_lkb(lkb);
4675        return 0;
4676}
4677
4678static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4679{
4680        struct dlm_rsb *r = lkb->lkb_resource;
4681        int error;
4682
4683        hold_rsb(r);
4684        lock_rsb(r);
4685
4686        error = validate_message(lkb, ms);
4687        if (error)
4688                goto out;
4689
4690        /* stub reply can happen with waiters_mutex held */
4691        error = remove_from_waiters_ms(lkb, ms);
4692        if (error)
4693                goto out;
4694
4695        /* this is the value returned from do_unlock() on the master */
4696
4697        switch (ms->m_result) {
4698        case -DLM_EUNLOCK:
4699                receive_flags_reply(lkb, ms);
4700                remove_lock_pc(r, lkb);
4701                queue_cast(r, lkb, -DLM_EUNLOCK);
4702                break;
4703        case -ENOENT:
4704                break;
4705        default:
4706                log_error(r->res_ls, "receive_unlock_reply %x error %d",
4707                          lkb->lkb_id, ms->m_result);
4708        }
4709 out:
4710        unlock_rsb(r);
4711        put_rsb(r);
4712}
4713
4714static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4715{
4716        struct dlm_lkb *lkb;
4717        int error;
4718
4719        error = find_lkb(ls, ms->m_remid, &lkb);
4720        if (error)
4721                return error;
4722
4723        _receive_unlock_reply(lkb, ms);
4724        dlm_put_lkb(lkb);
4725        return 0;
4726}
4727
4728static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4729{
4730        struct dlm_rsb *r = lkb->lkb_resource;
4731        int error;
4732
4733        hold_rsb(r);
4734        lock_rsb(r);
4735
4736        error = validate_message(lkb, ms);
4737        if (error)
4738                goto out;
4739
4740        /* stub reply can happen with waiters_mutex held */
4741        error = remove_from_waiters_ms(lkb, ms);
4742        if (error)
4743                goto out;
4744
4745        /* this is the value returned from do_cancel() on the master */
4746
4747        switch (ms->m_result) {
4748        case -DLM_ECANCEL:
4749                receive_flags_reply(lkb, ms);
4750                revert_lock_pc(r, lkb);
4751                queue_cast(r, lkb, -DLM_ECANCEL);
4752                break;
4753        case 0:
4754                break;
4755        default:
4756                log_error(r->res_ls, "receive_cancel_reply %x error %d",
4757                          lkb->lkb_id, ms->m_result);
4758        }
4759 out:
4760        unlock_rsb(r);
4761        put_rsb(r);
4762}
4763
4764static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4765{
4766        struct dlm_lkb *lkb;
4767        int error;
4768
4769        error = find_lkb(ls, ms->m_remid, &lkb);
4770        if (error)
4771                return error;
4772
4773        _receive_cancel_reply(lkb, ms);
4774        dlm_put_lkb(lkb);
4775        return 0;
4776}
4777
4778static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4779{
4780        struct dlm_lkb *lkb;
4781        struct dlm_rsb *r;
4782        int error, ret_nodeid;
4783        int do_lookup_list = 0;
4784
4785        error = find_lkb(ls, ms->m_lkid, &lkb);
4786        if (error) {
4787                log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4788                return;
4789        }
4790
4791        /* ms->m_result is the value returned by dlm_master_lookup on dir node
4792           FIXME: will a non-zero error ever be returned? */
4793
4794        r = lkb->lkb_resource;
4795        hold_rsb(r);
4796        lock_rsb(r);
4797
4798        error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4799        if (error)
4800                goto out;
4801
4802        ret_nodeid = ms->m_nodeid;
4803
4804        /* We sometimes receive a request from the dir node for this
4805           rsb before we've received the dir node's loookup_reply for it.
4806           The request from the dir node implies we're the master, so we set
4807           ourself as master in receive_request_reply, and verify here that
4808           we are indeed the master. */
4809
4810        if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4811                /* This should never happen */
4812                log_error(ls, "receive_lookup_reply %x from %d ret %d "
4813                          "master %d dir %d our %d first %x %s",
4814                          lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4815                          r->res_master_nodeid, r->res_dir_nodeid,
4816                          dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4817        }
4818
4819        if (ret_nodeid == dlm_our_nodeid()) {
4820                r->res_master_nodeid = ret_nodeid;
4821                r->res_nodeid = 0;
4822                do_lookup_list = 1;
4823                r->res_first_lkid = 0;
4824        } else if (ret_nodeid == -1) {
4825                /* the remote node doesn't believe it's the dir node */
4826                log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4827                          lkb->lkb_id, ms->m_header.h_nodeid);
4828                r->res_master_nodeid = 0;
4829                r->res_nodeid = -1;
4830                lkb->lkb_nodeid = -1;
4831        } else {
4832                /* set_master() will set lkb_nodeid from r */
4833                r->res_master_nodeid = ret_nodeid;
4834                r->res_nodeid = ret_nodeid;
4835        }
4836
4837        if (is_overlap(lkb)) {
4838                log_debug(ls, "receive_lookup_reply %x unlock %x",
4839                          lkb->lkb_id, lkb->lkb_flags);
4840                queue_cast_overlap(r, lkb);
4841                unhold_lkb(lkb); /* undoes create_lkb() */
4842                goto out_list;
4843        }
4844
4845        _request_lock(r, lkb);
4846
4847 out_list:
4848        if (do_lookup_list)
4849                process_lookup_list(r);
4850 out:
4851        unlock_rsb(r);
4852        put_rsb(r);
4853        dlm_put_lkb(lkb);
4854}
4855
4856static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4857                             uint32_t saved_seq)
4858{
4859        int error = 0, noent = 0;
4860
4861        if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4862                log_limit(ls, "receive %d from non-member %d %x %x %d",
4863                          ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4864                          ms->m_remid, ms->m_result);
4865                return;
4866        }
4867
4868        switch (ms->m_type) {
4869
4870        /* messages sent to a master node */
4871
4872        case DLM_MSG_REQUEST:
4873                error = receive_request(ls, ms);
4874                break;
4875
4876        case DLM_MSG_CONVERT:
4877                error = receive_convert(ls, ms);
4878                break;
4879
4880        case DLM_MSG_UNLOCK:
4881                error = receive_unlock(ls, ms);
4882                break;
4883
4884        case DLM_MSG_CANCEL:
4885                noent = 1;
4886                error = receive_cancel(ls, ms);
4887                break;
4888