root/fs/dlm/lock.c

/* [<][>][^][v][top][bottom][index][help] */

DEFINITIONS

This source file includes following definitions.
  1. dlm_modes_compat
  2. dlm_print_lkb
  3. dlm_print_rsb
  4. dlm_dump_rsb
  5. dlm_lock_recovery
  6. dlm_unlock_recovery
  7. dlm_lock_recovery_try
  8. can_be_queued
  9. force_blocking_asts
  10. is_demoted
  11. is_altmode
  12. is_granted
  13. is_remote
  14. is_process_copy
  15. is_master_copy
  16. middle_conversion
  17. down_conversion
  18. is_overlap_unlock
  19. is_overlap_cancel
  20. is_overlap
  21. queue_cast
  22. queue_cast_overlap
  23. queue_bast
  24. hold_rsb
  25. dlm_hold_rsb
  26. put_rsb
  27. dlm_put_rsb
  28. pre_rsb_struct
  29. get_rsb_struct
  30. rsb_cmp
  31. dlm_search_rsb_tree
  32. rsb_insert
  33. find_rsb_dir
  34. find_rsb_nodir
  35. find_rsb
  36. validate_master_nodeid
  37. dlm_master_lookup
  38. dlm_dump_rsb_hash
  39. dlm_dump_rsb_name
  40. toss_rsb
  41. unhold_rsb
  42. kill_rsb
  43. attach_lkb
  44. detach_lkb
  45. create_lkb
  46. find_lkb
  47. kill_lkb
  48. __put_lkb
  49. dlm_put_lkb
  50. hold_lkb
  51. unhold_lkb
  52. lkb_add_ordered
  53. add_lkb
  54. del_lkb
  55. move_lkb
  56. msg_reply_type
  57. nodeid_warned
  58. dlm_scan_waiters
  59. add_to_waiters
  60. _remove_from_waiters
  61. remove_from_waiters
  62. remove_from_waiters_ms
  63. wait_pending_remove
  64. shrink_bucket
  65. dlm_scan_rsbs
  66. add_timeout
  67. del_timeout
  68. dlm_scan_timeout
  69. dlm_adjust_timeouts
  70. set_lvb_lock
  71. set_lvb_unlock
  72. set_lvb_lock_pc
  73. _remove_lock
  74. remove_lock
  75. remove_lock_pc
  76. revert_lock
  77. revert_lock_pc
  78. _grant_lock
  79. grant_lock
  80. grant_lock_pc
  81. grant_lock_pending
  82. munge_demoted
  83. munge_altmode
  84. first_in_list
  85. queue_conflict
  86. conversion_deadlock_detect
  87. _can_be_granted
  88. can_be_granted
  89. grant_pending_convert
  90. grant_pending_wait
  91. lock_requires_bast
  92. grant_pending_locks
  93. modes_require_bast
  94. send_bast_queue
  95. send_blocking_asts
  96. send_blocking_asts_all
  97. set_master
  98. process_lookup_list
  99. confirm_master
  100. set_lock_args
  101. set_unlock_args
  102. validate_lock_args
  103. validate_unlock_args
  104. do_request
  105. do_request_effects
  106. do_convert
  107. do_convert_effects
  108. do_unlock
  109. do_unlock_effects
  110. do_cancel
  111. do_cancel_effects
  112. _request_lock
  113. _convert_lock
  114. _unlock_lock
  115. _cancel_lock
  116. request_lock
  117. convert_lock
  118. unlock_lock
  119. cancel_lock
  120. dlm_lock
  121. dlm_unlock
  122. _create_message
  123. create_message
  124. send_message
  125. send_args
  126. send_common
  127. send_request
  128. send_convert
  129. send_unlock
  130. send_cancel
  131. send_grant
  132. send_bast
  133. send_lookup
  134. send_remove
  135. send_common_reply
  136. send_request_reply
  137. send_convert_reply
  138. send_unlock_reply
  139. send_cancel_reply
  140. send_lookup_reply
  141. receive_flags
  142. receive_flags_reply
  143. receive_extralen
  144. receive_lvb
  145. fake_bastfn
  146. fake_astfn
  147. receive_request_args
  148. receive_convert_args
  149. receive_unlock_args
  150. setup_stub_lkb
  151. validate_message
  152. send_repeat_remove
  153. receive_request
  154. receive_convert
  155. receive_unlock
  156. receive_cancel
  157. receive_grant
  158. receive_bast
  159. receive_lookup
  160. receive_remove
  161. receive_purge
  162. receive_request_reply
  163. __receive_convert_reply
  164. _receive_convert_reply
  165. receive_convert_reply
  166. _receive_unlock_reply
  167. receive_unlock_reply
  168. _receive_cancel_reply
  169. receive_cancel_reply
  170. receive_lookup_reply
  171. _receive_message
  172. dlm_receive_message
  173. dlm_receive_message_saved
  174. dlm_receive_buffer
  175. recover_convert_waiter
  176. waiter_needs_recovery
  177. dlm_recover_waiters_pre
  178. find_resend_waiter
  179. dlm_recover_waiters_post
  180. purge_mstcpy_list
  181. dlm_purge_mstcpy_locks
  182. purge_dead_list
  183. dlm_recover_purge
  184. find_grant_rsb
  185. dlm_recover_grant
  186. search_remid_list
  187. search_remid
  188. receive_rcom_lock_args
  189. dlm_recover_master_copy
  190. dlm_recover_process_copy
  191. dlm_user_request
  192. dlm_user_convert
  193. dlm_user_adopt_orphan
  194. dlm_user_unlock
  195. dlm_user_cancel
  196. dlm_user_deadlock
  197. orphan_proc_lock
  198. unlock_proc_lock
  199. del_proc_lock
  200. dlm_clear_proc_locks
  201. purge_proc_locks
  202. do_purge
  203. send_purge
  204. dlm_user_purge

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

/* [<][>][^][v][top][bottom][index][help] */