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