err.no Git - linux-2.6/blob - fs/dlm/lock.c

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