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