linux/fs/dlm/lock.c
<<
>>
Prefs
   1/******************************************************************************
   2*******************************************************************************
   3**
   4**  Copyright (C) 2005-2010 Red Hat, Inc.  All rights reserved.
   5**
   6**  This copyrighted material is made available to anyone wishing to use,
   7**  modify, copy, or redistribute it subject to the terms and conditions
   8**  of the GNU General Public License v.2.
   9**
  10*******************************************************************************
  11******************************************************************************/
  12
  13/* Central locking logic has four stages:
  14
  15   dlm_lock()
  16   dlm_unlock()
  17
  18   request_lock(ls, lkb)
  19   convert_lock(ls, lkb)
  20   unlock_lock(ls, lkb)
  21   cancel_lock(ls, lkb)
  22
  23   _request_lock(r, lkb)
  24   _convert_lock(r, lkb)
  25   _unlock_lock(r, lkb)
  26   _cancel_lock(r, lkb)
  27
  28   do_request(r, lkb)
  29   do_convert(r, lkb)
  30   do_unlock(r, lkb)
  31   do_cancel(r, lkb)
  32
  33   Stage 1 (lock, unlock) is mainly about checking input args and
  34   splitting into one of the four main operations:
  35
  36       dlm_lock          = request_lock
  37       dlm_lock+CONVERT  = convert_lock
  38       dlm_unlock        = unlock_lock
  39       dlm_unlock+CANCEL = cancel_lock
  40
  41   Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
  42   provided to the next stage.
  43
  44   Stage 3, _xxxx_lock(), determines if the operation is local or remote.
  45   When remote, it calls send_xxxx(), when local it calls do_xxxx().
  46
  47   Stage 4, do_xxxx(), is the guts of the operation.  It manipulates the
  48   given rsb and lkb and queues callbacks.
  49
  50   For remote operations, send_xxxx() results in the corresponding do_xxxx()
  51   function being executed on the remote node.  The connecting send/receive
  52   calls on local (L) and remote (R) nodes:
  53
  54   L: send_xxxx()              ->  R: receive_xxxx()
  55                                   R: do_xxxx()
  56   L: receive_xxxx_reply()     <-  R: send_xxxx_reply()
  57*/
  58#include <linux/types.h>
  59#include <linux/rbtree.h>
  60#include <linux/slab.h>
  61#include "dlm_internal.h"
  62#include <linux/dlm_device.h>
  63#include "memory.h"
  64#include "lowcomms.h"
  65#include "requestqueue.h"
  66#include "util.h"
  67#include "dir.h"
  68#include "member.h"
  69#include "lockspace.h"
  70#include "ast.h"
  71#include "lock.h"
  72#include "rcom.h"
  73#include "recover.h"
  74#include "lvb_table.h"
  75#include "user.h"
  76#include "config.h"
  77
  78static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
  79static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
  80static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  81static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
  82static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
  83static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
  84static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
  85static int send_remove(struct dlm_rsb *r);
  86static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  87static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  88static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
  89                                    struct dlm_message *ms);
  90static int receive_extralen(struct dlm_message *ms);
  91static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
  92static void del_timeout(struct dlm_lkb *lkb);
  93static void toss_rsb(struct kref *kref);
  94
  95/*
  96 * Lock compatibilty matrix - thanks Steve
  97 * UN = Unlocked state. Not really a state, used as a flag
  98 * PD = Padding. Used to make the matrix a nice power of two in size
  99 * Other states are the same as the VMS DLM.
 100 * Usage: matrix[grmode+1][rqmode+1]  (although m[rq+1][gr+1] is the same)
 101 */
 102
 103static const int __dlm_compat_matrix[8][8] = {
 104      /* UN NL CR CW PR PW EX PD */
 105        {1, 1, 1, 1, 1, 1, 1, 0},       /* UN */
 106        {1, 1, 1, 1, 1, 1, 1, 0},       /* NL */
 107        {1, 1, 1, 1, 1, 1, 0, 0},       /* CR */
 108        {1, 1, 1, 1, 0, 0, 0, 0},       /* CW */
 109        {1, 1, 1, 0, 1, 0, 0, 0},       /* PR */
 110        {1, 1, 1, 0, 0, 0, 0, 0},       /* PW */
 111        {1, 1, 0, 0, 0, 0, 0, 0},       /* EX */
 112        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 113};
 114
 115/*
 116 * This defines the direction of transfer of LVB data.
 117 * Granted mode is the row; requested mode is the column.
 118 * Usage: matrix[grmode+1][rqmode+1]
 119 * 1 = LVB is returned to the caller
 120 * 0 = LVB is written to the resource
 121 * -1 = nothing happens to the LVB
 122 */
 123
 124const int dlm_lvb_operations[8][8] = {
 125        /* UN   NL  CR  CW  PR  PW  EX  PD*/
 126        {  -1,  1,  1,  1,  1,  1,  1, -1 }, /* UN */
 127        {  -1,  1,  1,  1,  1,  1,  1,  0 }, /* NL */
 128        {  -1, -1,  1,  1,  1,  1,  1,  0 }, /* CR */
 129        {  -1, -1, -1,  1,  1,  1,  1,  0 }, /* CW */
 130        {  -1, -1, -1, -1,  1,  1,  1,  0 }, /* PR */
 131        {  -1,  0,  0,  0,  0,  0,  1,  0 }, /* PW */
 132        {  -1,  0,  0,  0,  0,  0,  0,  0 }, /* EX */
 133        {  -1,  0,  0,  0,  0,  0,  0,  0 }  /* PD */
 134};
 135
 136#define modes_compat(gr, rq) \
 137        __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
 138
 139int dlm_modes_compat(int mode1, int mode2)
 140{
 141        return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
 142}
 143
 144/*
 145 * Compatibility matrix for conversions with QUECVT set.
 146 * Granted mode is the row; requested mode is the column.
 147 * Usage: matrix[grmode+1][rqmode+1]
 148 */
 149
 150static const int __quecvt_compat_matrix[8][8] = {
 151      /* UN NL CR CW PR PW EX PD */
 152        {0, 0, 0, 0, 0, 0, 0, 0},       /* UN */
 153        {0, 0, 1, 1, 1, 1, 1, 0},       /* NL */
 154        {0, 0, 0, 1, 1, 1, 1, 0},       /* CR */
 155        {0, 0, 0, 0, 1, 1, 1, 0},       /* CW */
 156        {0, 0, 0, 1, 0, 1, 1, 0},       /* PR */
 157        {0, 0, 0, 0, 0, 0, 1, 0},       /* PW */
 158        {0, 0, 0, 0, 0, 0, 0, 0},       /* EX */
 159        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 160};
 161
 162void dlm_print_lkb(struct dlm_lkb *lkb)
 163{
 164        printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
 165               "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
 166               lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
 167               lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
 168               lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
 169               (unsigned long long)lkb->lkb_recover_seq);
 170}
 171
 172static void dlm_print_rsb(struct dlm_rsb *r)
 173{
 174        printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
 175               "rlc %d name %s\n",
 176               r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
 177               r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
 178               r->res_name);
 179}
 180
 181void dlm_dump_rsb(struct dlm_rsb *r)
 182{
 183        struct dlm_lkb *lkb;
 184
 185        dlm_print_rsb(r);
 186
 187        printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
 188               list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
 189        printk(KERN_ERR "rsb lookup list\n");
 190        list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
 191                dlm_print_lkb(lkb);
 192        printk(KERN_ERR "rsb grant queue:\n");
 193        list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
 194                dlm_print_lkb(lkb);
 195        printk(KERN_ERR "rsb convert queue:\n");
 196        list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
 197                dlm_print_lkb(lkb);
 198        printk(KERN_ERR "rsb wait queue:\n");
 199        list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
 200                dlm_print_lkb(lkb);
 201}
 202
 203/* Threads cannot use the lockspace while it's being recovered */
 204
 205static inline void dlm_lock_recovery(struct dlm_ls *ls)
 206{
 207        down_read(&ls->ls_in_recovery);
 208}
 209
 210void dlm_unlock_recovery(struct dlm_ls *ls)
 211{
 212        up_read(&ls->ls_in_recovery);
 213}
 214
 215int dlm_lock_recovery_try(struct dlm_ls *ls)
 216{
 217        return down_read_trylock(&ls->ls_in_recovery);
 218}
 219
 220static inline int can_be_queued(struct dlm_lkb *lkb)
 221{
 222        return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
 223}
 224
 225static inline int force_blocking_asts(struct dlm_lkb *lkb)
 226{
 227        return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
 228}
 229
 230static inline int is_demoted(struct dlm_lkb *lkb)
 231{
 232        return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
 233}
 234
 235static inline int is_altmode(struct dlm_lkb *lkb)
 236{
 237        return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
 238}
 239
 240static inline int is_granted(struct dlm_lkb *lkb)
 241{
 242        return (lkb->lkb_status == DLM_LKSTS_GRANTED);
 243}
 244
 245static inline int is_remote(struct dlm_rsb *r)
 246{
 247        DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
 248        return !!r->res_nodeid;
 249}
 250
 251static inline int is_process_copy(struct dlm_lkb *lkb)
 252{
 253        return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
 254}
 255
 256static inline int is_master_copy(struct dlm_lkb *lkb)
 257{
 258        return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
 259}
 260
 261static inline int middle_conversion(struct dlm_lkb *lkb)
 262{
 263        if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
 264            (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
 265                return 1;
 266        return 0;
 267}
 268
 269static inline int down_conversion(struct dlm_lkb *lkb)
 270{
 271        return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
 272}
 273
 274static inline int is_overlap_unlock(struct dlm_lkb *lkb)
 275{
 276        return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
 277}
 278
 279static inline int is_overlap_cancel(struct dlm_lkb *lkb)
 280{
 281        return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
 282}
 283
 284static inline int is_overlap(struct dlm_lkb *lkb)
 285{
 286        return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
 287                                  DLM_IFL_OVERLAP_CANCEL));
 288}
 289
 290static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
 291{
 292        if (is_master_copy(lkb))
 293                return;
 294
 295        del_timeout(lkb);
 296
 297        DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
 298
 299        /* if the operation was a cancel, then return -DLM_ECANCEL, if a
 300           timeout caused the cancel then return -ETIMEDOUT */
 301        if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
 302                lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
 303                rv = -ETIMEDOUT;
 304        }
 305
 306        if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
 307                lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
 308                rv = -EDEADLK;
 309        }
 310
 311        dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
 312}
 313
 314static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
 315{
 316        queue_cast(r, lkb,
 317                   is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
 318}
 319
 320static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
 321{
 322        if (is_master_copy(lkb)) {
 323                send_bast(r, lkb, rqmode);
 324        } else {
 325                dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
 326        }
 327}
 328
 329/*
 330 * Basic operations on rsb's and lkb's
 331 */
 332
 333/* This is only called to add a reference when the code already holds
 334   a valid reference to the rsb, so there's no need for locking. */
 335
 336static inline void hold_rsb(struct dlm_rsb *r)
 337{
 338        kref_get(&r->res_ref);
 339}
 340
 341void dlm_hold_rsb(struct dlm_rsb *r)
 342{
 343        hold_rsb(r);
 344}
 345
 346/* When all references to the rsb are gone it's transferred to
 347   the tossed list for later disposal. */
 348
 349static void put_rsb(struct dlm_rsb *r)
 350{
 351        struct dlm_ls *ls = r->res_ls;
 352        uint32_t bucket = r->res_bucket;
 353
 354        spin_lock(&ls->ls_rsbtbl[bucket].lock);
 355        kref_put(&r->res_ref, toss_rsb);
 356        spin_unlock(&ls->ls_rsbtbl[bucket].lock);
 357}
 358
 359void dlm_put_rsb(struct dlm_rsb *r)
 360{
 361        put_rsb(r);
 362}
 363
 364static int pre_rsb_struct(struct dlm_ls *ls)
 365{
 366        struct dlm_rsb *r1, *r2;
 367        int count = 0;
 368
 369        spin_lock(&ls->ls_new_rsb_spin);
 370        if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
 371                spin_unlock(&ls->ls_new_rsb_spin);
 372                return 0;
 373        }
 374        spin_unlock(&ls->ls_new_rsb_spin);
 375
 376        r1 = dlm_allocate_rsb(ls);
 377        r2 = dlm_allocate_rsb(ls);
 378
 379        spin_lock(&ls->ls_new_rsb_spin);
 380        if (r1) {
 381                list_add(&r1->res_hashchain, &ls->ls_new_rsb);
 382                ls->ls_new_rsb_count++;
 383        }
 384        if (r2) {
 385                list_add(&r2->res_hashchain, &ls->ls_new_rsb);
 386                ls->ls_new_rsb_count++;
 387        }
 388        count = ls->ls_new_rsb_count;
 389        spin_unlock(&ls->ls_new_rsb_spin);
 390
 391        if (!count)
 392                return -ENOMEM;
 393        return 0;
 394}
 395
 396/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
 397   unlock any spinlocks, go back and call pre_rsb_struct again.
 398   Otherwise, take an rsb off the list and return it. */
 399
 400static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
 401                          struct dlm_rsb **r_ret)
 402{
 403        struct dlm_rsb *r;
 404        int count;
 405
 406        spin_lock(&ls->ls_new_rsb_spin);
 407        if (list_empty(&ls->ls_new_rsb)) {
 408                count = ls->ls_new_rsb_count;
 409                spin_unlock(&ls->ls_new_rsb_spin);
 410                log_debug(ls, "find_rsb retry %d %d %s",
 411                          count, dlm_config.ci_new_rsb_count, name);
 412                return -EAGAIN;
 413        }
 414
 415        r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
 416        list_del(&r->res_hashchain);
 417        /* Convert the empty list_head to a NULL rb_node for tree usage: */
 418        memset(&r->res_hashnode, 0, sizeof(struct rb_node));
 419        ls->ls_new_rsb_count--;
 420        spin_unlock(&ls->ls_new_rsb_spin);
 421
 422        r->res_ls = ls;
 423        r->res_length = len;
 424        memcpy(r->res_name, name, len);
 425        mutex_init(&r->res_mutex);
 426
 427        INIT_LIST_HEAD(&r->res_lookup);
 428        INIT_LIST_HEAD(&r->res_grantqueue);
 429        INIT_LIST_HEAD(&r->res_convertqueue);
 430        INIT_LIST_HEAD(&r->res_waitqueue);
 431        INIT_LIST_HEAD(&r->res_root_list);
 432        INIT_LIST_HEAD(&r->res_recover_list);
 433
 434        *r_ret = r;
 435        return 0;
 436}
 437
 438static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
 439{
 440        char maxname[DLM_RESNAME_MAXLEN];
 441
 442        memset(maxname, 0, DLM_RESNAME_MAXLEN);
 443        memcpy(maxname, name, nlen);
 444        return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
 445}
 446
 447int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
 448                        struct dlm_rsb **r_ret)
 449{
 450        struct rb_node *node = tree->rb_node;
 451        struct dlm_rsb *r;
 452        int rc;
 453
 454        while (node) {
 455                r = rb_entry(node, struct dlm_rsb, res_hashnode);
 456                rc = rsb_cmp(r, name, len);
 457                if (rc < 0)
 458                        node = node->rb_left;
 459                else if (rc > 0)
 460                        node = node->rb_right;
 461                else
 462                        goto found;
 463        }
 464        *r_ret = NULL;
 465        return -EBADR;
 466
 467 found:
 468        *r_ret = r;
 469        return 0;
 470}
 471
 472static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
 473{
 474        struct rb_node **newn = &tree->rb_node;
 475        struct rb_node *parent = NULL;
 476        int rc;
 477
 478        while (*newn) {
 479                struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
 480                                               res_hashnode);
 481
 482                parent = *newn;
 483                rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
 484                if (rc < 0)
 485                        newn = &parent->rb_left;
 486                else if (rc > 0)
 487                        newn = &parent->rb_right;
 488                else {
 489                        log_print("rsb_insert match");
 490                        dlm_dump_rsb(rsb);
 491                        dlm_dump_rsb(cur);
 492                        return -EEXIST;
 493                }
 494        }
 495
 496        rb_link_node(&rsb->res_hashnode, parent, newn);
 497        rb_insert_color(&rsb->res_hashnode, tree);
 498        return 0;
 499}
 500
 501/*
 502 * Find rsb in rsbtbl and potentially create/add one
 503 *
 504 * Delaying the release of rsb's has a similar benefit to applications keeping
 505 * NL locks on an rsb, but without the guarantee that the cached master value
 506 * will still be valid when the rsb is reused.  Apps aren't always smart enough
 507 * to keep NL locks on an rsb that they may lock again shortly; this can lead
 508 * to excessive master lookups and removals if we don't delay the release.
 509 *
 510 * Searching for an rsb means looking through both the normal list and toss
 511 * list.  When found on the toss list the rsb is moved to the normal list with
 512 * ref count of 1; when found on normal list the ref count is incremented.
 513 *
 514 * rsb's on the keep list are being used locally and refcounted.
 515 * rsb's on the toss list are not being used locally, and are not refcounted.
 516 *
 517 * The toss list rsb's were either
 518 * - previously used locally but not any more (were on keep list, then
 519 *   moved to toss list when last refcount dropped)
 520 * - created and put on toss list as a directory record for a lookup
 521 *   (we are the dir node for the res, but are not using the res right now,
 522 *   but some other node is)
 523 *
 524 * The purpose of find_rsb() is to return a refcounted rsb for local use.
 525 * So, if the given rsb is on the toss list, it is moved to the keep list
 526 * before being returned.
 527 *
 528 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
 529 * more refcounts exist, so the rsb is moved from the keep list to the
 530 * toss list.
 531 *
 532 * rsb's on both keep and toss lists are used for doing a name to master
 533 * lookups.  rsb's that are in use locally (and being refcounted) are on
 534 * the keep list, rsb's that are not in use locally (not refcounted) and
 535 * only exist for name/master lookups are on the toss list.
 536 *
 537 * rsb's on the toss list who's dir_nodeid is not local can have stale
 538 * name/master mappings.  So, remote requests on such rsb's can potentially
 539 * return with an error, which means the mapping is stale and needs to
 540 * be updated with a new lookup.  (The idea behind MASTER UNCERTAIN and
 541 * first_lkid is to keep only a single outstanding request on an rsb
 542 * while that rsb has a potentially stale master.)
 543 */
 544
 545static int find_rsb_dir(struct dlm_ls *ls, char *name, int len,
 546                        uint32_t hash, uint32_t b,
 547                        int dir_nodeid, int from_nodeid,
 548                        unsigned int flags, struct dlm_rsb **r_ret)
 549{
 550        struct dlm_rsb *r = NULL;
 551        int our_nodeid = dlm_our_nodeid();
 552        int from_local = 0;
 553        int from_other = 0;
 554        int from_dir = 0;
 555        int create = 0;
 556        int error;
 557
 558        if (flags & R_RECEIVE_REQUEST) {
 559                if (from_nodeid == dir_nodeid)
 560                        from_dir = 1;
 561                else
 562                        from_other = 1;
 563        } else if (flags & R_REQUEST) {
 564                from_local = 1;
 565        }
 566
 567        /*
 568         * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
 569         * from_nodeid has sent us a lock in dlm_recover_locks, believing
 570         * we're the new master.  Our local recovery may not have set
 571         * res_master_nodeid to our_nodeid yet, so allow either.  Don't
 572         * create the rsb; dlm_recover_process_copy() will handle EBADR
 573         * by resending.
 574         *
 575         * If someone sends us a request, we are the dir node, and we do
 576         * not find the rsb anywhere, then recreate it.  This happens if
 577         * someone sends us a request after we have removed/freed an rsb
 578         * from our toss list.  (They sent a request instead of lookup
 579         * because they are using an rsb from their toss list.)
 580         */
 581
 582        if (from_local || from_dir ||
 583            (from_other && (dir_nodeid == our_nodeid))) {
 584                create = 1;
 585        }
 586
 587 retry:
 588        if (create) {
 589                error = pre_rsb_struct(ls);
 590                if (error < 0)
 591                        goto out;
 592        }
 593
 594        spin_lock(&ls->ls_rsbtbl[b].lock);
 595
 596        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 597        if (error)
 598                goto do_toss;
 599        
 600        /*
 601         * rsb is active, so we can't check master_nodeid without lock_rsb.
 602         */
 603
 604        kref_get(&r->res_ref);
 605        error = 0;
 606        goto out_unlock;
 607
 608
 609 do_toss:
 610        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 611        if (error)
 612                goto do_new;
 613
 614        /*
 615         * rsb found inactive (master_nodeid may be out of date unless
 616         * we are the dir_nodeid or were the master)  No other thread
 617         * is using this rsb because it's on the toss list, so we can
 618         * look at or update res_master_nodeid without lock_rsb.
 619         */
 620
 621        if ((r->res_master_nodeid != our_nodeid) && from_other) {
 622                /* our rsb was not master, and another node (not the dir node)
 623                   has sent us a request */
 624                log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
 625                          from_nodeid, r->res_master_nodeid, dir_nodeid,
 626                          r->res_name);
 627                error = -ENOTBLK;
 628                goto out_unlock;
 629        }
 630
 631        if ((r->res_master_nodeid != our_nodeid) && from_dir) {
 632                /* don't think this should ever happen */
 633                log_error(ls, "find_rsb toss from_dir %d master %d",
 634                          from_nodeid, r->res_master_nodeid);
 635                dlm_print_rsb(r);
 636                /* fix it and go on */
 637                r->res_master_nodeid = our_nodeid;
 638                r->res_nodeid = 0;
 639                rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
 640                r->res_first_lkid = 0;
 641        }
 642
 643        if (from_local && (r->res_master_nodeid != our_nodeid)) {
 644                /* Because we have held no locks on this rsb,
 645                   res_master_nodeid could have become stale. */
 646                rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
 647                r->res_first_lkid = 0;
 648        }
 649
 650        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 651        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 652        goto out_unlock;
 653
 654
 655 do_new:
 656        /*
 657         * rsb not found
 658         */
 659
 660        if (error == -EBADR && !create)
 661                goto out_unlock;
 662
 663        error = get_rsb_struct(ls, name, len, &r);
 664        if (error == -EAGAIN) {
 665                spin_unlock(&ls->ls_rsbtbl[b].lock);
 666                goto retry;
 667        }
 668        if (error)
 669                goto out_unlock;
 670
 671        r->res_hash = hash;
 672        r->res_bucket = b;
 673        r->res_dir_nodeid = dir_nodeid;
 674        kref_init(&r->res_ref);
 675
 676        if (from_dir) {
 677                /* want to see how often this happens */
 678                log_debug(ls, "find_rsb new from_dir %d recreate %s",
 679                          from_nodeid, r->res_name);
 680                r->res_master_nodeid = our_nodeid;
 681                r->res_nodeid = 0;
 682                goto out_add;
 683        }
 684
 685        if (from_other && (dir_nodeid != our_nodeid)) {
 686                /* should never happen */
 687                log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
 688                          from_nodeid, dir_nodeid, our_nodeid, r->res_name);
 689                dlm_free_rsb(r);
 690                r = NULL;
 691                error = -ENOTBLK;
 692                goto out_unlock;
 693        }
 694
 695        if (from_other) {
 696                log_debug(ls, "find_rsb new from_other %d dir %d %s",
 697                          from_nodeid, dir_nodeid, r->res_name);
 698        }
 699
 700        if (dir_nodeid == our_nodeid) {
 701                /* When we are the dir nodeid, we can set the master
 702                   node immediately */
 703                r->res_master_nodeid = our_nodeid;
 704                r->res_nodeid = 0;
 705        } else {
 706                /* set_master will send_lookup to dir_nodeid */
 707                r->res_master_nodeid = 0;
 708                r->res_nodeid = -1;
 709        }
 710
 711 out_add:
 712        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 713 out_unlock:
 714        spin_unlock(&ls->ls_rsbtbl[b].lock);
 715 out:
 716        *r_ret = r;
 717        return error;
 718}
 719
 720/* During recovery, other nodes can send us new MSTCPY locks (from
 721   dlm_recover_locks) before we've made ourself master (in
 722   dlm_recover_masters). */
 723
 724static int find_rsb_nodir(struct dlm_ls *ls, char *name, int len,
 725                          uint32_t hash, uint32_t b,
 726                          int dir_nodeid, int from_nodeid,
 727                          unsigned int flags, struct dlm_rsb **r_ret)
 728{
 729        struct dlm_rsb *r = NULL;
 730        int our_nodeid = dlm_our_nodeid();
 731        int recover = (flags & R_RECEIVE_RECOVER);
 732        int error;
 733
 734 retry:
 735        error = pre_rsb_struct(ls);
 736        if (error < 0)
 737                goto out;
 738
 739        spin_lock(&ls->ls_rsbtbl[b].lock);
 740
 741        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 742        if (error)
 743                goto do_toss;
 744
 745        /*
 746         * rsb is active, so we can't check master_nodeid without lock_rsb.
 747         */
 748
 749        kref_get(&r->res_ref);
 750        goto out_unlock;
 751
 752
 753 do_toss:
 754        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 755        if (error)
 756                goto do_new;
 757
 758        /*
 759         * rsb found inactive. No other thread is using this rsb because
 760         * it's on the toss list, so we can look at or update
 761         * res_master_nodeid without lock_rsb.
 762         */
 763
 764        if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
 765                /* our rsb is not master, and another node has sent us a
 766                   request; this should never happen */
 767                log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
 768                          from_nodeid, r->res_master_nodeid, dir_nodeid);
 769                dlm_print_rsb(r);
 770                error = -ENOTBLK;
 771                goto out_unlock;
 772        }
 773
 774        if (!recover && (r->res_master_nodeid != our_nodeid) &&
 775            (dir_nodeid == our_nodeid)) {
 776                /* our rsb is not master, and we are dir; may as well fix it;
 777                   this should never happen */
 778                log_error(ls, "find_rsb toss our %d master %d dir %d",
 779                          our_nodeid, r->res_master_nodeid, dir_nodeid);
 780                dlm_print_rsb(r);
 781                r->res_master_nodeid = our_nodeid;
 782                r->res_nodeid = 0;
 783        }
 784
 785        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 786        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 787        goto out_unlock;
 788
 789
 790 do_new:
 791        /*
 792         * rsb not found
 793         */
 794
 795        error = get_rsb_struct(ls, name, len, &r);
 796        if (error == -EAGAIN) {
 797                spin_unlock(&ls->ls_rsbtbl[b].lock);
 798                goto retry;
 799        }
 800        if (error)
 801                goto out_unlock;
 802
 803        r->res_hash = hash;
 804        r->res_bucket = b;
 805        r->res_dir_nodeid = dir_nodeid;
 806        r->res_master_nodeid = dir_nodeid;
 807        r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
 808        kref_init(&r->res_ref);
 809
 810        error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 811 out_unlock:
 812        spin_unlock(&ls->ls_rsbtbl[b].lock);
 813 out:
 814        *r_ret = r;
 815        return error;
 816}
 817
 818static int find_rsb(struct dlm_ls *ls, char *name, int len, int from_nodeid,
 819                    unsigned int flags, struct dlm_rsb **r_ret)
 820{
 821        uint32_t hash, b;
 822        int dir_nodeid;
 823
 824        if (len > DLM_RESNAME_MAXLEN)
 825                return -EINVAL;
 826
 827        hash = jhash(name, len, 0);
 828        b = hash & (ls->ls_rsbtbl_size - 1);
 829
 830        dir_nodeid = dlm_hash2nodeid(ls, hash);
 831
 832        if (dlm_no_directory(ls))
 833                return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
 834                                      from_nodeid, flags, r_ret);
 835        else
 836                return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
 837                                      from_nodeid, flags, r_ret);
 838}
 839
 840/* we have received a request and found that res_master_nodeid != our_nodeid,
 841   so we need to return an error or make ourself the master */
 842
 843static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
 844                                  int from_nodeid)
 845{
 846        if (dlm_no_directory(ls)) {
 847                log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
 848                          from_nodeid, r->res_master_nodeid,
 849                          r->res_dir_nodeid);
 850                dlm_print_rsb(r);
 851                return -ENOTBLK;
 852        }
 853
 854        if (from_nodeid != r->res_dir_nodeid) {
 855                /* our rsb is not master, and another node (not the dir node)
 856                   has sent us a request.  this is much more common when our
 857                   master_nodeid is zero, so limit debug to non-zero.  */
 858
 859                if (r->res_master_nodeid) {
 860                        log_debug(ls, "validate master from_other %d master %d "
 861                                  "dir %d first %x %s", from_nodeid,
 862                                  r->res_master_nodeid, r->res_dir_nodeid,
 863                                  r->res_first_lkid, r->res_name);
 864                }
 865                return -ENOTBLK;
 866        } else {
 867                /* our rsb is not master, but the dir nodeid has sent us a
 868                   request; this could happen with master 0 / res_nodeid -1 */
 869
 870                if (r->res_master_nodeid) {
 871                        log_error(ls, "validate master from_dir %d master %d "
 872                                  "first %x %s",
 873                                  from_nodeid, r->res_master_nodeid,
 874                                  r->res_first_lkid, r->res_name);
 875                }
 876
 877                r->res_master_nodeid = dlm_our_nodeid();
 878                r->res_nodeid = 0;
 879                return 0;
 880        }
 881}
 882
 883/*
 884 * We're the dir node for this res and another node wants to know the
 885 * master nodeid.  During normal operation (non recovery) this is only
 886 * called from receive_lookup(); master lookups when the local node is
 887 * the dir node are done by find_rsb().
 888 *
 889 * normal operation, we are the dir node for a resource
 890 * . _request_lock
 891 * . set_master
 892 * . send_lookup
 893 * . receive_lookup
 894 * . dlm_master_lookup flags 0
 895 *
 896 * recover directory, we are rebuilding dir for all resources
 897 * . dlm_recover_directory
 898 * . dlm_rcom_names
 899 *   remote node sends back the rsb names it is master of and we are dir of
 900 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
 901 *   we either create new rsb setting remote node as master, or find existing
 902 *   rsb and set master to be the remote node.
 903 *
 904 * recover masters, we are finding the new master for resources
 905 * . dlm_recover_masters
 906 * . recover_master
 907 * . dlm_send_rcom_lookup
 908 * . receive_rcom_lookup
 909 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
 910 */
 911
 912int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
 913                      unsigned int flags, int *r_nodeid, int *result)
 914{
 915        struct dlm_rsb *r = NULL;
 916        uint32_t hash, b;
 917        int from_master = (flags & DLM_LU_RECOVER_DIR);
 918        int fix_master = (flags & DLM_LU_RECOVER_MASTER);
 919        int our_nodeid = dlm_our_nodeid();
 920        int dir_nodeid, error, toss_list = 0;
 921
 922        if (len > DLM_RESNAME_MAXLEN)
 923                return -EINVAL;
 924
 925        if (from_nodeid == our_nodeid) {
 926                log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
 927                          our_nodeid, flags);
 928                return -EINVAL;
 929        }
 930
 931        hash = jhash(name, len, 0);
 932        b = hash & (ls->ls_rsbtbl_size - 1);
 933
 934        dir_nodeid = dlm_hash2nodeid(ls, hash);
 935        if (dir_nodeid != our_nodeid) {
 936                log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
 937                          from_nodeid, dir_nodeid, our_nodeid, hash,
 938                          ls->ls_num_nodes);
 939                *r_nodeid = -1;
 940                return -EINVAL;
 941        }
 942
 943 retry:
 944        error = pre_rsb_struct(ls);
 945        if (error < 0)
 946                return error;
 947
 948        spin_lock(&ls->ls_rsbtbl[b].lock);
 949        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 950        if (!error) {
 951                /* because the rsb is active, we need to lock_rsb before
 952                   checking/changing re_master_nodeid */
 953
 954                hold_rsb(r);
 955                spin_unlock(&ls->ls_rsbtbl[b].lock);
 956                lock_rsb(r);
 957                goto found;
 958        }
 959
 960        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 961        if (error)
 962                goto not_found;
 963
 964        /* because the rsb is inactive (on toss list), it's not refcounted
 965           and lock_rsb is not used, but is protected by the rsbtbl lock */
 966
 967        toss_list = 1;
 968 found:
 969        if (r->res_dir_nodeid != our_nodeid) {
 970                /* should not happen, but may as well fix it and carry on */
 971                log_error(ls, "dlm_master_lookup res_dir %d our %d %s",
 972                          r->res_dir_nodeid, our_nodeid, r->res_name);
 973                r->res_dir_nodeid = our_nodeid;
 974        }
 975
 976        if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
 977                /* Recovery uses this function to set a new master when
 978                   the previous master failed.  Setting NEW_MASTER will
 979                   force dlm_recover_masters to call recover_master on this
 980                   rsb even though the res_nodeid is no longer removed. */
 981
 982                r->res_master_nodeid = from_nodeid;
 983                r->res_nodeid = from_nodeid;
 984                rsb_set_flag(r, RSB_NEW_MASTER);
 985
 986                if (toss_list) {
 987                        /* I don't think we should ever find it on toss list. */
 988                        log_error(ls, "dlm_master_lookup fix_master on toss");
 989                        dlm_dump_rsb(r);
 990                }
 991        }
 992
 993        if (from_master && (r->res_master_nodeid != from_nodeid)) {
 994                /* this will happen if from_nodeid became master during
 995                   a previous recovery cycle, and we aborted the previous
 996                   cycle before recovering this master value */
 997
 998                log_limit(ls, "dlm_master_lookup from_master %d "
 999                          "master_nodeid %d res_nodeid %d first %x %s",
1000                          from_nodeid, r->res_master_nodeid, r->res_nodeid,
1001                          r->res_first_lkid, r->res_name);
1002
1003                if (r->res_master_nodeid == our_nodeid) {
1004                        log_error(ls, "from_master %d our_master", from_nodeid);
1005                        dlm_dump_rsb(r);
1006                        dlm_send_rcom_lookup_dump(r, from_nodeid);
1007                        goto out_found;
1008                }
1009
1010                r->res_master_nodeid = from_nodeid;
1011                r->res_nodeid = from_nodeid;
1012                rsb_set_flag(r, RSB_NEW_MASTER);
1013        }
1014
1015        if (!r->res_master_nodeid) {
1016                /* this will happen if recovery happens while we're looking
1017                   up the master for this rsb */
1018
1019                log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s",
1020                          from_nodeid, r->res_first_lkid, r->res_name);
1021                r->res_master_nodeid = from_nodeid;
1022                r->res_nodeid = from_nodeid;
1023        }
1024
1025        if (!from_master && !fix_master &&
1026            (r->res_master_nodeid == from_nodeid)) {
1027                /* this can happen when the master sends remove, the dir node
1028                   finds the rsb on the keep list and ignores the remove,
1029                   and the former master sends a lookup */
1030
1031                log_limit(ls, "dlm_master_lookup from master %d flags %x "
1032                          "first %x %s", from_nodeid, flags,
1033                          r->res_first_lkid, r->res_name);
1034        }
1035
1036 out_found:
1037        *r_nodeid = r->res_master_nodeid;
1038        if (result)
1039                *result = DLM_LU_MATCH;
1040
1041        if (toss_list) {
1042                r->res_toss_time = jiffies;
1043                /* the rsb was inactive (on toss list) */
1044                spin_unlock(&ls->ls_rsbtbl[b].lock);
1045        } else {
1046                /* the rsb was active */
1047                unlock_rsb(r);
1048                put_rsb(r);
1049        }
1050        return 0;
1051
1052 not_found:
1053        error = get_rsb_struct(ls, name, len, &r);
1054        if (error == -EAGAIN) {
1055                spin_unlock(&ls->ls_rsbtbl[b].lock);
1056                goto retry;
1057        }
1058        if (error)
1059                goto out_unlock;
1060
1061        r->res_hash = hash;
1062        r->res_bucket = b;
1063        r->res_dir_nodeid = our_nodeid;
1064        r->res_master_nodeid = from_nodeid;
1065        r->res_nodeid = from_nodeid;
1066        kref_init(&r->res_ref);
1067        r->res_toss_time = jiffies;
1068
1069        error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1070        if (error) {
1071                /* should never happen */
1072                dlm_free_rsb(r);
1073                spin_unlock(&ls->ls_rsbtbl[b].lock);
1074                goto retry;
1075        }
1076
1077        if (result)
1078                *result = DLM_LU_ADD;
1079        *r_nodeid = from_nodeid;
1080        error = 0;
1081 out_unlock:
1082        spin_unlock(&ls->ls_rsbtbl[b].lock);
1083        return error;
1084}
1085
1086static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1087{
1088        struct rb_node *n;
1089        struct dlm_rsb *r;
1090        int i;
1091
1092        for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1093                spin_lock(&ls->ls_rsbtbl[i].lock);
1094                for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1095                        r = rb_entry(n, struct dlm_rsb, res_hashnode);
1096                        if (r->res_hash == hash)
1097                                dlm_dump_rsb(r);
1098                }
1099                spin_unlock(&ls->ls_rsbtbl[i].lock);
1100        }
1101}
1102
1103void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1104{
1105        struct dlm_rsb *r = NULL;
1106        uint32_t hash, b;
1107        int error;
1108
1109        hash = jhash(name, len, 0);
1110        b = hash & (ls->ls_rsbtbl_size - 1);
1111
1112        spin_lock(&ls->ls_rsbtbl[b].lock);
1113        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1114        if (!error)
1115                goto out_dump;
1116
1117        error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1118        if (error)
1119                goto out;
1120 out_dump:
1121        dlm_dump_rsb(r);
1122 out:
1123        spin_unlock(&ls->ls_rsbtbl[b].lock);
1124}
1125
1126static void toss_rsb(struct kref *kref)
1127{
1128        struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1129        struct dlm_ls *ls = r->res_ls;
1130
1131        DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1132        kref_init(&r->res_ref);
1133        rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1134        rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1135        r->res_toss_time = jiffies;
1136        ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1137        if (r->res_lvbptr) {
1138                dlm_free_lvb(r->res_lvbptr);
1139                r->res_lvbptr = NULL;
1140        }
1141}
1142
1143/* See comment for unhold_lkb */
1144
1145static void unhold_rsb(struct dlm_rsb *r)
1146{
1147        int rv;
1148        rv = kref_put(&r->res_ref, toss_rsb);
1149        DLM_ASSERT(!rv, dlm_dump_rsb(r););
1150}
1151
1152static void kill_rsb(struct kref *kref)
1153{
1154        struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1155
1156        /* All work is done after the return from kref_put() so we
1157           can release the write_lock before the remove and free. */
1158
1159        DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1160        DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1161        DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1162        DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1163        DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1164        DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1165}
1166
1167/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1168   The rsb must exist as long as any lkb's for it do. */
1169
1170static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1171{
1172        hold_rsb(r);
1173        lkb->lkb_resource = r;
1174}
1175
1176static void detach_lkb(struct dlm_lkb *lkb)
1177{
1178        if (lkb->lkb_resource) {
1179                put_rsb(lkb->lkb_resource);
1180                lkb->lkb_resource = NULL;
1181        }
1182}
1183
1184static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1185{
1186        struct dlm_lkb *lkb;
1187        int rv;
1188
1189        lkb = dlm_allocate_lkb(ls);
1190        if (!lkb)
1191                return -ENOMEM;
1192
1193        lkb->lkb_nodeid = -1;
1194        lkb->lkb_grmode = DLM_LOCK_IV;
1195        kref_init(&lkb->lkb_ref);
1196        INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1197        INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1198        INIT_LIST_HEAD(&lkb->lkb_time_list);
1199        INIT_LIST_HEAD(&lkb->lkb_cb_list);
1200        mutex_init(&lkb->lkb_cb_mutex);
1201        INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
1202
1203        idr_preload(GFP_NOFS);
1204        spin_lock(&ls->ls_lkbidr_spin);
1205        rv = idr_alloc(&ls->ls_lkbidr, lkb, 1, 0, GFP_NOWAIT);
1206        if (rv >= 0)
1207                lkb->lkb_id = rv;
1208        spin_unlock(&ls->ls_lkbidr_spin);
1209        idr_preload_end();
1210
1211        if (rv < 0) {
1212                log_error(ls, "create_lkb idr error %d", rv);
1213                return rv;
1214        }
1215
1216        *lkb_ret = lkb;
1217        return 0;
1218}
1219
1220static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1221{
1222        struct dlm_lkb *lkb;
1223
1224        spin_lock(&ls->ls_lkbidr_spin);
1225        lkb = idr_find(&ls->ls_lkbidr, lkid);
1226        if (lkb)
1227                kref_get(&lkb->lkb_ref);
1228        spin_unlock(&ls->ls_lkbidr_spin);
1229
1230        *lkb_ret = lkb;
1231        return lkb ? 0 : -ENOENT;
1232}
1233
1234static void kill_lkb(struct kref *kref)
1235{
1236        struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1237
1238        /* All work is done after the return from kref_put() so we
1239           can release the write_lock before the detach_lkb */
1240
1241        DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1242}
1243
1244/* __put_lkb() is used when an lkb may not have an rsb attached to
1245   it so we need to provide the lockspace explicitly */
1246
1247static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1248{
1249        uint32_t lkid = lkb->lkb_id;
1250
1251        spin_lock(&ls->ls_lkbidr_spin);
1252        if (kref_put(&lkb->lkb_ref, kill_lkb)) {
1253                idr_remove(&ls->ls_lkbidr, lkid);
1254                spin_unlock(&ls->ls_lkbidr_spin);
1255
1256                detach_lkb(lkb);
1257
1258                /* for local/process lkbs, lvbptr points to caller's lksb */
1259                if (lkb->lkb_lvbptr && is_master_copy(lkb))
1260                        dlm_free_lvb(lkb->lkb_lvbptr);
1261                dlm_free_lkb(lkb);
1262                return 1;
1263        } else {
1264                spin_unlock(&ls->ls_lkbidr_spin);
1265                return 0;
1266        }
1267}
1268
1269int dlm_put_lkb(struct dlm_lkb *lkb)
1270{
1271        struct dlm_ls *ls;
1272
1273        DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1274        DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1275
1276        ls = lkb->lkb_resource->res_ls;
1277        return __put_lkb(ls, lkb);
1278}
1279
1280/* This is only called to add a reference when the code already holds
1281   a valid reference to the lkb, so there's no need for locking. */
1282
1283static inline void hold_lkb(struct dlm_lkb *lkb)
1284{
1285        kref_get(&lkb->lkb_ref);
1286}
1287
1288/* This is called when we need to remove a reference and are certain
1289   it's not the last ref.  e.g. del_lkb is always called between a
1290   find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1291   put_lkb would work fine, but would involve unnecessary locking */
1292
1293static inline void unhold_lkb(struct dlm_lkb *lkb)
1294{
1295        int rv;
1296        rv = kref_put(&lkb->lkb_ref, kill_lkb);
1297        DLM_ASSERT(!rv, dlm_print_lkb(lkb););
1298}
1299
1300static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1301                            int mode)
1302{
1303        struct dlm_lkb *lkb = NULL;
1304
1305        list_for_each_entry(lkb, head, lkb_statequeue)
1306                if (lkb->lkb_rqmode < mode)
1307                        break;
1308
1309        __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
1310}
1311
1312/* add/remove lkb to rsb's grant/convert/wait queue */
1313
1314static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1315{
1316        kref_get(&lkb->lkb_ref);
1317
1318        DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1319
1320        lkb->lkb_timestamp = ktime_get();
1321
1322        lkb->lkb_status = status;
1323
1324        switch (status) {
1325        case DLM_LKSTS_WAITING:
1326                if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1327                        list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1328                else
1329                        list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1330                break;
1331        case DLM_LKSTS_GRANTED:
1332                /* convention says granted locks kept in order of grmode */
1333                lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1334                                lkb->lkb_grmode);
1335                break;
1336        case DLM_LKSTS_CONVERT:
1337                if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1338                        list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1339                else
1340                        list_add_tail(&lkb->lkb_statequeue,
1341                                      &r->res_convertqueue);
1342                break;
1343        default:
1344                DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1345        }
1346}
1347
1348static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1349{
1350        lkb->lkb_status = 0;
1351        list_del(&lkb->lkb_statequeue);
1352        unhold_lkb(lkb);
1353}
1354
1355static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1356{
1357        hold_lkb(lkb);
1358        del_lkb(r, lkb);
1359        add_lkb(r, lkb, sts);
1360        unhold_lkb(lkb);
1361}
1362
1363static int msg_reply_type(int mstype)
1364{
1365        switch (mstype) {
1366        case DLM_MSG_REQUEST:
1367                return DLM_MSG_REQUEST_REPLY;
1368        case DLM_MSG_CONVERT:
1369                return DLM_MSG_CONVERT_REPLY;
1370        case DLM_MSG_UNLOCK:
1371                return DLM_MSG_UNLOCK_REPLY;
1372        case DLM_MSG_CANCEL:
1373                return DLM_MSG_CANCEL_REPLY;
1374        case DLM_MSG_LOOKUP:
1375                return DLM_MSG_LOOKUP_REPLY;
1376        }
1377        return -1;
1378}
1379
1380static int nodeid_warned(int nodeid, int num_nodes, int *warned)
1381{
1382        int i;
1383
1384        for (i = 0; i < num_nodes; i++) {
1385                if (!warned[i]) {
1386                        warned[i] = nodeid;
1387                        return 0;
1388                }
1389                if (warned[i] == nodeid)
1390                        return 1;
1391        }
1392        return 0;
1393}
1394
1395void dlm_scan_waiters(struct dlm_ls *ls)
1396{
1397        struct dlm_lkb *lkb;
1398        s64 us;
1399        s64 debug_maxus = 0;
1400        u32 debug_scanned = 0;
1401        u32 debug_expired = 0;
1402        int num_nodes = 0;
1403        int *warned = NULL;
1404
1405        if (!dlm_config.ci_waitwarn_us)
1406                return;
1407
1408        mutex_lock(&ls->ls_waiters_mutex);
1409
1410        list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1411                if (!lkb->lkb_wait_time)
1412                        continue;
1413
1414                debug_scanned++;
1415
1416                us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
1417
1418                if (us < dlm_config.ci_waitwarn_us)
1419                        continue;
1420
1421                lkb->lkb_wait_time = 0;
1422
1423                debug_expired++;
1424                if (us > debug_maxus)
1425                        debug_maxus = us;
1426
1427                if (!num_nodes) {
1428                        num_nodes = ls->ls_num_nodes;
1429                        warned = kcalloc(num_nodes, sizeof(int), GFP_KERNEL);
1430                }
1431                if (!warned)
1432                        continue;
1433                if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
1434                        continue;
1435
1436                log_error(ls, "waitwarn %x %lld %d us check connection to "
1437                          "node %d", lkb->lkb_id, (long long)us,
1438                          dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
1439        }
1440        mutex_unlock(&ls->ls_waiters_mutex);
1441        kfree(warned);
1442
1443        if (debug_expired)
1444                log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
1445                          debug_scanned, debug_expired,
1446                          dlm_config.ci_waitwarn_us, (long long)debug_maxus);
1447}
1448
1449/* add/remove lkb from global waiters list of lkb's waiting for
1450   a reply from a remote node */
1451
1452static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1453{
1454        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1455        int error = 0;
1456
1457        mutex_lock(&ls->ls_waiters_mutex);
1458
1459        if (is_overlap_unlock(lkb) ||
1460            (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1461                error = -EINVAL;
1462                goto out;
1463        }
1464
1465        if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1466                switch (mstype) {
1467                case DLM_MSG_UNLOCK:
1468                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1469                        break;
1470                case DLM_MSG_CANCEL:
1471                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1472                        break;
1473                default:
1474                        error = -EBUSY;
1475                        goto out;
1476                }
1477                lkb->lkb_wait_count++;
1478                hold_lkb(lkb);
1479
1480                log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1481                          lkb->lkb_id, lkb->lkb_wait_type, mstype,
1482                          lkb->lkb_wait_count, lkb->lkb_flags);
1483                goto out;
1484        }
1485
1486        DLM_ASSERT(!lkb->lkb_wait_count,
1487                   dlm_print_lkb(lkb);
1488                   printk("wait_count %d\n", lkb->lkb_wait_count););
1489
1490        lkb->lkb_wait_count++;
1491        lkb->lkb_wait_type = mstype;
1492        lkb->lkb_wait_time = ktime_get();
1493        lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1494        hold_lkb(lkb);
1495        list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1496 out:
1497        if (error)
1498                log_error(ls, "addwait error %x %d flags %x %d %d %s",
1499                          lkb->lkb_id, error, lkb->lkb_flags, mstype,
1500                          lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1501        mutex_unlock(&ls->ls_waiters_mutex);
1502        return error;
1503}
1504
1505/* We clear the RESEND flag because we might be taking an lkb off the waiters
1506   list as part of process_requestqueue (e.g. a lookup that has an optimized
1507   request reply on the requestqueue) between dlm_recover_waiters_pre() which
1508   set RESEND and dlm_recover_waiters_post() */
1509
1510static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1511                                struct dlm_message *ms)
1512{
1513        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1514        int overlap_done = 0;
1515
1516        if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1517                log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1518                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1519                overlap_done = 1;
1520                goto out_del;
1521        }
1522
1523        if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1524                log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1525                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1526                overlap_done = 1;
1527                goto out_del;
1528        }
1529
1530        /* Cancel state was preemptively cleared by a successful convert,
1531           see next comment, nothing to do. */
1532
1533        if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1534            (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1535                log_debug(ls, "remwait %x cancel_reply wait_type %d",
1536                          lkb->lkb_id, lkb->lkb_wait_type);
1537                return -1;
1538        }
1539
1540        /* Remove for the convert reply, and premptively remove for the
1541           cancel reply.  A convert has been granted while there's still
1542           an outstanding cancel on it (the cancel is moot and the result
1543           in the cancel reply should be 0).  We preempt the cancel reply
1544           because the app gets the convert result and then can follow up
1545           with another op, like convert.  This subsequent op would see the
1546           lingering state of the cancel and fail with -EBUSY. */
1547
1548        if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1549            (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1550            is_overlap_cancel(lkb) && ms && !ms->m_result) {
1551                log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1552                          lkb->lkb_id);
1553                lkb->lkb_wait_type = 0;
1554                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1555                lkb->lkb_wait_count--;
1556                goto out_del;
1557        }
1558
1559        /* N.B. type of reply may not always correspond to type of original
1560           msg due to lookup->request optimization, verify others? */
1561
1562        if (lkb->lkb_wait_type) {
1563                lkb->lkb_wait_type = 0;
1564                goto out_del;
1565        }
1566
1567        log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1568                  lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid,
1569                  mstype, lkb->lkb_flags);
1570        return -1;
1571
1572 out_del:
1573        /* the force-unlock/cancel has completed and we haven't recvd a reply
1574           to the op that was in progress prior to the unlock/cancel; we
1575           give up on any reply to the earlier op.  FIXME: not sure when/how
1576           this would happen */
1577
1578        if (overlap_done && lkb->lkb_wait_type) {
1579                log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1580                          lkb->lkb_id, mstype, lkb->lkb_wait_type);
1581                lkb->lkb_wait_count--;
1582                lkb->lkb_wait_type = 0;
1583        }
1584
1585        DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1586
1587        lkb->lkb_flags &= ~DLM_IFL_RESEND;
1588        lkb->lkb_wait_count--;
1589        if (!lkb->lkb_wait_count)
1590                list_del_init(&lkb->lkb_wait_reply);
1591        unhold_lkb(lkb);
1592        return 0;
1593}
1594
1595static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1596{
1597        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1598        int error;
1599
1600        mutex_lock(&ls->ls_waiters_mutex);
1601        error = _remove_from_waiters(lkb, mstype, NULL);
1602        mutex_unlock(&ls->ls_waiters_mutex);
1603        return error;
1604}
1605
1606/* Handles situations where we might be processing a "fake" or "stub" reply in
1607   which we can't try to take waiters_mutex again. */
1608
1609static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1610{
1611        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1612        int error;
1613
1614        if (ms->m_flags != DLM_IFL_STUB_MS)
1615                mutex_lock(&ls->ls_waiters_mutex);
1616        error = _remove_from_waiters(lkb, ms->m_type, ms);
1617        if (ms->m_flags != DLM_IFL_STUB_MS)
1618                mutex_unlock(&ls->ls_waiters_mutex);
1619        return error;
1620}
1621
1622/* If there's an rsb for the same resource being removed, ensure
1623   that the remove message is sent before the new lookup message.
1624   It should be rare to need a delay here, but if not, then it may
1625   be worthwhile to add a proper wait mechanism rather than a delay. */
1626
1627static void wait_pending_remove(struct dlm_rsb *r)
1628{
1629        struct dlm_ls *ls = r->res_ls;
1630 restart:
1631        spin_lock(&ls->ls_remove_spin);
1632        if (ls->ls_remove_len &&
1633            !rsb_cmp(r, ls->ls_remove_name, ls->ls_remove_len)) {
1634                log_debug(ls, "delay lookup for remove dir %d %s",
1635                          r->res_dir_nodeid, r->res_name);
1636                spin_unlock(&ls->ls_remove_spin);
1637                msleep(1);
1638                goto restart;
1639        }
1640        spin_unlock(&ls->ls_remove_spin);
1641}
1642
1643/*
1644 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1645 * read by other threads in wait_pending_remove.  ls_remove_names
1646 * and ls_remove_lens are only used by the scan thread, so they do
1647 * not need protection.
1648 */
1649
1650static void shrink_bucket(struct dlm_ls *ls, int b)
1651{
1652        struct rb_node *n, *next;
1653        struct dlm_rsb *r;
1654        char *name;
1655        int our_nodeid = dlm_our_nodeid();
1656        int remote_count = 0;
1657        int need_shrink = 0;
1658        int i, len, rv;
1659
1660        memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1661
1662        spin_lock(&ls->ls_rsbtbl[b].lock);
1663
1664        if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1665                spin_unlock(&ls->ls_rsbtbl[b].lock);
1666                return;
1667        }
1668
1669        for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1670                next = rb_next(n);
1671                r = rb_entry(n, struct dlm_rsb, res_hashnode);
1672
1673                /* If we're the directory record for this rsb, and
1674                   we're not the master of it, then we need to wait
1675                   for the master node to send us a dir remove for
1676                   before removing the dir record. */
1677
1678                if (!dlm_no_directory(ls) &&
1679                    (r->res_master_nodeid != our_nodeid) &&
1680                    (dlm_dir_nodeid(r) == our_nodeid)) {
1681                        continue;
1682                }
1683
1684                need_shrink = 1;
1685
1686                if (!time_after_eq(jiffies, r->res_toss_time +
1687                                   dlm_config.ci_toss_secs * HZ)) {
1688                        continue;
1689                }
1690
1691                if (!dlm_no_directory(ls) &&
1692                    (r->res_master_nodeid == our_nodeid) &&
1693                    (dlm_dir_nodeid(r) != our_nodeid)) {
1694
1695                        /* We're the master of this rsb but we're not
1696                           the directory record, so we need to tell the
1697                           dir node to remove the dir record. */
1698
1699                        ls->ls_remove_lens[remote_count] = r->res_length;
1700                        memcpy(ls->ls_remove_names[remote_count], r->res_name,
1701                               DLM_RESNAME_MAXLEN);
1702                        remote_count++;
1703
1704                        if (remote_count >= DLM_REMOVE_NAMES_MAX)
1705                                break;
1706                        continue;
1707                }
1708
1709                if (!kref_put(&r->res_ref, kill_rsb)) {
1710                        log_error(ls, "tossed rsb in use %s", r->res_name);
1711                        continue;
1712                }
1713
1714                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1715                dlm_free_rsb(r);
1716        }
1717
1718        if (need_shrink)
1719                ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1720        else
1721                ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1722        spin_unlock(&ls->ls_rsbtbl[b].lock);
1723
1724        /*
1725         * While searching for rsb's to free, we found some that require
1726         * remote removal.  We leave them in place and find them again here
1727         * so there is a very small gap between removing them from the toss
1728         * list and sending the removal.  Keeping this gap small is
1729         * important to keep us (the master node) from being out of sync
1730         * with the remote dir node for very long.
1731         *
1732         * From the time the rsb is removed from toss until just after
1733         * send_remove, the rsb name is saved in ls_remove_name.  A new
1734         * lookup checks this to ensure that a new lookup message for the
1735         * same resource name is not sent just before the remove message.
1736         */
1737
1738        for (i = 0; i < remote_count; i++) {
1739                name = ls->ls_remove_names[i];
1740                len = ls->ls_remove_lens[i];
1741
1742                spin_lock(&ls->ls_rsbtbl[b].lock);
1743                rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1744                if (rv) {
1745                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1746                        log_debug(ls, "remove_name not toss %s", name);
1747                        continue;
1748                }
1749
1750                if (r->res_master_nodeid != our_nodeid) {
1751                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1752                        log_debug(ls, "remove_name master %d dir %d our %d %s",
1753                                  r->res_master_nodeid, r->res_dir_nodeid,
1754                                  our_nodeid, name);
1755                        continue;
1756                }
1757
1758                if (r->res_dir_nodeid == our_nodeid) {
1759                        /* should never happen */
1760                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1761                        log_error(ls, "remove_name dir %d master %d our %d %s",
1762                                  r->res_dir_nodeid, r->res_master_nodeid,
1763                                  our_nodeid, name);
1764                        continue;
1765                }
1766
1767                if (!time_after_eq(jiffies, r->res_toss_time +
1768                                   dlm_config.ci_toss_secs * HZ)) {
1769                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1770                        log_debug(ls, "remove_name toss_time %lu now %lu %s",
1771                                  r->res_toss_time, jiffies, name);
1772                        continue;
1773                }
1774
1775                if (!kref_put(&r->res_ref, kill_rsb)) {
1776                        spin_unlock(&ls->ls_rsbtbl[b].lock);
1777                        log_error(ls, "remove_name in use %s", name);
1778                        continue;
1779                }
1780
1781                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1782
1783                /* block lookup of same name until we've sent remove */
1784                spin_lock(&ls->ls_remove_spin);
1785                ls->ls_remove_len = len;
1786                memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1787                spin_unlock(&ls->ls_remove_spin);
1788                spin_unlock(&ls->ls_rsbtbl[b].lock);
1789
1790                send_remove(r);
1791
1792                /* allow lookup of name again */
1793                spin_lock(&ls->ls_remove_spin);
1794                ls->ls_remove_len = 0;
1795                memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1796                spin_unlock(&ls->ls_remove_spin);
1797
1798                dlm_free_rsb(r);
1799        }
1800}
1801
1802void dlm_scan_rsbs(struct dlm_ls *ls)
1803{
1804        int i;
1805
1806        for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1807                shrink_bucket(ls, i);
1808                if (dlm_locking_stopped(ls))
1809                        break;
1810                cond_resched();
1811        }
1812}
1813
1814static void add_timeout(struct dlm_lkb *lkb)
1815{
1816        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1817
1818        if (is_master_copy(lkb))
1819                return;
1820
1821        if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1822            !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1823                lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1824                goto add_it;
1825        }
1826        if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1827                goto add_it;
1828        return;
1829
1830 add_it:
1831        DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1832        mutex_lock(&ls->ls_timeout_mutex);
1833        hold_lkb(lkb);
1834        list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1835        mutex_unlock(&ls->ls_timeout_mutex);
1836}
1837
1838static void del_timeout(struct dlm_lkb *lkb)
1839{
1840        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1841
1842        mutex_lock(&ls->ls_timeout_mutex);
1843        if (!list_empty(&lkb->lkb_time_list)) {
1844                list_del_init(&lkb->lkb_time_list);
1845                unhold_lkb(lkb);
1846        }
1847        mutex_unlock(&ls->ls_timeout_mutex);
1848}
1849
1850/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1851   lkb_lksb_timeout without lock_rsb?  Note: we can't lock timeout_mutex
1852   and then lock rsb because of lock ordering in add_timeout.  We may need
1853   to specify some special timeout-related bits in the lkb that are just to
1854   be accessed under the timeout_mutex. */
1855
1856void dlm_scan_timeout(struct dlm_ls *ls)
1857{
1858        struct dlm_rsb *r;
1859        struct dlm_lkb *lkb;
1860        int do_cancel, do_warn;
1861        s64 wait_us;
1862
1863        for (;;) {
1864                if (dlm_locking_stopped(ls))
1865                        break;
1866
1867                do_cancel = 0;
1868                do_warn = 0;
1869                mutex_lock(&ls->ls_timeout_mutex);
1870                list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1871
1872                        wait_us = ktime_to_us(ktime_sub(ktime_get(),
1873                                                        lkb->lkb_timestamp));
1874
1875                        if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1876                            wait_us >= (lkb->lkb_timeout_cs * 10000))
1877                                do_cancel = 1;
1878
1879                        if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1880                            wait_us >= dlm_config.ci_timewarn_cs * 10000)
1881                                do_warn = 1;
1882
1883                        if (!do_cancel && !do_warn)
1884                                continue;
1885                        hold_lkb(lkb);
1886                        break;
1887                }
1888                mutex_unlock(&ls->ls_timeout_mutex);
1889
1890                if (!do_cancel && !do_warn)
1891                        break;
1892
1893                r = lkb->lkb_resource;
1894                hold_rsb(r);
1895                lock_rsb(r);
1896
1897                if (do_warn) {
1898                        /* clear flag so we only warn once */
1899                        lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1900                        if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1901                                del_timeout(lkb);
1902                        dlm_timeout_warn(lkb);
1903                }
1904
1905                if (do_cancel) {
1906                        log_debug(ls, "timeout cancel %x node %d %s",
1907                                  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1908                        lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1909                        lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1910                        del_timeout(lkb);
1911                        _cancel_lock(r, lkb);
1912                }
1913
1914                unlock_rsb(r);
1915                unhold_rsb(r);
1916                dlm_put_lkb(lkb);
1917        }
1918}
1919
1920/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1921   dlm_recoverd before checking/setting ls_recover_begin. */
1922
1923void dlm_adjust_timeouts(struct dlm_ls *ls)
1924{
1925        struct dlm_lkb *lkb;
1926        u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1927
1928        ls->ls_recover_begin = 0;
1929        mutex_lock(&ls->ls_timeout_mutex);
1930        list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1931                lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1932        mutex_unlock(&ls->ls_timeout_mutex);
1933
1934        if (!dlm_config.ci_waitwarn_us)
1935                return;
1936
1937        mutex_lock(&ls->ls_waiters_mutex);
1938        list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1939                if (ktime_to_us(lkb->lkb_wait_time))
1940                        lkb->lkb_wait_time = ktime_get();
1941        }
1942        mutex_unlock(&ls->ls_waiters_mutex);
1943}
1944
1945/* lkb is master or local copy */
1946
1947static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1948{
1949        int b, len = r->res_ls->ls_lvblen;
1950
1951        /* b=1 lvb returned to caller
1952           b=0 lvb written to rsb or invalidated
1953           b=-1 do nothing */
1954
1955        b =  dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1956
1957        if (b == 1) {
1958                if (!lkb->lkb_lvbptr)
1959                        return;
1960
1961                if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1962                        return;
1963
1964                if (!r->res_lvbptr)
1965                        return;
1966
1967                memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1968                lkb->lkb_lvbseq = r->res_lvbseq;
1969
1970        } else if (b == 0) {
1971                if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1972                        rsb_set_flag(r, RSB_VALNOTVALID);
1973                        return;
1974                }
1975
1976                if (!lkb->lkb_lvbptr)
1977                        return;
1978
1979                if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1980                        return;
1981
1982                if (!r->res_lvbptr)
1983                        r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1984
1985                if (!r->res_lvbptr)
1986                        return;
1987
1988                memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1989                r->res_lvbseq++;
1990                lkb->lkb_lvbseq = r->res_lvbseq;
1991                rsb_clear_flag(r, RSB_VALNOTVALID);
1992        }
1993
1994        if (rsb_flag(r, RSB_VALNOTVALID))
1995                lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1996}
1997
1998static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1999{
2000        if (lkb->lkb_grmode < DLM_LOCK_PW)
2001                return;
2002
2003        if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2004                rsb_set_flag(r, RSB_VALNOTVALID);
2005                return;
2006        }
2007
2008        if (!lkb->lkb_lvbptr)
2009                return;
2010
2011        if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2012                return;
2013
2014        if (!r->res_lvbptr)
2015                r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
2016
2017        if (!r->res_lvbptr)
2018                return;
2019
2020        memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2021        r->res_lvbseq++;
2022        rsb_clear_flag(r, RSB_VALNOTVALID);
2023}
2024
2025/* lkb is process copy (pc) */
2026
2027static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2028                            struct dlm_message *ms)
2029{
2030        int b;
2031
2032        if (!lkb->lkb_lvbptr)
2033                return;
2034
2035        if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2036                return;
2037
2038        b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2039        if (b == 1) {
2040                int len = receive_extralen(ms);
2041                if (len > r->res_ls->ls_lvblen)
2042                        len = r->res_ls->ls_lvblen;
2043                memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2044                lkb->lkb_lvbseq = ms->m_lvbseq;
2045        }
2046}
2047
2048/* Manipulate lkb's on rsb's convert/granted/waiting queues
2049   remove_lock -- used for unlock, removes lkb from granted
2050   revert_lock -- used for cancel, moves lkb from convert to granted
2051   grant_lock  -- used for request and convert, adds lkb to granted or
2052                  moves lkb from convert or waiting to granted
2053
2054   Each of these is used for master or local copy lkb's.  There is
2055   also a _pc() variation used to make the corresponding change on
2056   a process copy (pc) lkb. */
2057
2058static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2059{
2060        del_lkb(r, lkb);
2061        lkb->lkb_grmode = DLM_LOCK_IV;
2062        /* this unhold undoes the original ref from create_lkb()
2063           so this leads to the lkb being freed */
2064        unhold_lkb(lkb);
2065}
2066
2067static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2068{
2069        set_lvb_unlock(r, lkb);
2070        _remove_lock(r, lkb);
2071}
2072
2073static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2074{
2075        _remove_lock(r, lkb);
2076}
2077
2078/* returns: 0 did nothing
2079            1 moved lock to granted
2080           -1 removed lock */
2081
2082static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2083{
2084        int rv = 0;
2085
2086        lkb->lkb_rqmode = DLM_LOCK_IV;
2087
2088        switch (lkb->lkb_status) {
2089        case DLM_LKSTS_GRANTED:
2090                break;
2091        case DLM_LKSTS_CONVERT:
2092                move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2093                rv = 1;
2094                break;
2095        case DLM_LKSTS_WAITING:
2096                del_lkb(r, lkb);
2097                lkb->lkb_grmode = DLM_LOCK_IV;
2098                /* this unhold undoes the original ref from create_lkb()
2099                   so this leads to the lkb being freed */
2100                unhold_lkb(lkb);
2101                rv = -1;
2102                break;
2103        default:
2104                log_print("invalid status for revert %d", lkb->lkb_status);
2105        }
2106        return rv;
2107}
2108
2109static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2110{
2111        return revert_lock(r, lkb);
2112}
2113
2114static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2115{
2116        if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2117                lkb->lkb_grmode = lkb->lkb_rqmode;
2118                if (lkb->lkb_status)
2119                        move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2120                else
2121                        add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2122        }
2123
2124        lkb->lkb_rqmode = DLM_LOCK_IV;
2125        lkb->lkb_highbast = 0;
2126}
2127
2128static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2129{
2130        set_lvb_lock(r, lkb);
2131        _grant_lock(r, lkb);
2132}
2133
2134static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2135                          struct dlm_message *ms)
2136{
2137        set_lvb_lock_pc(r, lkb, ms);
2138        _grant_lock(r, lkb);
2139}
2140
2141/* called by grant_pending_locks() which means an async grant message must
2142   be sent to the requesting node in addition to granting the lock if the
2143   lkb belongs to a remote node. */
2144
2145static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2146{
2147        grant_lock(r, lkb);
2148        if (is_master_copy(lkb))
2149                send_grant(r, lkb);
2150        else
2151                queue_cast(r, lkb, 0);
2152}
2153
2154/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2155   change the granted/requested modes.  We're munging things accordingly in
2156   the process copy.
2157   CONVDEADLK: our grmode may have been forced down to NL to resolve a
2158   conversion deadlock
2159   ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2160   compatible with other granted locks */
2161
2162static void munge_demoted(struct dlm_lkb *lkb)
2163{
2164        if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2165                log_print("munge_demoted %x invalid modes gr %d rq %d",
2166                          lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2167                return;
2168        }
2169
2170        lkb->lkb_grmode = DLM_LOCK_NL;
2171}
2172
2173static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2174{
2175        if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
2176            ms->m_type != DLM_MSG_GRANT) {
2177                log_print("munge_altmode %x invalid reply type %d",
2178                          lkb->lkb_id, ms->m_type);
2179                return;
2180        }
2181
2182        if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2183                lkb->lkb_rqmode = DLM_LOCK_PR;
2184        else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2185                lkb->lkb_rqmode = DLM_LOCK_CW;
2186        else {
2187                log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2188                dlm_print_lkb(lkb);
2189        }
2190}
2191
2192static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2193{
2194        struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2195                                           lkb_statequeue);
2196        if (lkb->lkb_id == first->lkb_id)
2197                return 1;
2198
2199        return 0;
2200}
2201
2202/* Check if the given lkb conflicts with another lkb on the queue. */
2203
2204static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2205{
2206        struct dlm_lkb *this;
2207
2208        list_for_each_entry(this, head, lkb_statequeue) {
2209                if (this == lkb)
2210                        continue;
2211                if (!modes_compat(this, lkb))
2212                        return 1;
2213        }
2214        return 0;
2215}
2216
2217/*
2218 * "A conversion deadlock arises with a pair of lock requests in the converting
2219 * queue for one resource.  The granted mode of each lock blocks the requested
2220 * mode of the other lock."
2221 *
2222 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2223 * convert queue from being granted, then deadlk/demote lkb.
2224 *
2225 * Example:
2226 * Granted Queue: empty
2227 * Convert Queue: NL->EX (first lock)
2228 *                PR->EX (second lock)
2229 *
2230 * The first lock can't be granted because of the granted mode of the second
2231 * lock and the second lock can't be granted because it's not first in the
2232 * list.  We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2233 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2234 * flag set and return DEMOTED in the lksb flags.
2235 *
2236 * Originally, this function detected conv-deadlk in a more limited scope:
2237 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2238 * - if lkb1 was the first entry in the queue (not just earlier), and was
2239 *   blocked by the granted mode of lkb2, and there was nothing on the
2240 *   granted queue preventing lkb1 from being granted immediately, i.e.
2241 *   lkb2 was the only thing preventing lkb1 from being granted.
2242 *
2243 * That second condition meant we'd only say there was conv-deadlk if
2244 * resolving it (by demotion) would lead to the first lock on the convert
2245 * queue being granted right away.  It allowed conversion deadlocks to exist
2246 * between locks on the convert queue while they couldn't be granted anyway.
2247 *
2248 * Now, we detect and take action on conversion deadlocks immediately when
2249 * they're created, even if they may not be immediately consequential.  If
2250 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2251 * mode that would prevent lkb1's conversion from being granted, we do a
2252 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2253 * I think this means that the lkb_is_ahead condition below should always
2254 * be zero, i.e. there will never be conv-deadlk between two locks that are
2255 * both already on the convert queue.
2256 */
2257
2258static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2259{
2260        struct dlm_lkb *lkb1;
2261        int lkb_is_ahead = 0;
2262
2263        list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2264                if (lkb1 == lkb2) {
2265                        lkb_is_ahead = 1;
2266                        continue;
2267                }
2268
2269                if (!lkb_is_ahead) {
2270                        if (!modes_compat(lkb2, lkb1))
2271                                return 1;
2272                } else {
2273                        if (!modes_compat(lkb2, lkb1) &&
2274                            !modes_compat(lkb1, lkb2))
2275                                return 1;
2276                }
2277        }
2278        return 0;
2279}
2280
2281/*
2282 * Return 1 if the lock can be granted, 0 otherwise.
2283 * Also detect and resolve conversion deadlocks.
2284 *
2285 * lkb is the lock to be granted
2286 *
2287 * now is 1 if the function is being called in the context of the
2288 * immediate request, it is 0 if called later, after the lock has been
2289 * queued.
2290 *
2291 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2292 * after recovery.
2293 *
2294 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2295 */
2296
2297static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2298                           int recover)
2299{
2300        int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2301
2302        /*
2303         * 6-10: Version 5.4 introduced an option to address the phenomenon of
2304         * a new request for a NL mode lock being blocked.
2305         *
2306         * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2307         * request, then it would be granted.  In essence, the use of this flag
2308         * tells the Lock Manager to expedite theis request by not considering
2309         * what may be in the CONVERTING or WAITING queues...  As of this
2310         * writing, the EXPEDITE flag can be used only with new requests for NL
2311         * mode locks.  This flag is not valid for conversion requests.
2312         *
2313         * A shortcut.  Earlier checks return an error if EXPEDITE is used in a
2314         * conversion or used with a non-NL requested mode.  We also know an
2315         * EXPEDITE request is always granted immediately, so now must always
2316         * be 1.  The full condition to grant an expedite request: (now &&
2317         * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2318         * therefore be shortened to just checking the flag.
2319         */
2320
2321        if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2322                return 1;
2323
2324        /*
2325         * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2326         * added to the remaining conditions.
2327         */
2328
2329        if (queue_conflict(&r->res_grantqueue, lkb))
2330                return 0;
2331
2332        /*
2333         * 6-3: By default, a conversion request is immediately granted if the
2334         * requested mode is compatible with the modes of all other granted
2335         * locks
2336         */
2337
2338        if (queue_conflict(&r->res_convertqueue, lkb))
2339                return 0;
2340
2341        /*
2342         * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2343         * locks for a recovered rsb, on which lkb's have been rebuilt.
2344         * The lkb's may have been rebuilt on the queues in a different
2345         * order than they were in on the previous master.  So, granting
2346         * queued conversions in order after recovery doesn't make sense
2347         * since the order hasn't been preserved anyway.  The new order
2348         * could also have created a new "in place" conversion deadlock.
2349         * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2350         * After recovery, there would be no granted locks, and possibly
2351         * NL->EX, PR->EX, an in-place conversion deadlock.)  So, after
2352         * recovery, grant conversions without considering order.
2353         */
2354
2355        if (conv && recover)
2356                return 1;
2357
2358        /*
2359         * 6-5: But the default algorithm for deciding whether to grant or
2360         * queue conversion requests does not by itself guarantee that such
2361         * requests are serviced on a "first come first serve" basis.  This, in
2362         * turn, can lead to a phenomenon known as "indefinate postponement".
2363         *
2364         * 6-7: This issue is dealt with by using the optional QUECVT flag with
2365         * the system service employed to request a lock conversion.  This flag
2366         * forces certain conversion requests to be queued, even if they are
2367         * compatible with the granted modes of other locks on the same
2368         * resource.  Thus, the use of this flag results in conversion requests
2369         * being ordered on a "first come first servce" basis.
2370         *
2371         * DCT: This condition is all about new conversions being able to occur
2372         * "in place" while the lock remains on the granted queue (assuming
2373         * nothing else conflicts.)  IOW if QUECVT isn't set, a conversion
2374         * doesn't _have_ to go onto the convert queue where it's processed in
2375         * order.  The "now" variable is necessary to distinguish converts
2376         * being received and processed for the first time now, because once a
2377         * convert is moved to the conversion queue the condition below applies
2378         * requiring fifo granting.
2379         */
2380
2381        if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2382                return 1;
2383
2384        /*
2385         * Even if the convert is compat with all granted locks,
2386         * QUECVT forces it behind other locks on the convert queue.
2387         */
2388
2389        if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2390                if (list_empty(&r->res_convertqueue))
2391                        return 1;
2392                else
2393                        return 0;
2394        }
2395
2396        /*
2397         * The NOORDER flag is set to avoid the standard vms rules on grant
2398         * order.
2399         */
2400
2401        if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2402                return 1;
2403
2404        /*
2405         * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2406         * granted until all other conversion requests ahead of it are granted
2407         * and/or canceled.
2408         */
2409
2410        if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2411                return 1;
2412
2413        /*
2414         * 6-4: By default, a new request is immediately granted only if all
2415         * three of the following conditions are satisfied when the request is
2416         * issued:
2417         * - The queue of ungranted conversion requests for the resource is
2418         *   empty.
2419         * - The queue of ungranted new requests for the resource is empty.
2420         * - The mode of the new request is compatible with the most
2421         *   restrictive mode of all granted locks on the resource.
2422         */
2423
2424        if (now && !conv && list_empty(&r->res_convertqueue) &&
2425            list_empty(&r->res_waitqueue))
2426                return 1;
2427
2428        /*
2429         * 6-4: Once a lock request is in the queue of ungranted new requests,
2430         * it cannot be granted until the queue of ungranted conversion
2431         * requests is empty, all ungranted new requests ahead of it are
2432         * granted and/or canceled, and it is compatible with the granted mode
2433         * of the most restrictive lock granted on the resource.
2434         */
2435
2436        if (!now && !conv && list_empty(&r->res_convertqueue) &&
2437            first_in_list(lkb, &r->res_waitqueue))
2438                return 1;
2439
2440        return 0;
2441}
2442
2443static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2444                          int recover, int *err)
2445{
2446        int rv;
2447        int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2448        int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2449
2450        if (err)
2451                *err = 0;
2452
2453        rv = _can_be_granted(r, lkb, now, recover);
2454        if (rv)
2455                goto out;
2456
2457        /*
2458         * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2459         * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2460         * cancels one of the locks.
2461         */
2462
2463        if (is_convert && can_be_queued(lkb) &&
2464            conversion_deadlock_detect(r, lkb)) {
2465                if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2466                        lkb->lkb_grmode = DLM_LOCK_NL;
2467                        lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2468                } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
2469                        if (err)
2470                                *err = -EDEADLK;
2471                        else {
2472                                log_print("can_be_granted deadlock %x now %d",
2473                                          lkb->lkb_id, now);
2474                                dlm_dump_rsb(r);
2475                        }
2476                }
2477                goto out;
2478        }
2479
2480        /*
2481         * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2482         * to grant a request in a mode other than the normal rqmode.  It's a
2483         * simple way to provide a big optimization to applications that can
2484         * use them.
2485         */
2486
2487        if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2488                alt = DLM_LOCK_PR;
2489        else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2490                alt = DLM_LOCK_CW;
2491
2492        if (alt) {
2493                lkb->lkb_rqmode = alt;
2494                rv = _can_be_granted(r, lkb, now, 0);
2495                if (rv)
2496                        lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2497                else
2498                        lkb->lkb_rqmode = rqmode;
2499        }
2500 out:
2501        return rv;
2502}
2503
2504/* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
2505   for locks pending on the convert list.  Once verified (watch for these
2506   log_prints), we should be able to just call _can_be_granted() and not
2507   bother with the demote/deadlk cases here (and there's no easy way to deal
2508   with a deadlk here, we'd have to generate something like grant_lock with
2509   the deadlk error.) */
2510
2511/* Returns the highest requested mode of all blocked conversions; sets
2512   cw if there's a blocked conversion to DLM_LOCK_CW. */
2513
2514static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2515                                 unsigned int *count)
2516{
2517        struct dlm_lkb *lkb, *s;
2518        int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2519        int hi, demoted, quit, grant_restart, demote_restart;
2520        int deadlk;
2521
2522        quit = 0;
2523 restart:
2524        grant_restart = 0;
2525        demote_restart = 0;
2526        hi = DLM_LOCK_IV;
2527
2528        list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2529                demoted = is_demoted(lkb);
2530                deadlk = 0;
2531
2532                if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2533                        grant_lock_pending(r, lkb);
2534                        grant_restart = 1;
2535                        if (count)
2536                                (*count)++;
2537                        continue;
2538                }
2539
2540                if (!demoted && is_demoted(lkb)) {
2541                        log_print("WARN: pending demoted %x node %d %s",
2542                                  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2543                        demote_restart = 1;
2544                        continue;
2545                }
2546
2547                if (deadlk) {
2548                        log_print("WARN: pending deadlock %x node %d %s",
2549                                  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2550                        dlm_dump_rsb(r);
2551                        continue;
2552                }
2553
2554                hi = max_t(int, lkb->lkb_rqmode, hi);
2555
2556                if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2557                        *cw = 1;
2558        }
2559
2560        if (grant_restart)
2561                goto restart;
2562        if (demote_restart && !quit) {
2563                quit = 1;
2564                goto restart;
2565        }
2566
2567        return max_t(int, high, hi);
2568}
2569
2570static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2571                              unsigned int *count)
2572{
2573        struct dlm_lkb *lkb, *s;
2574
2575        list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2576                if (can_be_granted(r, lkb, 0, 0, NULL)) {
2577                        grant_lock_pending(r, lkb);
2578                        if (count)
2579                                (*count)++;
2580                } else {
2581                        high = max_t(int, lkb->lkb_rqmode, high);
2582                        if (lkb->lkb_rqmode == DLM_LOCK_CW)
2583                                *cw = 1;
2584                }
2585        }
2586
2587        return high;
2588}
2589
2590/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2591   on either the convert or waiting queue.
2592   high is the largest rqmode of all locks blocked on the convert or
2593   waiting queue. */
2594
2595static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2596{
2597        if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2598                if (gr->lkb_highbast < DLM_LOCK_EX)
2599                        return 1;
2600                return 0;
2601        }
2602
2603        if (gr->lkb_highbast < high &&
2604            !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2605                return 1;
2606        return 0;
2607}
2608
2609static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2610{
2611        struct dlm_lkb *lkb, *s;
2612        int high = DLM_LOCK_IV;
2613        int cw = 0;
2614
2615        if (!is_master(r)) {
2616                log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2617                dlm_dump_rsb(r);
2618                return;
2619        }
2620
2621        high = grant_pending_convert(r, high, &cw, count);
2622        high = grant_pending_wait(r, high, &cw, count);
2623
2624        if (high == DLM_LOCK_IV)
2625                return;
2626
2627        /*
2628         * If there are locks left on the wait/convert queue then send blocking
2629         * ASTs to granted locks based on the largest requested mode (high)
2630         * found above.
2631         */
2632
2633        list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2634                if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2635                        if (cw && high == DLM_LOCK_PR &&
2636                            lkb->lkb_grmode == DLM_LOCK_PR)
2637                                queue_bast(r, lkb, DLM_LOCK_CW);
2638                        else
2639                                queue_bast(r, lkb, high);
2640                        lkb->lkb_highbast = high;
2641                }
2642        }
2643}
2644
2645static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2646{
2647        if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2648            (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2649                if (gr->lkb_highbast < DLM_LOCK_EX)
2650                        return 1;
2651                return 0;
2652        }
2653
2654        if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2655                return 1;
2656        return 0;
2657}
2658
2659static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2660                            struct dlm_lkb *lkb)
2661{
2662        struct dlm_lkb *gr;
2663
2664        list_for_each_entry(gr, head, lkb_statequeue) {
2665                /* skip self when sending basts to convertqueue */
2666                if (gr == lkb)
2667                        continue;
2668                if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2669                        queue_bast(r, gr, lkb->lkb_rqmode);
2670                        gr->lkb_highbast = lkb->lkb_rqmode;
2671                }
2672        }
2673}
2674
2675static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2676{
2677        send_bast_queue(r, &r->res_grantqueue, lkb);
2678}
2679
2680static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2681{
2682        send_bast_queue(r, &r->res_grantqueue, lkb);
2683        send_bast_queue(r, &r->res_convertqueue, lkb);
2684}
2685
2686/* set_master(r, lkb) -- set the master nodeid of a resource
2687
2688   The purpose of this function is to set the nodeid field in the given
2689   lkb using the nodeid field in the given rsb.  If the rsb's nodeid is
2690   known, it can just be copied to the lkb and the function will return
2691   0.  If the rsb's nodeid is _not_ known, it needs to be looked up
2692   before it can be copied to the lkb.
2693
2694   When the rsb nodeid is being looked up remotely, the initial lkb
2695   causing the lookup is kept on the ls_waiters list waiting for the
2696   lookup reply.  Other lkb's waiting for the same rsb lookup are kept
2697   on the rsb's res_lookup list until the master is verified.
2698
2699   Return values:
2700   0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2701   1: the rsb master is not available and the lkb has been placed on
2702      a wait queue
2703*/
2704
2705static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2706{
2707        int our_nodeid = dlm_our_nodeid();
2708
2709        if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2710                rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2711                r->res_first_lkid = lkb->lkb_id;
2712                lkb->lkb_nodeid = r->res_nodeid;
2713                return 0;
2714        }
2715
2716        if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2717                list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2718                return 1;
2719        }
2720
2721        if (r->res_master_nodeid == our_nodeid) {
2722                lkb->lkb_nodeid = 0;
2723                return 0;
2724        }
2725
2726        if (r->res_master_nodeid) {
2727                lkb->lkb_nodeid = r->res_master_nodeid;
2728                return 0;
2729        }
2730
2731        if (dlm_dir_nodeid(r) == our_nodeid) {
2732                /* This is a somewhat unusual case; find_rsb will usually
2733                   have set res_master_nodeid when dir nodeid is local, but
2734                   there are cases where we become the dir node after we've
2735                   past find_rsb and go through _request_lock again.
2736                   confirm_master() or process_lookup_list() needs to be
2737                   called after this. */
2738                log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2739                          lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2740                          r->res_name);
2741                r->res_master_nodeid = our_nodeid;
2742                r->res_nodeid = 0;
2743                lkb->lkb_nodeid = 0;
2744                return 0;
2745        }
2746
2747        wait_pending_remove(r);
2748
2749        r->res_first_lkid = lkb->lkb_id;
2750        send_lookup(r, lkb);
2751        return 1;
2752}
2753
2754static void process_lookup_list(struct dlm_rsb *r)
2755{
2756        struct dlm_lkb *lkb, *safe;
2757
2758        list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2759                list_del_init(&lkb->lkb_rsb_lookup);
2760                _request_lock(r, lkb);
2761                schedule();
2762        }
2763}
2764
2765/* confirm_master -- confirm (or deny) an rsb's master nodeid */
2766
2767static void confirm_master(struct dlm_rsb *r, int error)
2768{
2769        struct dlm_lkb *lkb;
2770
2771        if (!r->res_first_lkid)
2772                return;
2773
2774        switch (error) {
2775        case 0:
2776        case -EINPROGRESS:
2777                r->res_first_lkid = 0;
2778                process_lookup_list(r);
2779                break;
2780
2781        case -EAGAIN:
2782        case -EBADR:
2783        case -ENOTBLK:
2784                /* the remote request failed and won't be retried (it was
2785                   a NOQUEUE, or has been canceled/unlocked); make a waiting
2786                   lkb the first_lkid */
2787
2788                r->res_first_lkid = 0;
2789
2790                if (!list_empty(&r->res_lookup)) {
2791                        lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2792                                         lkb_rsb_lookup);
2793                        list_del_init(&lkb->lkb_rsb_lookup);
2794                        r->res_first_lkid = lkb->lkb_id;
2795                        _request_lock(r, lkb);
2796                }
2797                break;
2798
2799        default:
2800                log_error(r->res_ls, "confirm_master unknown error %d", error);
2801        }
2802}
2803
2804static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2805                         int namelen, unsigned long timeout_cs,
2806                         void (*ast) (void *astparam),
2807                         void *astparam,
2808                         void (*bast) (void *astparam, int mode),
2809                         struct dlm_args *args)
2810{
2811        int rv = -EINVAL;
2812
2813        /* check for invalid arg usage */
2814
2815        if (mode < 0 || mode > DLM_LOCK_EX)
2816                goto out;
2817
2818        if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2819                goto out;
2820
2821        if (flags & DLM_LKF_CANCEL)
2822                goto out;
2823
2824        if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2825                goto out;
2826
2827        if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2828                goto out;
2829
2830        if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2831                goto out;
2832
2833        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2834                goto out;
2835
2836        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2837                goto out;
2838
2839        if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2840                goto out;
2841
2842        if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2843                goto out;
2844
2845        if (!ast || !lksb)
2846                goto out;
2847
2848        if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2849                goto out;
2850
2851        if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2852                goto out;
2853
2854        /* these args will be copied to the lkb in validate_lock_args,
2855           it cannot be done now because when converting locks, fields in
2856           an active lkb cannot be modified before locking the rsb */
2857
2858        args->flags = flags;
2859        args->astfn = ast;
2860        args->astparam = astparam;
2861        args->bastfn = bast;
2862        args->timeout = timeout_cs;
2863        args->mode = mode;
2864        args->lksb = lksb;
2865        rv = 0;
2866 out:
2867        return rv;
2868}
2869
2870static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2871{
2872        if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2873                      DLM_LKF_FORCEUNLOCK))
2874                return -EINVAL;
2875
2876        if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2877                return -EINVAL;
2878
2879        args->flags = flags;
2880        args->astparam = astarg;
2881        return 0;
2882}
2883
2884static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2885                              struct dlm_args *args)
2886{
2887        int rv = -EINVAL;
2888
2889        if (args->flags & DLM_LKF_CONVERT) {
2890                if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2891                        goto out;
2892
2893                if (args->flags & DLM_LKF_QUECVT &&
2894                    !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2895                        goto out;
2896
2897                rv = -EBUSY;
2898                if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2899                        goto out;
2900
2901                if (lkb->lkb_wait_type)
2902                        goto out;
2903
2904                if (is_overlap(lkb))
2905                        goto out;
2906        }
2907
2908        lkb->lkb_exflags = args->flags;
2909        lkb->lkb_sbflags = 0;
2910        lkb->lkb_astfn = args->astfn;
2911        lkb->lkb_astparam = args->astparam;
2912        lkb->lkb_bastfn = args->bastfn;
2913        lkb->lkb_rqmode = args->mode;
2914        lkb->lkb_lksb = args->lksb;
2915        lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2916        lkb->lkb_ownpid = (int) current->pid;
2917        lkb->lkb_timeout_cs = args->timeout;
2918        rv = 0;
2919 out:
2920        if (rv)
2921                log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2922                          rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2923                          lkb->lkb_status, lkb->lkb_wait_type,
2924                          lkb->lkb_resource->res_name);
2925        return rv;
2926}
2927
2928/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2929   for success */
2930
2931/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2932   because there may be a lookup in progress and it's valid to do
2933   cancel/unlockf on it */
2934
2935static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2936{
2937        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2938        int rv = -EINVAL;
2939
2940        if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2941                log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2942                dlm_print_lkb(lkb);
2943                goto out;
2944        }
2945
2946        /* an lkb may still exist even though the lock is EOL'ed due to a
2947           cancel, unlock or failed noqueue request; an app can't use these
2948           locks; return same error as if the lkid had not been found at all */
2949
2950        if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2951                log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2952                rv = -ENOENT;
2953                goto out;
2954        }
2955
2956        /* an lkb may be waiting for an rsb lookup to complete where the
2957           lookup was initiated by another lock */
2958
2959        if (!list_empty(&lkb->lkb_rsb_lookup)) {
2960                if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2961                        log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2962                        list_del_init(&lkb->lkb_rsb_lookup);
2963                        queue_cast(lkb->lkb_resource, lkb,
2964                                   args->flags & DLM_LKF_CANCEL ?
2965                                   -DLM_ECANCEL : -DLM_EUNLOCK);
2966                        unhold_lkb(lkb); /* undoes create_lkb() */
2967                }
2968                /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2969                rv = -EBUSY;
2970                goto out;
2971        }
2972
2973        /* cancel not allowed with another cancel/unlock in progress */
2974
2975        if (args->flags & DLM_LKF_CANCEL) {
2976                if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2977                        goto out;
2978
2979                if (is_overlap(lkb))
2980                        goto out;
2981
2982                /* don't let scand try to do a cancel */
2983                del_timeout(lkb);
2984
2985                if (lkb->lkb_flags & DLM_IFL_RESEND) {
2986                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2987                        rv = -EBUSY;
2988                        goto out;
2989                }
2990
2991                /* there's nothing to cancel */
2992                if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2993                    !lkb->lkb_wait_type) {
2994                        rv = -EBUSY;
2995                        goto out;
2996                }
2997
2998                switch (lkb->lkb_wait_type) {
2999                case DLM_MSG_LOOKUP:
3000                case DLM_MSG_REQUEST:
3001                        lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3002                        rv = -EBUSY;
3003                        goto out;
3004                case DLM_MSG_UNLOCK:
3005                case DLM_MSG_CANCEL:
3006                        goto out;
3007                }
3008                /* add_to_waiters() will set OVERLAP_CANCEL */
3009                goto out_ok;
3010        }
3011
3012        /* do we need to allow a force-unlock if there's a normal unlock
3013           already in progress?  in what conditions could the normal unlock
3014           fail such that we'd want to send a force-unlock to be sure? */
3015
3016        if (args->flags & DLM_LKF_FORCEUNLOCK) {
3017                if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3018                        goto out;
3019
3020                if (is_overlap_unlock(lkb))
3021                        goto out;
3022
3023                /* don't let scand try to do a cancel */
3024                del_timeout(lkb);
3025
3026                if (lkb->lkb_flags & DLM_IFL_RESEND) {
3027                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3028                        rv = -EBUSY;
3029                        goto out;
3030                }
3031
3032                switch (lkb->lkb_wait_type) {
3033                case DLM_MSG_LOOKUP:
3034                case DLM_MSG_REQUEST:
3035                        lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3036                        rv = -EBUSY;
3037                        goto out;
3038                case DLM_MSG_UNLOCK:
3039                        goto out;
3040                }
3041                /* add_to_waiters() will set OVERLAP_UNLOCK */
3042                goto out_ok;
3043        }
3044
3045        /* normal unlock not allowed if there's any op in progress */
3046        rv = -EBUSY;
3047        if (lkb->lkb_wait_type || lkb->lkb_wait_count)
3048                goto out;
3049
3050 out_ok:
3051        /* an overlapping op shouldn't blow away exflags from other op */
3052        lkb->lkb_exflags |= args->flags;
3053        lkb->lkb_sbflags = 0;
3054        lkb->lkb_astparam = args->astparam;
3055        rv = 0;
3056 out:
3057        if (rv)
3058                log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3059                          lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3060                          args->flags, lkb->lkb_wait_type,
3061                          lkb->lkb_resource->res_name);
3062        return rv;
3063}
3064
3065/*
3066 * Four stage 4 varieties:
3067 * do_request(), do_convert(), do_unlock(), do_cancel()
3068 * These are called on the master node for the given lock and
3069 * from the central locking logic.
3070 */
3071
3072static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3073{
3074        int error = 0;
3075
3076        if (can_be_granted(r, lkb, 1, 0, NULL)) {
3077                grant_lock(r, lkb);
3078                queue_cast(r, lkb, 0);
3079                goto out;
3080        }
3081
3082        if (can_be_queued(lkb)) {
3083                error = -EINPROGRESS;
3084                add_lkb(r, lkb, DLM_LKSTS_WAITING);
3085                add_timeout(lkb);
3086                goto out;
3087        }
3088
3089        error = -EAGAIN;
3090        queue_cast(r, lkb, -EAGAIN);
3091 out:
3092        return error;
3093}
3094
3095static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3096                               int error)
3097{
3098        switch (error) {
3099        case -EAGAIN:
3100                if (force_blocking_asts(lkb))
3101                        send_blocking_asts_all(r, lkb);
3102                break;
3103        case -EINPROGRESS:
3104                send_blocking_asts(r, lkb);
3105                break;
3106        }
3107}
3108
3109static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3110{
3111        int error = 0;
3112        int deadlk = 0;
3113
3114        /* changing an existing lock may allow others to be granted */
3115
3116        if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3117                grant_lock(r, lkb);
3118                queue_cast(r, lkb, 0);
3119                goto out;
3120        }
3121
3122        /* can_be_granted() detected that this lock would block in a conversion
3123           deadlock, so we leave it on the granted queue and return EDEADLK in
3124           the ast for the convert. */
3125
3126        if (deadlk) {
3127                /* it's left on the granted queue */
3128                revert_lock(r, lkb);
3129                queue_cast(r, lkb, -EDEADLK);
3130                error = -EDEADLK;
3131                goto out;
3132        }
3133
3134        /* is_demoted() means the can_be_granted() above set the grmode
3135           to NL, and left us on the granted queue.  This auto-demotion
3136           (due to CONVDEADLK) might mean other locks, and/or this lock, are
3137           now grantable.  We have to try to grant other converting locks
3138           before we try again to grant this one. */
3139
3140        if (is_demoted(lkb)) {
3141                grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3142                if (_can_be_granted(r, lkb, 1, 0)) {
3143                        grant_lock(r, lkb);
3144                        queue_cast(r, lkb, 0);
3145                        goto out;
3146                }
3147                /* else fall through and move to convert queue */
3148        }
3149
3150        if (can_be_queued(lkb)) {
3151                error = -EINPROGRESS;
3152                del_lkb(r, lkb);
3153                add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3154                add_timeout(lkb);
3155                goto out;
3156        }
3157
3158        error = -EAGAIN;
3159        queue_cast(r, lkb, -EAGAIN);
3160 out:
3161        return error;
3162}
3163
3164static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3165                               int error)
3166{
3167        switch (error) {
3168        case 0:
3169                grant_pending_locks(r, NULL);
3170                /* grant_pending_locks also sends basts */
3171                break;
3172        case -EAGAIN:
3173                if (force_blocking_asts(lkb))
3174                        send_blocking_asts_all(r, lkb);
3175                break;
3176        case -EINPROGRESS:
3177                send_blocking_asts(r, lkb);
3178                break;
3179        }
3180}
3181
3182static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3183{
3184        remove_lock(r, lkb);
3185        queue_cast(r, lkb, -DLM_EUNLOCK);
3186        return -DLM_EUNLOCK;
3187}
3188
3189static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3190                              int error)
3191{
3192        grant_pending_locks(r, NULL);
3193}
3194
3195/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3196
3197static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3198{
3199        int error;
3200
3201        error = revert_lock(r, lkb);
3202        if (error) {
3203                queue_cast(r, lkb, -DLM_ECANCEL);
3204                return -DLM_ECANCEL;
3205        }
3206        return 0;
3207}
3208
3209static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3210                              int error)
3211{
3212        if (error)
3213                grant_pending_locks(r, NULL);
3214}
3215
3216/*
3217 * Four stage 3 varieties:
3218 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3219 */
3220
3221/* add a new lkb to a possibly new rsb, called by requesting process */
3222
3223static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3224{
3225        int error;
3226
3227        /* set_master: sets lkb nodeid from r */
3228
3229        error = set_master(r, lkb);
3230        if (error < 0)
3231                goto out;
3232        if (error) {
3233                error = 0;
3234                goto out;
3235        }
3236
3237        if (is_remote(r)) {
3238                /* receive_request() calls do_request() on remote node */
3239                error = send_request(r, lkb);
3240        } else {
3241                error = do_request(r, lkb);
3242                /* for remote locks the request_reply is sent
3243                   between do_request and do_request_effects */
3244                do_request_effects(r, lkb, error);
3245        }
3246 out:
3247        return error;
3248}
3249
3250/* change some property of an existing lkb, e.g. mode */
3251
3252static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3253{
3254        int error;
3255
3256        if (is_remote(r)) {
3257                /* receive_convert() calls do_convert() on remote node */
3258                error = send_convert(r, lkb);
3259        } else {
3260                error = do_convert(r, lkb);
3261                /* for remote locks the convert_reply is sent
3262                   between do_convert and do_convert_effects */
3263                do_convert_effects(r, lkb, error);
3264        }
3265
3266        return error;
3267}
3268
3269/* remove an existing lkb from the granted queue */
3270
3271static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3272{
3273        int error;
3274
3275        if (is_remote(r)) {
3276                /* receive_unlock() calls do_unlock() on remote node */
3277                error = send_unlock(r, lkb);
3278        } else {
3279                error = do_unlock(r, lkb);
3280                /* for remote locks the unlock_reply is sent
3281                   between do_unlock and do_unlock_effects */
3282                do_unlock_effects(r, lkb, error);
3283        }
3284
3285        return error;
3286}
3287
3288/* remove an existing lkb from the convert or wait queue */
3289
3290static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3291{
3292        int error;
3293
3294        if (is_remote(r)) {
3295                /* receive_cancel() calls do_cancel() on remote node */
3296                error = send_cancel(r, lkb);
3297        } else {
3298                error = do_cancel(r, lkb);
3299                /* for remote locks the cancel_reply is sent
3300                   between do_cancel and do_cancel_effects */
3301                do_cancel_effects(r, lkb, error);
3302        }
3303
3304        return error;
3305}
3306
3307/*
3308 * Four stage 2 varieties:
3309 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3310 */
3311
3312static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3313                        int len, struct dlm_args *args)
3314{
3315        struct dlm_rsb *r;
3316        int error;
3317
3318        error = validate_lock_args(ls, lkb, args);
3319        if (error)
3320                return error;
3321
3322        error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3323        if (error)
3324                return error;
3325
3326        lock_rsb(r);
3327
3328        attach_lkb(r, lkb);
3329        lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3330
3331        error = _request_lock(r, lkb);
3332
3333        unlock_rsb(r);
3334        put_rsb(r);
3335        return error;
3336}
3337
3338static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3339                        struct dlm_args *args)
3340{
3341        struct dlm_rsb *r;
3342        int error;
3343
3344        r = lkb->lkb_resource;
3345
3346        hold_rsb(r);
3347        lock_rsb(r);
3348
3349        error = validate_lock_args(ls, lkb, args);
3350        if (error)
3351                goto out;
3352
3353        error = _convert_lock(r, lkb);
3354 out:
3355        unlock_rsb(r);
3356        put_rsb(r);
3357        return error;
3358}
3359
3360static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3361                       struct dlm_args *args)
3362{
3363        struct dlm_rsb *r;
3364        int error;
3365
3366        r = lkb->lkb_resource;
3367
3368        hold_rsb(r);
3369        lock_rsb(r);
3370
3371        error = validate_unlock_args(lkb, args);
3372        if (error)
3373                goto out;
3374
3375        error = _unlock_lock(r, lkb);
3376 out:
3377        unlock_rsb(r);
3378        put_rsb(r);
3379        return error;
3380}
3381
3382static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3383                       struct dlm_args *args)
3384{
3385        struct dlm_rsb *r;
3386        int error;
3387
3388        r = lkb->lkb_resource;
3389
3390        hold_rsb(r);
3391        lock_rsb(r);
3392
3393        error = validate_unlock_args(lkb, args);
3394        if (error)
3395                goto out;
3396
3397        error = _cancel_lock(r, lkb);
3398 out:
3399        unlock_rsb(r);
3400        put_rsb(r);
3401        return error;
3402}
3403
3404/*
3405 * Two stage 1 varieties:  dlm_lock() and dlm_unlock()
3406 */
3407
3408int dlm_lock(dlm_lockspace_t *lockspace,
3409             int mode,
3410             struct dlm_lksb *lksb,
3411             uint32_t flags,
3412             void *name,
3413             unsigned int namelen,
3414             uint32_t parent_lkid,
3415             void (*ast) (void *astarg),
3416             void *astarg,
3417             void (*bast) (void *astarg, int mode))
3418{
3419        struct dlm_ls *ls;
3420        struct dlm_lkb *lkb;
3421        struct dlm_args args;
3422        int error, convert = flags & DLM_LKF_CONVERT;
3423
3424        ls = dlm_find_lockspace_local(lockspace);
3425        if (!ls)
3426                return -EINVAL;
3427
3428        dlm_lock_recovery(ls);
3429
3430        if (convert)
3431                error = find_lkb(ls, lksb->sb_lkid, &lkb);
3432        else
3433                error = create_lkb(ls, &lkb);
3434
3435        if (error)
3436                goto out;
3437
3438        error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3439                              astarg, bast, &args);
3440        if (error)
3441                goto out_put;
3442
3443        if (convert)
3444                error = convert_lock(ls, lkb, &args);
3445        else
3446                error = request_lock(ls, lkb, name, namelen, &args);
3447
3448        if (error == -EINPROGRESS)
3449                error = 0;
3450 out_put:
3451        if (convert || error)
3452                __put_lkb(ls, lkb);
3453        if (error == -EAGAIN || error == -EDEADLK)
3454                error = 0;
3455 out:
3456        dlm_unlock_recovery(ls);
3457        dlm_put_lockspace(ls);
3458        return error;
3459}
3460
3461int dlm_unlock(dlm_lockspace_t *lockspace,
3462               uint32_t lkid,
3463               uint32_t flags,
3464               struct dlm_lksb *lksb,
3465               void *astarg)
3466{
3467        struct dlm_ls *ls;
3468        struct dlm_lkb *lkb;
3469        struct dlm_args args;
3470        int error;
3471
3472        ls = dlm_find_lockspace_local(lockspace);
3473        if (!ls)
3474                return -EINVAL;
3475
3476        dlm_lock_recovery(ls);
3477
3478        error = find_lkb(ls, lkid, &lkb);
3479        if (error)
3480                goto out;
3481
3482        error = set_unlock_args(flags, astarg, &args);
3483        if (error)
3484                goto out_put;
3485
3486        if (flags & DLM_LKF_CANCEL)
3487                error = cancel_lock(ls, lkb, &args);
3488        else
3489                error = unlock_lock(ls, lkb, &args);
3490
3491        if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3492                error = 0;
3493        if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3494                error = 0;
3495 out_put:
3496        dlm_put_lkb(lkb);
3497 out:
3498        dlm_unlock_recovery(ls);
3499        dlm_put_lockspace(ls);
3500        return error;
3501}
3502
3503/*
3504 * send/receive routines for remote operations and replies
3505 *
3506 * send_args
3507 * send_common
3508 * send_request                 receive_request
3509 * send_convert                 receive_convert
3510 * send_unlock                  receive_unlock
3511 * send_cancel                  receive_cancel
3512 * send_grant                   receive_grant
3513 * send_bast                    receive_bast
3514 * send_lookup                  receive_lookup
3515 * send_remove                  receive_remove
3516 *
3517 *                              send_common_reply
3518 * receive_request_reply        send_request_reply
3519 * receive_convert_reply        send_convert_reply
3520 * receive_unlock_reply         send_unlock_reply
3521 * receive_cancel_reply         send_cancel_reply
3522 * receive_lookup_reply         send_lookup_reply
3523 */
3524
3525static int _create_message(struct dlm_ls *ls, int mb_len,
3526                           int to_nodeid, int mstype,
3527                           struct dlm_message **ms_ret,
3528                           struct dlm_mhandle **mh_ret)
3529{
3530        struct dlm_message *ms;
3531        struct dlm_mhandle *mh;
3532        char *mb;
3533
3534        /* get_buffer gives us a message handle (mh) that we need to
3535           pass into lowcomms_commit and a message buffer (mb) that we
3536           write our data into */
3537
3538        mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
3539        if (!mh)
3540                return -ENOBUFS;
3541
3542        memset(mb, 0, mb_len);
3543
3544        ms = (struct dlm_message *) mb;
3545
3546        ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3547        ms->m_header.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_lowcomms_commit_buffer(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                goto fail;
4180        }
4181
4182        r = lkb->lkb_resource;
4183
4184        hold_rsb(r);
4185        lock_rsb(r);
4186
4187        error = validate_message(lkb, ms);
4188        if (error)
4189                goto out;
4190
4191        receive_flags(lkb, ms);
4192
4193        error = receive_convert_args(ls, lkb, ms);
4194        if (error) {
4195                send_convert_reply(r, lkb, error);
4196                goto out;
4197        }
4198
4199        reply = !down_conversion(lkb);
4200
4201        error = do_convert(r, lkb);
4202        if (reply)
4203                send_convert_reply(r, lkb, error);
4204        do_convert_effects(r, lkb, error);
4205 out:
4206        unlock_rsb(r);
4207        put_rsb(r);
4208        dlm_put_lkb(lkb);
4209        return 0;
4210
4211 fail:
4212        setup_stub_lkb(ls, ms);
4213        send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4214        return error;
4215}
4216
4217static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4218{
4219        struct dlm_lkb *lkb;
4220        struct dlm_rsb *r;
4221        int error;
4222
4223        error = find_lkb(ls, ms->m_remid, &lkb);
4224        if (error)
4225                goto fail;
4226
4227        if (lkb->lkb_remid != ms->m_lkid) {
4228                log_error(ls, "receive_unlock %x remid %x remote %d %x",
4229                          lkb->lkb_id, lkb->lkb_remid,
4230                          ms->m_header.h_nodeid, ms->m_lkid);
4231                error = -ENOENT;
4232                goto fail;
4233        }
4234
4235        r = lkb->lkb_resource;
4236
4237        hold_rsb(r);
4238        lock_rsb(r);
4239
4240        error = validate_message(lkb, ms);
4241        if (error)
4242                goto out;
4243
4244        receive_flags(lkb, ms);
4245
4246        error = receive_unlock_args(ls, lkb, ms);
4247        if (error) {
4248                send_unlock_reply(r, lkb, error);
4249                goto out;
4250        }
4251
4252        error = do_unlock(r, lkb);
4253        send_unlock_reply(r, lkb, error);
4254        do_unlock_effects(r, lkb, error);
4255 out:
4256        unlock_rsb(r);
4257        put_rsb(r);
4258        dlm_put_lkb(lkb);
4259        return 0;
4260
4261 fail:
4262        setup_stub_lkb(ls, ms);
4263        send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4264        return error;
4265}
4266
4267static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4268{
4269        struct dlm_lkb *lkb;
4270        struct dlm_rsb *r;
4271        int error;
4272
4273        error = find_lkb(ls, ms->m_remid, &lkb);
4274        if (error)
4275                goto fail;
4276
4277        receive_flags(lkb, ms);
4278
4279        r = lkb->lkb_resource;
4280
4281        hold_rsb(r);
4282        lock_rsb(r);
4283
4284        error = validate_message(lkb, ms);
4285        if (error)
4286                goto out;
4287
4288        error = do_cancel(r, lkb);
4289        send_cancel_reply(r, lkb, error);
4290        do_cancel_effects(r, lkb, error);
4291 out:
4292        unlock_rsb(r);
4293        put_rsb(r);
4294        dlm_put_lkb(lkb);
4295        return 0;
4296
4297 fail:
4298        setup_stub_lkb(ls, ms);
4299        send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4300        return error;
4301}
4302
4303static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4304{
4305        struct dlm_lkb *lkb;
4306        struct dlm_rsb *r;
4307        int error;
4308
4309        error = find_lkb(ls, ms->m_remid, &lkb);
4310        if (error)
4311                return error;
4312
4313        r = lkb->lkb_resource;
4314
4315        hold_rsb(r);
4316        lock_rsb(r);
4317
4318        error = validate_message(lkb, ms);
4319        if (error)
4320                goto out;
4321
4322        receive_flags_reply(lkb, ms);
4323        if (is_altmode(lkb))
4324                munge_altmode(lkb, ms);
4325        grant_lock_pc(r, lkb, ms);
4326        queue_cast(r, lkb, 0);
4327 out:
4328        unlock_rsb(r);
4329        put_rsb(r);
4330        dlm_put_lkb(lkb);
4331        return 0;
4332}
4333
4334static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4335{
4336        struct dlm_lkb *lkb;
4337        struct dlm_rsb *r;
4338        int error;
4339
4340        error = find_lkb(ls, ms->m_remid, &lkb);
4341        if (error)
4342                return error;
4343
4344        r = lkb->lkb_resource;
4345
4346        hold_rsb(r);
4347        lock_rsb(r);
4348
4349        error = validate_message(lkb, ms);
4350        if (error)
4351                goto out;
4352
4353        queue_bast(r, lkb, ms->m_bastmode);
4354        lkb->lkb_highbast = ms->m_bastmode;
4355 out:
4356        unlock_rsb(r);
4357        put_rsb(r);
4358        dlm_put_lkb(lkb);
4359        return 0;
4360}
4361
4362static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4363{
4364        int len, error, ret_nodeid, from_nodeid, our_nodeid;
4365
4366        from_nodeid = ms->m_header.h_nodeid;
4367        our_nodeid = dlm_our_nodeid();
4368
4369        len = receive_extralen(ms);
4370
4371        error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4372                                  &ret_nodeid, NULL);
4373
4374        /* Optimization: we're master so treat lookup as a request */
4375        if (!error && ret_nodeid == our_nodeid) {
4376                receive_request(ls, ms);
4377                return;
4378        }
4379        send_lookup_reply(ls, ms, ret_nodeid, error);
4380}
4381
4382static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4383{
4384        char name[DLM_RESNAME_MAXLEN+1];
4385        struct dlm_rsb *r;
4386        uint32_t hash, b;
4387        int rv, len, dir_nodeid, from_nodeid;
4388
4389        from_nodeid = ms->m_header.h_nodeid;
4390
4391        len = receive_extralen(ms);
4392
4393        if (len > DLM_RESNAME_MAXLEN) {
4394                log_error(ls, "receive_remove from %d bad len %d",
4395                          from_nodeid, len);
4396                return;
4397        }
4398
4399        dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4400        if (dir_nodeid != dlm_our_nodeid()) {
4401                log_error(ls, "receive_remove from %d bad nodeid %d",
4402                          from_nodeid, dir_nodeid);
4403                return;
4404        }
4405
4406        /* Look for name on rsbtbl.toss, if it's there, kill it.
4407           If it's on rsbtbl.keep, it's being used, and we should ignore this
4408           message.  This is an expected race between the dir node sending a
4409           request to the master node at the same time as the master node sends
4410           a remove to the dir node.  The resolution to that race is for the
4411           dir node to ignore the remove message, and the master node to
4412           recreate the master rsb when it gets a request from the dir node for
4413           an rsb it doesn't have. */
4414
4415        memset(name, 0, sizeof(name));
4416        memcpy(name, ms->m_extra, len);
4417
4418        hash = jhash(name, len, 0);
4419        b = hash & (ls->ls_rsbtbl_size - 1);
4420
4421        spin_lock(&ls->ls_rsbtbl[b].lock);
4422
4423        rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4424        if (rv) {
4425                /* verify the rsb is on keep list per comment above */
4426                rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4427                if (rv) {
4428                        /* should not happen */
4429                        log_error(ls, "receive_remove from %d not found %s",
4430                                  from_nodeid, name);
4431                        spin_unlock(&ls->ls_rsbtbl[b].lock);
4432                        return;
4433                }
4434                if (r->res_master_nodeid != from_nodeid) {
4435                        /* should not happen */
4436                        log_error(ls, "receive_remove keep from %d master %d",
4437                                  from_nodeid, r->res_master_nodeid);
4438                        dlm_print_rsb(r);
4439                        spin_unlock(&ls->ls_rsbtbl[b].lock);
4440                        return;
4441                }
4442
4443                log_debug(ls, "receive_remove from %d master %d first %x %s",
4444                          from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4445                          name);
4446                spin_unlock(&ls->ls_rsbtbl[b].lock);
4447                return;
4448        }
4449
4450        if (r->res_master_nodeid != from_nodeid) {
4451                log_error(ls, "receive_remove toss from %d master %d",
4452                          from_nodeid, r->res_master_nodeid);
4453                dlm_print_rsb(r);
4454                spin_unlock(&ls->ls_rsbtbl[b].lock);
4455                return;
4456        }
4457
4458        if (kref_put(&r->res_ref, kill_rsb)) {
4459                rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4460                spin_unlock(&ls->ls_rsbtbl[b].lock);
4461                dlm_free_rsb(r);
4462        } else {
4463                log_error(ls, "receive_remove from %d rsb ref error",
4464                          from_nodeid);
4465                dlm_print_rsb(r);
4466                spin_unlock(&ls->ls_rsbtbl[b].lock);
4467        }
4468}
4469
4470static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4471{
4472        do_purge(ls, ms->m_nodeid, ms->m_pid);
4473}
4474
4475static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4476{
4477        struct dlm_lkb *lkb;
4478        struct dlm_rsb *r;
4479        int error, mstype, result;
4480        int from_nodeid = ms->m_header.h_nodeid;
4481
4482        error = find_lkb(ls, ms->m_remid, &lkb);
4483        if (error)
4484                return error;
4485
4486        r = lkb->lkb_resource;
4487        hold_rsb(r);
4488        lock_rsb(r);
4489
4490        error = validate_message(lkb, ms);
4491        if (error)
4492                goto out;
4493
4494        mstype = lkb->lkb_wait_type;
4495        error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4496        if (error) {
4497                log_error(ls, "receive_request_reply %x remote %d %x result %d",
4498                          lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4499                dlm_dump_rsb(r);
4500                goto out;
4501        }
4502
4503        /* Optimization: the dir node was also the master, so it took our
4504           lookup as a request and sent request reply instead of lookup reply */
4505        if (mstype == DLM_MSG_LOOKUP) {
4506                r->res_master_nodeid = from_nodeid;
4507                r->res_nodeid = from_nodeid;
4508                lkb->lkb_nodeid = from_nodeid;
4509        }
4510
4511        /* this is the value returned from do_request() on the master */
4512        result = ms->m_result;
4513
4514        switch (result) {
4515        case -EAGAIN:
4516                /* request would block (be queued) on remote master */
4517                queue_cast(r, lkb, -EAGAIN);
4518                confirm_master(r, -EAGAIN);
4519                unhold_lkb(lkb); /* undoes create_lkb() */
4520                break;
4521
4522        case -EINPROGRESS:
4523        case 0:
4524                /* request was queued or granted on remote master */
4525                receive_flags_reply(lkb, ms);
4526                lkb->lkb_remid = ms->m_lkid;
4527                if (is_altmode(lkb))
4528                        munge_altmode(lkb, ms);
4529                if (result) {
4530                        add_lkb(r, lkb, DLM_LKSTS_WAITING);
4531                        add_timeout(lkb);
4532                } else {
4533                        grant_lock_pc(r, lkb, ms);
4534                        queue_cast(r, lkb, 0);
4535                }
4536                confirm_master(r, result);
4537                break;
4538
4539        case -EBADR:
4540        case -ENOTBLK:
4541                /* find_rsb failed to find rsb or rsb wasn't master */
4542                log_limit(ls, "receive_request_reply %x from %d %d "
4543                          "master %d dir %d first %x %s", lkb->lkb_id,
4544                          from_nodeid, result, r->res_master_nodeid,
4545                          r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4546
4547                if (r->res_dir_nodeid != dlm_our_nodeid() &&
4548                    r->res_master_nodeid != dlm_our_nodeid()) {
4549                        /* cause _request_lock->set_master->send_lookup */
4550                        r->res_master_nodeid = 0;
4551                        r->res_nodeid = -1;
4552                        lkb->lkb_nodeid = -1;
4553                }
4554
4555                if (is_overlap(lkb)) {
4556                        /* we'll ignore error in cancel/unlock reply */
4557                        queue_cast_overlap(r, lkb);
4558                        confirm_master(r, result);
4559                        unhold_lkb(lkb); /* undoes create_lkb() */
4560                } else {
4561                        _request_lock(r, lkb);
4562
4563                        if (r->res_master_nodeid == dlm_our_nodeid())
4564                                confirm_master(r, 0);
4565                }
4566                break;
4567
4568        default:
4569                log_error(ls, "receive_request_reply %x error %d",
4570                          lkb->lkb_id, result);
4571        }
4572
4573        if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4574                log_debug(ls, "receive_request_reply %x result %d unlock",
4575                          lkb->lkb_id, result);
4576                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4577                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4578                send_unlock(r, lkb);
4579        } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4580                log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4581                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4582                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4583                send_cancel(r, lkb);
4584        } else {
4585                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4586                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4587        }
4588 out:
4589        unlock_rsb(r);
4590        put_rsb(r);
4591        dlm_put_lkb(lkb);
4592        return 0;
4593}
4594
4595static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4596                                    struct dlm_message *ms)
4597{
4598        /* this is the value returned from do_convert() on the master */
4599        switch (ms->m_result) {
4600        case -EAGAIN:
4601                /* convert would block (be queued) on remote master */
4602                queue_cast(r, lkb, -EAGAIN);
4603                break;
4604
4605        case -EDEADLK:
4606                receive_flags_reply(lkb, ms);
4607                revert_lock_pc(r, lkb);
4608                queue_cast(r, lkb, -EDEADLK);
4609                break;
4610
4611        case -EINPROGRESS:
4612                /* convert was queued on remote master */
4613                receive_flags_reply(lkb, ms);
4614                if (is_demoted(lkb))
4615                        munge_demoted(lkb);
4616                del_lkb(r, lkb);
4617                add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4618                add_timeout(lkb);
4619                break;
4620
4621        case 0:
4622                /* convert was granted on remote master */
4623                receive_flags_reply(lkb, ms);
4624                if (is_demoted(lkb))
4625                        munge_demoted(lkb);
4626                grant_lock_pc(r, lkb, ms);
4627                queue_cast(r, lkb, 0);
4628                break;
4629
4630        default:
4631                log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4632                          lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4633                          ms->m_result);
4634                dlm_print_rsb(r);
4635                dlm_print_lkb(lkb);
4636        }
4637}
4638
4639static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4640{
4641        struct dlm_rsb *r = lkb->lkb_resource;
4642        int error;
4643
4644        hold_rsb(r);
4645        lock_rsb(r);
4646
4647        error = validate_message(lkb, ms);
4648        if (error)
4649                goto out;
4650
4651        /* stub reply can happen with waiters_mutex held */
4652        error = remove_from_waiters_ms(lkb, ms);
4653        if (error)
4654                goto out;
4655
4656        __receive_convert_reply(r, lkb, ms);
4657 out:
4658        unlock_rsb(r);
4659        put_rsb(r);
4660}
4661
4662static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4663{
4664        struct dlm_lkb *lkb;
4665        int error;
4666
4667        error = find_lkb(ls, ms->m_remid, &lkb);
4668        if (error)
4669                return error;
4670
4671        _receive_convert_reply(lkb, ms);
4672        dlm_put_lkb(lkb);
4673        return 0;
4674}
4675
4676static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4677{
4678        struct dlm_rsb *r = lkb->lkb_resource;
4679        int error;
4680
4681        hold_rsb(r);
4682        lock_rsb(r);
4683
4684        error = validate_message(lkb, ms);
4685        if (error)
4686                goto out;
4687
4688        /* stub reply can happen with waiters_mutex held */
4689        error = remove_from_waiters_ms(lkb, ms);
4690        if (error)
4691                goto out;
4692
4693        /* this is the value returned from do_unlock() on the master */
4694
4695        switch (ms->m_result) {
4696        case -DLM_EUNLOCK:
4697                receive_flags_reply(lkb, ms);
4698                remove_lock_pc(r, lkb);
4699                queue_cast(r, lkb, -DLM_EUNLOCK);
4700                break;
4701        case -ENOENT:
4702                break;
4703        default:
4704                log_error(r->res_ls, "receive_unlock_reply %x error %d",
4705                          lkb->lkb_id, ms->m_result);
4706        }
4707 out:
4708        unlock_rsb(r);
4709        put_rsb(r);
4710}
4711
4712static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4713{
4714        struct dlm_lkb *lkb;
4715        int error;
4716
4717        error = find_lkb(ls, ms->m_remid, &lkb);
4718        if (error)
4719                return error;
4720
4721        _receive_unlock_reply(lkb, ms);
4722        dlm_put_lkb(lkb);
4723        return 0;
4724}
4725
4726static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4727{
4728        struct dlm_rsb *r = lkb->lkb_resource;
4729        int error;
4730
4731        hold_rsb(r);
4732        lock_rsb(r);
4733
4734        error = validate_message(lkb, ms);
4735        if (error)
4736                goto out;
4737
4738        /* stub reply can happen with waiters_mutex held */
4739        error = remove_from_waiters_ms(lkb, ms);
4740        if (error)
4741                goto out;
4742
4743        /* this is the value returned from do_cancel() on the master */
4744
4745        switch (ms->m_result) {
4746        case -DLM_ECANCEL:
4747                receive_flags_reply(lkb, ms);
4748                revert_lock_pc(r, lkb);
4749                queue_cast(r, lkb, -DLM_ECANCEL);
4750                break;
4751        case 0:
4752                break;
4753        default:
4754                log_error(r->res_ls, "receive_cancel_reply %x error %d",
4755                          lkb->lkb_id, ms->m_result);
4756        }
4757 out:
4758        unlock_rsb(r);
4759        put_rsb(r);
4760}
4761
4762static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4763{
4764        struct dlm_lkb *lkb;
4765        int error;
4766
4767        error = find_lkb(ls, ms->m_remid, &lkb);
4768        if (error)
4769                return error;
4770
4771        _receive_cancel_reply(lkb, ms);
4772        dlm_put_lkb(lkb);
4773        return 0;
4774}
4775
4776static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4777{
4778        struct dlm_lkb *lkb;
4779        struct dlm_rsb *r;
4780        int error, ret_nodeid;
4781        int do_lookup_list = 0;
4782
4783        error = find_lkb(ls, ms->m_lkid, &lkb);
4784        if (error) {
4785                log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4786                return;
4787        }
4788
4789        /* ms->m_result is the value returned by dlm_master_lookup on dir node
4790           FIXME: will a non-zero error ever be returned? */
4791
4792        r = lkb->lkb_resource;
4793        hold_rsb(r);
4794        lock_rsb(r);
4795
4796        error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4797        if (error)
4798                goto out;
4799
4800        ret_nodeid = ms->m_nodeid;
4801
4802        /* We sometimes receive a request from the dir node for this
4803           rsb before we've received the dir node's loookup_reply for it.
4804           The request from the dir node implies we're the master, so we set
4805           ourself as master in receive_request_reply, and verify here that
4806           we are indeed the master. */
4807
4808        if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4809                /* This should never happen */
4810                log_error(ls, "receive_lookup_reply %x from %d ret %d "
4811                          "master %d dir %d our %d first %x %s",
4812                          lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4813                          r->res_master_nodeid, r->res_dir_nodeid,
4814                          dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4815        }
4816
4817        if (ret_nodeid == dlm_our_nodeid()) {
4818                r->res_master_nodeid = ret_nodeid;
4819                r->res_nodeid = 0;
4820                do_lookup_list = 1;
4821                r->res_first_lkid = 0;
4822        } else if (ret_nodeid == -1) {
4823                /* the remote node doesn't believe it's the dir node */
4824                log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4825                          lkb->lkb_id, ms->m_header.h_nodeid);
4826                r->res_master_nodeid = 0;
4827                r->res_nodeid = -1;
4828                lkb->lkb_nodeid = -1;
4829        } else {
4830                /* set_master() will set lkb_nodeid from r */
4831                r->res_master_nodeid = ret_nodeid;
4832                r->res_nodeid = ret_nodeid;
4833        }
4834
4835        if (is_overlap(lkb)) {
4836                log_debug(ls, "receive_lookup_reply %x unlock %x",
4837                          lkb->lkb_id, lkb->lkb_flags);
4838                queue_cast_overlap(r, lkb);
4839                unhold_lkb(lkb); /* undoes create_lkb() */
4840                goto out_list;
4841        }
4842
4843        _request_lock(r, lkb);
4844
4845 out_list:
4846        if (do_lookup_list)
4847                process_lookup_list(r);
4848 out:
4849        unlock_rsb(r);
4850        put_rsb(r);
4851        dlm_put_lkb(lkb);
4852}
4853
4854static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4855                             uint32_t saved_seq)
4856{
4857        int error = 0, noent = 0;
4858
4859        if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4860                log_limit(ls, "receive %d from non-member %d %x %x %d",
4861                          ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4862                          ms->m_remid, ms->m_result);
4863                return;
4864        }
4865
4866        switch (ms->m_type) {
4867
4868        /* messages sent to a master node */
4869
4870        case DLM_MSG_REQUEST:
4871                error = receive_request(ls, ms);
4872                break;
4873
4874        case DLM_MSG_CONVERT:
4875                error = receive_convert(ls, ms);
4876                break;
4877
4878        case DLM_MSG_UNLOCK:
4879                error = receive_unlock(ls, ms);
4880                break;
4881
4882        case DLM_MSG_CANCEL:
4883                noent = 1;
4884                error = receive_cancel(ls, ms);
4885                break;
4886
4887        /* messages sent from a master node (replies to above) */
4888
4889        case DLM_MSG_REQUEST_REPLY:
4890                error = receive_request_reply(ls, ms);
4891                break;
4892
4893        case DLM_MSG_CONVERT_REPLY:
4894                error = receive_convert_reply(ls, ms);
4895                break;
4896
4897        case DLM_MSG_UNLOCK_REPLY:
4898                error = receive_unlock_reply(ls, ms);
4899                break;
4900
4901        case DLM_MSG_CANCEL_REPLY:
4902                error = receive_cancel_reply(ls, ms);
4903                break;
4904
4905        /* messages sent from a master node (only two types of async msg) */
4906
4907        case DLM_MSG_GRANT:
4908                noent = 1;
4909                error = receive_grant(ls, ms);
4910                break;
4911
4912        case DLM_MSG_BAST:
4913                noent = 1;
4914                error = receive_bast(ls, ms);
4915                break;
4916
4917        /* messages sent to a dir node */
4918
4919        case DLM_MSG_LOOKUP:
4920                receive_lookup(ls, ms);
4921                break;
4922
4923        case DLM_MSG_REMOVE:
4924                receive_remove(ls, ms);
4925                break;
4926
4927        /* messages sent from a dir node (remove has no reply) */
4928
4929        case DLM_MSG_LOOKUP_REPLY:
4930                receive_lookup_reply(ls, ms);
4931                break;
4932
4933        /* other messages */
4934
4935        case DLM_MSG_PURGE:
4936                receive_purge(ls, ms);
4937                break;
4938
4939        default:
4940                log_error(ls, "unknown message type %d", ms->m_type);
4941        }
4942
4943        /*
4944         * When checking for ENOENT, we're checking the result of
4945         * find_lkb(m_remid):
4946         *
4947         * The lock id referenced in the message wasn't found.  This may
4948         * happen in normal usage for the async messages and cancel, so
4949         * only use log_debug for them.
4950         *
4951         * Some errors are expected and normal.
4952         */
4953
4954        if (error == -ENOENT && noent) {
4955                log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4956                          ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4957                          ms->m_lkid, saved_seq);
4958        } else if (error == -ENOENT) {
4959                log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4960                          ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4961                          ms->m_lkid, saved_seq);
4962
4963                if (ms->m_type == DLM_MSG_CONVERT)
4964                        dlm_dump_rsb_hash(ls, ms->m_hash);
4965        }
4966
4967        if (error == -EINVAL) {
4968                log_error(ls, "receive %d inval from %d lkid %x remid %x "
4969                          "saved_seq %u",
4970                          ms->m_type, ms->m_header.h_nodeid,
4971                          ms->m_lkid, ms->m_remid, saved_seq);
4972        }
4973}
4974
4975/* If the lockspace is in recovery mode (locking stopped), then normal
4976   messages are saved on the requestqueue for processing after recovery is
4977   done.  When not in recovery mode, we wait for dlm_recoverd to drain saved
4978   messages off the requestqueue before we process new ones. This occurs right
4979   after recovery completes when we transition from saving all messages on
4980   requestqueue, to processing all the saved messages, to processing new
4981   messages as they arrive. */
4982
4983static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4984                                int nodeid)
4985{
4986        if (dlm_locking_stopped(ls)) {
4987                /* If we were a member of this lockspace, left, and rejoined,
4988                   other nodes may still be sending us messages from the
4989                   lockspace generation before we left. */
4990                if (!ls->ls_generation) {
4991                        log_limit(ls, "receive %d from %d ignore old gen",
4992                                  ms->m_type, nodeid);
4993                        return;
4994                }
4995
4996                dlm_add_requestqueue(ls, nodeid, ms);
4997        } else {
4998                dlm_wait_requestqueue(ls);
4999                _receive_message(ls, ms, 0);
5000        }
5001}
5002
5003/* This is called by dlm_recoverd to process messages that were saved on
5004   the requestqueue. */
5005
5006void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5007                               uint32_t saved_seq)
5008{
5009        _receive_message(ls, ms, saved_seq);
5010}
5011
5012/* This is called by the midcomms layer when something is received for
5013   the lockspace.  It could be either a MSG (normal message sent as part of
5014   standard locking activity) or an RCOM (recovery message sent as part of
5015   lockspace recovery). */
5016
5017void dlm_receive_buffer(union dlm_packet *p, int nodeid)
5018{
5019        struct dlm_header *hd = &p->header;
5020        struct dlm_ls *ls;
5021        int type = 0;
5022
5023        switch (hd->h_cmd) {
5024        case DLM_MSG:
5025                dlm_message_in(&p->message);
5026                type = p->message.m_type;
5027                break;
5028        case DLM_RCOM:
5029                dlm_rcom_in(&p->rcom);
5030                type = p->rcom.rc_type;
5031                break;
5032        default:
5033                log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5034                return;
5035        }
5036
5037        if (hd->h_nodeid != nodeid) {
5038                log_print("invalid h_nodeid %d from %d lockspace %x",
5039                          hd->h_nodeid, nodeid, hd->h_lockspace);
5040                return;
5041        }
5042
5043        ls = dlm_find_lockspace_global(hd->h_lockspace);
5044        if (!ls) {
5045                if (dlm_config.ci_log_debug) {
5046                        printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5047                                "%u from %d cmd %d type %d\n",
5048                                hd->h_lockspace, nodeid, hd->h_cmd, type);
5049                }
5050
5051                if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5052                        dlm_send_ls_not_ready(nodeid, &p->rcom);
5053                return;
5054        }
5055
5056        /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5057           be inactive (in this ls) before transitioning to recovery mode */
5058
5059        down_read(&ls->ls_recv_active);
5060        if (hd->h_cmd == DLM_MSG)
5061                dlm_receive_message(ls, &p->message, nodeid);
5062        else
5063                dlm_receive_rcom(ls, &p->rcom, nodeid);
5064        up_read(&ls->ls_recv_active);
5065
5066        dlm_put_lockspace(ls);
5067}
5068
5069static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5070                                   struct dlm_message *ms_stub)
5071{
5072        if (middle_conversion(lkb)) {
5073                hold_lkb(lkb);
5074                memset(ms_stub, 0, sizeof(struct dlm_message));
5075                ms_stub->m_flags = DLM_IFL_STUB_MS;
5076                ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
5077                ms_stub->m_result = -EINPROGRESS;
5078                ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5079                _receive_convert_reply(lkb, ms_stub);
5080
5081                /* Same special case as in receive_rcom_lock_args() */
5082                lkb->lkb_grmode = DLM_LOCK_IV;
5083                rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5084                unhold_lkb(lkb);
5085
5086        } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5087                lkb->lkb_flags |= DLM_IFL_RESEND;
5088        }
5089
5090        /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5091           conversions are async; there's no reply from the remote master */
5092}
5093
5094/* A waiting lkb needs recovery if the master node has failed, or
5095   the master node is changing (only when no directory is used) */
5096
5097static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5098                                 int dir_nodeid)
5099{
5100        if (dlm_no_directory(ls))
5101                return 1;
5102
5103        if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5104                return 1;
5105
5106        return 0;
5107}
5108
5109/* Recovery for locks that are waiting for replies from nodes that are now
5110   gone.  We can just complete unlocks and cancels by faking a reply from the
5111   dead node.  Requests and up-conversions we flag to be resent after
5112   recovery.  Down-conversions can just be completed with a fake reply like
5113   unlocks.  Conversions between PR and CW need special attention. */
5114
5115void dlm_recover_waiters_pre(struct dlm_ls *ls)
5116{
5117        struct dlm_lkb *lkb, *safe;
5118        struct dlm_message *ms_stub;
5119        int wait_type, stub_unlock_result, stub_cancel_result;
5120        int dir_nodeid;
5121
5122        ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5123        if (!ms_stub)
5124                return;
5125
5126        mutex_lock(&ls->ls_waiters_mutex);
5127
5128        list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5129
5130                dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5131
5132                /* exclude debug messages about unlocks because there can be so
5133                   many and they aren't very interesting */
5134
5135                if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5136                        log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5137                                  "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5138                                  lkb->lkb_id,
5139                                  lkb->lkb_remid,
5140                                  lkb->lkb_wait_type,
5141                                  lkb->lkb_resource->res_nodeid,
5142                                  lkb->lkb_nodeid,
5143                                  lkb->lkb_wait_nodeid,
5144                                  dir_nodeid);
5145                }
5146
5147                /* all outstanding lookups, regardless of destination  will be
5148                   resent after recovery is done */
5149
5150                if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5151                        lkb->lkb_flags |= DLM_IFL_RESEND;
5152                        continue;
5153                }
5154
5155                if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5156                        continue;
5157
5158                wait_type = lkb->lkb_wait_type;
5159                stub_unlock_result = -DLM_EUNLOCK;
5160                stub_cancel_result = -DLM_ECANCEL;
5161
5162                /* Main reply may have been received leaving a zero wait_type,
5163                   but a reply for the overlapping op may not have been
5164                   received.  In that case we need to fake the appropriate
5165                   reply for the overlap op. */
5166
5167                if (!wait_type) {
5168                        if (is_overlap_cancel(lkb)) {
5169                                wait_type = DLM_MSG_CANCEL;
5170                                if (lkb->lkb_grmode == DLM_LOCK_IV)
5171                                        stub_cancel_result = 0;
5172                        }
5173                        if (is_overlap_unlock(lkb)) {
5174                                wait_type = DLM_MSG_UNLOCK;
5175                                if (lkb->lkb_grmode == DLM_LOCK_IV)
5176                                        stub_unlock_result = -ENOENT;
5177                        }
5178
5179                        log_debug(ls, "rwpre overlap %x %x %d %d %d",
5180                                  lkb->lkb_id, lkb->lkb_flags, wait_type,
5181                                  stub_cancel_result, stub_unlock_result);
5182                }
5183
5184                switch (wait_type) {
5185
5186                case DLM_MSG_REQUEST:
5187                        lkb->lkb_flags |= DLM_IFL_RESEND;
5188                        break;
5189
5190                case DLM_MSG_CONVERT:
5191                        recover_convert_waiter(ls, lkb, ms_stub);
5192                        break;
5193
5194                case DLM_MSG_UNLOCK:
5195                        hold_lkb(lkb);
5196                        memset(ms_stub, 0, sizeof(struct dlm_message));
5197                        ms_stub->m_flags = DLM_IFL_STUB_MS;
5198                        ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
5199                        ms_stub->m_result = stub_unlock_result;
5200                        ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5201                        _receive_unlock_reply(lkb, ms_stub);
5202                        dlm_put_lkb(lkb);
5203                        break;
5204
5205                case DLM_MSG_CANCEL:
5206                        hold_lkb(lkb);
5207                        memset(ms_stub, 0, sizeof(struct dlm_message));
5208                        ms_stub->m_flags = DLM_IFL_STUB_MS;
5209                        ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
5210                        ms_stub->m_result = stub_cancel_result;
5211                        ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5212                        _receive_cancel_reply(lkb, ms_stub);
5213                        dlm_put_lkb(lkb);
5214                        break;
5215
5216                default:
5217                        log_error(ls, "invalid lkb wait_type %d %d",
5218                                  lkb->lkb_wait_type, wait_type);
5219                }
5220                schedule();
5221        }
5222        mutex_unlock(&ls->ls_waiters_mutex);
5223        kfree(ms_stub);
5224}
5225
5226static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5227{
5228        struct dlm_lkb *lkb;
5229        int found = 0;
5230
5231        mutex_lock(&ls->ls_waiters_mutex);
5232        list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5233                if (lkb->lkb_flags & DLM_IFL_RESEND) {
5234                        hold_lkb(lkb);
5235                        found = 1;
5236                        break;
5237                }
5238        }
5239        mutex_unlock(&ls->ls_waiters_mutex);
5240
5241        if (!found)
5242                lkb = NULL;
5243        return lkb;
5244}
5245
5246/* Deal with lookups and lkb's marked RESEND from _pre.  We may now be the
5247   master or dir-node for r.  Processing the lkb may result in it being placed
5248   back on waiters. */
5249
5250/* We do this after normal locking has been enabled and any saved messages
5251   (in requestqueue) have been processed.  We should be confident that at
5252   this point we won't get or process a reply to any of these waiting
5253   operations.  But, new ops may be coming in on the rsbs/locks here from
5254   userspace or remotely. */
5255
5256/* there may have been an overlap unlock/cancel prior to recovery or after
5257   recovery.  if before, the lkb may still have a pos wait_count; if after, the
5258   overlap flag would just have been set and nothing new sent.  we can be
5259   confident here than any replies to either the initial op or overlap ops
5260   prior to recovery have been received. */
5261
5262int dlm_recover_waiters_post(struct dlm_ls *ls)
5263{
5264        struct dlm_lkb *lkb;
5265        struct dlm_rsb *r;
5266        int error = 0, mstype, err, oc, ou;
5267
5268        while (1) {
5269                if (dlm_locking_stopped(ls)) {
5270                        log_debug(ls, "recover_waiters_post aborted");
5271                        error = -EINTR;
5272                        break;
5273                }
5274
5275                lkb = find_resend_waiter(ls);
5276                if (!lkb)
5277                        break;
5278
5279                r = lkb->lkb_resource;
5280                hold_rsb(r);
5281                lock_rsb(r);
5282
5283                mstype = lkb->lkb_wait_type;
5284                oc = is_overlap_cancel(lkb);
5285                ou = is_overlap_unlock(lkb);
5286                err = 0;
5287
5288                log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5289                          "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5290                          "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5291                          r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5292                          dlm_dir_nodeid(r), oc, ou);
5293
5294                /* At this point we assume that we won't get a reply to any
5295                   previous op or overlap op on this lock.  First, do a big
5296                   remove_from_waiters() for all previous ops. */
5297
5298                lkb->lkb_flags &= ~DLM_IFL_RESEND;
5299                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5300                lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5301                lkb->lkb_wait_type = 0;
5302                lkb->lkb_wait_count = 0;
5303                mutex_lock(&ls->ls_waiters_mutex);
5304                list_del_init(&lkb->lkb_wait_reply);
5305                mutex_unlock(&ls->ls_waiters_mutex);
5306                unhold_lkb(lkb); /* for waiters list */
5307
5308                if (oc || ou) {
5309                        /* do an unlock or cancel instead of resending */
5310                        switch (mstype) {
5311                        case DLM_MSG_LOOKUP:
5312                        case DLM_MSG_REQUEST:
5313                                queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5314                                                        -DLM_ECANCEL);
5315                                unhold_lkb(lkb); /* undoes create_lkb() */
5316                                break;
5317                        case DLM_MSG_CONVERT:
5318                                if (oc) {
5319                                        queue_cast(r, lkb, -DLM_ECANCEL);
5320                                } else {
5321                                        lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5322                                        _unlock_lock(r, lkb);
5323                                }
5324                                break;
5325                        default:
5326                                err = 1;
5327                        }
5328                } else {
5329                        switch (mstype) {
5330                        case DLM_MSG_LOOKUP:
5331                        case DLM_MSG_REQUEST:
5332                                _request_lock(r, lkb);
5333                                if (is_master(r))
5334                                        confirm_master(r, 0);
5335                                break;
5336                        case DLM_MSG_CONVERT:
5337                                _convert_lock(r, lkb);
5338                                break;
5339                        default:
5340                                err = 1;
5341                        }
5342                }
5343
5344                if (err) {
5345                        log_error(ls, "waiter %x msg %d r_nodeid %d "
5346                                  "dir_nodeid %d overlap %d %d",
5347                                  lkb->lkb_id, mstype, r->res_nodeid,
5348                                  dlm_dir_nodeid(r), oc, ou);
5349                }
5350                unlock_rsb(r);
5351                put_rsb(r);
5352                dlm_put_lkb(lkb);
5353        }
5354
5355        return error;
5356}
5357
5358static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5359                              struct list_head *list)
5360{
5361        struct dlm_lkb *lkb, *safe;
5362
5363        list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5364                if (!is_master_copy(lkb))
5365                        continue;
5366
5367                /* don't purge lkbs we've added in recover_master_copy for
5368                   the current recovery seq */
5369
5370                if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5371                        continue;
5372
5373                del_lkb(r, lkb);
5374
5375                /* this put should free the lkb */
5376                if (!dlm_put_lkb(lkb))
5377                        log_error(ls, "purged mstcpy lkb not released");
5378        }
5379}
5380
5381void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5382{
5383        struct dlm_ls *ls = r->res_ls;
5384
5385        purge_mstcpy_list(ls, r, &r->res_grantqueue);
5386        purge_mstcpy_list(ls, r, &r->res_convertqueue);
5387        purge_mstcpy_list(ls, r, &r->res_waitqueue);
5388}
5389
5390static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5391                            struct list_head *list,
5392                            int nodeid_gone, unsigned int *count)
5393{
5394        struct dlm_lkb *lkb, *safe;
5395
5396        list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5397                if (!is_master_copy(lkb))
5398                        continue;
5399
5400                if ((lkb->lkb_nodeid == nodeid_gone) ||
5401                    dlm_is_removed(ls, lkb->lkb_nodeid)) {
5402
5403                        /* tell recover_lvb to invalidate the lvb
5404                           because a node holding EX/PW failed */
5405                        if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5406                            (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5407                                rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5408                        }
5409
5410                        del_lkb(r, lkb);
5411
5412                        /* this put should free the lkb */
5413                        if (!dlm_put_lkb(lkb))
5414                                log_error(ls, "purged dead lkb not released");
5415
5416                        rsb_set_flag(r, RSB_RECOVER_GRANT);
5417
5418                        (*count)++;
5419                }
5420        }
5421}
5422
5423/* Get rid of locks held by nodes that are gone. */
5424
5425void dlm_recover_purge(struct dlm_ls *ls)
5426{
5427        struct dlm_rsb *r;
5428        struct dlm_member *memb;
5429        int nodes_count = 0;
5430        int nodeid_gone = 0;
5431        unsigned int lkb_count = 0;
5432
5433        /* cache one removed nodeid to optimize the common
5434           case of a single node removed */
5435
5436        list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5437                nodes_count++;
5438                nodeid_gone = memb->nodeid;
5439        }
5440
5441        if (!nodes_count)
5442                return;
5443
5444        down_write(&ls->ls_root_sem);
5445        list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5446                hold_rsb(r);
5447                lock_rsb(r);
5448                if (is_master(r)) {
5449                        purge_dead_list(ls, r, &r->res_grantqueue,
5450                                        nodeid_gone, &lkb_count);
5451                        purge_dead_list(ls, r, &r->res_convertqueue,
5452                                        nodeid_gone, &lkb_count);
5453                        purge_dead_list(ls, r, &r->res_waitqueue,
5454                                        nodeid_gone, &lkb_count);
5455                }
5456                unlock_rsb(r);
5457                unhold_rsb(r);
5458                cond_resched();
5459        }
5460        up_write(&ls->ls_root_sem);
5461
5462        if (lkb_count)
5463                log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5464                          lkb_count, nodes_count);
5465}
5466
5467static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5468{
5469        struct rb_node *n;
5470        struct dlm_rsb *r;
5471
5472        spin_lock(&ls->ls_rsbtbl[bucket].lock);
5473        for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5474                r = rb_entry(n, struct dlm_rsb, res_hashnode);
5475
5476                if (!rsb_flag(r, RSB_RECOVER_GRANT))
5477                        continue;
5478                if (!is_master(r)) {
5479                        rsb_clear_flag(r, RSB_RECOVER_GRANT);
5480                        continue;
5481                }
5482                hold_rsb(r);
5483                spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5484                return r;
5485        }
5486        spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5487        return NULL;
5488}
5489
5490/*
5491 * Attempt to grant locks on resources that we are the master of.
5492 * Locks may have become grantable during recovery because locks
5493 * from departed nodes have been purged (or not rebuilt), allowing
5494 * previously blocked locks to now be granted.  The subset of rsb's
5495 * we are interested in are those with lkb's on either the convert or
5496 * waiting queues.
5497 *
5498 * Simplest would be to go through each master rsb and check for non-empty
5499 * convert or waiting queues, and attempt to grant on those rsbs.
5500 * Checking the queues requires lock_rsb, though, for which we'd need
5501 * to release the rsbtbl lock.  This would make iterating through all
5502 * rsb's very inefficient.  So, we rely on earlier recovery routines
5503 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5504 * locks for.
5505 */
5506
5507void dlm_recover_grant(struct dlm_ls *ls)
5508{
5509        struct dlm_rsb *r;
5510        int bucket = 0;
5511        unsigned int count = 0;
5512        unsigned int rsb_count = 0;
5513        unsigned int lkb_count = 0;
5514
5515        while (1) {
5516                r = find_grant_rsb(ls, bucket);
5517                if (!r) {
5518                        if (bucket == ls->ls_rsbtbl_size - 1)
5519                                break;
5520                        bucket++;
5521                        continue;
5522                }
5523                rsb_count++;
5524                count = 0;
5525                lock_rsb(r);
5526                /* the RECOVER_GRANT flag is checked in the grant path */
5527                grant_pending_locks(r, &count);
5528                rsb_clear_flag(r, RSB_RECOVER_GRANT);
5529                lkb_count += count;
5530                confirm_master(r, 0);
5531                unlock_rsb(r);
5532                put_rsb(r);
5533                cond_resched();
5534        }
5535
5536        if (lkb_count)
5537                log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5538                          lkb_count, rsb_count);
5539}
5540
5541static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5542                                         uint32_t remid)
5543{
5544        struct dlm_lkb *lkb;
5545
5546        list_for_each_entry(lkb, head, lkb_statequeue) {
5547                if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5548                        return lkb;
5549        }
5550        return NULL;
5551}
5552
5553static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5554                                    uint32_t remid)
5555{
5556        struct dlm_lkb *lkb;
5557
5558        lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5559        if (lkb)
5560                return lkb;
5561        lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5562        if (lkb)
5563                return lkb;
5564        lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5565        if (lkb)
5566                return lkb;
5567        return NULL;
5568}
5569
5570/* needs at least dlm_rcom + rcom_lock */
5571static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5572                                  struct dlm_rsb *r, struct dlm_rcom *rc)
5573{
5574        struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5575
5576        lkb->lkb_nodeid = rc->rc_header.h_nodeid;
5577        lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5578        lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5579        lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5580        lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5581        lkb->lkb_flags |= DLM_IFL_MSTCPY;
5582        lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5583        lkb->lkb_rqmode = rl->rl_rqmode;
5584        lkb->lkb_grmode = rl->rl_grmode;
5585        /* don't set lkb_status because add_lkb wants to itself */
5586
5587        lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5588        lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5589
5590        if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5591                int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
5592                         sizeof(struct rcom_lock);
5593                if (lvblen > ls->ls_lvblen)
5594                        return -EINVAL;
5595                lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5596                if (!lkb->lkb_lvbptr)
5597                        return -ENOMEM;
5598                memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5599        }
5600
5601        /* Conversions between PR and CW (middle modes) need special handling.
5602           The real granted mode of these converting locks cannot be determined
5603           until all locks have been rebuilt on the rsb (recover_conversion) */
5604
5605        if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5606            middle_conversion(lkb)) {
5607                rl->rl_status = DLM_LKSTS_CONVERT;
5608                lkb->lkb_grmode = DLM_LOCK_IV;
5609                rsb_set_flag(r, RSB_RECOVER_CONVERT);
5610        }
5611
5612        return 0;
5613}
5614
5615/* This lkb may have been recovered in a previous aborted recovery so we need
5616   to check if the rsb already has an lkb with the given remote nodeid/lkid.
5617   If so we just send back a standard reply.  If not, we create a new lkb with
5618   the given values and send back our lkid.  We send back our lkid by sending
5619   back the rcom_lock struct we got but with the remid field filled in. */
5620
5621/* needs at least dlm_rcom + rcom_lock */
5622int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5623{
5624        struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5625        struct dlm_rsb *r;
5626        struct dlm_lkb *lkb;
5627        uint32_t remid = 0;
5628        int from_nodeid = rc->rc_header.h_nodeid;
5629        int error;
5630
5631        if (rl->rl_parent_lkid) {
5632                error = -EOPNOTSUPP;
5633                goto out;
5634        }
5635
5636        remid = le32_to_cpu(rl->rl_lkid);
5637
5638        /* In general we expect the rsb returned to be R_MASTER, but we don't
5639           have to require it.  Recovery of masters on one node can overlap
5640           recovery of locks on another node, so one node can send us MSTCPY
5641           locks before we've made ourselves master of this rsb.  We can still
5642           add new MSTCPY locks that we receive here without any harm; when
5643           we make ourselves master, dlm_recover_masters() won't touch the
5644           MSTCPY locks we've received early. */
5645
5646        error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5647                         from_nodeid, R_RECEIVE_RECOVER, &r);
5648        if (error)
5649                goto out;
5650
5651        lock_rsb(r);
5652
5653        if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5654                log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5655                          from_nodeid, remid);
5656                error = -EBADR;
5657                goto out_unlock;
5658        }
5659
5660        lkb = search_remid(r, from_nodeid, remid);
5661        if (lkb) {
5662                error = -EEXIST;
5663                goto out_remid;
5664        }
5665
5666        error = create_lkb(ls, &lkb);
5667        if (error)
5668                goto out_unlock;
5669
5670        error = receive_rcom_lock_args(ls, lkb, r, rc);
5671        if (error) {
5672                __put_lkb(ls, lkb);
5673                goto out_unlock;
5674        }
5675
5676        attach_lkb(r, lkb);
5677        add_lkb(r, lkb, rl->rl_status);
5678        error = 0;
5679        ls->ls_recover_locks_in++;
5680
5681        if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5682                rsb_set_flag(r, RSB_RECOVER_GRANT);
5683
5684 out_remid:
5685        /* this is the new value returned to the lock holder for
5686           saving in its process-copy lkb */
5687        rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5688
5689        lkb->lkb_recover_seq = ls->ls_recover_seq;
5690
5691 out_unlock:
5692        unlock_rsb(r);
5693        put_rsb(r);
5694 out:
5695        if (error && error != -EEXIST)
5696                log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5697                          from_nodeid, remid, error);
5698        rl->rl_result = cpu_to_le32(error);
5699        return error;
5700}
5701
5702/* needs at least dlm_rcom + rcom_lock */
5703int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5704{
5705        struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5706        struct dlm_rsb *r;
5707        struct dlm_lkb *lkb;
5708        uint32_t lkid, remid;
5709        int error, result;
5710
5711        lkid = le32_to_cpu(rl->rl_lkid);
5712        remid = le32_to_cpu(rl->rl_remid);
5713        result = le32_to_cpu(rl->rl_result);
5714
5715        error = find_lkb(ls, lkid, &lkb);
5716        if (error) {
5717                log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5718                          lkid, rc->rc_header.h_nodeid, remid, result);
5719                return error;
5720        }
5721
5722        r = lkb->lkb_resource;
5723        hold_rsb(r);
5724        lock_rsb(r);
5725
5726        if (!is_process_copy(lkb)) {
5727                log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5728                          lkid, rc->rc_header.h_nodeid, remid, result);
5729                dlm_dump_rsb(r);
5730                unlock_rsb(r);
5731                put_rsb(r);
5732                dlm_put_lkb(lkb);
5733                return -EINVAL;
5734        }
5735
5736        switch (result) {
5737        case -EBADR:
5738                /* There's a chance the new master received our lock before
5739                   dlm_recover_master_reply(), this wouldn't happen if we did
5740                   a barrier between recover_masters and recover_locks. */
5741
5742                log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5743                          lkid, rc->rc_header.h_nodeid, remid, result);
5744        
5745                dlm_send_rcom_lock(r, lkb);
5746                goto out;
5747        case -EEXIST:
5748        case 0:
5749                lkb->lkb_remid = remid;
5750                break;
5751        default:
5752                log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5753                          lkid, rc->rc_header.h_nodeid, remid, result);
5754        }
5755
5756        /* an ack for dlm_recover_locks() which waits for replies from
5757           all the locks it sends to new masters */
5758        dlm_recovered_lock(r);
5759 out:
5760        unlock_rsb(r);
5761        put_rsb(r);
5762        dlm_put_lkb(lkb);
5763
5764        return 0;
5765}
5766
5767int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5768                     int mode, uint32_t flags, void *name, unsigned int namelen,
5769                     unsigned long timeout_cs)
5770{
5771        struct dlm_lkb *lkb;
5772        struct dlm_args args;
5773        int error;
5774
5775        dlm_lock_recovery(ls);
5776
5777        error = create_lkb(ls, &lkb);
5778        if (error) {
5779                kfree(ua);
5780                goto out;
5781        }
5782
5783        if (flags & DLM_LKF_VALBLK) {
5784                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5785                if (!ua->lksb.sb_lvbptr) {
5786                        kfree(ua);
5787                        __put_lkb(ls, lkb);
5788                        error = -ENOMEM;
5789                        goto out;
5790                }
5791        }
5792
5793        /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5794           When DLM_IFL_USER is set, the dlm knows that this is a userspace
5795           lock and that lkb_astparam is the dlm_user_args structure. */
5796
5797        error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5798                              fake_astfn, ua, fake_bastfn, &args);
5799        lkb->lkb_flags |= DLM_IFL_USER;
5800
5801        if (error) {
5802                __put_lkb(ls, lkb);
5803                goto out;
5804        }
5805
5806        error = request_lock(ls, lkb, name, namelen, &args);
5807
5808        switch (error) {
5809        case 0:
5810                break;
5811        case -EINPROGRESS:
5812                error = 0;
5813                break;
5814        case -EAGAIN:
5815                error = 0;
5816                /* fall through */
5817        default:
5818                __put_lkb(ls, lkb);
5819                goto out;
5820        }
5821
5822        /* add this new lkb to the per-process list of locks */
5823        spin_lock(&ua->proc->locks_spin);
5824        hold_lkb(lkb);
5825        list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5826        spin_unlock(&ua->proc->locks_spin);
5827 out:
5828        dlm_unlock_recovery(ls);
5829        return error;
5830}
5831
5832int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5833                     int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5834                     unsigned long timeout_cs)
5835{
5836        struct dlm_lkb *lkb;
5837        struct dlm_args args;
5838        struct dlm_user_args *ua;
5839        int error;
5840
5841        dlm_lock_recovery(ls);
5842
5843        error = find_lkb(ls, lkid, &lkb);
5844        if (error)
5845                goto out;
5846
5847        /* user can change the params on its lock when it converts it, or
5848           add an lvb that didn't exist before */
5849
5850        ua = lkb->lkb_ua;
5851
5852        if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5853                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5854                if (!ua->lksb.sb_lvbptr) {
5855                        error = -ENOMEM;
5856                        goto out_put;
5857                }
5858        }
5859        if (lvb_in && ua->lksb.sb_lvbptr)
5860                memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5861
5862        ua->xid = ua_tmp->xid;
5863        ua->castparam = ua_tmp->castparam;
5864        ua->castaddr = ua_tmp->castaddr;
5865        ua->bastparam = ua_tmp->bastparam;
5866        ua->bastaddr = ua_tmp->bastaddr;
5867        ua->user_lksb = ua_tmp->user_lksb;
5868
5869        error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5870                              fake_astfn, ua, fake_bastfn, &args);
5871        if (error)
5872                goto out_put;
5873
5874        error = convert_lock(ls, lkb, &args);
5875
5876        if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5877                error = 0;
5878 out_put:
5879        dlm_put_lkb(lkb);
5880 out:
5881        dlm_unlock_recovery(ls);
5882        kfree(ua_tmp);
5883        return error;
5884}
5885
5886/*
5887 * The caller asks for an orphan lock on a given resource with a given mode.
5888 * If a matching lock exists, it's moved to the owner's list of locks and
5889 * the lkid is returned.
5890 */
5891
5892int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5893                     int mode, uint32_t flags, void *name, unsigned int namelen,
5894                     unsigned long timeout_cs, uint32_t *lkid)
5895{
5896        struct dlm_lkb *lkb;
5897        struct dlm_user_args *ua;
5898        int found_other_mode = 0;
5899        int found = 0;
5900        int rv = 0;
5901
5902        mutex_lock(&ls->ls_orphans_mutex);
5903        list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5904                if (lkb->lkb_resource->res_length != namelen)
5905                        continue;
5906                if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5907                        continue;
5908                if (lkb->lkb_grmode != mode) {
5909                        found_other_mode = 1;
5910                        continue;
5911                }
5912
5913                found = 1;
5914                list_del_init(&lkb->lkb_ownqueue);
5915                lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5916                *lkid = lkb->lkb_id;
5917                break;
5918        }
5919        mutex_unlock(&ls->ls_orphans_mutex);
5920
5921        if (!found && found_other_mode) {
5922                rv = -EAGAIN;
5923                goto out;
5924        }
5925
5926        if (!found) {
5927                rv = -ENOENT;
5928                goto out;
5929        }
5930
5931        lkb->lkb_exflags = flags;
5932        lkb->lkb_ownpid = (int) current->pid;
5933
5934        ua = lkb->lkb_ua;
5935
5936        ua->proc = ua_tmp->proc;
5937        ua->xid = ua_tmp->xid;
5938        ua->castparam = ua_tmp->castparam;
5939        ua->castaddr = ua_tmp->castaddr;
5940        ua->bastparam = ua_tmp->bastparam;
5941        ua->bastaddr = ua_tmp->bastaddr;
5942        ua->user_lksb = ua_tmp->user_lksb;
5943
5944        /*
5945         * The lkb reference from the ls_orphans list was not
5946         * removed above, and is now considered the reference
5947         * for the proc locks list.
5948         */
5949
5950        spin_lock(&ua->proc->locks_spin);
5951        list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5952        spin_unlock(&ua->proc->locks_spin);
5953 out:
5954        kfree(ua_tmp);
5955        return rv;
5956}
5957
5958int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5959                    uint32_t flags, uint32_t lkid, char *lvb_in)
5960{
5961        struct dlm_lkb *lkb;
5962        struct dlm_args args;
5963        struct dlm_user_args *ua;
5964        int error;
5965
5966        dlm_lock_recovery(ls);
5967
5968        error = find_lkb(ls, lkid, &lkb);
5969        if (error)
5970                goto out;
5971
5972        ua = lkb->lkb_ua;
5973
5974        if (lvb_in && ua->lksb.sb_lvbptr)
5975                memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5976        if (ua_tmp->castparam)
5977                ua->castparam = ua_tmp->castparam;
5978        ua->user_lksb = ua_tmp->user_lksb;
5979
5980        error = set_unlock_args(flags, ua, &args);
5981        if (error)
5982                goto out_put;
5983
5984        error = unlock_lock(ls, lkb, &args);
5985
5986        if (error == -DLM_EUNLOCK)
5987                error = 0;
5988        /* from validate_unlock_args() */
5989        if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
5990                error = 0;
5991        if (error)
5992                goto out_put;
5993
5994        spin_lock(&ua->proc->locks_spin);
5995        /* dlm_user_add_cb() may have already taken lkb off the proc list */
5996        if (!list_empty(&lkb->lkb_ownqueue))
5997                list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
5998        spin_unlock(&ua->proc->locks_spin);
5999 out_put:
6000        dlm_put_lkb(lkb);
6001 out:
6002        dlm_unlock_recovery(ls);
6003        kfree(ua_tmp);
6004        return error;
6005}
6006
6007int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6008                    uint32_t flags, uint32_t lkid)
6009{
6010        struct dlm_lkb *lkb;
6011        struct dlm_args args;
6012        struct dlm_user_args *ua;
6013        int error;
6014
6015        dlm_lock_recovery(ls);
6016
6017        error = find_lkb(ls, lkid, &lkb);
6018        if (error)
6019                goto out;
6020
6021        ua = lkb->lkb_ua;
6022        if (ua_tmp->castparam)
6023                ua->castparam = ua_tmp->castparam;
6024        ua->user_lksb = ua_tmp->user_lksb;
6025
6026        error = set_unlock_args(flags, ua, &args);
6027        if (error)
6028                goto out_put;
6029
6030        error = cancel_lock(ls, lkb, &args);
6031
6032        if (error == -DLM_ECANCEL)
6033                error = 0;
6034        /* from validate_unlock_args() */
6035        if (error == -EBUSY)
6036                error = 0;
6037 out_put:
6038        dlm_put_lkb(lkb);
6039 out:
6040        dlm_unlock_recovery(ls);
6041        kfree(ua_tmp);
6042        return error;
6043}
6044
6045int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6046{
6047        struct dlm_lkb *lkb;
6048        struct dlm_args args;
6049        struct dlm_user_args *ua;
6050        struct dlm_rsb *r;
6051        int error;
6052
6053        dlm_lock_recovery(ls);
6054
6055        error = find_lkb(ls, lkid, &lkb);
6056        if (error)
6057                goto out;
6058
6059        ua = lkb->lkb_ua;
6060
6061        error = set_unlock_args(flags, ua, &args);
6062        if (error)
6063                goto out_put;
6064
6065        /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6066
6067        r = lkb->lkb_resource;
6068        hold_rsb(r);
6069        lock_rsb(r);
6070
6071        error = validate_unlock_args(lkb, &args);
6072        if (error)
6073                goto out_r;
6074        lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6075
6076        error = _cancel_lock(r, lkb);
6077 out_r:
6078        unlock_rsb(r);
6079        put_rsb(r);
6080
6081        if (error == -DLM_ECANCEL)
6082                error = 0;
6083        /* from validate_unlock_args() */
6084        if (error == -EBUSY)
6085                error = 0;
6086 out_put:
6087        dlm_put_lkb(lkb);
6088 out:
6089        dlm_unlock_recovery(ls);
6090        return error;
6091}
6092
6093/* lkb's that are removed from the waiters list by revert are just left on the
6094   orphans list with the granted orphan locks, to be freed by purge */
6095
6096static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6097{
6098        struct dlm_args args;
6099        int error;
6100
6101        hold_lkb(lkb); /* reference for the ls_orphans list */
6102        mutex_lock(&ls->ls_orphans_mutex);
6103        list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6104        mutex_unlock(&ls->ls_orphans_mutex);
6105
6106        set_unlock_args(0, lkb->lkb_ua, &args);
6107
6108        error = cancel_lock(ls, lkb, &args);
6109        if (error == -DLM_ECANCEL)
6110                error = 0;
6111        return error;
6112}
6113
6114/* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6115   granted.  Regardless of what rsb queue the lock is on, it's removed and
6116   freed.  The IVVALBLK flag causes the lvb on the resource to be invalidated
6117   if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6118
6119static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6120{
6121        struct dlm_args args;
6122        int error;
6123
6124        set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6125                        lkb->lkb_ua, &args);
6126
6127        error = unlock_lock(ls, lkb, &args);
6128        if (error == -DLM_EUNLOCK)
6129                error = 0;
6130        return error;
6131}
6132
6133/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6134   (which does lock_rsb) due to deadlock with receiving a message that does
6135   lock_rsb followed by dlm_user_add_cb() */
6136
6137static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6138                                     struct dlm_user_proc *proc)
6139{
6140        struct dlm_lkb *lkb = NULL;
6141
6142        mutex_lock(&ls->ls_clear_proc_locks);
6143        if (list_empty(&proc->locks))
6144                goto out;
6145
6146        lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6147        list_del_init(&lkb->lkb_ownqueue);
6148
6149        if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6150                lkb->lkb_flags |= DLM_IFL_ORPHAN;
6151        else
6152                lkb->lkb_flags |= DLM_IFL_DEAD;
6153 out:
6154        mutex_unlock(&ls->ls_clear_proc_locks);
6155        return lkb;
6156}
6157
6158/* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6159   1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6160   which we clear here. */
6161
6162/* proc CLOSING flag is set so no more device_reads should look at proc->asts
6163   list, and no more device_writes should add lkb's to proc->locks list; so we
6164   shouldn't need to take asts_spin or locks_spin here.  this assumes that
6165   device reads/writes/closes are serialized -- FIXME: we may need to serialize
6166   them ourself. */
6167
6168void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6169{
6170        struct dlm_lkb *lkb, *safe;
6171
6172        dlm_lock_recovery(ls);
6173
6174        while (1) {
6175                lkb = del_proc_lock(ls, proc);
6176                if (!lkb)
6177                        break;
6178                del_timeout(lkb);
6179                if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6180                        orphan_proc_lock(ls, lkb);
6181                else
6182                        unlock_proc_lock(ls, lkb);
6183
6184                /* this removes the reference for the proc->locks list
6185                   added by dlm_user_request, it may result in the lkb
6186                   being freed */
6187
6188                dlm_put_lkb(lkb);
6189        }
6190
6191        mutex_lock(&ls->ls_clear_proc_locks);
6192
6193        /* in-progress unlocks */
6194        list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6195                list_del_init(&lkb->lkb_ownqueue);
6196                lkb->lkb_flags |= DLM_IFL_DEAD;
6197                dlm_put_lkb(lkb);
6198        }
6199
6200        list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6201                memset(&lkb->lkb_callbacks, 0,
6202                       sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6203                list_del_init(&lkb->lkb_cb_list);
6204                dlm_put_lkb(lkb);
6205        }
6206
6207        mutex_unlock(&ls->ls_clear_proc_locks);
6208        dlm_unlock_recovery(ls);
6209}
6210
6211static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6212{
6213        struct dlm_lkb *lkb, *safe;
6214
6215        while (1) {
6216                lkb = NULL;
6217                spin_lock(&proc->locks_spin);
6218                if (!list_empty(&proc->locks)) {
6219                        lkb = list_entry(proc->locks.next, struct dlm_lkb,
6220                                         lkb_ownqueue);
6221                        list_del_init(&lkb->lkb_ownqueue);
6222                }
6223                spin_unlock(&proc->locks_spin);
6224
6225                if (!lkb)
6226                        break;
6227
6228                lkb->lkb_flags |= DLM_IFL_DEAD;
6229                unlock_proc_lock(ls, lkb);
6230                dlm_put_lkb(lkb); /* ref from proc->locks list */
6231        }
6232
6233        spin_lock(&proc->locks_spin);
6234        list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6235                list_del_init(&lkb->lkb_ownqueue);
6236                lkb->lkb_flags |= DLM_IFL_DEAD;
6237                dlm_put_lkb(lkb);
6238        }
6239        spin_unlock(&proc->locks_spin);
6240
6241        spin_lock(&proc->asts_spin);
6242        list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6243                memset(&lkb->lkb_callbacks, 0,
6244                       sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6245                list_del_init(&lkb->lkb_cb_list);
6246                dlm_put_lkb(lkb);
6247        }
6248        spin_unlock(&proc->asts_spin);
6249}
6250
6251/* pid of 0 means purge all orphans */
6252
6253static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6254{
6255        struct dlm_lkb *lkb, *safe;
6256
6257        mutex_lock(&ls->ls_orphans_mutex);
6258        list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6259                if (pid && lkb->lkb_ownpid != pid)
6260                        continue;
6261                unlock_proc_lock(ls, lkb);
6262                list_del_init(&lkb->lkb_ownqueue);
6263                dlm_put_lkb(lkb);
6264        }
6265        mutex_unlock(&ls->ls_orphans_mutex);
6266}
6267
6268static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6269{
6270        struct dlm_message *ms;
6271        struct dlm_mhandle *mh;
6272        int error;
6273
6274        error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6275                                DLM_MSG_PURGE, &ms, &mh);
6276        if (error)
6277                return error;
6278        ms->m_nodeid = nodeid;
6279        ms->m_pid = pid;
6280
6281        return send_message(mh, ms);
6282}
6283
6284int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6285                   int nodeid, int pid)
6286{
6287        int error = 0;
6288
6289        if (nodeid && (nodeid != dlm_our_nodeid())) {
6290                error = send_purge(ls, nodeid, pid);
6291        } else {
6292                dlm_lock_recovery(ls);
6293                if (pid == current->pid)
6294                        purge_proc_locks(ls, proc);
6295                else
6296                        do_purge(ls, nodeid, pid);
6297                dlm_unlock_recovery(ls);
6298        }
6299        return error;
6300}
6301
6302