git.oblomov.eu Git - linux-2.6/blob - fs/dlm/lock.c

   1 /******************************************************************************
   2 *******************************************************************************
   3 **
   4 **  Copyright (C) 2005 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 "dlm_internal.h"
  60 #include <linux/dlm_device.h>
  61 #include "memory.h"
  62 #include "lowcomms.h"
  63 #include "requestqueue.h"
  64 #include "util.h"
  65 #include "dir.h"
  66 #include "member.h"
  67 #include "lockspace.h"
  68 #include "ast.h"
  69 #include "lock.h"
  70 #include "rcom.h"
  71 #include "recover.h"
  72 #include "lvb_table.h"
  73 #include "user.h"
  74 #include "config.h"
  75
  76 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
  77 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
  78 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  79 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
  80 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
  81 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
  82 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
  83 static int send_remove(struct dlm_rsb *r);
  84 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  85 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
  86                                     struct dlm_message *ms);
  87 static int receive_extralen(struct dlm_message *ms);
  88
  89 /*
  90  * Lock compatibilty matrix - thanks Steve
  91  * UN = Unlocked state. Not really a state, used as a flag
  92  * PD = Padding. Used to make the matrix a nice power of two in size
  93  * Other states are the same as the VMS DLM.
  94  * Usage: matrix[grmode+1][rqmode+1]  (although m[rq+1][gr+1] is the same)
  95  */
  96
  97 static const int __dlm_compat_matrix[8][8] = {
  98       /* UN NL CR CW PR PW EX PD */
  99         {1, 1, 1, 1, 1, 1, 1, 0},       /* UN */
 100         {1, 1, 1, 1, 1, 1, 1, 0},       /* NL */
 101         {1, 1, 1, 1, 1, 1, 0, 0},       /* CR */
 102         {1, 1, 1, 1, 0, 0, 0, 0},       /* CW */
 103         {1, 1, 1, 0, 1, 0, 0, 0},       /* PR */
 104         {1, 1, 1, 0, 0, 0, 0, 0},       /* PW */
 105         {1, 1, 0, 0, 0, 0, 0, 0},       /* EX */
 106         {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 107 };
 108
 109 /*
 110  * This defines the direction of transfer of LVB data.
 111  * Granted mode is the row; requested mode is the column.
 112  * Usage: matrix[grmode+1][rqmode+1]
 113  * 1 = LVB is returned to the caller
 114  * 0 = LVB is written to the resource
 115  * -1 = nothing happens to the LVB
 116  */
 117
 118 const int dlm_lvb_operations[8][8] = {
 119         /* UN   NL  CR  CW  PR  PW  EX  PD*/
 120         {  -1,  1,  1,  1,  1,  1,  1, -1 }, /* UN */
 121         {  -1,  1,  1,  1,  1,  1,  1,  0 }, /* NL */
 122         {  -1, -1,  1,  1,  1,  1,  1,  0 }, /* CR */
 123         {  -1, -1, -1,  1,  1,  1,  1,  0 }, /* CW */
 124         {  -1, -1, -1, -1,  1,  1,  1,  0 }, /* PR */
 125         {  -1,  0,  0,  0,  0,  0,  1,  0 }, /* PW */
 126         {  -1,  0,  0,  0,  0,  0,  0,  0 }, /* EX */
 127         {  -1,  0,  0,  0,  0,  0,  0,  0 }  /* PD */
 128 };
 129
 130 #define modes_compat(gr, rq) \
 131         __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
 132
 133 int dlm_modes_compat(int mode1, int mode2)
 134 {
 135         return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
 136 }
 137
 138 /*
 139  * Compatibility matrix for conversions with QUECVT set.
 140  * Granted mode is the row; requested mode is the column.
 141  * Usage: matrix[grmode+1][rqmode+1]
 142  */
 143
 144 static const int __quecvt_compat_matrix[8][8] = {
 145       /* UN NL CR CW PR PW EX PD */
 146         {0, 0, 0, 0, 0, 0, 0, 0},       /* UN */
 147         {0, 0, 1, 1, 1, 1, 1, 0},       /* NL */
 148         {0, 0, 0, 1, 1, 1, 1, 0},       /* CR */
 149         {0, 0, 0, 0, 1, 1, 1, 0},       /* CW */
 150         {0, 0, 0, 1, 0, 1, 1, 0},       /* PR */
 151         {0, 0, 0, 0, 0, 0, 1, 0},       /* PW */
 152         {0, 0, 0, 0, 0, 0, 0, 0},       /* EX */
 153         {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 154 };
 155
 156 void dlm_print_lkb(struct dlm_lkb *lkb)
 157 {
 158         printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
 159                "     status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
 160                lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
 161                lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
 162                lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
 163 }
 164
 165 void dlm_print_rsb(struct dlm_rsb *r)
 166 {
 167         printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
 168                r->res_nodeid, r->res_flags, r->res_first_lkid,
 169                r->res_recover_locks_count, r->res_name);
 170 }
 171
 172 void dlm_dump_rsb(struct dlm_rsb *r)
 173 {
 174         struct dlm_lkb *lkb;
 175
 176         dlm_print_rsb(r);
 177
 178         printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
 179                list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
 180         printk(KERN_ERR "rsb lookup list\n");
 181         list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
 182                 dlm_print_lkb(lkb);
 183         printk(KERN_ERR "rsb grant queue:\n");
 184         list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
 185                 dlm_print_lkb(lkb);
 186         printk(KERN_ERR "rsb convert queue:\n");
 187         list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
 188                 dlm_print_lkb(lkb);
 189         printk(KERN_ERR "rsb wait queue:\n");
 190         list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
 191                 dlm_print_lkb(lkb);
 192 }
 193
 194 /* Threads cannot use the lockspace while it's being recovered */
 195
 196 static inline void lock_recovery(struct dlm_ls *ls)
 197 {
 198         down_read(&ls->ls_in_recovery);
 199 }
 200
 201 static inline void unlock_recovery(struct dlm_ls *ls)
 202 {
 203         up_read(&ls->ls_in_recovery);
 204 }
 205
 206 static inline int lock_recovery_try(struct dlm_ls *ls)
 207 {
 208         return down_read_trylock(&ls->ls_in_recovery);
 209 }
 210
 211 static inline int can_be_queued(struct dlm_lkb *lkb)
 212 {
 213         return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
 214 }
 215
 216 static inline int force_blocking_asts(struct dlm_lkb *lkb)
 217 {
 218         return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
 219 }
 220
 221 static inline int is_demoted(struct dlm_lkb *lkb)
 222 {
 223         return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
 224 }
 225
 226 static inline int is_remote(struct dlm_rsb *r)
 227 {
 228         DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
 229         return !!r->res_nodeid;
 230 }
 231
 232 static inline int is_process_copy(struct dlm_lkb *lkb)
 233 {
 234         return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
 235 }
 236
 237 static inline int is_master_copy(struct dlm_lkb *lkb)
 238 {
 239         if (lkb->lkb_flags & DLM_IFL_MSTCPY)
 240                 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
 241         return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
 242 }
 243
 244 static inline int middle_conversion(struct dlm_lkb *lkb)
 245 {
 246         if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
 247             (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
 248                 return 1;
 249         return 0;
 250 }
 251
 252 static inline int down_conversion(struct dlm_lkb *lkb)
 253 {
 254         return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
 255 }
 256
 257 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
 258 {
 259         if (is_master_copy(lkb))
 260                 return;
 261
 262         DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
 263
 264         lkb->lkb_lksb->sb_status = rv;
 265         lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
 266
 267         dlm_add_ast(lkb, AST_COMP);
 268 }
 269
 270 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
 271 {
 272         if (is_master_copy(lkb))
 273                 send_bast(r, lkb, rqmode);
 274         else {
 275                 lkb->lkb_bastmode = rqmode;
 276                 dlm_add_ast(lkb, AST_BAST);
 277         }
 278 }
 279
 280 /*
 281  * Basic operations on rsb's and lkb's
 282  */
 283
 284 static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
 285 {
 286         struct dlm_rsb *r;
 287
 288         r = allocate_rsb(ls, len);
 289         if (!r)
 290                 return NULL;
 291
 292         r->res_ls = ls;
 293         r->res_length = len;
 294         memcpy(r->res_name, name, len);
 295         mutex_init(&r->res_mutex);
 296
 297         INIT_LIST_HEAD(&r->res_lookup);
 298         INIT_LIST_HEAD(&r->res_grantqueue);
 299         INIT_LIST_HEAD(&r->res_convertqueue);
 300         INIT_LIST_HEAD(&r->res_waitqueue);
 301         INIT_LIST_HEAD(&r->res_root_list);
 302         INIT_LIST_HEAD(&r->res_recover_list);
 303
 304         return r;
 305 }
 306
 307 static int search_rsb_list(struct list_head *head, char *name, int len,
 308                            unsigned int flags, struct dlm_rsb **r_ret)
 309 {
 310         struct dlm_rsb *r;
 311         int error = 0;
 312
 313         list_for_each_entry(r, head, res_hashchain) {
 314                 if (len == r->res_length && !memcmp(name, r->res_name, len))
 315                         goto found;
 316         }
 317         return -EBADR;
 318
 319  found:
 320         if (r->res_nodeid && (flags & R_MASTER))
 321                 error = -ENOTBLK;
 322         *r_ret = r;
 323         return error;
 324 }
 325
 326 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
 327                        unsigned int flags, struct dlm_rsb **r_ret)
 328 {
 329         struct dlm_rsb *r;
 330         int error;
 331
 332         error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
 333         if (!error) {
 334                 kref_get(&r->res_ref);
 335                 goto out;
 336         }
 337         error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
 338         if (error)
 339                 goto out;
 340
 341         list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
 342
 343         if (dlm_no_directory(ls))
 344                 goto out;
 345
 346         if (r->res_nodeid == -1) {
 347                 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
 348                 r->res_first_lkid = 0;
 349         } else if (r->res_nodeid > 0) {
 350                 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
 351                 r->res_first_lkid = 0;
 352         } else {
 353                 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
 354                 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
 355         }
 356  out:
 357         *r_ret = r;
 358         return error;
 359 }
 360
 361 static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
 362                       unsigned int flags, struct dlm_rsb **r_ret)
 363 {
 364         int error;
 365         write_lock(&ls->ls_rsbtbl[b].lock);
 366         error = _search_rsb(ls, name, len, b, flags, r_ret);
 367         write_unlock(&ls->ls_rsbtbl[b].lock);
 368         return error;
 369 }
 370
 371 /*
 372  * Find rsb in rsbtbl and potentially create/add one
 373  *
 374  * Delaying the release of rsb's has a similar benefit to applications keeping
 375  * NL locks on an rsb, but without the guarantee that the cached master value
 376  * will still be valid when the rsb is reused.  Apps aren't always smart enough
 377  * to keep NL locks on an rsb that they may lock again shortly; this can lead
 378  * to excessive master lookups and removals if we don't delay the release.
 379  *
 380  * Searching for an rsb means looking through both the normal list and toss
 381  * list.  When found on the toss list the rsb is moved to the normal list with
 382  * ref count of 1; when found on normal list the ref count is incremented.
 383  */
 384
 385 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
 386                     unsigned int flags, struct dlm_rsb **r_ret)
 387 {
 388         struct dlm_rsb *r, *tmp;
 389         uint32_t hash, bucket;
 390         int error = 0;
 391
 392         if (dlm_no_directory(ls))
 393                 flags |= R_CREATE;
 394
 395         hash = jhash(name, namelen, 0);
 396         bucket = hash & (ls->ls_rsbtbl_size - 1);
 397
 398         error = search_rsb(ls, name, namelen, bucket, flags, &r);
 399         if (!error)
 400                 goto out;
 401
 402         if (error == -EBADR && !(flags & R_CREATE))
 403                 goto out;
 404
 405         /* the rsb was found but wasn't a master copy */
 406         if (error == -ENOTBLK)
 407                 goto out;
 408
 409         error = -ENOMEM;
 410         r = create_rsb(ls, name, namelen);
 411         if (!r)
 412                 goto out;
 413
 414         r->res_hash = hash;
 415         r->res_bucket = bucket;
 416         r->res_nodeid = -1;
 417         kref_init(&r->res_ref);
 418
 419         /* With no directory, the master can be set immediately */
 420         if (dlm_no_directory(ls)) {
 421                 int nodeid = dlm_dir_nodeid(r);
 422                 if (nodeid == dlm_our_nodeid())
 423                         nodeid = 0;
 424                 r->res_nodeid = nodeid;
 425         }
 426
 427         write_lock(&ls->ls_rsbtbl[bucket].lock);
 428         error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
 429         if (!error) {
 430                 write_unlock(&ls->ls_rsbtbl[bucket].lock);
 431                 free_rsb(r);
 432                 r = tmp;
 433                 goto out;
 434         }
 435         list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
 436         write_unlock(&ls->ls_rsbtbl[bucket].lock);
 437         error = 0;
 438  out:
 439         *r_ret = r;
 440         return error;
 441 }
 442
 443 int dlm_find_rsb(struct dlm_ls *ls, char *name, int namelen,
 444                  unsigned int flags, struct dlm_rsb **r_ret)
 445 {
 446         return find_rsb(ls, name, namelen, flags, r_ret);
 447 }
 448
 449 /* This is only called to add a reference when the code already holds
 450    a valid reference to the rsb, so there's no need for locking. */
 451
 452 static inline void hold_rsb(struct dlm_rsb *r)
 453 {
 454         kref_get(&r->res_ref);
 455 }
 456
 457 void dlm_hold_rsb(struct dlm_rsb *r)
 458 {
 459         hold_rsb(r);
 460 }
 461
 462 static void toss_rsb(struct kref *kref)
 463 {
 464         struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
 465         struct dlm_ls *ls = r->res_ls;
 466
 467         DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
 468         kref_init(&r->res_ref);
 469         list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
 470         r->res_toss_time = jiffies;
 471         if (r->res_lvbptr) {
 472                 free_lvb(r->res_lvbptr);
 473                 r->res_lvbptr = NULL;
 474         }
 475 }
 476
 477 /* When all references to the rsb are gone it's transfered to
 478    the tossed list for later disposal. */
 479
 480 static void put_rsb(struct dlm_rsb *r)
 481 {
 482         struct dlm_ls *ls = r->res_ls;
 483         uint32_t bucket = r->res_bucket;
 484
 485         write_lock(&ls->ls_rsbtbl[bucket].lock);
 486         kref_put(&r->res_ref, toss_rsb);
 487         write_unlock(&ls->ls_rsbtbl[bucket].lock);
 488 }
 489
 490 void dlm_put_rsb(struct dlm_rsb *r)
 491 {
 492         put_rsb(r);
 493 }
 494
 495 /* See comment for unhold_lkb */
 496
 497 static void unhold_rsb(struct dlm_rsb *r)
 498 {
 499         int rv;
 500         rv = kref_put(&r->res_ref, toss_rsb);
 501         DLM_ASSERT(!rv, dlm_dump_rsb(r););
 502 }
 503
 504 static void kill_rsb(struct kref *kref)
 505 {
 506         struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
 507
 508         /* All work is done after the return from kref_put() so we
 509            can release the write_lock before the remove and free. */
 510
 511         DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
 512         DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
 513         DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
 514         DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
 515         DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
 516         DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
 517 }
 518
 519 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
 520    The rsb must exist as long as any lkb's for it do. */
 521
 522 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
 523 {
 524         hold_rsb(r);
 525         lkb->lkb_resource = r;
 526 }
 527
 528 static void detach_lkb(struct dlm_lkb *lkb)
 529 {
 530         if (lkb->lkb_resource) {
 531                 put_rsb(lkb->lkb_resource);
 532                 lkb->lkb_resource = NULL;
 533         }
 534 }
 535
 536 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
 537 {
 538         struct dlm_lkb *lkb, *tmp;
 539         uint32_t lkid = 0;
 540         uint16_t bucket;
 541
 542         lkb = allocate_lkb(ls);
 543         if (!lkb)
 544                 return -ENOMEM;
 545
 546         lkb->lkb_nodeid = -1;
 547         lkb->lkb_grmode = DLM_LOCK_IV;
 548         kref_init(&lkb->lkb_ref);
 549         INIT_LIST_HEAD(&lkb->lkb_ownqueue);
 550
 551         get_random_bytes(&bucket, sizeof(bucket));
 552         bucket &= (ls->ls_lkbtbl_size - 1);
 553
 554         write_lock(&ls->ls_lkbtbl[bucket].lock);
 555
 556         /* counter can roll over so we must verify lkid is not in use */
 557
 558         while (lkid == 0) {
 559                 lkid = bucket | (ls->ls_lkbtbl[bucket].counter++ << 16);
 560
 561                 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
 562                                     lkb_idtbl_list) {
 563                         if (tmp->lkb_id != lkid)
 564                                 continue;
 565                         lkid = 0;
 566                         break;
 567                 }
 568         }
 569
 570         lkb->lkb_id = lkid;
 571         list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
 572         write_unlock(&ls->ls_lkbtbl[bucket].lock);
 573
 574         *lkb_ret = lkb;
 575         return 0;
 576 }
 577
 578 static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
 579 {
 580         uint16_t bucket = lkid & 0xFFFF;
 581         struct dlm_lkb *lkb;
 582
 583         list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
 584                 if (lkb->lkb_id == lkid)
 585                         return lkb;
 586         }
 587         return NULL;
 588 }
 589
 590 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
 591 {
 592         struct dlm_lkb *lkb;
 593         uint16_t bucket = lkid & 0xFFFF;
 594
 595         if (bucket >= ls->ls_lkbtbl_size)
 596                 return -EBADSLT;
 597
 598         read_lock(&ls->ls_lkbtbl[bucket].lock);
 599         lkb = __find_lkb(ls, lkid);
 600         if (lkb)
 601                 kref_get(&lkb->lkb_ref);
 602         read_unlock(&ls->ls_lkbtbl[bucket].lock);
 603
 604         *lkb_ret = lkb;
 605         return lkb ? 0 : -ENOENT;
 606 }
 607
 608 static void kill_lkb(struct kref *kref)
 609 {
 610         struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
 611
 612         /* All work is done after the return from kref_put() so we
 613            can release the write_lock before the detach_lkb */
 614
 615         DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
 616 }
 617
 618 /* __put_lkb() is used when an lkb may not have an rsb attached to
 619    it so we need to provide the lockspace explicitly */
 620
 621 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
 622 {
 623         uint16_t bucket = lkb->lkb_id & 0xFFFF;
 624
 625         write_lock(&ls->ls_lkbtbl[bucket].lock);
 626         if (kref_put(&lkb->lkb_ref, kill_lkb)) {
 627                 list_del(&lkb->lkb_idtbl_list);
 628                 write_unlock(&ls->ls_lkbtbl[bucket].lock);
 629
 630                 detach_lkb(lkb);
 631
 632                 /* for local/process lkbs, lvbptr points to caller's lksb */
 633                 if (lkb->lkb_lvbptr && is_master_copy(lkb))
 634                         free_lvb(lkb->lkb_lvbptr);
 635                 free_lkb(lkb);
 636                 return 1;
 637         } else {
 638                 write_unlock(&ls->ls_lkbtbl[bucket].lock);
 639                 return 0;
 640         }
 641 }
 642
 643 int dlm_put_lkb(struct dlm_lkb *lkb)
 644 {
 645         struct dlm_ls *ls;
 646
 647         DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
 648         DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
 649
 650         ls = lkb->lkb_resource->res_ls;
 651         return __put_lkb(ls, lkb);
 652 }
 653
 654 /* This is only called to add a reference when the code already holds
 655    a valid reference to the lkb, so there's no need for locking. */
 656
 657 static inline void hold_lkb(struct dlm_lkb *lkb)
 658 {
 659         kref_get(&lkb->lkb_ref);
 660 }
 661
 662 /* This is called when we need to remove a reference and are certain
 663    it's not the last ref.  e.g. del_lkb is always called between a
 664    find_lkb/put_lkb and is always the inverse of a previous add_lkb.
 665    put_lkb would work fine, but would involve unnecessary locking */
 666
 667 static inline void unhold_lkb(struct dlm_lkb *lkb)
 668 {
 669         int rv;
 670         rv = kref_put(&lkb->lkb_ref, kill_lkb);
 671         DLM_ASSERT(!rv, dlm_print_lkb(lkb););
 672 }
 673
 674 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
 675                             int mode)
 676 {
 677         struct dlm_lkb *lkb = NULL;
 678
 679         list_for_each_entry(lkb, head, lkb_statequeue)
 680                 if (lkb->lkb_rqmode < mode)
 681                         break;
 682
 683         if (!lkb)
 684                 list_add_tail(new, head);
 685         else
 686                 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
 687 }
 688
 689 /* add/remove lkb to rsb's grant/convert/wait queue */
 690
 691 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
 692 {
 693         kref_get(&lkb->lkb_ref);
 694
 695         DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
 696
 697         lkb->lkb_status = status;
 698
 699         switch (status) {
 700         case DLM_LKSTS_WAITING:
 701                 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
 702                         list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
 703                 else
 704                         list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
 705                 break;
 706         case DLM_LKSTS_GRANTED:
 707                 /* convention says granted locks kept in order of grmode */
 708                 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
 709                                 lkb->lkb_grmode);
 710                 break;
 711         case DLM_LKSTS_CONVERT:
 712                 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
 713                         list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
 714                 else
 715                         list_add_tail(&lkb->lkb_statequeue,
 716                                       &r->res_convertqueue);
 717                 break;
 718         default:
 719                 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
 720         }
 721 }
 722
 723 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
 724 {
 725         lkb->lkb_status = 0;
 726         list_del(&lkb->lkb_statequeue);
 727         unhold_lkb(lkb);
 728 }
 729
 730 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
 731 {
 732         hold_lkb(lkb);
 733         del_lkb(r, lkb);
 734         add_lkb(r, lkb, sts);
 735         unhold_lkb(lkb);
 736 }
 737
 738 /* add/remove lkb from global waiters list of lkb's waiting for
 739    a reply from a remote node */
 740
 741 static void add_to_waiters(struct dlm_lkb *lkb, int mstype)
 742 {
 743         struct dlm_ls *ls = lkb->lkb_resource->res_ls;
 744
 745         mutex_lock(&ls->ls_waiters_mutex);
 746         if (lkb->lkb_wait_type) {
 747                 log_print("add_to_waiters error %d", lkb->lkb_wait_type);
 748                 goto out;
 749         }
 750         lkb->lkb_wait_type = mstype;
 751         kref_get(&lkb->lkb_ref);
 752         list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
 753  out:
 754         mutex_unlock(&ls->ls_waiters_mutex);
 755 }
 756
 757 /* We clear the RESEND flag because we might be taking an lkb off the waiters
 758    list as part of process_requestqueue (e.g. a lookup that has an optimized
 759    request reply on the requestqueue) between dlm_recover_waiters_pre() which
 760    set RESEND and dlm_recover_waiters_post() */
 761
 762 static int _remove_from_waiters(struct dlm_lkb *lkb)
 763 {
 764         int error = 0;
 765
 766         if (!lkb->lkb_wait_type) {
 767                 log_print("remove_from_waiters error");
 768                 error = -EINVAL;
 769                 goto out;
 770         }
 771         lkb->lkb_wait_type = 0;
 772         lkb->lkb_flags &= ~DLM_IFL_RESEND;
 773         list_del(&lkb->lkb_wait_reply);
 774         unhold_lkb(lkb);
 775  out:
 776         return error;
 777 }
 778
 779 static int remove_from_waiters(struct dlm_lkb *lkb)
 780 {
 781         struct dlm_ls *ls = lkb->lkb_resource->res_ls;
 782         int error;
 783
 784         mutex_lock(&ls->ls_waiters_mutex);
 785         error = _remove_from_waiters(lkb);
 786         mutex_unlock(&ls->ls_waiters_mutex);
 787         return error;
 788 }
 789
 790 static void dir_remove(struct dlm_rsb *r)
 791 {
 792         int to_nodeid;
 793
 794         if (dlm_no_directory(r->res_ls))
 795                 return;
 796
 797         to_nodeid = dlm_dir_nodeid(r);
 798         if (to_nodeid != dlm_our_nodeid())
 799                 send_remove(r);
 800         else
 801                 dlm_dir_remove_entry(r->res_ls, to_nodeid,
 802                                      r->res_name, r->res_length);
 803 }
 804
 805 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
 806    found since they are in order of newest to oldest? */
 807
 808 static int shrink_bucket(struct dlm_ls *ls, int b)
 809 {
 810         struct dlm_rsb *r;
 811         int count = 0, found;
 812
 813         for (;;) {
 814                 found = 0;
 815                 write_lock(&ls->ls_rsbtbl[b].lock);
 816                 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
 817                                             res_hashchain) {
 818                         if (!time_after_eq(jiffies, r->res_toss_time +
 819                                            dlm_config.ci_toss_secs * HZ))
 820                                 continue;
 821                         found = 1;
 822                         break;
 823                 }
 824
 825                 if (!found) {
 826                         write_unlock(&ls->ls_rsbtbl[b].lock);
 827                         break;
 828                 }
 829
 830                 if (kref_put(&r->res_ref, kill_rsb)) {
 831                         list_del(&r->res_hashchain);
 832                         write_unlock(&ls->ls_rsbtbl[b].lock);
 833
 834                         if (is_master(r))
 835                                 dir_remove(r);
 836                         free_rsb(r);
 837                         count++;
 838                 } else {
 839                         write_unlock(&ls->ls_rsbtbl[b].lock);
 840                         log_error(ls, "tossed rsb in use %s", r->res_name);
 841                 }
 842         }
 843
 844         return count;
 845 }
 846
 847 void dlm_scan_rsbs(struct dlm_ls *ls)
 848 {
 849         int i;
 850
 851         if (dlm_locking_stopped(ls))
 852                 return;
 853
 854         for (i = 0; i < ls->ls_rsbtbl_size; i++) {
 855                 shrink_bucket(ls, i);
 856                 cond_resched();
 857         }
 858 }
 859
 860 /* lkb is master or local copy */
 861
 862 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 863 {
 864         int b, len = r->res_ls->ls_lvblen;
 865
 866         /* b=1 lvb returned to caller
 867            b=0 lvb written to rsb or invalidated
 868            b=-1 do nothing */
 869
 870         b =  dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
 871
 872         if (b == 1) {
 873                 if (!lkb->lkb_lvbptr)
 874                         return;
 875
 876                 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
 877                         return;
 878
 879                 if (!r->res_lvbptr)
 880                         return;
 881
 882                 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
 883                 lkb->lkb_lvbseq = r->res_lvbseq;
 884
 885         } else if (b == 0) {
 886                 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
 887                         rsb_set_flag(r, RSB_VALNOTVALID);
 888                         return;
 889                 }
 890
 891                 if (!lkb->lkb_lvbptr)
 892                         return;
 893
 894                 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
 895                         return;
 896
 897                 if (!r->res_lvbptr)
 898                         r->res_lvbptr = allocate_lvb(r->res_ls);
 899
 900                 if (!r->res_lvbptr)
 901                         return;
 902
 903                 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
 904                 r->res_lvbseq++;
 905                 lkb->lkb_lvbseq = r->res_lvbseq;
 906                 rsb_clear_flag(r, RSB_VALNOTVALID);
 907         }
 908
 909         if (rsb_flag(r, RSB_VALNOTVALID))
 910                 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
 911 }
 912
 913 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 914 {
 915         if (lkb->lkb_grmode < DLM_LOCK_PW)
 916                 return;
 917
 918         if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
 919                 rsb_set_flag(r, RSB_VALNOTVALID);
 920                 return;
 921         }
 922
 923         if (!lkb->lkb_lvbptr)
 924                 return;
 925
 926         if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
 927                 return;
 928
 929         if (!r->res_lvbptr)
 930                 r->res_lvbptr = allocate_lvb(r->res_ls);
 931
 932         if (!r->res_lvbptr)
 933                 return;
 934
 935         memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
 936         r->res_lvbseq++;
 937         rsb_clear_flag(r, RSB_VALNOTVALID);
 938 }
 939
 940 /* lkb is process copy (pc) */
 941
 942 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
 943                             struct dlm_message *ms)
 944 {
 945         int b;
 946
 947         if (!lkb->lkb_lvbptr)
 948                 return;
 949
 950         if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
 951                 return;
 952
 953         b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
 954         if (b == 1) {
 955                 int len = receive_extralen(ms);
 956                 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
 957                 lkb->lkb_lvbseq = ms->m_lvbseq;
 958         }
 959 }
 960
 961 /* Manipulate lkb's on rsb's convert/granted/waiting queues
 962    remove_lock -- used for unlock, removes lkb from granted
 963    revert_lock -- used for cancel, moves lkb from convert to granted
 964    grant_lock  -- used for request and convert, adds lkb to granted or
 965                   moves lkb from convert or waiting to granted
 966
 967    Each of these is used for master or local copy lkb's.  There is
 968    also a _pc() variation used to make the corresponding change on
 969    a process copy (pc) lkb. */
 970
 971 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 972 {
 973         del_lkb(r, lkb);
 974         lkb->lkb_grmode = DLM_LOCK_IV;
 975         /* this unhold undoes the original ref from create_lkb()
 976            so this leads to the lkb being freed */
 977         unhold_lkb(lkb);
 978 }
 979
 980 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 981 {
 982         set_lvb_unlock(r, lkb);
 983         _remove_lock(r, lkb);
 984 }
 985
 986 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
 987 {
 988         _remove_lock(r, lkb);
 989 }
 990
 991 static void revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 992 {
 993         lkb->lkb_rqmode = DLM_LOCK_IV;
 994
 995         switch (lkb->lkb_status) {
 996         case DLM_LKSTS_GRANTED:
 997                 break;
 998         case DLM_LKSTS_CONVERT:
 999                 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1000                 break;
1001         case DLM_LKSTS_WAITING:
1002                 del_lkb(r, lkb);
1003                 lkb->lkb_grmode = DLM_LOCK_IV;
1004                 /* this unhold undoes the original ref from create_lkb()
1005                    so this leads to the lkb being freed */
1006                 unhold_lkb(lkb);
1007                 break;
1008         default:
1009                 log_print("invalid status for revert %d", lkb->lkb_status);
1010         }
1011 }
1012
1013 static void revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1014 {
1015         revert_lock(r, lkb);
1016 }
1017
1018 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1019 {
1020         if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1021                 lkb->lkb_grmode = lkb->lkb_rqmode;
1022                 if (lkb->lkb_status)
1023                         move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1024                 else
1025                         add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1026         }
1027
1028         lkb->lkb_rqmode = DLM_LOCK_IV;
1029 }
1030
1031 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1032 {
1033         set_lvb_lock(r, lkb);
1034         _grant_lock(r, lkb);
1035         lkb->lkb_highbast = 0;
1036 }
1037
1038 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1039                           struct dlm_message *ms)
1040 {
1041         set_lvb_lock_pc(r, lkb, ms);
1042         _grant_lock(r, lkb);
1043 }
1044
1045 /* called by grant_pending_locks() which means an async grant message must
1046    be sent to the requesting node in addition to granting the lock if the
1047    lkb belongs to a remote node. */
1048
1049 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1050 {
1051         grant_lock(r, lkb);
1052         if (is_master_copy(lkb))
1053                 send_grant(r, lkb);
1054         else
1055                 queue_cast(r, lkb, 0);
1056 }
1057
1058 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1059 {
1060         struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1061                                            lkb_statequeue);
1062         if (lkb->lkb_id == first->lkb_id)
1063                 return 1;
1064
1065         return 0;
1066 }
1067
1068 /* Check if the given lkb conflicts with another lkb on the queue. */
1069
1070 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1071 {
1072         struct dlm_lkb *this;
1073
1074         list_for_each_entry(this, head, lkb_statequeue) {
1075                 if (this == lkb)
1076                         continue;
1077                 if (!modes_compat(this, lkb))
1078                         return 1;
1079         }
1080         return 0;
1081 }
1082
1083 /*
1084  * "A conversion deadlock arises with a pair of lock requests in the converting
1085  * queue for one resource.  The granted mode of each lock blocks the requested
1086  * mode of the other lock."
1087  *
1088  * Part 2: if the granted mode of lkb is preventing the first lkb in the
1089  * convert queue from being granted, then demote lkb (set grmode to NL).
1090  * This second form requires that we check for conv-deadlk even when
1091  * now == 0 in _can_be_granted().
1092  *
1093  * Example:
1094  * Granted Queue: empty
1095  * Convert Queue: NL->EX (first lock)
1096  *                PR->EX (second lock)
1097  *
1098  * The first lock can't be granted because of the granted mode of the second
1099  * lock and the second lock can't be granted because it's not first in the
1100  * list.  We demote the granted mode of the second lock (the lkb passed to this
1101  * function).
1102  *
1103  * After the resolution, the "grant pending" function needs to go back and try
1104  * to grant locks on the convert queue again since the first lock can now be
1105  * granted.
1106  */
1107
1108 static int conversion_deadlock_detect(struct dlm_rsb *rsb, struct dlm_lkb *lkb)
1109 {
1110         struct dlm_lkb *this, *first = NULL, *self = NULL;
1111
1112         list_for_each_entry(this, &rsb->res_convertqueue, lkb_statequeue) {
1113                 if (!first)
1114                         first = this;
1115                 if (this == lkb) {
1116                         self = lkb;
1117                         continue;
1118                 }
1119
1120                 if (!modes_compat(this, lkb) && !modes_compat(lkb, this))
1121                         return 1;
1122         }
1123
1124         /* if lkb is on the convert queue and is preventing the first
1125            from being granted, then there's deadlock and we demote lkb.
1126            multiple converting locks may need to do this before the first
1127            converting lock can be granted. */
1128
1129         if (self && self != first) {
1130                 if (!modes_compat(lkb, first) &&
1131                     !queue_conflict(&rsb->res_grantqueue, first))
1132                         return 1;
1133         }
1134
1135         return 0;
1136 }
1137
1138 /*
1139  * Return 1 if the lock can be granted, 0 otherwise.
1140  * Also detect and resolve conversion deadlocks.
1141  *
1142  * lkb is the lock to be granted
1143  *
1144  * now is 1 if the function is being called in the context of the
1145  * immediate request, it is 0 if called later, after the lock has been
1146  * queued.
1147  *
1148  * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1149  */
1150
1151 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1152 {
1153         int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1154
1155         /*
1156          * 6-10: Version 5.4 introduced an option to address the phenomenon of
1157          * a new request for a NL mode lock being blocked.
1158          *
1159          * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1160          * request, then it would be granted.  In essence, the use of this flag
1161          * tells the Lock Manager to expedite theis request by not considering
1162          * what may be in the CONVERTING or WAITING queues...  As of this
1163          * writing, the EXPEDITE flag can be used only with new requests for NL
1164          * mode locks.  This flag is not valid for conversion requests.
1165          *
1166          * A shortcut.  Earlier checks return an error if EXPEDITE is used in a
1167          * conversion or used with a non-NL requested mode.  We also know an
1168          * EXPEDITE request is always granted immediately, so now must always
1169          * be 1.  The full condition to grant an expedite request: (now &&
1170          * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1171          * therefore be shortened to just checking the flag.
1172          */
1173
1174         if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1175                 return 1;
1176
1177         /*
1178          * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1179          * added to the remaining conditions.
1180          */
1181
1182         if (queue_conflict(&r->res_grantqueue, lkb))
1183                 goto out;
1184
1185         /*
1186          * 6-3: By default, a conversion request is immediately granted if the
1187          * requested mode is compatible with the modes of all other granted
1188          * locks
1189          */
1190
1191         if (queue_conflict(&r->res_convertqueue, lkb))
1192                 goto out;
1193
1194         /*
1195          * 6-5: But the default algorithm for deciding whether to grant or
1196          * queue conversion requests does not by itself guarantee that such
1197          * requests are serviced on a "first come first serve" basis.  This, in
1198          * turn, can lead to a phenomenon known as "indefinate postponement".
1199          *
1200          * 6-7: This issue is dealt with by using the optional QUECVT flag with
1201          * the system service employed to request a lock conversion.  This flag
1202          * forces certain conversion requests to be queued, even if they are
1203          * compatible with the granted modes of other locks on the same
1204          * resource.  Thus, the use of this flag results in conversion requests
1205          * being ordered on a "first come first servce" basis.
1206          *
1207          * DCT: This condition is all about new conversions being able to occur
1208          * "in place" while the lock remains on the granted queue (assuming
1209          * nothing else conflicts.)  IOW if QUECVT isn't set, a conversion
1210          * doesn't _have_ to go onto the convert queue where it's processed in
1211          * order.  The "now" variable is necessary to distinguish converts
1212          * being received and processed for the first time now, because once a
1213          * convert is moved to the conversion queue the condition below applies
1214          * requiring fifo granting.
1215          */
1216
1217         if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1218                 return 1;
1219
1220         /*
1221          * The NOORDER flag is set to avoid the standard vms rules on grant
1222          * order.
1223          */
1224
1225         if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1226                 return 1;
1227
1228         /*
1229          * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1230          * granted until all other conversion requests ahead of it are granted
1231          * and/or canceled.
1232          */
1233
1234         if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1235                 return 1;
1236
1237         /*
1238          * 6-4: By default, a new request is immediately granted only if all
1239          * three of the following conditions are satisfied when the request is
1240          * issued:
1241          * - The queue of ungranted conversion requests for the resource is
1242          *   empty.
1243          * - The queue of ungranted new requests for the resource is empty.
1244          * - The mode of the new request is compatible with the most
1245          *   restrictive mode of all granted locks on the resource.
1246          */
1247
1248         if (now && !conv && list_empty(&r->res_convertqueue) &&
1249             list_empty(&r->res_waitqueue))
1250                 return 1;
1251
1252         /*
1253          * 6-4: Once a lock request is in the queue of ungranted new requests,
1254          * it cannot be granted until the queue of ungranted conversion
1255          * requests is empty, all ungranted new requests ahead of it are
1256          * granted and/or canceled, and it is compatible with the granted mode
1257          * of the most restrictive lock granted on the resource.
1258          */
1259
1260         if (!now && !conv && list_empty(&r->res_convertqueue) &&
1261             first_in_list(lkb, &r->res_waitqueue))
1262                 return 1;
1263
1264  out:
1265         /*
1266          * The following, enabled by CONVDEADLK, departs from VMS.
1267          */
1268
1269         if (conv && (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) &&
1270             conversion_deadlock_detect(r, lkb)) {
1271                 lkb->lkb_grmode = DLM_LOCK_NL;
1272                 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1273         }
1274
1275         return 0;
1276 }
1277
1278 /*
1279  * The ALTPR and ALTCW flags aren't traditional lock manager flags, but are a
1280  * simple way to provide a big optimization to applications that can use them.
1281  */
1282
1283 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1284 {
1285         uint32_t flags = lkb->lkb_exflags;
1286         int rv;
1287         int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1288
1289         rv = _can_be_granted(r, lkb, now);
1290         if (rv)
1291                 goto out;
1292
1293         if (lkb->lkb_sbflags & DLM_SBF_DEMOTED)
1294                 goto out;
1295
1296         if (rqmode != DLM_LOCK_PR && flags & DLM_LKF_ALTPR)
1297                 alt = DLM_LOCK_PR;
1298         else if (rqmode != DLM_LOCK_CW && flags & DLM_LKF_ALTCW)
1299                 alt = DLM_LOCK_CW;
1300
1301         if (alt) {
1302                 lkb->lkb_rqmode = alt;
1303                 rv = _can_be_granted(r, lkb, now);
1304                 if (rv)
1305                         lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1306                 else
1307                         lkb->lkb_rqmode = rqmode;
1308         }
1309  out:
1310         return rv;
1311 }
1312
1313 static int grant_pending_convert(struct dlm_rsb *r, int high)
1314 {
1315         struct dlm_lkb *lkb, *s;
1316         int hi, demoted, quit, grant_restart, demote_restart;
1317
1318         quit = 0;
1319  restart:
1320         grant_restart = 0;
1321         demote_restart = 0;
1322         hi = DLM_LOCK_IV;
1323
1324         list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1325                 demoted = is_demoted(lkb);
1326                 if (can_be_granted(r, lkb, 0)) {
1327                         grant_lock_pending(r, lkb);
1328                         grant_restart = 1;
1329                 } else {
1330                         hi = max_t(int, lkb->lkb_rqmode, hi);
1331                         if (!demoted && is_demoted(lkb))
1332                                 demote_restart = 1;
1333                 }
1334         }
1335
1336         if (grant_restart)
1337                 goto restart;
1338         if (demote_restart && !quit) {
1339                 quit = 1;
1340                 goto restart;
1341         }
1342
1343         return max_t(int, high, hi);
1344 }
1345
1346 static int grant_pending_wait(struct dlm_rsb *r, int high)
1347 {
1348         struct dlm_lkb *lkb, *s;
1349
1350         list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1351                 if (can_be_granted(r, lkb, 0))
1352                         grant_lock_pending(r, lkb);
1353                 else
1354                         high = max_t(int, lkb->lkb_rqmode, high);
1355         }
1356
1357         return high;
1358 }
1359
1360 static void grant_pending_locks(struct dlm_rsb *r)
1361 {
1362         struct dlm_lkb *lkb, *s;
1363         int high = DLM_LOCK_IV;
1364
1365         DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1366
1367         high = grant_pending_convert(r, high);
1368         high = grant_pending_wait(r, high);
1369
1370         if (high == DLM_LOCK_IV)
1371                 return;
1372
1373         /*
1374          * If there are locks left on the wait/convert queue then send blocking
1375          * ASTs to granted locks based on the largest requested mode (high)
1376          * found above. FIXME: highbast < high comparison not valid for PR/CW.
1377          */
1378
1379         list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1380                 if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
1381                     !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
1382                         queue_bast(r, lkb, high);
1383                         lkb->lkb_highbast = high;
1384                 }
1385         }
1386 }
1387
1388 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1389                             struct dlm_lkb *lkb)
1390 {
1391         struct dlm_lkb *gr;
1392
1393         list_for_each_entry(gr, head, lkb_statequeue) {
1394                 if (gr->lkb_bastaddr &&
1395                     gr->lkb_highbast < lkb->lkb_rqmode &&
1396                     !modes_compat(gr, lkb)) {
1397                         queue_bast(r, gr, lkb->lkb_rqmode);
1398                         gr->lkb_highbast = lkb->lkb_rqmode;
1399                 }
1400         }
1401 }
1402
1403 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1404 {
1405         send_bast_queue(r, &r->res_grantqueue, lkb);
1406 }
1407
1408 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1409 {
1410         send_bast_queue(r, &r->res_grantqueue, lkb);
1411         send_bast_queue(r, &r->res_convertqueue, lkb);
1412 }
1413
1414 /* set_master(r, lkb) -- set the master nodeid of a resource
1415
1416    The purpose of this function is to set the nodeid field in the given
1417    lkb using the nodeid field in the given rsb.  If the rsb's nodeid is
1418    known, it can just be copied to the lkb and the function will return
1419    0.  If the rsb's nodeid is _not_ known, it needs to be looked up
1420    before it can be copied to the lkb.
1421
1422    When the rsb nodeid is being looked up remotely, the initial lkb
1423    causing the lookup is kept on the ls_waiters list waiting for the
1424    lookup reply.  Other lkb's waiting for the same rsb lookup are kept
1425    on the rsb's res_lookup list until the master is verified.
1426
1427    Return values:
1428    0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1429    1: the rsb master is not available and the lkb has been placed on
1430       a wait queue
1431 */
1432
1433 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1434 {
1435         struct dlm_ls *ls = r->res_ls;
1436         int error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1437
1438         if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1439                 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1440                 r->res_first_lkid = lkb->lkb_id;
1441                 lkb->lkb_nodeid = r->res_nodeid;
1442                 return 0;
1443         }
1444
1445         if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1446                 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1447                 return 1;
1448         }
1449
1450         if (r->res_nodeid == 0) {
1451                 lkb->lkb_nodeid = 0;
1452                 return 0;
1453         }
1454
1455         if (r->res_nodeid > 0) {
1456                 lkb->lkb_nodeid = r->res_nodeid;
1457                 return 0;
1458         }
1459
1460         DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
1461
1462         dir_nodeid = dlm_dir_nodeid(r);
1463
1464         if (dir_nodeid != our_nodeid) {
1465                 r->res_first_lkid = lkb->lkb_id;
1466                 send_lookup(r, lkb);
1467                 return 1;
1468         }
1469
1470         for (;;) {
1471                 /* It's possible for dlm_scand to remove an old rsb for
1472                    this same resource from the toss list, us to create
1473                    a new one, look up the master locally, and find it
1474                    already exists just before dlm_scand does the
1475                    dir_remove() on the previous rsb. */
1476
1477                 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1478                                        r->res_length, &ret_nodeid);
1479                 if (!error)
1480                         break;
1481                 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1482                 schedule();
1483         }
1484
1485         if (ret_nodeid == our_nodeid) {
1486                 r->res_first_lkid = 0;
1487                 r->res_nodeid = 0;
1488                 lkb->lkb_nodeid = 0;
1489         } else {
1490                 r->res_first_lkid = lkb->lkb_id;
1491                 r->res_nodeid = ret_nodeid;
1492                 lkb->lkb_nodeid = ret_nodeid;
1493         }
1494         return 0;
1495 }
1496
1497 static void process_lookup_list(struct dlm_rsb *r)
1498 {
1499         struct dlm_lkb *lkb, *safe;
1500
1501         list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1502                 list_del(&lkb->lkb_rsb_lookup);
1503                 _request_lock(r, lkb);
1504                 schedule();
1505         }
1506 }
1507
1508 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1509
1510 static void confirm_master(struct dlm_rsb *r, int error)
1511 {
1512         struct dlm_lkb *lkb;
1513
1514         if (!r->res_first_lkid)
1515                 return;
1516
1517         switch (error) {
1518         case 0:
1519         case -EINPROGRESS:
1520                 r->res_first_lkid = 0;
1521                 process_lookup_list(r);
1522                 break;
1523
1524         case -EAGAIN:
1525                 /* the remote master didn't queue our NOQUEUE request;
1526                    make a waiting lkb the first_lkid */
1527
1528                 r->res_first_lkid = 0;
1529
1530                 if (!list_empty(&r->res_lookup)) {
1531                         lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1532                                          lkb_rsb_lookup);
1533                         list_del(&lkb->lkb_rsb_lookup);
1534                         r->res_first_lkid = lkb->lkb_id;
1535                         _request_lock(r, lkb);
1536                 } else
1537                         r->res_nodeid = -1;
1538                 break;
1539
1540         default:
1541                 log_error(r->res_ls, "confirm_master unknown error %d", error);
1542         }
1543 }
1544
1545 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1546                          int namelen, uint32_t parent_lkid, void *ast,
1547                          void *astarg, void *bast, struct dlm_args *args)
1548 {
1549         int rv = -EINVAL;
1550
1551         /* check for invalid arg usage */
1552
1553         if (mode < 0 || mode > DLM_LOCK_EX)
1554                 goto out;
1555
1556         if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1557                 goto out;
1558
1559         if (flags & DLM_LKF_CANCEL)
1560                 goto out;
1561
1562         if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1563                 goto out;
1564
1565         if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1566                 goto out;
1567
1568         if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1569                 goto out;
1570
1571         if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1572                 goto out;
1573
1574         if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1575                 goto out;
1576
1577         if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
1578                 goto out;
1579
1580         if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
1581                 goto out;
1582
1583         if (!ast || !lksb)
1584                 goto out;
1585
1586         if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
1587                 goto out;
1588
1589         /* parent/child locks not yet supported */
1590         if (parent_lkid)
1591                 goto out;
1592
1593         if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
1594                 goto out;
1595
1596         /* these args will be copied to the lkb in validate_lock_args,
1597            it cannot be done now because when converting locks, fields in
1598            an active lkb cannot be modified before locking the rsb */
1599
1600         args->flags = flags;
1601         args->astaddr = ast;
1602         args->astparam = (long) astarg;
1603         args->bastaddr = bast;
1604         args->mode = mode;
1605         args->lksb = lksb;
1606         rv = 0;
1607  out:
1608         return rv;
1609 }
1610
1611 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
1612 {
1613         if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
1614                       DLM_LKF_FORCEUNLOCK))
1615                 return -EINVAL;
1616
1617         args->flags = flags;
1618         args->astparam = (long) astarg;
1619         return 0;
1620 }
1621
1622 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
1623                               struct dlm_args *args)
1624 {
1625         int rv = -EINVAL;
1626
1627         if (args->flags & DLM_LKF_CONVERT) {
1628                 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1629                         goto out;
1630
1631                 if (args->flags & DLM_LKF_QUECVT &&
1632                     !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
1633                         goto out;
1634
1635                 rv = -EBUSY;
1636                 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
1637                         goto out;
1638
1639                 if (lkb->lkb_wait_type)
1640                         goto out;
1641         }
1642
1643         lkb->lkb_exflags = args->flags;
1644         lkb->lkb_sbflags = 0;
1645         lkb->lkb_astaddr = args->astaddr;
1646         lkb->lkb_astparam = args->astparam;
1647         lkb->lkb_bastaddr = args->bastaddr;
1648         lkb->lkb_rqmode = args->mode;
1649         lkb->lkb_lksb = args->lksb;
1650         lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
1651         lkb->lkb_ownpid = (int) current->pid;
1652         rv = 0;
1653  out:
1654         return rv;
1655 }
1656
1657 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
1658 {
1659         int rv = -EINVAL;
1660
1661         if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1662                 goto out;
1663
1664         if (args->flags & DLM_LKF_FORCEUNLOCK)
1665                 goto out_ok;
1666
1667         if (args->flags & DLM_LKF_CANCEL &&
1668             lkb->lkb_status == DLM_LKSTS_GRANTED)
1669                 goto out;
1670
1671         if (!(args->flags & DLM_LKF_CANCEL) &&
1672             lkb->lkb_status != DLM_LKSTS_GRANTED)
1673                 goto out;
1674
1675         rv = -EBUSY;
1676         if (lkb->lkb_wait_type)
1677                 goto out;
1678
1679  out_ok:
1680         lkb->lkb_exflags = args->flags;
1681         lkb->lkb_sbflags = 0;
1682         lkb->lkb_astparam = args->astparam;
1683
1684         rv = 0;
1685  out:
1686         return rv;
1687 }
1688
1689 /*
1690  * Four stage 4 varieties:
1691  * do_request(), do_convert(), do_unlock(), do_cancel()
1692  * These are called on the master node for the given lock and
1693  * from the central locking logic.
1694  */
1695
1696 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
1697 {
1698         int error = 0;
1699
1700         if (can_be_granted(r, lkb, 1)) {
1701                 grant_lock(r, lkb);
1702                 queue_cast(r, lkb, 0);
1703                 goto out;
1704         }
1705
1706         if (can_be_queued(lkb)) {
1707                 error = -EINPROGRESS;
1708                 add_lkb(r, lkb, DLM_LKSTS_WAITING);
1709                 send_blocking_asts(r, lkb);
1710                 goto out;
1711         }
1712
1713         error = -EAGAIN;
1714         if (force_blocking_asts(lkb))
1715                 send_blocking_asts_all(r, lkb);
1716         queue_cast(r, lkb, -EAGAIN);
1717
1718  out:
1719         return error;
1720 }
1721
1722 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
1723 {
1724         int error = 0;
1725
1726         /* changing an existing lock may allow others to be granted */
1727
1728         if (can_be_granted(r, lkb, 1)) {
1729                 grant_lock(r, lkb);
1730                 queue_cast(r, lkb, 0);
1731                 grant_pending_locks(r);
1732                 goto out;
1733         }
1734
1735         if (can_be_queued(lkb)) {
1736                 if (is_demoted(lkb))
1737                         grant_pending_locks(r);
1738                 error = -EINPROGRESS;
1739                 del_lkb(r, lkb);
1740                 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
1741                 send_blocking_asts(r, lkb);
1742                 goto out;
1743         }
1744
1745         error = -EAGAIN;
1746         if (force_blocking_asts(lkb))
1747                 send_blocking_asts_all(r, lkb);
1748         queue_cast(r, lkb, -EAGAIN);
1749
1750  out:
1751         return error;
1752 }
1753
1754 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1755 {
1756         remove_lock(r, lkb);
1757         queue_cast(r, lkb, -DLM_EUNLOCK);
1758         grant_pending_locks(r);
1759         return -DLM_EUNLOCK;
1760 }
1761
1762 /* FIXME: if revert_lock() finds that the lkb is granted, we should
1763    skip the queue_cast(ECANCEL).  It indicates that the request/convert
1764    completed (and queued a normal ast) just before the cancel; we don't
1765    want to clobber the sb_result for the normal ast with ECANCEL. */
1766
1767 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
1768 {
1769         revert_lock(r, lkb);
1770         queue_cast(r, lkb, -DLM_ECANCEL);
1771         grant_pending_locks(r);
1772         return -DLM_ECANCEL;
1773 }
1774
1775 /*
1776  * Four stage 3 varieties:
1777  * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
1778  */
1779
1780 /* add a new lkb to a possibly new rsb, called by requesting process */
1781
1782 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1783 {
1784         int error;
1785
1786         /* set_master: sets lkb nodeid from r */
1787
1788         error = set_master(r, lkb);
1789         if (error < 0)
1790                 goto out;
1791         if (error) {
1792                 error = 0;
1793                 goto out;
1794         }
1795
1796         if (is_remote(r))
1797                 /* receive_request() calls do_request() on remote node */
1798                 error = send_request(r, lkb);
1799         else
1800                 error = do_request(r, lkb);
1801  out:
1802         return error;
1803 }
1804
1805 /* change some property of an existing lkb, e.g. mode */
1806
1807 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1808 {
1809         int error;
1810
1811         if (is_remote(r))
1812                 /* receive_convert() calls do_convert() on remote node */
1813                 error = send_convert(r, lkb);
1814         else
1815                 error = do_convert(r, lkb);
1816
1817         return error;
1818 }
1819
1820 /* remove an existing lkb from the granted queue */
1821
1822 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1823 {
1824         int error;
1825
1826         if (is_remote(r))
1827                 /* receive_unlock() calls do_unlock() on remote node */
1828                 error = send_unlock(r, lkb);
1829         else
1830                 error = do_unlock(r, lkb);
1831
1832         return error;
1833 }
1834
1835 /* remove an existing lkb from the convert or wait queue */
1836
1837 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1838 {
1839         int error;
1840
1841         if (is_remote(r))
1842                 /* receive_cancel() calls do_cancel() on remote node */
1843                 error = send_cancel(r, lkb);
1844         else
1845                 error = do_cancel(r, lkb);
1846
1847         return error;
1848 }
1849
1850 /*
1851  * Four stage 2 varieties:
1852  * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
1853  */
1854
1855 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
1856                         int len, struct dlm_args *args)
1857 {
1858         struct dlm_rsb *r;
1859         int error;
1860
1861         error = validate_lock_args(ls, lkb, args);
1862         if (error)
1863                 goto out;
1864
1865         error = find_rsb(ls, name, len, R_CREATE, &r);
1866         if (error)
1867                 goto out;
1868
1869         lock_rsb(r);
1870
1871         attach_lkb(r, lkb);
1872         lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
1873
1874         error = _request_lock(r, lkb);
1875
1876         unlock_rsb(r);
1877         put_rsb(r);
1878
1879  out:
1880         return error;
1881 }
1882
1883 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1884                         struct dlm_args *args)
1885 {
1886         struct dlm_rsb *r;
1887         int error;
1888
1889         r = lkb->lkb_resource;
1890
1891         hold_rsb(r);
1892         lock_rsb(r);
1893
1894         error = validate_lock_args(ls, lkb, args);
1895         if (error)
1896                 goto out;
1897
1898         error = _convert_lock(r, lkb);
1899  out:
1900         unlock_rsb(r);
1901         put_rsb(r);
1902         return error;
1903 }
1904
1905 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1906                        struct dlm_args *args)
1907 {
1908         struct dlm_rsb *r;
1909         int error;
1910
1911         r = lkb->lkb_resource;
1912
1913         hold_rsb(r);
1914         lock_rsb(r);
1915
1916         error = validate_unlock_args(lkb, args);
1917         if (error)
1918                 goto out;
1919
1920         error = _unlock_lock(r, lkb);
1921  out:
1922         unlock_rsb(r);
1923         put_rsb(r);
1924         return error;
1925 }
1926
1927 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1928                        struct dlm_args *args)
1929 {
1930         struct dlm_rsb *r;
1931         int error;
1932
1933         r = lkb->lkb_resource;
1934
1935         hold_rsb(r);
1936         lock_rsb(r);
1937
1938         error = validate_unlock_args(lkb, args);
1939         if (error)
1940                 goto out;
1941
1942         error = _cancel_lock(r, lkb);
1943  out:
1944         unlock_rsb(r);
1945         put_rsb(r);
1946         return error;
1947 }
1948
1949 /*
1950  * Two stage 1 varieties:  dlm_lock() and dlm_unlock()
1951  */
1952
1953 int dlm_lock(dlm_lockspace_t *lockspace,
1954              int mode,
1955              struct dlm_lksb *lksb,
1956              uint32_t flags,
1957              void *name,
1958              unsigned int namelen,
1959              uint32_t parent_lkid,
1960              void (*ast) (void *astarg),
1961              void *astarg,
1962              void (*bast) (void *astarg, int mode))
1963 {
1964         struct dlm_ls *ls;
1965         struct dlm_lkb *lkb;
1966         struct dlm_args args;
1967         int error, convert = flags & DLM_LKF_CONVERT;
1968
1969         ls = dlm_find_lockspace_local(lockspace);
1970         if (!ls)
1971                 return -EINVAL;
1972
1973         lock_recovery(ls);
1974
1975         if (convert)
1976                 error = find_lkb(ls, lksb->sb_lkid, &lkb);
1977         else
1978                 error = create_lkb(ls, &lkb);
1979
1980         if (error)
1981                 goto out;
1982
1983         error = set_lock_args(mode, lksb, flags, namelen, parent_lkid, ast,
1984                               astarg, bast, &args);
1985         if (error)
1986                 goto out_put;
1987
1988         if (convert)
1989                 error = convert_lock(ls, lkb, &args);
1990         else
1991                 error = request_lock(ls, lkb, name, namelen, &args);
1992
1993         if (error == -EINPROGRESS)
1994                 error = 0;
1995  out_put:
1996         if (convert || error)
1997                 __put_lkb(ls, lkb);
1998         if (error == -EAGAIN)
1999                 error = 0;
2000  out:
2001         unlock_recovery(ls);
2002         dlm_put_lockspace(ls);
2003         return error;
2004 }
2005
2006 int dlm_unlock(dlm_lockspace_t *lockspace,
2007                uint32_t lkid,
2008                uint32_t flags,
2009                struct dlm_lksb *lksb,
2010                void *astarg)
2011 {
2012         struct dlm_ls *ls;
2013         struct dlm_lkb *lkb;
2014         struct dlm_args args;
2015         int error;
2016
2017         ls = dlm_find_lockspace_local(lockspace);
2018         if (!ls)
2019                 return -EINVAL;
2020
2021         lock_recovery(ls);
2022
2023         error = find_lkb(ls, lkid, &lkb);
2024         if (error)
2025                 goto out;
2026
2027         error = set_unlock_args(flags, astarg, &args);
2028         if (error)
2029                 goto out_put;
2030
2031         if (flags & DLM_LKF_CANCEL)
2032                 error = cancel_lock(ls, lkb, &args);
2033         else
2034                 error = unlock_lock(ls, lkb, &args);
2035
2036         if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2037                 error = 0;
2038  out_put:
2039         dlm_put_lkb(lkb);
2040  out:
2041         unlock_recovery(ls);
2042         dlm_put_lockspace(ls);
2043         return error;
2044 }
2045
2046 /*
2047  * send/receive routines for remote operations and replies
2048  *
2049  * send_args
2050  * send_common
2051  * send_request                 receive_request
2052  * send_convert                 receive_convert
2053  * send_unlock                  receive_unlock
2054  * send_cancel                  receive_cancel
2055  * send_grant                   receive_grant
2056  * send_bast                    receive_bast
2057  * send_lookup                  receive_lookup
2058  * send_remove                  receive_remove
2059  *
2060  *                              send_common_reply
2061  * receive_request_reply        send_request_reply
2062  * receive_convert_reply        send_convert_reply
2063  * receive_unlock_reply         send_unlock_reply
2064  * receive_cancel_reply         send_cancel_reply
2065  * receive_lookup_reply         send_lookup_reply
2066  */
2067
2068 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2069                           int to_nodeid, int mstype,
2070                           struct dlm_message **ms_ret,
2071                           struct dlm_mhandle **mh_ret)
2072 {
2073         struct dlm_message *ms;
2074         struct dlm_mhandle *mh;
2075         char *mb;
2076         int mb_len = sizeof(struct dlm_message);
2077
2078         switch (mstype) {
2079         case DLM_MSG_REQUEST:
2080         case DLM_MSG_LOOKUP:
2081         case DLM_MSG_REMOVE:
2082                 mb_len += r->res_length;
2083                 break;
2084         case DLM_MSG_CONVERT:
2085         case DLM_MSG_UNLOCK:
2086         case DLM_MSG_REQUEST_REPLY:
2087         case DLM_MSG_CONVERT_REPLY:
2088         case DLM_MSG_GRANT:
2089                 if (lkb && lkb->lkb_lvbptr)
2090                         mb_len += r->res_ls->ls_lvblen;
2091                 break;
2092         }
2093
2094         /* get_buffer gives us a message handle (mh) that we need to
2095            pass into lowcomms_commit and a message buffer (mb) that we
2096            write our data into */
2097
2098         mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_KERNEL, &mb);
2099         if (!mh)
2100                 return -ENOBUFS;
2101
2102         memset(mb, 0, mb_len);
2103
2104         ms = (struct dlm_message *) mb;
2105
2106         ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2107         ms->m_header.h_lockspace = r->res_ls->ls_global_id;
2108         ms->m_header.h_nodeid = dlm_our_nodeid();
2109         ms->m_header.h_length = mb_len;
2110         ms->m_header.h_cmd = DLM_MSG;
2111
2112         ms->m_type = mstype;
2113
2114         *mh_ret = mh;
2115         *ms_ret = ms;
2116         return 0;
2117 }
2118
2119 /* further lowcomms enhancements or alternate implementations may make
2120    the return value from this function useful at some point */
2121
2122 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2123 {
2124         dlm_message_out(ms);
2125         dlm_lowcomms_commit_buffer(mh);
2126         return 0;
2127 }
2128
2129 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2130                       struct dlm_message *ms)
2131 {
2132         ms->m_nodeid   = lkb->lkb_nodeid;
2133         ms->m_pid      = lkb->lkb_ownpid;
2134         ms->m_lkid     = lkb->lkb_id;
2135         ms->m_remid    = lkb->lkb_remid;
2136         ms->m_exflags  = lkb->lkb_exflags;
2137         ms->m_sbflags  = lkb->lkb_sbflags;
2138         ms->m_flags    = lkb->lkb_flags;
2139         ms->m_lvbseq   = lkb->lkb_lvbseq;
2140         ms->m_status   = lkb->lkb_status;
2141         ms->m_grmode   = lkb->lkb_grmode;
2142         ms->m_rqmode   = lkb->lkb_rqmode;
2143         ms->m_hash     = r->res_hash;
2144
2145         /* m_result and m_bastmode are set from function args,
2146            not from lkb fields */
2147
2148         if (lkb->lkb_bastaddr)
2149                 ms->m_asts |= AST_BAST;
2150         if (lkb->lkb_astaddr)
2151                 ms->m_asts |= AST_COMP;
2152
2153         /* compare with switch in create_message; send_remove() doesn't
2154            use send_args() */
2155
2156         switch (ms->m_type) {
2157         case DLM_MSG_REQUEST:
2158         case DLM_MSG_LOOKUP:
2159                 memcpy(ms->m_extra, r->res_name, r->res_length);
2160                 break;
2161         case DLM_MSG_CONVERT:
2162         case DLM_MSG_UNLOCK:
2163         case DLM_MSG_REQUEST_REPLY:
2164         case DLM_MSG_CONVERT_REPLY:
2165         case DLM_MSG_GRANT:
2166                 if (!lkb->lkb_lvbptr)
2167                         break;
2168                 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2169                 break;
2170         }
2171 }
2172
2173 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2174 {
2175         struct dlm_message *ms;
2176         struct dlm_mhandle *mh;
2177         int to_nodeid, error;
2178
2179         add_to_waiters(lkb, mstype);
2180
2181         to_nodeid = r->res_nodeid;
2182
2183         error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2184         if (error)
2185                 goto fail;
2186
2187         send_args(r, lkb, ms);
2188
2189         error = send_message(mh, ms);
2190         if (error)
2191                 goto fail;
2192         return 0;
2193
2194  fail:
2195         remove_from_waiters(lkb);
2196         return error;
2197 }
2198
2199 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2200 {
2201         return send_common(r, lkb, DLM_MSG_REQUEST);
2202 }
2203
2204 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2205 {
2206         int error;
2207
2208         error = send_common(r, lkb, DLM_MSG_CONVERT);
2209
2210         /* down conversions go without a reply from the master */
2211         if (!error && down_conversion(lkb)) {
2212                 remove_from_waiters(lkb);
2213                 r->res_ls->ls_stub_ms.m_result = 0;
2214                 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
2215                 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2216         }
2217
2218         return error;
2219 }
2220
2221 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2222    MASTER_UNCERTAIN to force the next request on the rsb to confirm
2223    that the master is still correct. */
2224
2225 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2226 {
2227         return send_common(r, lkb, DLM_MSG_UNLOCK);
2228 }
2229
2230 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2231 {
2232         return send_common(r, lkb, DLM_MSG_CANCEL);
2233 }
2234
2235 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2236 {
2237         struct dlm_message *ms;
2238         struct dlm_mhandle *mh;
2239         int to_nodeid, error;
2240
2241         to_nodeid = lkb->lkb_nodeid;
2242
2243         error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2244         if (error)
2245                 goto out;
2246
2247         send_args(r, lkb, ms);
2248
2249         ms->m_result = 0;
2250
2251         error = send_message(mh, ms);
2252  out:
2253         return error;
2254 }
2255
2256 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2257 {
2258         struct dlm_message *ms;
2259         struct dlm_mhandle *mh;
2260         int to_nodeid, error;
2261
2262         to_nodeid = lkb->lkb_nodeid;
2263
2264         error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2265         if (error)
2266                 goto out;
2267
2268         send_args(r, lkb, ms);
2269
2270         ms->m_bastmode = mode;
2271
2272         error = send_message(mh, ms);
2273  out:
2274         return error;
2275 }
2276
2277 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2278 {
2279         struct dlm_message *ms;
2280         struct dlm_mhandle *mh;
2281         int to_nodeid, error;
2282
2283         add_to_waiters(lkb, DLM_MSG_LOOKUP);
2284
2285         to_nodeid = dlm_dir_nodeid(r);
2286
2287         error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2288         if (error)
2289                 goto fail;
2290
2291         send_args(r, lkb, ms);
2292
2293         error = send_message(mh, ms);
2294         if (error)
2295                 goto fail;
2296         return 0;
2297
2298  fail:
2299         remove_from_waiters(lkb);
2300         return error;
2301 }
2302
2303 static int send_remove(struct dlm_rsb *r)
2304 {
2305         struct dlm_message *ms;
2306         struct dlm_mhandle *mh;
2307         int to_nodeid, error;
2308
2309         to_nodeid = dlm_dir_nodeid(r);
2310
2311         error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2312         if (error)
2313                 goto out;
2314
2315         memcpy(ms->m_extra, r->res_name, r->res_length);
2316         ms->m_hash = r->res_hash;
2317
2318         error = send_message(mh, ms);
2319  out:
2320         return error;
2321 }
2322
2323 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2324                              int mstype, int rv)
2325 {
2326         struct dlm_message *ms;
2327         struct dlm_mhandle *mh;
2328         int to_nodeid, error;
2329
2330         to_nodeid = lkb->lkb_nodeid;
2331
2332         error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2333         if (error)
2334                 goto out;
2335
2336         send_args(r, lkb, ms);
2337
2338         ms->m_result = rv;
2339
2340         error = send_message(mh, ms);
2341  out:
2342         return error;
2343 }
2344
2345 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2346 {
2347         return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2348 }
2349
2350 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2351 {
2352         return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2353 }
2354
2355 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2356 {
2357         return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2358 }
2359
2360 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2361 {
2362         return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2363 }
2364
2365 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2366                              int ret_nodeid, int rv)
2367 {
2368         struct dlm_rsb *r = &ls->ls_stub_rsb;
2369         struct dlm_message *ms;
2370         struct dlm_mhandle *mh;
2371         int error, nodeid = ms_in->m_header.h_nodeid;
2372
2373         error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2374         if (error)
2375                 goto out;
2376
2377         ms->m_lkid = ms_in->m_lkid;
2378         ms->m_result = rv;
2379         ms->m_nodeid = ret_nodeid;
2380
2381         error = send_message(mh, ms);
2382  out:
2383         return error;
2384 }
2385
2386 /* which args we save from a received message depends heavily on the type
2387    of message, unlike the send side where we can safely send everything about
2388    the lkb for any type of message */
2389
2390 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2391 {
2392         lkb->lkb_exflags = ms->m_exflags;
2393         lkb->lkb_sbflags = ms->m_sbflags;
2394         lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2395                          (ms->m_flags & 0x0000FFFF);
2396 }
2397
2398 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2399 {
2400         lkb->lkb_sbflags = ms->m_sbflags;
2401         lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2402                          (ms->m_flags & 0x0000FFFF);
2403 }
2404
2405 static int receive_extralen(struct dlm_message *ms)
2406 {
2407         return (ms->m_header.h_length - sizeof(struct dlm_message));
2408 }
2409
2410 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2411                        struct dlm_message *ms)
2412 {
2413         int len;
2414
2415         if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2416                 if (!lkb->lkb_lvbptr)
2417                         lkb->lkb_lvbptr = allocate_lvb(ls);
2418                 if (!lkb->lkb_lvbptr)
2419                         return -ENOMEM;
2420                 len = receive_extralen(ms);
2421                 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2422         }
2423         return 0;
2424 }
2425
2426 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2427                                 struct dlm_message *ms)
2428 {
2429         lkb->lkb_nodeid = ms->m_header.h_nodeid;
2430         lkb->lkb_ownpid = ms->m_pid;
2431         lkb->lkb_remid = ms->m_lkid;
2432         lkb->lkb_grmode = DLM_LOCK_IV;
2433         lkb->lkb_rqmode = ms->m_rqmode;
2434         lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
2435         lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
2436
2437         DLM_ASSERT(is_master_copy(lkb), dlm_print_lkb(lkb););
2438
2439         if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2440                 /* lkb was just created so there won't be an lvb yet */
2441                 lkb->lkb_lvbptr = allocate_lvb(ls);
2442                 if (!lkb->lkb_lvbptr)
2443                         return -ENOMEM;
2444         }
2445
2446         return 0;
2447 }
2448
2449 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2450                                 struct dlm_message *ms)
2451 {
2452         if (lkb->lkb_nodeid != ms->m_header.h_nodeid) {
2453                 log_error(ls, "convert_args nodeid %d %d lkid %x %x",
2454                           lkb->lkb_nodeid, ms->m_header.h_nodeid,
2455                           lkb->lkb_id, lkb->lkb_remid);
2456                 return -EINVAL;
2457         }
2458
2459         if (!is_master_copy(lkb))
2460                 return -EINVAL;
2461
2462         if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2463                 return -EBUSY;
2464
2465         if (receive_lvb(ls, lkb, ms))
2466                 return -ENOMEM;
2467
2468         lkb->lkb_rqmode = ms->m_rqmode;
2469         lkb->lkb_lvbseq = ms->m_lvbseq;
2470
2471         return 0;
2472 }
2473
2474 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2475                                struct dlm_message *ms)
2476 {
2477         if (!is_master_copy(lkb))
2478                 return -EINVAL;
2479         if (receive_lvb(ls, lkb, ms))
2480                 return -ENOMEM;
2481         return 0;
2482 }
2483
2484 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
2485    uses to send a reply and that the remote end uses to process the reply. */
2486
2487 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
2488 {
2489         struct dlm_lkb *lkb = &ls->ls_stub_lkb;
2490         lkb->lkb_nodeid = ms->m_header.h_nodeid;
2491         lkb->lkb_remid = ms->m_lkid;
2492 }
2493
2494 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
2495 {
2496         struct dlm_lkb *lkb;
2497         struct dlm_rsb *r;
2498         int error, namelen;
2499
2500         error = create_lkb(ls, &lkb);
2501         if (error)
2502                 goto fail;
2503
2504         receive_flags(lkb, ms);
2505         lkb->lkb_flags |= DLM_IFL_MSTCPY;
2506         error = receive_request_args(ls, lkb, ms);
2507         if (error) {
2508                 __put_lkb(ls, lkb);
2509                 goto fail;
2510         }
2511
2512         namelen = receive_extralen(ms);
2513
2514         error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
2515         if (error) {
2516                 __put_lkb(ls, lkb);
2517                 goto fail;
2518         }
2519
2520         lock_rsb(r);
2521
2522         attach_lkb(r, lkb);
2523         error = do_request(r, lkb);
2524         send_request_reply(r, lkb, error);
2525
2526         unlock_rsb(r);
2527         put_rsb(r);
2528
2529         if (error == -EINPROGRESS)
2530                 error = 0;
2531         if (error)
2532                 dlm_put_lkb(lkb);
2533         return;
2534
2535  fail:
2536         setup_stub_lkb(ls, ms);
2537         send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2538 }
2539
2540 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
2541 {
2542         struct dlm_lkb *lkb;
2543         struct dlm_rsb *r;
2544         int error, reply = 1;
2545
2546         error = find_lkb(ls, ms->m_remid, &lkb);
2547         if (error)
2548                 goto fail;
2549
2550         r = lkb->lkb_resource;
2551
2552         hold_rsb(r);
2553         lock_rsb(r);
2554
2555         receive_flags(lkb, ms);
2556         error = receive_convert_args(ls, lkb, ms);
2557         if (error)
2558                 goto out;
2559         reply = !down_conversion(lkb);
2560
2561         error = do_convert(r, lkb);
2562  out:
2563         if (reply)
2564                 send_convert_reply(r, lkb, error);
2565
2566         unlock_rsb(r);
2567         put_rsb(r);
2568         dlm_put_lkb(lkb);
2569         return;
2570
2571  fail:
2572         setup_stub_lkb(ls, ms);
2573         send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2574 }
2575
2576 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
2577 {
2578         struct dlm_lkb *lkb;
2579         struct dlm_rsb *r;
2580         int error;
2581
2582         error = find_lkb(ls, ms->m_remid, &lkb);
2583         if (error)
2584                 goto fail;
2585
2586         r = lkb->lkb_resource;
2587
2588         hold_rsb(r);
2589         lock_rsb(r);
2590
2591         receive_flags(lkb, ms);
2592         error = receive_unlock_args(ls, lkb, ms);
2593         if (error)
2594                 goto out;
2595
2596         error = do_unlock(r, lkb);
2597  out:
2598         send_unlock_reply(r, lkb, error);
2599
2600         unlock_rsb(r);
2601         put_rsb(r);
2602         dlm_put_lkb(lkb);
2603         return;
2604
2605  fail:
2606         setup_stub_lkb(ls, ms);
2607         send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2608 }
2609
2610 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
2611 {
2612         struct dlm_lkb *lkb;
2613         struct dlm_rsb *r;
2614         int error;
2615
2616         error = find_lkb(ls, ms->m_remid, &lkb);
2617         if (error)
2618                 goto fail;
2619
2620         receive_flags(lkb, ms);
2621
2622         r = lkb->lkb_resource;
2623
2624         hold_rsb(r);
2625         lock_rsb(r);
2626
2627         error = do_cancel(r, lkb);
2628         send_cancel_reply(r, lkb, error);
2629
2630         unlock_rsb(r);
2631         put_rsb(r);
2632         dlm_put_lkb(lkb);
2633         return;
2634
2635  fail:
2636         setup_stub_lkb(ls, ms);
2637         send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2638 }
2639
2640 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
2641 {
2642         struct dlm_lkb *lkb;
2643         struct dlm_rsb *r;
2644         int error;
2645
2646         error = find_lkb(ls, ms->m_remid, &lkb);
2647         if (error) {
2648                 log_error(ls, "receive_grant no lkb");
2649                 return;
2650         }
2651         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2652
2653         r = lkb->lkb_resource;
2654
2655         hold_rsb(r);
2656         lock_rsb(r);
2657
2658         receive_flags_reply(lkb, ms);
2659         grant_lock_pc(r, lkb, ms);
2660         queue_cast(r, lkb, 0);
2661
2662         unlock_rsb(r);
2663         put_rsb(r);
2664         dlm_put_lkb(lkb);
2665 }
2666
2667 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
2668 {
2669         struct dlm_lkb *lkb;
2670         struct dlm_rsb *r;
2671         int error;
2672
2673         error = find_lkb(ls, ms->m_remid, &lkb);
2674         if (error) {
2675                 log_error(ls, "receive_bast no lkb");
2676                 return;
2677         }
2678         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2679
2680         r = lkb->lkb_resource;
2681
2682         hold_rsb(r);
2683         lock_rsb(r);
2684
2685         queue_bast(r, lkb, ms->m_bastmode);
2686
2687         unlock_rsb(r);
2688         put_rsb(r);
2689         dlm_put_lkb(lkb);
2690 }
2691
2692 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
2693 {
2694         int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
2695
2696         from_nodeid = ms->m_header.h_nodeid;
2697         our_nodeid = dlm_our_nodeid();
2698
2699         len = receive_extralen(ms);
2700
2701         dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2702         if (dir_nodeid != our_nodeid) {
2703                 log_error(ls, "lookup dir_nodeid %d from %d",
2704                           dir_nodeid, from_nodeid);
2705                 error = -EINVAL;
2706                 ret_nodeid = -1;
2707                 goto out;
2708         }
2709
2710         error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
2711
2712         /* Optimization: we're master so treat lookup as a request */
2713         if (!error && ret_nodeid == our_nodeid) {
2714                 receive_request(ls, ms);
2715                 return;
2716         }
2717  out:
2718         send_lookup_reply(ls, ms, ret_nodeid, error);
2719 }
2720
2721 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
2722 {
2723         int len, dir_nodeid, from_nodeid;
2724
2725         from_nodeid = ms->m_header.h_nodeid;
2726
2727         len = receive_extralen(ms);
2728
2729         dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2730         if (dir_nodeid != dlm_our_nodeid()) {
2731                 log_error(ls, "remove dir entry dir_nodeid %d from %d",
2732                           dir_nodeid, from_nodeid);
2733                 return;
2734         }
2735
2736         dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
2737 }
2738
2739 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
2740 {
2741         struct dlm_lkb *lkb;
2742         struct dlm_rsb *r;
2743         int error, mstype;
2744
2745         error = find_lkb(ls, ms->m_remid, &lkb);
2746         if (error) {
2747                 log_error(ls, "receive_request_reply no lkb");
2748                 return;
2749         }
2750         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2751
2752         mstype = lkb->lkb_wait_type;
2753         error = remove_from_waiters(lkb);
2754         if (error) {
2755                 log_error(ls, "receive_request_reply not on waiters");
2756                 goto out;
2757         }
2758
2759         /* this is the value returned from do_request() on the master */
2760         error = ms->m_result;
2761
2762         r = lkb->lkb_resource;
2763         hold_rsb(r);
2764         lock_rsb(r);
2765
2766         /* Optimization: the dir node was also the master, so it took our
2767            lookup as a request and sent request reply instead of lookup reply */
2768         if (mstype == DLM_MSG_LOOKUP) {
2769                 r->res_nodeid = ms->m_header.h_nodeid;
2770                 lkb->lkb_nodeid = r->res_nodeid;
2771         }
2772
2773         switch (error) {
2774         case -EAGAIN:
2775                 /* request would block (be queued) on remote master;
2776                    the unhold undoes the original ref from create_lkb()
2777                    so it leads to the lkb being freed */
2778                 queue_cast(r, lkb, -EAGAIN);
2779                 confirm_master(r, -EAGAIN);
2780                 unhold_lkb(lkb);
2781                 break;
2782
2783         case -EINPROGRESS:
2784         case 0:
2785                 /* request was queued or granted on remote master */
2786                 receive_flags_reply(lkb, ms);
2787                 lkb->lkb_remid = ms->m_lkid;
2788                 if (error)
2789                         add_lkb(r, lkb, DLM_LKSTS_WAITING);
2790                 else {
2791                         grant_lock_pc(r, lkb, ms);
2792                         queue_cast(r, lkb, 0);
2793                 }
2794                 confirm_master(r, error);
2795                 break;
2796
2797         case -EBADR:
2798         case -ENOTBLK:
2799                 /* find_rsb failed to find rsb or rsb wasn't master */
2800                 r->res_nodeid = -1;
2801                 lkb->lkb_nodeid = -1;
2802                 _request_lock(r, lkb);
2803                 break;
2804
2805         default:
2806                 log_error(ls, "receive_request_reply error %d", error);
2807         }
2808
2809         unlock_rsb(r);
2810         put_rsb(r);
2811  out:
2812         dlm_put_lkb(lkb);
2813 }
2814
2815 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2816                                     struct dlm_message *ms)
2817 {
2818         int error = ms->m_result;
2819
2820         /* this is the value returned from do_convert() on the master */
2821
2822         switch (error) {
2823         case -EAGAIN:
2824                 /* convert would block (be queued) on remote master */
2825                 queue_cast(r, lkb, -EAGAIN);
2826                 break;
2827
2828         case -EINPROGRESS:
2829                 /* convert was queued on remote master */
2830                 del_lkb(r, lkb);
2831                 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2832                 break;
2833
2834         case 0:
2835                 /* convert was granted on remote master */
2836                 receive_flags_reply(lkb, ms);
2837                 grant_lock_pc(r, lkb, ms);
2838                 queue_cast(r, lkb, 0);
2839                 break;
2840
2841         default:
2842                 log_error(r->res_ls, "receive_convert_reply error %d", error);
2843         }
2844 }
2845
2846 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2847 {
2848         struct dlm_rsb *r = lkb->lkb_resource;
2849
2850         hold_rsb(r);
2851         lock_rsb(r);
2852
2853         __receive_convert_reply(r, lkb, ms);
2854
2855         unlock_rsb(r);
2856         put_rsb(r);
2857 }
2858
2859 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
2860 {
2861         struct dlm_lkb *lkb;
2862         int error;
2863
2864         error = find_lkb(ls, ms->m_remid, &lkb);
2865         if (error) {
2866                 log_error(ls, "receive_convert_reply no lkb");
2867                 return;
2868         }
2869         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2870
2871         error = remove_from_waiters(lkb);
2872         if (error) {
2873                 log_error(ls, "receive_convert_reply not on waiters");
2874                 goto out;
2875         }
2876
2877         _receive_convert_reply(lkb, ms);
2878  out:
2879         dlm_put_lkb(lkb);
2880 }
2881
2882 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2883 {
2884         struct dlm_rsb *r = lkb->lkb_resource;
2885         int error = ms->m_result;
2886
2887         hold_rsb(r);
2888         lock_rsb(r);
2889
2890         /* this is the value returned from do_unlock() on the master */
2891
2892         switch (error) {
2893         case -DLM_EUNLOCK:
2894                 receive_flags_reply(lkb, ms);
2895                 remove_lock_pc(r, lkb);
2896                 queue_cast(r, lkb, -DLM_EUNLOCK);
2897                 break;
2898         default:
2899                 log_error(r->res_ls, "receive_unlock_reply error %d", error);
2900         }
2901
2902         unlock_rsb(r);
2903         put_rsb(r);
2904 }
2905
2906 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
2907 {
2908         struct dlm_lkb *lkb;
2909         int error;
2910
2911         error = find_lkb(ls, ms->m_remid, &lkb);
2912         if (error) {
2913                 log_error(ls, "receive_unlock_reply no lkb");
2914                 return;
2915         }
2916         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2917
2918         error = remove_from_waiters(lkb);
2919         if (error) {
2920                 log_error(ls, "receive_unlock_reply not on waiters");
2921                 goto out;
2922         }
2923
2924         _receive_unlock_reply(lkb, ms);
2925  out:
2926         dlm_put_lkb(lkb);
2927 }
2928
2929 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2930 {
2931         struct dlm_rsb *r = lkb->lkb_resource;
2932         int error = ms->m_result;
2933
2934         hold_rsb(r);
2935         lock_rsb(r);
2936
2937         /* this is the value returned from do_cancel() on the master */
2938
2939         switch (error) {
2940         case -DLM_ECANCEL:
2941                 receive_flags_reply(lkb, ms);
2942                 revert_lock_pc(r, lkb);
2943                 queue_cast(r, lkb, -DLM_ECANCEL);
2944                 break;
2945         default:
2946                 log_error(r->res_ls, "receive_cancel_reply error %d", error);
2947         }
2948
2949         unlock_rsb(r);
2950         put_rsb(r);
2951 }
2952
2953 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
2954 {
2955         struct dlm_lkb *lkb;
2956         int error;
2957
2958         error = find_lkb(ls, ms->m_remid, &lkb);
2959         if (error) {
2960                 log_error(ls, "receive_cancel_reply no lkb");
2961                 return;
2962         }
2963         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2964
2965         error = remove_from_waiters(lkb);
2966         if (error) {
2967                 log_error(ls, "receive_cancel_reply not on waiters");
2968                 goto out;
2969         }
2970
2971         _receive_cancel_reply(lkb, ms);
2972  out:
2973         dlm_put_lkb(lkb);
2974 }
2975
2976 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
2977 {
2978         struct dlm_lkb *lkb;
2979         struct dlm_rsb *r;
2980         int error, ret_nodeid;
2981
2982         error = find_lkb(ls, ms->m_lkid, &lkb);
2983         if (error) {
2984                 log_error(ls, "receive_lookup_reply no lkb");
2985                 return;
2986         }
2987
2988         error = remove_from_waiters(lkb);
2989         if (error) {
2990                 log_error(ls, "receive_lookup_reply not on waiters");
2991                 goto out;
2992         }
2993
2994         /* this is the value returned by dlm_dir_lookup on dir node
2995            FIXME: will a non-zero error ever be returned? */
2996         error = ms->m_result;
2997
2998         r = lkb->lkb_resource;
2999         hold_rsb(r);
3000         lock_rsb(r);
3001
3002         ret_nodeid = ms->m_nodeid;
3003         if (ret_nodeid == dlm_our_nodeid()) {
3004                 r->res_nodeid = 0;
3005                 ret_nodeid = 0;
3006                 r->res_first_lkid = 0;
3007         } else {
3008                 /* set_master() will copy res_nodeid to lkb_nodeid */
3009                 r->res_nodeid = ret_nodeid;
3010         }
3011
3012         _request_lock(r, lkb);
3013
3014         if (!ret_nodeid)
3015                 process_lookup_list(r);
3016
3017         unlock_rsb(r);
3018         put_rsb(r);
3019  out:
3020         dlm_put_lkb(lkb);
3021 }
3022
3023 int dlm_receive_message(struct dlm_header *hd, int nodeid, int recovery)
3024 {
3025         struct dlm_message *ms = (struct dlm_message *) hd;
3026         struct dlm_ls *ls;
3027         int error = 0;
3028
3029         if (!recovery)
3030                 dlm_message_in(ms);
3031
3032         ls = dlm_find_lockspace_global(hd->h_lockspace);
3033         if (!ls) {
3034                 log_print("drop message %d from %d for unknown lockspace %d",
3035                           ms->m_type, nodeid, hd->h_lockspace);
3036                 return -EINVAL;
3037         }
3038
3039         /* recovery may have just ended leaving a bunch of backed-up requests
3040            in the requestqueue; wait while dlm_recoverd clears them */
3041
3042         if (!recovery)
3043                 dlm_wait_requestqueue(ls);
3044
3045         /* recovery may have just started while there were a bunch of
3046            in-flight requests -- save them in requestqueue to be processed
3047            after recovery.  we can't let dlm_recvd block on the recovery
3048            lock.  if dlm_recoverd is calling this function to clear the
3049            requestqueue, it needs to be interrupted (-EINTR) if another
3050            recovery operation is starting. */
3051
3052         while (1) {
3053                 if (dlm_locking_stopped(ls)) {
3054                         if (recovery) {
3055                                 error = -EINTR;
3056                                 goto out;
3057                         }
3058                         error = dlm_add_requestqueue(ls, nodeid, hd);
3059                         if (error == -EAGAIN)
3060                                 continue;
3061                         else {
3062                                 error = -EINTR;
3063                                 goto out;
3064                         }
3065                 }
3066
3067                 if (lock_recovery_try(ls))
3068                         break;
3069                 schedule();
3070         }
3071
3072         switch (ms->m_type) {
3073
3074         /* messages sent to a master node */
3075
3076         case DLM_MSG_REQUEST:
3077                 receive_request(ls, ms);
3078                 break;
3079
3080         case DLM_MSG_CONVERT:
3081                 receive_convert(ls, ms);
3082                 break;
3083
3084         case DLM_MSG_UNLOCK:
3085                 receive_unlock(ls, ms);
3086                 break;
3087
3088         case DLM_MSG_CANCEL:
3089                 receive_cancel(ls, ms);
3090                 break;
3091
3092         /* messages sent from a master node (replies to above) */
3093
3094         case DLM_MSG_REQUEST_REPLY:
3095                 receive_request_reply(ls, ms);
3096                 break;
3097
3098         case DLM_MSG_CONVERT_REPLY:
3099                 receive_convert_reply(ls, ms);
3100                 break;
3101
3102         case DLM_MSG_UNLOCK_REPLY:
3103                 receive_unlock_reply(ls, ms);
3104                 break;
3105
3106         case DLM_MSG_CANCEL_REPLY:
3107                 receive_cancel_reply(ls, ms);
3108                 break;
3109
3110         /* messages sent from a master node (only two types of async msg) */
3111
3112         case DLM_MSG_GRANT:
3113                 receive_grant(ls, ms);
3114                 break;
3115
3116         case DLM_MSG_BAST:
3117                 receive_bast(ls, ms);
3118                 break;
3119
3120         /* messages sent to a dir node */
3121
3122         case DLM_MSG_LOOKUP:
3123                 receive_lookup(ls, ms);
3124                 break;
3125
3126         case DLM_MSG_REMOVE:
3127                 receive_remove(ls, ms);
3128                 break;
3129
3130         /* messages sent from a dir node (remove has no reply) */
3131
3132         case DLM_MSG_LOOKUP_REPLY:
3133                 receive_lookup_reply(ls, ms);
3134                 break;
3135
3136         default:
3137                 log_error(ls, "unknown message type %d", ms->m_type);
3138         }
3139
3140         unlock_recovery(ls);
3141  out:
3142         dlm_put_lockspace(ls);
3143         dlm_astd_wake();
3144         return error;
3145 }
3146
3147
3148 /*
3149  * Recovery related
3150  */
3151
3152 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3153 {
3154         if (middle_conversion(lkb)) {
3155                 hold_lkb(lkb);
3156                 ls->ls_stub_ms.m_result = -EINPROGRESS;
3157                 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3158                 _remove_from_waiters(lkb);
3159                 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3160
3161                 /* Same special case as in receive_rcom_lock_args() */
3162                 lkb->lkb_grmode = DLM_LOCK_IV;
3163                 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3164                 unhold_lkb(lkb);
3165
3166         } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3167                 lkb->lkb_flags |= DLM_IFL_RESEND;
3168         }
3169
3170         /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3171            conversions are async; there's no reply from the remote master */
3172 }
3173
3174 /* A waiting lkb needs recovery if the master node has failed, or
3175    the master node is changing (only when no directory is used) */
3176
3177 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3178 {
3179         if (dlm_is_removed(ls, lkb->lkb_nodeid))
3180                 return 1;
3181
3182         if (!dlm_no_directory(ls))
3183                 return 0;
3184
3185         if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3186                 return 1;
3187
3188         return 0;
3189 }
3190
3191 /* Recovery for locks that are waiting for replies from nodes that are now
3192    gone.  We can just complete unlocks and cancels by faking a reply from the
3193    dead node.  Requests and up-conversions we flag to be resent after
3194    recovery.  Down-conversions can just be completed with a fake reply like
3195    unlocks.  Conversions between PR and CW need special attention. */
3196
3197 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3198 {
3199         struct dlm_lkb *lkb, *safe;
3200
3201         mutex_lock(&ls->ls_waiters_mutex);
3202
3203         list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3204                 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3205                           lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3206
3207                 /* all outstanding lookups, regardless of destination  will be
3208                    resent after recovery is done */
3209
3210                 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3211                         lkb->lkb_flags |= DLM_IFL_RESEND;
3212                         continue;
3213                 }
3214
3215                 if (!waiter_needs_recovery(ls, lkb))
3216                         continue;
3217
3218                 switch (lkb->lkb_wait_type) {
3219
3220                 case DLM_MSG_REQUEST:
3221                         lkb->lkb_flags |= DLM_IFL_RESEND;
3222                         break;
3223
3224                 case DLM_MSG_CONVERT:
3225                         recover_convert_waiter(ls, lkb);
3226                         break;
3227
3228                 case DLM_MSG_UNLOCK:
3229                         hold_lkb(lkb);
3230                         ls->ls_stub_ms.m_result = -DLM_EUNLOCK;
3231                         ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3232                         _remove_from_waiters(lkb);
3233                         _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3234                         dlm_put_lkb(lkb);
3235                         break;
3236
3237                 case DLM_MSG_CANCEL:
3238                         hold_lkb(lkb);
3239                         ls->ls_stub_ms.m_result = -DLM_ECANCEL;
3240                         ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3241                         _remove_from_waiters(lkb);
3242                         _receive_cancel_reply(lkb, &ls->ls_stub_ms);
3243                         dlm_put_lkb(lkb);
3244                         break;
3245
3246                 default:
3247                         log_error(ls, "invalid lkb wait_type %d",
3248                                   lkb->lkb_wait_type);
3249                 }
3250                 schedule();
3251         }
3252         mutex_unlock(&ls->ls_waiters_mutex);
3253 }
3254
3255 static int remove_resend_waiter(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
3256 {
3257         struct dlm_lkb *lkb;
3258         int rv = 0;
3259
3260         mutex_lock(&ls->ls_waiters_mutex);
3261         list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
3262                 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3263                         rv = lkb->lkb_wait_type;
3264                         _remove_from_waiters(lkb);
3265                         lkb->lkb_flags &= ~DLM_IFL_RESEND;
3266                         break;
3267                 }
3268         }
3269         mutex_unlock(&ls->ls_waiters_mutex);
3270
3271         if (!rv)
3272                 lkb = NULL;
3273         *lkb_ret = lkb;
3274         return rv;
3275 }
3276
3277 /* Deal with lookups and lkb's marked RESEND from _pre.  We may now be the
3278    master or dir-node for r.  Processing the lkb may result in it being placed
3279    back on waiters. */
3280
3281 int dlm_recover_waiters_post(struct dlm_ls *ls)
3282 {
3283         struct dlm_lkb *lkb;
3284         struct dlm_rsb *r;
3285         int error = 0, mstype;
3286
3287         while (1) {
3288                 if (dlm_locking_stopped(ls)) {
3289                         log_debug(ls, "recover_waiters_post aborted");
3290                         error = -EINTR;
3291                         break;
3292                 }
3293
3294                 mstype = remove_resend_waiter(ls, &lkb);
3295                 if (!mstype)
3296                         break;
3297
3298                 r = lkb->lkb_resource;
3299
3300                 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
3301                           lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
3302
3303                 switch (mstype) {
3304
3305                 case DLM_MSG_LOOKUP:
3306                         hold_rsb(r);
3307                         lock_rsb(r);
3308                         _request_lock(r, lkb);
3309                         if (is_master(r))
3310                                 confirm_master(r, 0);
3311                         unlock_rsb(r);
3312                         put_rsb(r);
3313                         break;
3314
3315                 case DLM_MSG_REQUEST:
3316                         hold_rsb(r);
3317                         lock_rsb(r);
3318                         _request_lock(r, lkb);
3319                         if (is_master(r))
3320                                 confirm_master(r, 0);
3321                         unlock_rsb(r);
3322                         put_rsb(r);
3323                         break;
3324
3325                 case DLM_MSG_CONVERT:
3326                         hold_rsb(r);
3327                         lock_rsb(r);
3328                         _convert_lock(r, lkb);
3329                         unlock_rsb(r);
3330                         put_rsb(r);
3331                         break;
3332
3333                 default:
3334                         log_error(ls, "recover_waiters_post type %d", mstype);
3335                 }
3336         }
3337
3338         return error;
3339 }
3340
3341 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
3342                         int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
3343 {
3344         struct dlm_ls *ls = r->res_ls;
3345         struct dlm_lkb *lkb, *safe;
3346
3347         list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
3348                 if (test(ls, lkb)) {
3349                         rsb_set_flag(r, RSB_LOCKS_PURGED);
3350                         del_lkb(r, lkb);
3351                         /* this put should free the lkb */
3352                         if (!dlm_put_lkb(lkb))
3353                                 log_error(ls, "purged lkb not released");
3354                 }
3355         }
3356 }
3357
3358 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3359 {
3360         return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
3361 }
3362
3363 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3364 {
3365         return is_master_copy(lkb);
3366 }
3367
3368 static void purge_dead_locks(struct dlm_rsb *r)
3369 {
3370         purge_queue(r, &r->res_grantqueue, &purge_dead_test);
3371         purge_queue(r, &r->res_convertqueue, &purge_dead_test);
3372         purge_queue(r, &r->res_waitqueue, &purge_dead_test);
3373 }
3374
3375 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
3376 {
3377         purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
3378         purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
3379         purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
3380 }
3381
3382 /* Get rid of locks held by nodes that are gone. */
3383
3384 int dlm_purge_locks(struct dlm_ls *ls)
3385 {
3386         struct dlm_rsb *r;
3387
3388         log_debug(ls, "dlm_purge_locks");
3389
3390         down_write(&ls->ls_root_sem);
3391         list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
3392                 hold_rsb(r);
3393                 lock_rsb(r);
3394                 if (is_master(r))
3395                         purge_dead_locks(r);
3396                 unlock_rsb(r);
3397                 unhold_rsb(r);
3398
3399                 schedule();
3400         }
3401         up_write(&ls->ls_root_sem);
3402
3403         return 0;
3404 }
3405
3406 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
3407 {
3408         struct dlm_rsb *r, *r_ret = NULL;
3409
3410         read_lock(&ls->ls_rsbtbl[bucket].lock);
3411         list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
3412                 if (!rsb_flag(r, RSB_LOCKS_PURGED))
3413                         continue;
3414                 hold_rsb(r);
3415                 rsb_clear_flag(r, RSB_LOCKS_PURGED);
3416                 r_ret = r;
3417                 break;
3418         }
3419         read_unlock(&ls->ls_rsbtbl[bucket].lock);
3420         return r_ret;
3421 }
3422
3423 void dlm_grant_after_purge(struct dlm_ls *ls)
3424 {
3425         struct dlm_rsb *r;
3426         int bucket = 0;
3427
3428         while (1) {
3429                 r = find_purged_rsb(ls, bucket);
3430                 if (!r) {
3431                         if (bucket == ls->ls_rsbtbl_size - 1)
3432                                 break;
3433                         bucket++;
3434                         continue;
3435                 }
3436                 lock_rsb(r);
3437                 if (is_master(r)) {
3438                         grant_pending_locks(r);
3439                         confirm_master(r, 0);
3440                 }
3441                 unlock_rsb(r);
3442                 put_rsb(r);
3443                 schedule();
3444         }
3445 }
3446
3447 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
3448                                          uint32_t remid)
3449 {
3450         struct dlm_lkb *lkb;
3451
3452         list_for_each_entry(lkb, head, lkb_statequeue) {
3453                 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
3454                         return lkb;
3455         }
3456         return NULL;
3457 }
3458
3459 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
3460                                     uint32_t remid)
3461 {
3462         struct dlm_lkb *lkb;
3463
3464         lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
3465         if (lkb)
3466                 return lkb;
3467         lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
3468         if (lkb)
3469                 return lkb;
3470         lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
3471         if (lkb)
3472                 return lkb;
3473         return NULL;
3474 }
3475
3476 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3477                                   struct dlm_rsb *r, struct dlm_rcom *rc)
3478 {
3479         struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3480         int lvblen;
3481
3482         lkb->lkb_nodeid = rc->rc_header.h_nodeid;
3483         lkb->lkb_ownpid = rl->rl_ownpid;
3484         lkb->lkb_remid = rl->rl_lkid;
3485         lkb->lkb_exflags = rl->rl_exflags;
3486         lkb->lkb_flags = rl->rl_flags & 0x0000FFFF;
3487         lkb->lkb_flags |= DLM_IFL_MSTCPY;
3488         lkb->lkb_lvbseq = rl->rl_lvbseq;
3489         lkb->lkb_rqmode = rl->rl_rqmode;
3490         lkb->lkb_grmode = rl->rl_grmode;
3491         /* don't set lkb_status because add_lkb wants to itself */
3492
3493         lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
3494         lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
3495
3496         if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3497                 lkb->lkb_lvbptr = allocate_lvb(ls);
3498                 if (!lkb->lkb_lvbptr)
3499                         return -ENOMEM;
3500                 lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
3501                          sizeof(struct rcom_lock);
3502                 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
3503         }
3504
3505         /* Conversions between PR and CW (middle modes) need special handling.
3506            The real granted mode of these converting locks cannot be determined
3507            until all locks have been rebuilt on the rsb (recover_conversion) */
3508
3509         if (rl->rl_wait_type == DLM_MSG_CONVERT && middle_conversion(lkb)) {
3510                 rl->rl_status = DLM_LKSTS_CONVERT;
3511                 lkb->lkb_grmode = DLM_LOCK_IV;
3512                 rsb_set_flag(r, RSB_RECOVER_CONVERT);
3513         }
3514
3515         return 0;
3516 }
3517
3518 /* This lkb may have been recovered in a previous aborted recovery so we need
3519    to check if the rsb already has an lkb with the given remote nodeid/lkid.
3520    If so we just send back a standard reply.  If not, we create a new lkb with
3521    the given values and send back our lkid.  We send back our lkid by sending
3522    back the rcom_lock struct we got but with the remid field filled in. */
3523
3524 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3525 {
3526         struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3527         struct dlm_rsb *r;
3528         struct dlm_lkb *lkb;
3529         int error;
3530
3531         if (rl->rl_parent_lkid) {
3532                 error = -EOPNOTSUPP;
3533                 goto out;
3534         }
3535
3536         error = find_rsb(ls, rl->rl_name, rl->rl_namelen, R_MASTER, &r);
3537         if (error)
3538                 goto out;
3539
3540         lock_rsb(r);
3541
3542         lkb = search_remid(r, rc->rc_header.h_nodeid, rl->rl_lkid);
3543         if (lkb) {
3544                 error = -EEXIST;
3545                 goto out_remid;
3546         }
3547
3548         error = create_lkb(ls, &lkb);
3549         if (error)
3550                 goto out_unlock;
3551
3552         error = receive_rcom_lock_args(ls, lkb, r, rc);
3553         if (error) {
3554                 __put_lkb(ls, lkb);
3555                 goto out_unlock;
3556         }
3557
3558         attach_lkb(r, lkb);
3559         add_lkb(r, lkb, rl->rl_status);
3560         error = 0;
3561
3562  out_remid:
3563         /* this is the new value returned to the lock holder for
3564            saving in its process-copy lkb */
3565         rl->rl_remid = lkb->lkb_id;
3566
3567  out_unlock:
3568         unlock_rsb(r);
3569         put_rsb(r);
3570  out:
3571         if (error)
3572                 log_print("recover_master_copy %d %x", error, rl->rl_lkid);
3573         rl->rl_result = error;
3574         return error;
3575 }
3576
3577 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3578 {
3579         struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3580         struct dlm_rsb *r;
3581         struct dlm_lkb *lkb;
3582         int error;
3583
3584         error = find_lkb(ls, rl->rl_lkid, &lkb);
3585         if (error) {
3586                 log_error(ls, "recover_process_copy no lkid %x", rl->rl_lkid);
3587                 return error;
3588         }
3589
3590         DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3591
3592         error = rl->rl_result;
3593
3594         r = lkb->lkb_resource;
3595         hold_rsb(r);
3596         lock_rsb(r);
3597
3598         switch (error) {
3599         case -EBADR:
3600                 /* There's a chance the new master received our lock before
3601                    dlm_recover_master_reply(), this wouldn't happen if we did
3602                    a barrier between recover_masters and recover_locks. */
3603                 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
3604                           (unsigned long)r, r->res_name);
3605                 dlm_send_rcom_lock(r, lkb);
3606                 goto out;
3607         case -EEXIST:
3608                 log_debug(ls, "master copy exists %x", lkb->lkb_id);
3609                 /* fall through */
3610         case 0:
3611                 lkb->lkb_remid = rl->rl_remid;
3612                 break;
3613         default:
3614                 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
3615                           error, lkb->lkb_id);
3616         }
3617
3618         /* an ack for dlm_recover_locks() which waits for replies from
3619            all the locks it sends to new masters */
3620         dlm_recovered_lock(r);
3621  out:
3622         unlock_rsb(r);
3623         put_rsb(r);
3624         dlm_put_lkb(lkb);
3625
3626         return 0;
3627 }
3628
3629 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
3630                      int mode, uint32_t flags, void *name, unsigned int namelen,
3631                      uint32_t parent_lkid)
3632 {
3633         struct dlm_lkb *lkb;
3634         struct dlm_args args;
3635         int error;
3636
3637         lock_recovery(ls);
3638
3639         error = create_lkb(ls, &lkb);
3640         if (error) {
3641                 kfree(ua);
3642                 goto out;
3643         }
3644
3645         if (flags & DLM_LKF_VALBLK) {
3646                 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3647                 if (!ua->lksb.sb_lvbptr) {
3648                         kfree(ua);
3649                         __put_lkb(ls, lkb);
3650                         error = -ENOMEM;
3651                         goto out;
3652                 }
3653         }
3654
3655         /* After ua is attached to lkb it will be freed by free_lkb().
3656            When DLM_IFL_USER is set, the dlm knows that this is a userspace
3657            lock and that lkb_astparam is the dlm_user_args structure. */
3658
3659         error = set_lock_args(mode, &ua->lksb, flags, namelen, parent_lkid,
3660                               DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
3661         lkb->lkb_flags |= DLM_IFL_USER;
3662         ua->old_mode = DLM_LOCK_IV;
3663
3664         if (error) {
3665                 __put_lkb(ls, lkb);
3666                 goto out;
3667         }
3668
3669         error = request_lock(ls, lkb, name, namelen, &args);
3670
3671         switch (error) {
3672         case 0:
3673                 break;
3674         case -EINPROGRESS:
3675                 error = 0;
3676                 break;
3677         case -EAGAIN:
3678                 error = 0;
3679                 /* fall through */
3680         default:
3681                 __put_lkb(ls, lkb);
3682                 goto out;
3683         }
3684
3685         /* add this new lkb to the per-process list of locks */
3686         spin_lock(&ua->proc->locks_spin);
3687         kref_get(&lkb->lkb_ref);
3688         list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
3689         spin_unlock(&ua->proc->locks_spin);
3690  out:
3691         unlock_recovery(ls);
3692         return error;
3693 }
3694
3695 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3696                      int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
3697 {
3698         struct dlm_lkb *lkb;
3699         struct dlm_args args;
3700         struct dlm_user_args *ua;
3701         int error;
3702
3703         lock_recovery(ls);
3704
3705         error = find_lkb(ls, lkid, &lkb);
3706         if (error)
3707                 goto out;
3708
3709         /* user can change the params on its lock when it converts it, or
3710            add an lvb that didn't exist before */
3711
3712         ua = (struct dlm_user_args *)lkb->lkb_astparam;
3713
3714         if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
3715                 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3716                 if (!ua->lksb.sb_lvbptr) {
3717                         error = -ENOMEM;
3718                         goto out_put;
3719                 }
3720         }
3721         if (lvb_in && ua->lksb.sb_lvbptr)
3722                 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3723
3724         ua->castparam = ua_tmp->castparam;
3725         ua->castaddr = ua_tmp->castaddr;
3726         ua->bastparam = ua_tmp->bastparam;
3727         ua->bastaddr = ua_tmp->bastaddr;
3728         ua->user_lksb = ua_tmp->user_lksb;
3729         ua->old_mode = lkb->lkb_grmode;
3730
3731         error = set_lock_args(mode, &ua->lksb, flags, 0, 0, DLM_FAKE_USER_AST,
3732                               ua, DLM_FAKE_USER_AST, &args);
3733         if (error)
3734                 goto out_put;
3735
3736         error = convert_lock(ls, lkb, &args);
3737
3738         if (error == -EINPROGRESS || error == -EAGAIN)
3739                 error = 0;
3740  out_put:
3741         dlm_put_lkb(lkb);
3742  out:
3743         unlock_recovery(ls);
3744         kfree(ua_tmp);
3745         return error;
3746 }
3747
3748 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3749                     uint32_t flags, uint32_t lkid, char *lvb_in)
3750 {
3751         struct dlm_lkb *lkb;
3752         struct dlm_args args;
3753         struct dlm_user_args *ua;
3754         int error;
3755
3756         lock_recovery(ls);
3757
3758         error = find_lkb(ls, lkid, &lkb);
3759         if (error)
3760                 goto out;
3761
3762         ua = (struct dlm_user_args *)lkb->lkb_astparam;
3763
3764         if (lvb_in && ua->lksb.sb_lvbptr)
3765                 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3766         ua->castparam = ua_tmp->castparam;
3767         ua->user_lksb = ua_tmp->user_lksb;
3768
3769         error = set_unlock_args(flags, ua, &args);
3770         if (error)
3771                 goto out_put;
3772
3773         error = unlock_lock(ls, lkb, &args);
3774
3775         if (error == -DLM_EUNLOCK)
3776                 error = 0;
3777         if (error)
3778                 goto out_put;
3779
3780         spin_lock(&ua->proc->locks_spin);
3781         /* dlm_user_add_ast() may have already taken lkb off the proc list */
3782         if (!list_empty(&lkb->lkb_ownqueue))
3783                 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
3784         spin_unlock(&ua->proc->locks_spin);
3785  out_put:
3786         dlm_put_lkb(lkb);
3787  out:
3788         unlock_recovery(ls);
3789         return error;
3790 }
3791
3792 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3793                     uint32_t flags, uint32_t lkid)
3794 {
3795         struct dlm_lkb *lkb;
3796         struct dlm_args args;
3797         struct dlm_user_args *ua;
3798         int error;
3799
3800         lock_recovery(ls);
3801
3802         error = find_lkb(ls, lkid, &lkb);
3803         if (error)
3804                 goto out;
3805
3806         ua = (struct dlm_user_args *)lkb->lkb_astparam;
3807         ua->castparam = ua_tmp->castparam;
3808         ua->user_lksb = ua_tmp->user_lksb;
3809
3810         error = set_unlock_args(flags, ua, &args);
3811         if (error)
3812                 goto out_put;
3813
3814         error = cancel_lock(ls, lkb, &args);
3815
3816         if (error == -DLM_ECANCEL)
3817                 error = 0;
3818         if (error)
3819                 goto out_put;
3820
3821         /* this lkb was removed from the WAITING queue */
3822         if (lkb->lkb_grmode == DLM_LOCK_IV) {
3823                 spin_lock(&ua->proc->locks_spin);
3824                 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
3825                 spin_unlock(&ua->proc->locks_spin);
3826         }
3827  out_put:
3828         dlm_put_lkb(lkb);
3829  out:
3830         unlock_recovery(ls);
3831         return error;
3832 }
3833
3834 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3835 {
3836         struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3837
3838         if (ua->lksb.sb_lvbptr)
3839                 kfree(ua->lksb.sb_lvbptr);
3840         kfree(ua);
3841         lkb->lkb_astparam = (long)NULL;
3842
3843         /* TODO: propogate to master if needed */
3844         return 0;
3845 }
3846
3847 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
3848    Regardless of what rsb queue the lock is on, it's removed and freed. */
3849
3850 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3851 {
3852         struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3853         struct dlm_args args;
3854         int error;
3855
3856         /* FIXME: we need to handle the case where the lkb is in limbo
3857            while the rsb is being looked up, currently we assert in
3858            _unlock_lock/is_remote because rsb nodeid is -1. */
3859
3860         set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
3861
3862         error = unlock_lock(ls, lkb, &args);
3863         if (error == -DLM_EUNLOCK)
3864                 error = 0;
3865         return error;
3866 }
3867
3868 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
3869    1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
3870    which we clear here. */
3871
3872 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
3873    list, and no more device_writes should add lkb's to proc->locks list; so we
3874    shouldn't need to take asts_spin or locks_spin here.  this assumes that
3875    device reads/writes/closes are serialized -- FIXME: we may need to serialize
3876    them ourself. */
3877
3878 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
3879 {
3880         struct dlm_lkb *lkb, *safe;
3881
3882         lock_recovery(ls);
3883         mutex_lock(&ls->ls_clear_proc_locks);
3884
3885         list_for_each_entry_safe(lkb, safe, &proc->locks, lkb_ownqueue) {
3886                 list_del_init(&lkb->lkb_ownqueue);
3887
3888                 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT) {
3889                         lkb->lkb_flags |= DLM_IFL_ORPHAN;
3890                         orphan_proc_lock(ls, lkb);
3891                 } else {
3892                         lkb->lkb_flags |= DLM_IFL_DEAD;
3893                         unlock_proc_lock(ls, lkb);
3894                 }
3895
3896                 /* this removes the reference for the proc->locks list
3897                    added by dlm_user_request, it may result in the lkb
3898                    being freed */
3899
3900                 dlm_put_lkb(lkb);
3901         }
3902
3903         /* in-progress unlocks */
3904         list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
3905                 list_del_init(&lkb->lkb_ownqueue);
3906                 lkb->lkb_flags |= DLM_IFL_DEAD;
3907                 dlm_put_lkb(lkb);
3908         }
3909
3910         list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
3911                 list_del(&lkb->lkb_astqueue);
3912                 dlm_put_lkb(lkb);
3913         }
3914
3915         mutex_unlock(&ls->ls_clear_proc_locks);
3916         unlock_recovery(ls);
3917 }
3918