blob: 01e7d39c5fba8d55a4fa1a4bd8f64e13d4c32a95 [file] [log] [blame]
David Teiglande7fd4172006-01-18 09:30:29 +00001/******************************************************************************
2*******************************************************************************
3**
David Teigland46b43ee2008-01-08 16:24:00 -06004** Copyright (C) 2005-2008 Red Hat, Inc. All rights reserved.
David Teiglande7fd4172006-01-18 09:30:29 +00005**
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*/
David Teigland597d0ca2006-07-12 16:44:04 -050058#include <linux/types.h>
David Teiglande7fd4172006-01-18 09:30:29 +000059#include "dlm_internal.h"
David Teigland597d0ca2006-07-12 16:44:04 -050060#include <linux/dlm_device.h>
David Teiglande7fd4172006-01-18 09:30:29 +000061#include "memory.h"
62#include "lowcomms.h"
63#include "requestqueue.h"
64#include "util.h"
65#include "dir.h"
66#include "member.h"
67#include "lockspace.h"
68#include "ast.h"
69#include "lock.h"
70#include "rcom.h"
71#include "recover.h"
72#include "lvb_table.h"
David Teigland597d0ca2006-07-12 16:44:04 -050073#include "user.h"
David Teiglande7fd4172006-01-18 09:30:29 +000074#include "config.h"
75
76static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
77static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
78static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
79static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
80static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
81static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
82static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
83static int send_remove(struct dlm_rsb *r);
84static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
David Teigland3ae1acf2007-05-18 08:59:31 -050085static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
David Teiglande7fd4172006-01-18 09:30:29 +000086static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
87 struct dlm_message *ms);
88static int receive_extralen(struct dlm_message *ms);
David Teigland84991372007-03-30 15:02:40 -050089static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
David Teigland3ae1acf2007-05-18 08:59:31 -050090static void del_timeout(struct dlm_lkb *lkb);
David Teiglande7fd4172006-01-18 09:30:29 +000091
92/*
93 * Lock compatibilty matrix - thanks Steve
94 * UN = Unlocked state. Not really a state, used as a flag
95 * PD = Padding. Used to make the matrix a nice power of two in size
96 * Other states are the same as the VMS DLM.
97 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
98 */
99
100static const int __dlm_compat_matrix[8][8] = {
101 /* UN NL CR CW PR PW EX PD */
102 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
104 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
105 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
106 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
107 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
108 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
109 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
110};
111
112/*
113 * This defines the direction of transfer of LVB data.
114 * Granted mode is the row; requested mode is the column.
115 * Usage: matrix[grmode+1][rqmode+1]
116 * 1 = LVB is returned to the caller
117 * 0 = LVB is written to the resource
118 * -1 = nothing happens to the LVB
119 */
120
121const int dlm_lvb_operations[8][8] = {
122 /* UN NL CR CW PR PW EX PD*/
123 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
124 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
125 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
126 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
127 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
128 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
129 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
130 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
131};
David Teiglande7fd4172006-01-18 09:30:29 +0000132
133#define modes_compat(gr, rq) \
134 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
135
136int dlm_modes_compat(int mode1, int mode2)
137{
138 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
139}
140
141/*
142 * Compatibility matrix for conversions with QUECVT set.
143 * Granted mode is the row; requested mode is the column.
144 * Usage: matrix[grmode+1][rqmode+1]
145 */
146
147static const int __quecvt_compat_matrix[8][8] = {
148 /* UN NL CR CW PR PW EX PD */
149 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
150 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
151 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
152 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
153 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
154 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
155 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
156 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
157};
158
David Teigland597d0ca2006-07-12 16:44:04 -0500159void dlm_print_lkb(struct dlm_lkb *lkb)
David Teiglande7fd4172006-01-18 09:30:29 +0000160{
161 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
162 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
163 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
164 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
165 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
166}
167
Adrian Bunk170e19a2008-02-13 23:29:38 +0200168static void dlm_print_rsb(struct dlm_rsb *r)
David Teiglande7fd4172006-01-18 09:30:29 +0000169{
170 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
171 r->res_nodeid, r->res_flags, r->res_first_lkid,
172 r->res_recover_locks_count, r->res_name);
173}
174
David Teiglanda345da32006-08-18 11:54:25 -0500175void dlm_dump_rsb(struct dlm_rsb *r)
176{
177 struct dlm_lkb *lkb;
178
179 dlm_print_rsb(r);
180
181 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
182 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
183 printk(KERN_ERR "rsb lookup list\n");
184 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
185 dlm_print_lkb(lkb);
186 printk(KERN_ERR "rsb grant queue:\n");
187 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
188 dlm_print_lkb(lkb);
189 printk(KERN_ERR "rsb convert queue:\n");
190 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
191 dlm_print_lkb(lkb);
192 printk(KERN_ERR "rsb wait queue:\n");
193 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
194 dlm_print_lkb(lkb);
195}
196
David Teiglande7fd4172006-01-18 09:30:29 +0000197/* Threads cannot use the lockspace while it's being recovered */
198
David Teigland85e86ed2007-05-18 08:58:15 -0500199static inline void dlm_lock_recovery(struct dlm_ls *ls)
David Teiglande7fd4172006-01-18 09:30:29 +0000200{
201 down_read(&ls->ls_in_recovery);
202}
203
David Teigland85e86ed2007-05-18 08:58:15 -0500204void dlm_unlock_recovery(struct dlm_ls *ls)
David Teiglande7fd4172006-01-18 09:30:29 +0000205{
206 up_read(&ls->ls_in_recovery);
207}
208
David Teigland85e86ed2007-05-18 08:58:15 -0500209int dlm_lock_recovery_try(struct dlm_ls *ls)
David Teiglande7fd4172006-01-18 09:30:29 +0000210{
211 return down_read_trylock(&ls->ls_in_recovery);
212}
213
214static inline int can_be_queued(struct dlm_lkb *lkb)
215{
216 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
217}
218
219static inline int force_blocking_asts(struct dlm_lkb *lkb)
220{
221 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
222}
223
224static inline int is_demoted(struct dlm_lkb *lkb)
225{
226 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
227}
228
David Teigland7d3c1fe2007-04-19 10:30:41 -0500229static inline int is_altmode(struct dlm_lkb *lkb)
230{
231 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
232}
233
234static inline int is_granted(struct dlm_lkb *lkb)
235{
236 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
237}
238
David Teiglande7fd4172006-01-18 09:30:29 +0000239static inline int is_remote(struct dlm_rsb *r)
240{
241 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
242 return !!r->res_nodeid;
243}
244
245static inline int is_process_copy(struct dlm_lkb *lkb)
246{
247 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
248}
249
250static inline int is_master_copy(struct dlm_lkb *lkb)
251{
252 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
253 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
David Teigland90135922006-01-20 08:47:07 +0000254 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000255}
256
257static inline int middle_conversion(struct dlm_lkb *lkb)
258{
259 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
260 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
David Teigland90135922006-01-20 08:47:07 +0000261 return 1;
262 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000263}
264
265static inline int down_conversion(struct dlm_lkb *lkb)
266{
267 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
268}
269
David Teiglandef0c2bb2007-03-28 09:56:46 -0500270static inline int is_overlap_unlock(struct dlm_lkb *lkb)
271{
272 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
273}
274
275static inline int is_overlap_cancel(struct dlm_lkb *lkb)
276{
277 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
278}
279
280static inline int is_overlap(struct dlm_lkb *lkb)
281{
282 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
283 DLM_IFL_OVERLAP_CANCEL));
284}
285
David Teiglande7fd4172006-01-18 09:30:29 +0000286static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
287{
288 if (is_master_copy(lkb))
289 return;
290
David Teigland3ae1acf2007-05-18 08:59:31 -0500291 del_timeout(lkb);
292
David Teiglande7fd4172006-01-18 09:30:29 +0000293 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
294
David Teigland3ae1acf2007-05-18 08:59:31 -0500295 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
296 timeout caused the cancel then return -ETIMEDOUT */
297 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
298 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
299 rv = -ETIMEDOUT;
300 }
301
David Teigland8b4021f2007-05-29 08:46:00 -0500302 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
303 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
304 rv = -EDEADLK;
305 }
306
David Teiglande7fd4172006-01-18 09:30:29 +0000307 lkb->lkb_lksb->sb_status = rv;
308 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
309
David Teiglandfd22a512008-12-09 11:55:46 -0600310 dlm_add_ast(lkb, AST_COMP, 0);
David Teiglande7fd4172006-01-18 09:30:29 +0000311}
312
David Teiglandef0c2bb2007-03-28 09:56:46 -0500313static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
314{
315 queue_cast(r, lkb,
316 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
317}
318
David Teiglande7fd4172006-01-18 09:30:29 +0000319static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
320{
David Teiglande3a84ad2008-12-09 14:47:29 -0600321 lkb->lkb_time_bast = ktime_get();
322
David Teiglande7fd4172006-01-18 09:30:29 +0000323 if (is_master_copy(lkb))
324 send_bast(r, lkb, rqmode);
David Teiglandfd22a512008-12-09 11:55:46 -0600325 else
326 dlm_add_ast(lkb, AST_BAST, rqmode);
David Teiglande7fd4172006-01-18 09:30:29 +0000327}
328
329/*
330 * Basic operations on rsb's and lkb's
331 */
332
333static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
334{
335 struct dlm_rsb *r;
336
David Teigland52bda2b2007-11-07 09:06:49 -0600337 r = dlm_allocate_rsb(ls, len);
David Teiglande7fd4172006-01-18 09:30:29 +0000338 if (!r)
339 return NULL;
340
341 r->res_ls = ls;
342 r->res_length = len;
343 memcpy(r->res_name, name, len);
David Teigland90135922006-01-20 08:47:07 +0000344 mutex_init(&r->res_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +0000345
346 INIT_LIST_HEAD(&r->res_lookup);
347 INIT_LIST_HEAD(&r->res_grantqueue);
348 INIT_LIST_HEAD(&r->res_convertqueue);
349 INIT_LIST_HEAD(&r->res_waitqueue);
350 INIT_LIST_HEAD(&r->res_root_list);
351 INIT_LIST_HEAD(&r->res_recover_list);
352
353 return r;
354}
355
356static int search_rsb_list(struct list_head *head, char *name, int len,
357 unsigned int flags, struct dlm_rsb **r_ret)
358{
359 struct dlm_rsb *r;
360 int error = 0;
361
362 list_for_each_entry(r, head, res_hashchain) {
363 if (len == r->res_length && !memcmp(name, r->res_name, len))
364 goto found;
365 }
Benny Halevy18c60c02008-06-30 19:59:14 +0300366 *r_ret = NULL;
David Teigland597d0ca2006-07-12 16:44:04 -0500367 return -EBADR;
David Teiglande7fd4172006-01-18 09:30:29 +0000368
369 found:
370 if (r->res_nodeid && (flags & R_MASTER))
371 error = -ENOTBLK;
372 *r_ret = r;
373 return error;
374}
375
376static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
377 unsigned int flags, struct dlm_rsb **r_ret)
378{
379 struct dlm_rsb *r;
380 int error;
381
382 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
383 if (!error) {
384 kref_get(&r->res_ref);
385 goto out;
386 }
387 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
388 if (error)
389 goto out;
390
391 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
392
393 if (dlm_no_directory(ls))
394 goto out;
395
396 if (r->res_nodeid == -1) {
397 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
398 r->res_first_lkid = 0;
399 } else if (r->res_nodeid > 0) {
400 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
401 r->res_first_lkid = 0;
402 } else {
403 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
404 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
405 }
406 out:
407 *r_ret = r;
408 return error;
409}
410
411static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
412 unsigned int flags, struct dlm_rsb **r_ret)
413{
414 int error;
David Teiglandc7be7612009-01-07 16:50:41 -0600415 spin_lock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000416 error = _search_rsb(ls, name, len, b, flags, r_ret);
David Teiglandc7be7612009-01-07 16:50:41 -0600417 spin_unlock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000418 return error;
419}
420
421/*
422 * Find rsb in rsbtbl and potentially create/add one
423 *
424 * Delaying the release of rsb's has a similar benefit to applications keeping
425 * NL locks on an rsb, but without the guarantee that the cached master value
426 * will still be valid when the rsb is reused. Apps aren't always smart enough
427 * to keep NL locks on an rsb that they may lock again shortly; this can lead
428 * to excessive master lookups and removals if we don't delay the release.
429 *
430 * Searching for an rsb means looking through both the normal list and toss
431 * list. When found on the toss list the rsb is moved to the normal list with
432 * ref count of 1; when found on normal list the ref count is incremented.
433 */
434
435static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
436 unsigned int flags, struct dlm_rsb **r_ret)
437{
438 struct dlm_rsb *r, *tmp;
439 uint32_t hash, bucket;
Al Viroef58bcc2008-01-25 23:22:26 -0500440 int error = -EINVAL;
441
442 if (namelen > DLM_RESNAME_MAXLEN)
443 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +0000444
445 if (dlm_no_directory(ls))
446 flags |= R_CREATE;
447
Al Viroef58bcc2008-01-25 23:22:26 -0500448 error = 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000449 hash = jhash(name, namelen, 0);
450 bucket = hash & (ls->ls_rsbtbl_size - 1);
451
452 error = search_rsb(ls, name, namelen, bucket, flags, &r);
453 if (!error)
454 goto out;
455
David Teigland597d0ca2006-07-12 16:44:04 -0500456 if (error == -EBADR && !(flags & R_CREATE))
David Teiglande7fd4172006-01-18 09:30:29 +0000457 goto out;
458
459 /* the rsb was found but wasn't a master copy */
460 if (error == -ENOTBLK)
461 goto out;
462
463 error = -ENOMEM;
464 r = create_rsb(ls, name, namelen);
465 if (!r)
466 goto out;
467
468 r->res_hash = hash;
469 r->res_bucket = bucket;
470 r->res_nodeid = -1;
471 kref_init(&r->res_ref);
472
473 /* With no directory, the master can be set immediately */
474 if (dlm_no_directory(ls)) {
475 int nodeid = dlm_dir_nodeid(r);
476 if (nodeid == dlm_our_nodeid())
477 nodeid = 0;
478 r->res_nodeid = nodeid;
479 }
480
David Teiglandc7be7612009-01-07 16:50:41 -0600481 spin_lock(&ls->ls_rsbtbl[bucket].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000482 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
483 if (!error) {
David Teiglandc7be7612009-01-07 16:50:41 -0600484 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
David Teigland52bda2b2007-11-07 09:06:49 -0600485 dlm_free_rsb(r);
David Teiglande7fd4172006-01-18 09:30:29 +0000486 r = tmp;
487 goto out;
488 }
489 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
David Teiglandc7be7612009-01-07 16:50:41 -0600490 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000491 error = 0;
492 out:
493 *r_ret = r;
494 return error;
495}
496
David Teiglande7fd4172006-01-18 09:30:29 +0000497/* This is only called to add a reference when the code already holds
498 a valid reference to the rsb, so there's no need for locking. */
499
500static inline void hold_rsb(struct dlm_rsb *r)
501{
502 kref_get(&r->res_ref);
503}
504
505void dlm_hold_rsb(struct dlm_rsb *r)
506{
507 hold_rsb(r);
508}
509
510static void toss_rsb(struct kref *kref)
511{
512 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
513 struct dlm_ls *ls = r->res_ls;
514
515 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
516 kref_init(&r->res_ref);
517 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
518 r->res_toss_time = jiffies;
519 if (r->res_lvbptr) {
David Teigland52bda2b2007-11-07 09:06:49 -0600520 dlm_free_lvb(r->res_lvbptr);
David Teiglande7fd4172006-01-18 09:30:29 +0000521 r->res_lvbptr = NULL;
522 }
523}
524
525/* When all references to the rsb are gone it's transfered to
526 the tossed list for later disposal. */
527
528static void put_rsb(struct dlm_rsb *r)
529{
530 struct dlm_ls *ls = r->res_ls;
531 uint32_t bucket = r->res_bucket;
532
David Teiglandc7be7612009-01-07 16:50:41 -0600533 spin_lock(&ls->ls_rsbtbl[bucket].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000534 kref_put(&r->res_ref, toss_rsb);
David Teiglandc7be7612009-01-07 16:50:41 -0600535 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000536}
537
538void dlm_put_rsb(struct dlm_rsb *r)
539{
540 put_rsb(r);
541}
542
543/* See comment for unhold_lkb */
544
545static void unhold_rsb(struct dlm_rsb *r)
546{
547 int rv;
548 rv = kref_put(&r->res_ref, toss_rsb);
David Teiglanda345da32006-08-18 11:54:25 -0500549 DLM_ASSERT(!rv, dlm_dump_rsb(r););
David Teiglande7fd4172006-01-18 09:30:29 +0000550}
551
552static void kill_rsb(struct kref *kref)
553{
554 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
555
556 /* All work is done after the return from kref_put() so we
557 can release the write_lock before the remove and free. */
558
David Teiglanda345da32006-08-18 11:54:25 -0500559 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
560 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
561 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
562 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
563 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
564 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
David Teiglande7fd4172006-01-18 09:30:29 +0000565}
566
567/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
568 The rsb must exist as long as any lkb's for it do. */
569
570static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
571{
572 hold_rsb(r);
573 lkb->lkb_resource = r;
574}
575
576static void detach_lkb(struct dlm_lkb *lkb)
577{
578 if (lkb->lkb_resource) {
579 put_rsb(lkb->lkb_resource);
580 lkb->lkb_resource = NULL;
581 }
582}
583
584static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
585{
586 struct dlm_lkb *lkb, *tmp;
587 uint32_t lkid = 0;
588 uint16_t bucket;
589
David Teigland52bda2b2007-11-07 09:06:49 -0600590 lkb = dlm_allocate_lkb(ls);
David Teiglande7fd4172006-01-18 09:30:29 +0000591 if (!lkb)
592 return -ENOMEM;
593
594 lkb->lkb_nodeid = -1;
595 lkb->lkb_grmode = DLM_LOCK_IV;
596 kref_init(&lkb->lkb_ref);
David Teigland34e22be2006-07-18 11:24:04 -0500597 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
David Teiglandef0c2bb2007-03-28 09:56:46 -0500598 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
David Teigland3ae1acf2007-05-18 08:59:31 -0500599 INIT_LIST_HEAD(&lkb->lkb_time_list);
David Teiglande7fd4172006-01-18 09:30:29 +0000600
601 get_random_bytes(&bucket, sizeof(bucket));
602 bucket &= (ls->ls_lkbtbl_size - 1);
603
604 write_lock(&ls->ls_lkbtbl[bucket].lock);
605
606 /* counter can roll over so we must verify lkid is not in use */
607
608 while (lkid == 0) {
David Teiglandce03f122007-04-02 12:12:55 -0500609 lkid = (bucket << 16) | ls->ls_lkbtbl[bucket].counter++;
David Teiglande7fd4172006-01-18 09:30:29 +0000610
611 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
612 lkb_idtbl_list) {
613 if (tmp->lkb_id != lkid)
614 continue;
615 lkid = 0;
616 break;
617 }
618 }
619
620 lkb->lkb_id = lkid;
621 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
622 write_unlock(&ls->ls_lkbtbl[bucket].lock);
623
624 *lkb_ret = lkb;
625 return 0;
626}
627
628static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
629{
David Teiglande7fd4172006-01-18 09:30:29 +0000630 struct dlm_lkb *lkb;
David Teiglandce03f122007-04-02 12:12:55 -0500631 uint16_t bucket = (lkid >> 16);
David Teiglande7fd4172006-01-18 09:30:29 +0000632
633 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
634 if (lkb->lkb_id == lkid)
635 return lkb;
636 }
637 return NULL;
638}
639
640static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
641{
642 struct dlm_lkb *lkb;
David Teiglandce03f122007-04-02 12:12:55 -0500643 uint16_t bucket = (lkid >> 16);
David Teiglande7fd4172006-01-18 09:30:29 +0000644
645 if (bucket >= ls->ls_lkbtbl_size)
646 return -EBADSLT;
647
648 read_lock(&ls->ls_lkbtbl[bucket].lock);
649 lkb = __find_lkb(ls, lkid);
650 if (lkb)
651 kref_get(&lkb->lkb_ref);
652 read_unlock(&ls->ls_lkbtbl[bucket].lock);
653
654 *lkb_ret = lkb;
655 return lkb ? 0 : -ENOENT;
656}
657
658static void kill_lkb(struct kref *kref)
659{
660 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
661
662 /* All work is done after the return from kref_put() so we
663 can release the write_lock before the detach_lkb */
664
665 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
666}
667
David Teiglandb3f58d82006-02-28 11:16:37 -0500668/* __put_lkb() is used when an lkb may not have an rsb attached to
669 it so we need to provide the lockspace explicitly */
670
671static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
David Teiglande7fd4172006-01-18 09:30:29 +0000672{
David Teiglandce03f122007-04-02 12:12:55 -0500673 uint16_t bucket = (lkb->lkb_id >> 16);
David Teiglande7fd4172006-01-18 09:30:29 +0000674
675 write_lock(&ls->ls_lkbtbl[bucket].lock);
676 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
677 list_del(&lkb->lkb_idtbl_list);
678 write_unlock(&ls->ls_lkbtbl[bucket].lock);
679
680 detach_lkb(lkb);
681
682 /* for local/process lkbs, lvbptr points to caller's lksb */
683 if (lkb->lkb_lvbptr && is_master_copy(lkb))
David Teigland52bda2b2007-11-07 09:06:49 -0600684 dlm_free_lvb(lkb->lkb_lvbptr);
685 dlm_free_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +0000686 return 1;
687 } else {
688 write_unlock(&ls->ls_lkbtbl[bucket].lock);
689 return 0;
690 }
691}
692
693int dlm_put_lkb(struct dlm_lkb *lkb)
694{
David Teiglandb3f58d82006-02-28 11:16:37 -0500695 struct dlm_ls *ls;
696
697 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
698 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
699
700 ls = lkb->lkb_resource->res_ls;
701 return __put_lkb(ls, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +0000702}
703
704/* This is only called to add a reference when the code already holds
705 a valid reference to the lkb, so there's no need for locking. */
706
707static inline void hold_lkb(struct dlm_lkb *lkb)
708{
709 kref_get(&lkb->lkb_ref);
710}
711
712/* This is called when we need to remove a reference and are certain
713 it's not the last ref. e.g. del_lkb is always called between a
714 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
715 put_lkb would work fine, but would involve unnecessary locking */
716
717static inline void unhold_lkb(struct dlm_lkb *lkb)
718{
719 int rv;
720 rv = kref_put(&lkb->lkb_ref, kill_lkb);
721 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
722}
723
724static void lkb_add_ordered(struct list_head *new, struct list_head *head,
725 int mode)
726{
727 struct dlm_lkb *lkb = NULL;
728
729 list_for_each_entry(lkb, head, lkb_statequeue)
730 if (lkb->lkb_rqmode < mode)
731 break;
732
733 if (!lkb)
734 list_add_tail(new, head);
735 else
736 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
737}
738
739/* add/remove lkb to rsb's grant/convert/wait queue */
740
741static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
742{
743 kref_get(&lkb->lkb_ref);
744
745 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
746
David Teiglandeeda4182008-12-09 14:12:21 -0600747 lkb->lkb_timestamp = ktime_get();
748
David Teiglande7fd4172006-01-18 09:30:29 +0000749 lkb->lkb_status = status;
750
751 switch (status) {
752 case DLM_LKSTS_WAITING:
753 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
754 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
755 else
756 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
757 break;
758 case DLM_LKSTS_GRANTED:
759 /* convention says granted locks kept in order of grmode */
760 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
761 lkb->lkb_grmode);
762 break;
763 case DLM_LKSTS_CONVERT:
764 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
765 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
766 else
767 list_add_tail(&lkb->lkb_statequeue,
768 &r->res_convertqueue);
769 break;
770 default:
771 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
772 }
773}
774
775static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
776{
777 lkb->lkb_status = 0;
778 list_del(&lkb->lkb_statequeue);
779 unhold_lkb(lkb);
780}
781
782static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
783{
784 hold_lkb(lkb);
785 del_lkb(r, lkb);
786 add_lkb(r, lkb, sts);
787 unhold_lkb(lkb);
788}
789
David Teiglandef0c2bb2007-03-28 09:56:46 -0500790static int msg_reply_type(int mstype)
791{
792 switch (mstype) {
793 case DLM_MSG_REQUEST:
794 return DLM_MSG_REQUEST_REPLY;
795 case DLM_MSG_CONVERT:
796 return DLM_MSG_CONVERT_REPLY;
797 case DLM_MSG_UNLOCK:
798 return DLM_MSG_UNLOCK_REPLY;
799 case DLM_MSG_CANCEL:
800 return DLM_MSG_CANCEL_REPLY;
801 case DLM_MSG_LOOKUP:
802 return DLM_MSG_LOOKUP_REPLY;
803 }
804 return -1;
805}
806
David Teiglande7fd4172006-01-18 09:30:29 +0000807/* add/remove lkb from global waiters list of lkb's waiting for
808 a reply from a remote node */
809
David Teiglandef0c2bb2007-03-28 09:56:46 -0500810static int add_to_waiters(struct dlm_lkb *lkb, int mstype)
David Teiglande7fd4172006-01-18 09:30:29 +0000811{
812 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
David Teiglandef0c2bb2007-03-28 09:56:46 -0500813 int error = 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000814
David Teigland90135922006-01-20 08:47:07 +0000815 mutex_lock(&ls->ls_waiters_mutex);
David Teiglandef0c2bb2007-03-28 09:56:46 -0500816
817 if (is_overlap_unlock(lkb) ||
818 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
819 error = -EINVAL;
David Teiglande7fd4172006-01-18 09:30:29 +0000820 goto out;
821 }
David Teiglandef0c2bb2007-03-28 09:56:46 -0500822
823 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
824 switch (mstype) {
825 case DLM_MSG_UNLOCK:
826 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
827 break;
828 case DLM_MSG_CANCEL:
829 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
830 break;
831 default:
832 error = -EBUSY;
833 goto out;
834 }
835 lkb->lkb_wait_count++;
836 hold_lkb(lkb);
837
838 log_debug(ls, "add overlap %x cur %d new %d count %d flags %x",
839 lkb->lkb_id, lkb->lkb_wait_type, mstype,
840 lkb->lkb_wait_count, lkb->lkb_flags);
841 goto out;
842 }
843
844 DLM_ASSERT(!lkb->lkb_wait_count,
845 dlm_print_lkb(lkb);
846 printk("wait_count %d\n", lkb->lkb_wait_count););
847
848 lkb->lkb_wait_count++;
David Teiglande7fd4172006-01-18 09:30:29 +0000849 lkb->lkb_wait_type = mstype;
David Teiglandef0c2bb2007-03-28 09:56:46 -0500850 hold_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +0000851 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
852 out:
David Teiglandef0c2bb2007-03-28 09:56:46 -0500853 if (error)
854 log_error(ls, "add_to_waiters %x error %d flags %x %d %d %s",
855 lkb->lkb_id, error, lkb->lkb_flags, mstype,
856 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
David Teigland90135922006-01-20 08:47:07 +0000857 mutex_unlock(&ls->ls_waiters_mutex);
David Teiglandef0c2bb2007-03-28 09:56:46 -0500858 return error;
David Teiglande7fd4172006-01-18 09:30:29 +0000859}
860
David Teiglandb790c3b2007-01-24 10:21:33 -0600861/* We clear the RESEND flag because we might be taking an lkb off the waiters
862 list as part of process_requestqueue (e.g. a lookup that has an optimized
863 request reply on the requestqueue) between dlm_recover_waiters_pre() which
864 set RESEND and dlm_recover_waiters_post() */
865
David Teiglandef0c2bb2007-03-28 09:56:46 -0500866static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype)
David Teiglande7fd4172006-01-18 09:30:29 +0000867{
David Teiglandef0c2bb2007-03-28 09:56:46 -0500868 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
869 int overlap_done = 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000870
David Teiglandef0c2bb2007-03-28 09:56:46 -0500871 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
872 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
873 overlap_done = 1;
874 goto out_del;
David Teiglande7fd4172006-01-18 09:30:29 +0000875 }
David Teiglandef0c2bb2007-03-28 09:56:46 -0500876
877 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
878 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
879 overlap_done = 1;
880 goto out_del;
881 }
882
883 /* N.B. type of reply may not always correspond to type of original
884 msg due to lookup->request optimization, verify others? */
885
886 if (lkb->lkb_wait_type) {
887 lkb->lkb_wait_type = 0;
888 goto out_del;
889 }
890
891 log_error(ls, "remove_from_waiters lkid %x flags %x types %d %d",
892 lkb->lkb_id, lkb->lkb_flags, mstype, lkb->lkb_wait_type);
893 return -1;
894
895 out_del:
896 /* the force-unlock/cancel has completed and we haven't recvd a reply
897 to the op that was in progress prior to the unlock/cancel; we
898 give up on any reply to the earlier op. FIXME: not sure when/how
899 this would happen */
900
901 if (overlap_done && lkb->lkb_wait_type) {
902 log_error(ls, "remove_from_waiters %x reply %d give up on %d",
903 lkb->lkb_id, mstype, lkb->lkb_wait_type);
904 lkb->lkb_wait_count--;
905 lkb->lkb_wait_type = 0;
906 }
907
908 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
909
David Teiglandb790c3b2007-01-24 10:21:33 -0600910 lkb->lkb_flags &= ~DLM_IFL_RESEND;
David Teiglandef0c2bb2007-03-28 09:56:46 -0500911 lkb->lkb_wait_count--;
912 if (!lkb->lkb_wait_count)
913 list_del_init(&lkb->lkb_wait_reply);
David Teiglande7fd4172006-01-18 09:30:29 +0000914 unhold_lkb(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -0500915 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +0000916}
917
David Teiglandef0c2bb2007-03-28 09:56:46 -0500918static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
David Teiglande7fd4172006-01-18 09:30:29 +0000919{
920 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
921 int error;
922
David Teigland90135922006-01-20 08:47:07 +0000923 mutex_lock(&ls->ls_waiters_mutex);
David Teiglandef0c2bb2007-03-28 09:56:46 -0500924 error = _remove_from_waiters(lkb, mstype);
David Teigland90135922006-01-20 08:47:07 +0000925 mutex_unlock(&ls->ls_waiters_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +0000926 return error;
927}
928
David Teiglandef0c2bb2007-03-28 09:56:46 -0500929/* Handles situations where we might be processing a "fake" or "stub" reply in
930 which we can't try to take waiters_mutex again. */
931
932static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
933{
934 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
935 int error;
936
937 if (ms != &ls->ls_stub_ms)
938 mutex_lock(&ls->ls_waiters_mutex);
939 error = _remove_from_waiters(lkb, ms->m_type);
940 if (ms != &ls->ls_stub_ms)
941 mutex_unlock(&ls->ls_waiters_mutex);
942 return error;
943}
944
David Teiglande7fd4172006-01-18 09:30:29 +0000945static void dir_remove(struct dlm_rsb *r)
946{
947 int to_nodeid;
948
949 if (dlm_no_directory(r->res_ls))
950 return;
951
952 to_nodeid = dlm_dir_nodeid(r);
953 if (to_nodeid != dlm_our_nodeid())
954 send_remove(r);
955 else
956 dlm_dir_remove_entry(r->res_ls, to_nodeid,
957 r->res_name, r->res_length);
958}
959
960/* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
961 found since they are in order of newest to oldest? */
962
963static int shrink_bucket(struct dlm_ls *ls, int b)
964{
965 struct dlm_rsb *r;
966 int count = 0, found;
967
968 for (;;) {
David Teigland90135922006-01-20 08:47:07 +0000969 found = 0;
David Teiglandc7be7612009-01-07 16:50:41 -0600970 spin_lock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000971 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
972 res_hashchain) {
973 if (!time_after_eq(jiffies, r->res_toss_time +
David Teigland68c817a2007-01-09 09:41:48 -0600974 dlm_config.ci_toss_secs * HZ))
David Teiglande7fd4172006-01-18 09:30:29 +0000975 continue;
David Teigland90135922006-01-20 08:47:07 +0000976 found = 1;
David Teiglande7fd4172006-01-18 09:30:29 +0000977 break;
978 }
979
980 if (!found) {
David Teiglandc7be7612009-01-07 16:50:41 -0600981 spin_unlock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000982 break;
983 }
984
985 if (kref_put(&r->res_ref, kill_rsb)) {
986 list_del(&r->res_hashchain);
David Teiglandc7be7612009-01-07 16:50:41 -0600987 spin_unlock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000988
989 if (is_master(r))
990 dir_remove(r);
David Teigland52bda2b2007-11-07 09:06:49 -0600991 dlm_free_rsb(r);
David Teiglande7fd4172006-01-18 09:30:29 +0000992 count++;
993 } else {
David Teiglandc7be7612009-01-07 16:50:41 -0600994 spin_unlock(&ls->ls_rsbtbl[b].lock);
David Teiglande7fd4172006-01-18 09:30:29 +0000995 log_error(ls, "tossed rsb in use %s", r->res_name);
996 }
997 }
998
999 return count;
1000}
1001
1002void dlm_scan_rsbs(struct dlm_ls *ls)
1003{
1004 int i;
1005
David Teiglande7fd4172006-01-18 09:30:29 +00001006 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1007 shrink_bucket(ls, i);
David Teigland85e86ed2007-05-18 08:58:15 -05001008 if (dlm_locking_stopped(ls))
1009 break;
David Teiglande7fd4172006-01-18 09:30:29 +00001010 cond_resched();
1011 }
1012}
1013
David Teigland3ae1acf2007-05-18 08:59:31 -05001014static void add_timeout(struct dlm_lkb *lkb)
1015{
1016 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1017
David Teiglandeeda4182008-12-09 14:12:21 -06001018 if (is_master_copy(lkb))
David Teigland3ae1acf2007-05-18 08:59:31 -05001019 return;
David Teigland3ae1acf2007-05-18 08:59:31 -05001020
1021 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1022 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1023 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1024 goto add_it;
1025 }
David Teigland84d8cd62007-05-29 08:44:23 -05001026 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1027 goto add_it;
David Teigland3ae1acf2007-05-18 08:59:31 -05001028 return;
1029
1030 add_it:
1031 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1032 mutex_lock(&ls->ls_timeout_mutex);
1033 hold_lkb(lkb);
David Teigland3ae1acf2007-05-18 08:59:31 -05001034 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1035 mutex_unlock(&ls->ls_timeout_mutex);
1036}
1037
1038static void del_timeout(struct dlm_lkb *lkb)
1039{
1040 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1041
1042 mutex_lock(&ls->ls_timeout_mutex);
1043 if (!list_empty(&lkb->lkb_time_list)) {
1044 list_del_init(&lkb->lkb_time_list);
1045 unhold_lkb(lkb);
1046 }
1047 mutex_unlock(&ls->ls_timeout_mutex);
1048}
1049
1050/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1051 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1052 and then lock rsb because of lock ordering in add_timeout. We may need
1053 to specify some special timeout-related bits in the lkb that are just to
1054 be accessed under the timeout_mutex. */
1055
1056void dlm_scan_timeout(struct dlm_ls *ls)
1057{
1058 struct dlm_rsb *r;
1059 struct dlm_lkb *lkb;
1060 int do_cancel, do_warn;
David Teiglandeeda4182008-12-09 14:12:21 -06001061 s64 wait_us;
David Teigland3ae1acf2007-05-18 08:59:31 -05001062
1063 for (;;) {
1064 if (dlm_locking_stopped(ls))
1065 break;
1066
1067 do_cancel = 0;
1068 do_warn = 0;
1069 mutex_lock(&ls->ls_timeout_mutex);
1070 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1071
David Teiglandeeda4182008-12-09 14:12:21 -06001072 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1073 lkb->lkb_timestamp));
1074
David Teigland3ae1acf2007-05-18 08:59:31 -05001075 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
David Teiglandeeda4182008-12-09 14:12:21 -06001076 wait_us >= (lkb->lkb_timeout_cs * 10000))
David Teigland3ae1acf2007-05-18 08:59:31 -05001077 do_cancel = 1;
1078
1079 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
David Teiglandeeda4182008-12-09 14:12:21 -06001080 wait_us >= dlm_config.ci_timewarn_cs * 10000)
David Teigland3ae1acf2007-05-18 08:59:31 -05001081 do_warn = 1;
1082
1083 if (!do_cancel && !do_warn)
1084 continue;
1085 hold_lkb(lkb);
1086 break;
1087 }
1088 mutex_unlock(&ls->ls_timeout_mutex);
1089
1090 if (!do_cancel && !do_warn)
1091 break;
1092
1093 r = lkb->lkb_resource;
1094 hold_rsb(r);
1095 lock_rsb(r);
1096
1097 if (do_warn) {
1098 /* clear flag so we only warn once */
1099 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1100 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1101 del_timeout(lkb);
1102 dlm_timeout_warn(lkb);
1103 }
1104
1105 if (do_cancel) {
Steven Whitehouseb3cab7b2007-05-29 11:14:21 +01001106 log_debug(ls, "timeout cancel %x node %d %s",
David Teigland639aca42007-05-18 16:02:57 -05001107 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
David Teigland3ae1acf2007-05-18 08:59:31 -05001108 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1109 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1110 del_timeout(lkb);
1111 _cancel_lock(r, lkb);
1112 }
1113
1114 unlock_rsb(r);
1115 unhold_rsb(r);
1116 dlm_put_lkb(lkb);
1117 }
1118}
1119
1120/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1121 dlm_recoverd before checking/setting ls_recover_begin. */
1122
1123void dlm_adjust_timeouts(struct dlm_ls *ls)
1124{
1125 struct dlm_lkb *lkb;
David Teiglandeeda4182008-12-09 14:12:21 -06001126 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
David Teigland3ae1acf2007-05-18 08:59:31 -05001127
1128 ls->ls_recover_begin = 0;
1129 mutex_lock(&ls->ls_timeout_mutex);
1130 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
David Teiglandeeda4182008-12-09 14:12:21 -06001131 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
David Teigland3ae1acf2007-05-18 08:59:31 -05001132 mutex_unlock(&ls->ls_timeout_mutex);
1133}
1134
David Teiglande7fd4172006-01-18 09:30:29 +00001135/* lkb is master or local copy */
1136
1137static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1138{
1139 int b, len = r->res_ls->ls_lvblen;
1140
1141 /* b=1 lvb returned to caller
1142 b=0 lvb written to rsb or invalidated
1143 b=-1 do nothing */
1144
1145 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1146
1147 if (b == 1) {
1148 if (!lkb->lkb_lvbptr)
1149 return;
1150
1151 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1152 return;
1153
1154 if (!r->res_lvbptr)
1155 return;
1156
1157 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1158 lkb->lkb_lvbseq = r->res_lvbseq;
1159
1160 } else if (b == 0) {
1161 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1162 rsb_set_flag(r, RSB_VALNOTVALID);
1163 return;
1164 }
1165
1166 if (!lkb->lkb_lvbptr)
1167 return;
1168
1169 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1170 return;
1171
1172 if (!r->res_lvbptr)
David Teigland52bda2b2007-11-07 09:06:49 -06001173 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
David Teiglande7fd4172006-01-18 09:30:29 +00001174
1175 if (!r->res_lvbptr)
1176 return;
1177
1178 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1179 r->res_lvbseq++;
1180 lkb->lkb_lvbseq = r->res_lvbseq;
1181 rsb_clear_flag(r, RSB_VALNOTVALID);
1182 }
1183
1184 if (rsb_flag(r, RSB_VALNOTVALID))
1185 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1186}
1187
1188static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1189{
1190 if (lkb->lkb_grmode < DLM_LOCK_PW)
1191 return;
1192
1193 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1194 rsb_set_flag(r, RSB_VALNOTVALID);
1195 return;
1196 }
1197
1198 if (!lkb->lkb_lvbptr)
1199 return;
1200
1201 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1202 return;
1203
1204 if (!r->res_lvbptr)
David Teigland52bda2b2007-11-07 09:06:49 -06001205 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
David Teiglande7fd4172006-01-18 09:30:29 +00001206
1207 if (!r->res_lvbptr)
1208 return;
1209
1210 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1211 r->res_lvbseq++;
1212 rsb_clear_flag(r, RSB_VALNOTVALID);
1213}
1214
1215/* lkb is process copy (pc) */
1216
1217static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1218 struct dlm_message *ms)
1219{
1220 int b;
1221
1222 if (!lkb->lkb_lvbptr)
1223 return;
1224
1225 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1226 return;
1227
David Teigland597d0ca2006-07-12 16:44:04 -05001228 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
David Teiglande7fd4172006-01-18 09:30:29 +00001229 if (b == 1) {
1230 int len = receive_extralen(ms);
Al Viroa9cc9152008-01-26 00:02:29 -05001231 if (len > DLM_RESNAME_MAXLEN)
1232 len = DLM_RESNAME_MAXLEN;
David Teiglande7fd4172006-01-18 09:30:29 +00001233 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1234 lkb->lkb_lvbseq = ms->m_lvbseq;
1235 }
1236}
1237
1238/* Manipulate lkb's on rsb's convert/granted/waiting queues
1239 remove_lock -- used for unlock, removes lkb from granted
1240 revert_lock -- used for cancel, moves lkb from convert to granted
1241 grant_lock -- used for request and convert, adds lkb to granted or
1242 moves lkb from convert or waiting to granted
1243
1244 Each of these is used for master or local copy lkb's. There is
1245 also a _pc() variation used to make the corresponding change on
1246 a process copy (pc) lkb. */
1247
1248static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1249{
1250 del_lkb(r, lkb);
1251 lkb->lkb_grmode = DLM_LOCK_IV;
1252 /* this unhold undoes the original ref from create_lkb()
1253 so this leads to the lkb being freed */
1254 unhold_lkb(lkb);
1255}
1256
1257static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1258{
1259 set_lvb_unlock(r, lkb);
1260 _remove_lock(r, lkb);
1261}
1262
1263static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1264{
1265 _remove_lock(r, lkb);
1266}
1267
David Teiglandef0c2bb2007-03-28 09:56:46 -05001268/* returns: 0 did nothing
1269 1 moved lock to granted
1270 -1 removed lock */
1271
1272static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
David Teiglande7fd4172006-01-18 09:30:29 +00001273{
David Teiglandef0c2bb2007-03-28 09:56:46 -05001274 int rv = 0;
1275
David Teiglande7fd4172006-01-18 09:30:29 +00001276 lkb->lkb_rqmode = DLM_LOCK_IV;
1277
1278 switch (lkb->lkb_status) {
David Teigland597d0ca2006-07-12 16:44:04 -05001279 case DLM_LKSTS_GRANTED:
1280 break;
David Teiglande7fd4172006-01-18 09:30:29 +00001281 case DLM_LKSTS_CONVERT:
1282 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
David Teiglandef0c2bb2007-03-28 09:56:46 -05001283 rv = 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001284 break;
1285 case DLM_LKSTS_WAITING:
1286 del_lkb(r, lkb);
1287 lkb->lkb_grmode = DLM_LOCK_IV;
1288 /* this unhold undoes the original ref from create_lkb()
1289 so this leads to the lkb being freed */
1290 unhold_lkb(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05001291 rv = -1;
David Teiglande7fd4172006-01-18 09:30:29 +00001292 break;
1293 default:
1294 log_print("invalid status for revert %d", lkb->lkb_status);
1295 }
David Teiglandef0c2bb2007-03-28 09:56:46 -05001296 return rv;
David Teiglande7fd4172006-01-18 09:30:29 +00001297}
1298
David Teiglandef0c2bb2007-03-28 09:56:46 -05001299static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
David Teiglande7fd4172006-01-18 09:30:29 +00001300{
David Teiglandef0c2bb2007-03-28 09:56:46 -05001301 return revert_lock(r, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00001302}
1303
1304static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1305{
1306 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1307 lkb->lkb_grmode = lkb->lkb_rqmode;
1308 if (lkb->lkb_status)
1309 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1310 else
1311 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1312 }
1313
1314 lkb->lkb_rqmode = DLM_LOCK_IV;
David Teiglande7fd4172006-01-18 09:30:29 +00001315}
1316
1317static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1318{
1319 set_lvb_lock(r, lkb);
1320 _grant_lock(r, lkb);
1321 lkb->lkb_highbast = 0;
1322}
1323
1324static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1325 struct dlm_message *ms)
1326{
1327 set_lvb_lock_pc(r, lkb, ms);
1328 _grant_lock(r, lkb);
1329}
1330
1331/* called by grant_pending_locks() which means an async grant message must
1332 be sent to the requesting node in addition to granting the lock if the
1333 lkb belongs to a remote node. */
1334
1335static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1336{
1337 grant_lock(r, lkb);
1338 if (is_master_copy(lkb))
1339 send_grant(r, lkb);
1340 else
1341 queue_cast(r, lkb, 0);
1342}
1343
David Teigland7d3c1fe2007-04-19 10:30:41 -05001344/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1345 change the granted/requested modes. We're munging things accordingly in
1346 the process copy.
1347 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1348 conversion deadlock
1349 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1350 compatible with other granted locks */
1351
1352static void munge_demoted(struct dlm_lkb *lkb, struct dlm_message *ms)
1353{
1354 if (ms->m_type != DLM_MSG_CONVERT_REPLY) {
1355 log_print("munge_demoted %x invalid reply type %d",
1356 lkb->lkb_id, ms->m_type);
1357 return;
1358 }
1359
1360 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1361 log_print("munge_demoted %x invalid modes gr %d rq %d",
1362 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1363 return;
1364 }
1365
1366 lkb->lkb_grmode = DLM_LOCK_NL;
1367}
1368
1369static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1370{
1371 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1372 ms->m_type != DLM_MSG_GRANT) {
1373 log_print("munge_altmode %x invalid reply type %d",
1374 lkb->lkb_id, ms->m_type);
1375 return;
1376 }
1377
1378 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1379 lkb->lkb_rqmode = DLM_LOCK_PR;
1380 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1381 lkb->lkb_rqmode = DLM_LOCK_CW;
1382 else {
1383 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1384 dlm_print_lkb(lkb);
1385 }
1386}
1387
David Teiglande7fd4172006-01-18 09:30:29 +00001388static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1389{
1390 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1391 lkb_statequeue);
1392 if (lkb->lkb_id == first->lkb_id)
David Teigland90135922006-01-20 08:47:07 +00001393 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001394
David Teigland90135922006-01-20 08:47:07 +00001395 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001396}
1397
David Teiglande7fd4172006-01-18 09:30:29 +00001398/* Check if the given lkb conflicts with another lkb on the queue. */
1399
1400static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1401{
1402 struct dlm_lkb *this;
1403
1404 list_for_each_entry(this, head, lkb_statequeue) {
1405 if (this == lkb)
1406 continue;
David Teigland3bcd3682006-02-23 09:56:38 +00001407 if (!modes_compat(this, lkb))
David Teigland90135922006-01-20 08:47:07 +00001408 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001409 }
David Teigland90135922006-01-20 08:47:07 +00001410 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001411}
1412
1413/*
1414 * "A conversion deadlock arises with a pair of lock requests in the converting
1415 * queue for one resource. The granted mode of each lock blocks the requested
1416 * mode of the other lock."
1417 *
David Teiglandc85d65e2007-05-18 09:01:26 -05001418 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1419 * convert queue from being granted, then deadlk/demote lkb.
David Teiglande7fd4172006-01-18 09:30:29 +00001420 *
1421 * Example:
1422 * Granted Queue: empty
1423 * Convert Queue: NL->EX (first lock)
1424 * PR->EX (second lock)
1425 *
1426 * The first lock can't be granted because of the granted mode of the second
1427 * lock and the second lock can't be granted because it's not first in the
David Teiglandc85d65e2007-05-18 09:01:26 -05001428 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1429 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1430 * flag set and return DEMOTED in the lksb flags.
David Teiglande7fd4172006-01-18 09:30:29 +00001431 *
David Teiglandc85d65e2007-05-18 09:01:26 -05001432 * Originally, this function detected conv-deadlk in a more limited scope:
1433 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1434 * - if lkb1 was the first entry in the queue (not just earlier), and was
1435 * blocked by the granted mode of lkb2, and there was nothing on the
1436 * granted queue preventing lkb1 from being granted immediately, i.e.
1437 * lkb2 was the only thing preventing lkb1 from being granted.
1438 *
1439 * That second condition meant we'd only say there was conv-deadlk if
1440 * resolving it (by demotion) would lead to the first lock on the convert
1441 * queue being granted right away. It allowed conversion deadlocks to exist
1442 * between locks on the convert queue while they couldn't be granted anyway.
1443 *
1444 * Now, we detect and take action on conversion deadlocks immediately when
1445 * they're created, even if they may not be immediately consequential. If
1446 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1447 * mode that would prevent lkb1's conversion from being granted, we do a
1448 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1449 * I think this means that the lkb_is_ahead condition below should always
1450 * be zero, i.e. there will never be conv-deadlk between two locks that are
1451 * both already on the convert queue.
David Teiglande7fd4172006-01-18 09:30:29 +00001452 */
1453
David Teiglandc85d65e2007-05-18 09:01:26 -05001454static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
David Teiglande7fd4172006-01-18 09:30:29 +00001455{
David Teiglandc85d65e2007-05-18 09:01:26 -05001456 struct dlm_lkb *lkb1;
1457 int lkb_is_ahead = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001458
David Teiglandc85d65e2007-05-18 09:01:26 -05001459 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1460 if (lkb1 == lkb2) {
1461 lkb_is_ahead = 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001462 continue;
1463 }
1464
David Teiglandc85d65e2007-05-18 09:01:26 -05001465 if (!lkb_is_ahead) {
1466 if (!modes_compat(lkb2, lkb1))
1467 return 1;
1468 } else {
1469 if (!modes_compat(lkb2, lkb1) &&
1470 !modes_compat(lkb1, lkb2))
1471 return 1;
1472 }
David Teiglande7fd4172006-01-18 09:30:29 +00001473 }
David Teigland90135922006-01-20 08:47:07 +00001474 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001475}
1476
1477/*
1478 * Return 1 if the lock can be granted, 0 otherwise.
1479 * Also detect and resolve conversion deadlocks.
1480 *
1481 * lkb is the lock to be granted
1482 *
1483 * now is 1 if the function is being called in the context of the
1484 * immediate request, it is 0 if called later, after the lock has been
1485 * queued.
1486 *
1487 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1488 */
1489
1490static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1491{
1492 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1493
1494 /*
1495 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1496 * a new request for a NL mode lock being blocked.
1497 *
1498 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1499 * request, then it would be granted. In essence, the use of this flag
1500 * tells the Lock Manager to expedite theis request by not considering
1501 * what may be in the CONVERTING or WAITING queues... As of this
1502 * writing, the EXPEDITE flag can be used only with new requests for NL
1503 * mode locks. This flag is not valid for conversion requests.
1504 *
1505 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1506 * conversion or used with a non-NL requested mode. We also know an
1507 * EXPEDITE request is always granted immediately, so now must always
1508 * be 1. The full condition to grant an expedite request: (now &&
1509 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1510 * therefore be shortened to just checking the flag.
1511 */
1512
1513 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
David Teigland90135922006-01-20 08:47:07 +00001514 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001515
1516 /*
1517 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1518 * added to the remaining conditions.
1519 */
1520
1521 if (queue_conflict(&r->res_grantqueue, lkb))
1522 goto out;
1523
1524 /*
1525 * 6-3: By default, a conversion request is immediately granted if the
1526 * requested mode is compatible with the modes of all other granted
1527 * locks
1528 */
1529
1530 if (queue_conflict(&r->res_convertqueue, lkb))
1531 goto out;
1532
1533 /*
1534 * 6-5: But the default algorithm for deciding whether to grant or
1535 * queue conversion requests does not by itself guarantee that such
1536 * requests are serviced on a "first come first serve" basis. This, in
1537 * turn, can lead to a phenomenon known as "indefinate postponement".
1538 *
1539 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1540 * the system service employed to request a lock conversion. This flag
1541 * forces certain conversion requests to be queued, even if they are
1542 * compatible with the granted modes of other locks on the same
1543 * resource. Thus, the use of this flag results in conversion requests
1544 * being ordered on a "first come first servce" basis.
1545 *
1546 * DCT: This condition is all about new conversions being able to occur
1547 * "in place" while the lock remains on the granted queue (assuming
1548 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1549 * doesn't _have_ to go onto the convert queue where it's processed in
1550 * order. The "now" variable is necessary to distinguish converts
1551 * being received and processed for the first time now, because once a
1552 * convert is moved to the conversion queue the condition below applies
1553 * requiring fifo granting.
1554 */
1555
1556 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
David Teigland90135922006-01-20 08:47:07 +00001557 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001558
1559 /*
David Teigland3bcd3682006-02-23 09:56:38 +00001560 * The NOORDER flag is set to avoid the standard vms rules on grant
1561 * order.
David Teiglande7fd4172006-01-18 09:30:29 +00001562 */
1563
1564 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
David Teigland90135922006-01-20 08:47:07 +00001565 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001566
1567 /*
1568 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1569 * granted until all other conversion requests ahead of it are granted
1570 * and/or canceled.
1571 */
1572
1573 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
David Teigland90135922006-01-20 08:47:07 +00001574 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001575
1576 /*
1577 * 6-4: By default, a new request is immediately granted only if all
1578 * three of the following conditions are satisfied when the request is
1579 * issued:
1580 * - The queue of ungranted conversion requests for the resource is
1581 * empty.
1582 * - The queue of ungranted new requests for the resource is empty.
1583 * - The mode of the new request is compatible with the most
1584 * restrictive mode of all granted locks on the resource.
1585 */
1586
1587 if (now && !conv && list_empty(&r->res_convertqueue) &&
1588 list_empty(&r->res_waitqueue))
David Teigland90135922006-01-20 08:47:07 +00001589 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001590
1591 /*
1592 * 6-4: Once a lock request is in the queue of ungranted new requests,
1593 * it cannot be granted until the queue of ungranted conversion
1594 * requests is empty, all ungranted new requests ahead of it are
1595 * granted and/or canceled, and it is compatible with the granted mode
1596 * of the most restrictive lock granted on the resource.
1597 */
1598
1599 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1600 first_in_list(lkb, &r->res_waitqueue))
David Teigland90135922006-01-20 08:47:07 +00001601 return 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001602 out:
David Teigland90135922006-01-20 08:47:07 +00001603 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001604}
1605
David Teiglandc85d65e2007-05-18 09:01:26 -05001606static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1607 int *err)
David Teiglande7fd4172006-01-18 09:30:29 +00001608{
David Teiglande7fd4172006-01-18 09:30:29 +00001609 int rv;
1610 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
David Teiglandc85d65e2007-05-18 09:01:26 -05001611 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1612
1613 if (err)
1614 *err = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001615
1616 rv = _can_be_granted(r, lkb, now);
1617 if (rv)
1618 goto out;
1619
David Teiglandc85d65e2007-05-18 09:01:26 -05001620 /*
1621 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1622 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1623 * cancels one of the locks.
1624 */
David Teiglande7fd4172006-01-18 09:30:29 +00001625
David Teiglandc85d65e2007-05-18 09:01:26 -05001626 if (is_convert && can_be_queued(lkb) &&
1627 conversion_deadlock_detect(r, lkb)) {
1628 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1629 lkb->lkb_grmode = DLM_LOCK_NL;
1630 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1631 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1632 if (err)
1633 *err = -EDEADLK;
1634 else {
1635 log_print("can_be_granted deadlock %x now %d",
1636 lkb->lkb_id, now);
1637 dlm_dump_rsb(r);
1638 }
1639 }
1640 goto out;
1641 }
1642
1643 /*
1644 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1645 * to grant a request in a mode other than the normal rqmode. It's a
1646 * simple way to provide a big optimization to applications that can
1647 * use them.
1648 */
1649
1650 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
David Teiglande7fd4172006-01-18 09:30:29 +00001651 alt = DLM_LOCK_PR;
David Teiglandc85d65e2007-05-18 09:01:26 -05001652 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
David Teiglande7fd4172006-01-18 09:30:29 +00001653 alt = DLM_LOCK_CW;
1654
1655 if (alt) {
1656 lkb->lkb_rqmode = alt;
1657 rv = _can_be_granted(r, lkb, now);
1658 if (rv)
1659 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1660 else
1661 lkb->lkb_rqmode = rqmode;
1662 }
1663 out:
1664 return rv;
1665}
1666
David Teiglandc85d65e2007-05-18 09:01:26 -05001667/* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1668 for locks pending on the convert list. Once verified (watch for these
1669 log_prints), we should be able to just call _can_be_granted() and not
1670 bother with the demote/deadlk cases here (and there's no easy way to deal
1671 with a deadlk here, we'd have to generate something like grant_lock with
1672 the deadlk error.) */
1673
David Teigland36509252007-08-07 09:44:48 -05001674/* Returns the highest requested mode of all blocked conversions; sets
1675 cw if there's a blocked conversion to DLM_LOCK_CW. */
David Teiglandc85d65e2007-05-18 09:01:26 -05001676
David Teigland36509252007-08-07 09:44:48 -05001677static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
David Teiglande7fd4172006-01-18 09:30:29 +00001678{
1679 struct dlm_lkb *lkb, *s;
1680 int hi, demoted, quit, grant_restart, demote_restart;
David Teiglandc85d65e2007-05-18 09:01:26 -05001681 int deadlk;
David Teiglande7fd4172006-01-18 09:30:29 +00001682
1683 quit = 0;
1684 restart:
1685 grant_restart = 0;
1686 demote_restart = 0;
1687 hi = DLM_LOCK_IV;
1688
1689 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1690 demoted = is_demoted(lkb);
David Teiglandc85d65e2007-05-18 09:01:26 -05001691 deadlk = 0;
1692
1693 if (can_be_granted(r, lkb, 0, &deadlk)) {
David Teiglande7fd4172006-01-18 09:30:29 +00001694 grant_lock_pending(r, lkb);
1695 grant_restart = 1;
David Teiglandc85d65e2007-05-18 09:01:26 -05001696 continue;
David Teiglande7fd4172006-01-18 09:30:29 +00001697 }
David Teiglandc85d65e2007-05-18 09:01:26 -05001698
1699 if (!demoted && is_demoted(lkb)) {
1700 log_print("WARN: pending demoted %x node %d %s",
1701 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1702 demote_restart = 1;
1703 continue;
1704 }
1705
1706 if (deadlk) {
1707 log_print("WARN: pending deadlock %x node %d %s",
1708 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1709 dlm_dump_rsb(r);
1710 continue;
1711 }
1712
1713 hi = max_t(int, lkb->lkb_rqmode, hi);
David Teigland36509252007-08-07 09:44:48 -05001714
1715 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
1716 *cw = 1;
David Teiglande7fd4172006-01-18 09:30:29 +00001717 }
1718
1719 if (grant_restart)
1720 goto restart;
1721 if (demote_restart && !quit) {
1722 quit = 1;
1723 goto restart;
1724 }
1725
1726 return max_t(int, high, hi);
1727}
1728
David Teigland36509252007-08-07 09:44:48 -05001729static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
David Teiglande7fd4172006-01-18 09:30:29 +00001730{
1731 struct dlm_lkb *lkb, *s;
1732
1733 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
David Teiglandc85d65e2007-05-18 09:01:26 -05001734 if (can_be_granted(r, lkb, 0, NULL))
David Teiglande7fd4172006-01-18 09:30:29 +00001735 grant_lock_pending(r, lkb);
David Teigland36509252007-08-07 09:44:48 -05001736 else {
David Teiglande7fd4172006-01-18 09:30:29 +00001737 high = max_t(int, lkb->lkb_rqmode, high);
David Teigland36509252007-08-07 09:44:48 -05001738 if (lkb->lkb_rqmode == DLM_LOCK_CW)
1739 *cw = 1;
1740 }
David Teiglande7fd4172006-01-18 09:30:29 +00001741 }
1742
1743 return high;
1744}
1745
David Teigland36509252007-08-07 09:44:48 -05001746/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1747 on either the convert or waiting queue.
1748 high is the largest rqmode of all locks blocked on the convert or
1749 waiting queue. */
1750
1751static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
1752{
1753 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
1754 if (gr->lkb_highbast < DLM_LOCK_EX)
1755 return 1;
1756 return 0;
1757 }
1758
1759 if (gr->lkb_highbast < high &&
1760 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
1761 return 1;
1762 return 0;
1763}
1764
David Teiglande7fd4172006-01-18 09:30:29 +00001765static void grant_pending_locks(struct dlm_rsb *r)
1766{
1767 struct dlm_lkb *lkb, *s;
1768 int high = DLM_LOCK_IV;
David Teigland36509252007-08-07 09:44:48 -05001769 int cw = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00001770
David Teiglanda345da32006-08-18 11:54:25 -05001771 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
David Teiglande7fd4172006-01-18 09:30:29 +00001772
David Teigland36509252007-08-07 09:44:48 -05001773 high = grant_pending_convert(r, high, &cw);
1774 high = grant_pending_wait(r, high, &cw);
David Teiglande7fd4172006-01-18 09:30:29 +00001775
1776 if (high == DLM_LOCK_IV)
1777 return;
1778
1779 /*
1780 * If there are locks left on the wait/convert queue then send blocking
1781 * ASTs to granted locks based on the largest requested mode (high)
David Teigland36509252007-08-07 09:44:48 -05001782 * found above.
David Teiglande7fd4172006-01-18 09:30:29 +00001783 */
1784
1785 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
David Teiglande5dae542008-02-06 00:35:45 -06001786 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
David Teigland329fc4c2008-05-20 12:18:10 -05001787 if (cw && high == DLM_LOCK_PR &&
1788 lkb->lkb_grmode == DLM_LOCK_PR)
David Teigland36509252007-08-07 09:44:48 -05001789 queue_bast(r, lkb, DLM_LOCK_CW);
1790 else
1791 queue_bast(r, lkb, high);
David Teiglande7fd4172006-01-18 09:30:29 +00001792 lkb->lkb_highbast = high;
1793 }
1794 }
1795}
1796
David Teigland36509252007-08-07 09:44:48 -05001797static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
1798{
1799 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
1800 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
1801 if (gr->lkb_highbast < DLM_LOCK_EX)
1802 return 1;
1803 return 0;
1804 }
1805
1806 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
1807 return 1;
1808 return 0;
1809}
1810
David Teiglande7fd4172006-01-18 09:30:29 +00001811static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1812 struct dlm_lkb *lkb)
1813{
1814 struct dlm_lkb *gr;
1815
1816 list_for_each_entry(gr, head, lkb_statequeue) {
David Teiglande5dae542008-02-06 00:35:45 -06001817 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
David Teiglande7fd4172006-01-18 09:30:29 +00001818 queue_bast(r, gr, lkb->lkb_rqmode);
1819 gr->lkb_highbast = lkb->lkb_rqmode;
1820 }
1821 }
1822}
1823
1824static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1825{
1826 send_bast_queue(r, &r->res_grantqueue, lkb);
1827}
1828
1829static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1830{
1831 send_bast_queue(r, &r->res_grantqueue, lkb);
1832 send_bast_queue(r, &r->res_convertqueue, lkb);
1833}
1834
1835/* set_master(r, lkb) -- set the master nodeid of a resource
1836
1837 The purpose of this function is to set the nodeid field in the given
1838 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1839 known, it can just be copied to the lkb and the function will return
1840 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1841 before it can be copied to the lkb.
1842
1843 When the rsb nodeid is being looked up remotely, the initial lkb
1844 causing the lookup is kept on the ls_waiters list waiting for the
1845 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1846 on the rsb's res_lookup list until the master is verified.
1847
1848 Return values:
1849 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1850 1: the rsb master is not available and the lkb has been placed on
1851 a wait queue
1852*/
1853
1854static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1855{
1856 struct dlm_ls *ls = r->res_ls;
David Teigland755b5eb2008-01-09 10:37:39 -06001857 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
David Teiglande7fd4172006-01-18 09:30:29 +00001858
1859 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1860 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1861 r->res_first_lkid = lkb->lkb_id;
1862 lkb->lkb_nodeid = r->res_nodeid;
1863 return 0;
1864 }
1865
1866 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1867 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1868 return 1;
1869 }
1870
1871 if (r->res_nodeid == 0) {
1872 lkb->lkb_nodeid = 0;
1873 return 0;
1874 }
1875
1876 if (r->res_nodeid > 0) {
1877 lkb->lkb_nodeid = r->res_nodeid;
1878 return 0;
1879 }
1880
David Teiglanda345da32006-08-18 11:54:25 -05001881 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
David Teiglande7fd4172006-01-18 09:30:29 +00001882
1883 dir_nodeid = dlm_dir_nodeid(r);
1884
1885 if (dir_nodeid != our_nodeid) {
1886 r->res_first_lkid = lkb->lkb_id;
1887 send_lookup(r, lkb);
1888 return 1;
1889 }
1890
David Teigland755b5eb2008-01-09 10:37:39 -06001891 for (i = 0; i < 2; i++) {
David Teiglande7fd4172006-01-18 09:30:29 +00001892 /* It's possible for dlm_scand to remove an old rsb for
1893 this same resource from the toss list, us to create
1894 a new one, look up the master locally, and find it
1895 already exists just before dlm_scand does the
1896 dir_remove() on the previous rsb. */
1897
1898 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1899 r->res_length, &ret_nodeid);
1900 if (!error)
1901 break;
1902 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1903 schedule();
1904 }
David Teigland755b5eb2008-01-09 10:37:39 -06001905 if (error && error != -EEXIST)
1906 return error;
David Teiglande7fd4172006-01-18 09:30:29 +00001907
1908 if (ret_nodeid == our_nodeid) {
1909 r->res_first_lkid = 0;
1910 r->res_nodeid = 0;
1911 lkb->lkb_nodeid = 0;
1912 } else {
1913 r->res_first_lkid = lkb->lkb_id;
1914 r->res_nodeid = ret_nodeid;
1915 lkb->lkb_nodeid = ret_nodeid;
1916 }
1917 return 0;
1918}
1919
1920static void process_lookup_list(struct dlm_rsb *r)
1921{
1922 struct dlm_lkb *lkb, *safe;
1923
1924 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
David Teiglandef0c2bb2007-03-28 09:56:46 -05001925 list_del_init(&lkb->lkb_rsb_lookup);
David Teiglande7fd4172006-01-18 09:30:29 +00001926 _request_lock(r, lkb);
1927 schedule();
1928 }
1929}
1930
1931/* confirm_master -- confirm (or deny) an rsb's master nodeid */
1932
1933static void confirm_master(struct dlm_rsb *r, int error)
1934{
1935 struct dlm_lkb *lkb;
1936
1937 if (!r->res_first_lkid)
1938 return;
1939
1940 switch (error) {
1941 case 0:
1942 case -EINPROGRESS:
1943 r->res_first_lkid = 0;
1944 process_lookup_list(r);
1945 break;
1946
1947 case -EAGAIN:
David Teiglandaec64e12008-01-08 15:37:47 -06001948 case -EBADR:
1949 case -ENOTBLK:
1950 /* the remote request failed and won't be retried (it was
1951 a NOQUEUE, or has been canceled/unlocked); make a waiting
1952 lkb the first_lkid */
David Teiglande7fd4172006-01-18 09:30:29 +00001953
1954 r->res_first_lkid = 0;
1955
1956 if (!list_empty(&r->res_lookup)) {
1957 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1958 lkb_rsb_lookup);
David Teiglandef0c2bb2007-03-28 09:56:46 -05001959 list_del_init(&lkb->lkb_rsb_lookup);
David Teiglande7fd4172006-01-18 09:30:29 +00001960 r->res_first_lkid = lkb->lkb_id;
1961 _request_lock(r, lkb);
David Teigland761b9d32008-02-21 11:25:42 -06001962 }
David Teiglande7fd4172006-01-18 09:30:29 +00001963 break;
1964
1965 default:
1966 log_error(r->res_ls, "confirm_master unknown error %d", error);
1967 }
1968}
1969
1970static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
David Teiglande5dae542008-02-06 00:35:45 -06001971 int namelen, unsigned long timeout_cs,
1972 void (*ast) (void *astparam),
1973 void *astparam,
1974 void (*bast) (void *astparam, int mode),
1975 struct dlm_args *args)
David Teiglande7fd4172006-01-18 09:30:29 +00001976{
1977 int rv = -EINVAL;
1978
1979 /* check for invalid arg usage */
1980
1981 if (mode < 0 || mode > DLM_LOCK_EX)
1982 goto out;
1983
1984 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1985 goto out;
1986
1987 if (flags & DLM_LKF_CANCEL)
1988 goto out;
1989
1990 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1991 goto out;
1992
1993 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1994 goto out;
1995
1996 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1997 goto out;
1998
1999 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2000 goto out;
2001
2002 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2003 goto out;
2004
2005 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2006 goto out;
2007
2008 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2009 goto out;
2010
2011 if (!ast || !lksb)
2012 goto out;
2013
2014 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2015 goto out;
2016
David Teiglande7fd4172006-01-18 09:30:29 +00002017 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2018 goto out;
2019
2020 /* these args will be copied to the lkb in validate_lock_args,
2021 it cannot be done now because when converting locks, fields in
2022 an active lkb cannot be modified before locking the rsb */
2023
2024 args->flags = flags;
David Teiglande5dae542008-02-06 00:35:45 -06002025 args->astfn = ast;
2026 args->astparam = astparam;
2027 args->bastfn = bast;
David Teiglandd7db9232007-05-18 09:00:32 -05002028 args->timeout = timeout_cs;
David Teiglande7fd4172006-01-18 09:30:29 +00002029 args->mode = mode;
2030 args->lksb = lksb;
David Teiglande7fd4172006-01-18 09:30:29 +00002031 rv = 0;
2032 out:
2033 return rv;
2034}
2035
2036static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2037{
2038 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2039 DLM_LKF_FORCEUNLOCK))
2040 return -EINVAL;
2041
David Teiglandef0c2bb2007-03-28 09:56:46 -05002042 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2043 return -EINVAL;
2044
David Teiglande7fd4172006-01-18 09:30:29 +00002045 args->flags = flags;
David Teiglande5dae542008-02-06 00:35:45 -06002046 args->astparam = astarg;
David Teiglande7fd4172006-01-18 09:30:29 +00002047 return 0;
2048}
2049
2050static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2051 struct dlm_args *args)
2052{
2053 int rv = -EINVAL;
2054
2055 if (args->flags & DLM_LKF_CONVERT) {
2056 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2057 goto out;
2058
2059 if (args->flags & DLM_LKF_QUECVT &&
2060 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2061 goto out;
2062
2063 rv = -EBUSY;
2064 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2065 goto out;
2066
2067 if (lkb->lkb_wait_type)
2068 goto out;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002069
2070 if (is_overlap(lkb))
2071 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +00002072 }
2073
2074 lkb->lkb_exflags = args->flags;
2075 lkb->lkb_sbflags = 0;
David Teiglande5dae542008-02-06 00:35:45 -06002076 lkb->lkb_astfn = args->astfn;
David Teiglande7fd4172006-01-18 09:30:29 +00002077 lkb->lkb_astparam = args->astparam;
David Teiglande5dae542008-02-06 00:35:45 -06002078 lkb->lkb_bastfn = args->bastfn;
David Teiglande7fd4172006-01-18 09:30:29 +00002079 lkb->lkb_rqmode = args->mode;
2080 lkb->lkb_lksb = args->lksb;
2081 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2082 lkb->lkb_ownpid = (int) current->pid;
David Teiglandd7db9232007-05-18 09:00:32 -05002083 lkb->lkb_timeout_cs = args->timeout;
David Teiglande7fd4172006-01-18 09:30:29 +00002084 rv = 0;
2085 out:
2086 return rv;
2087}
2088
David Teiglandef0c2bb2007-03-28 09:56:46 -05002089/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2090 for success */
2091
2092/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2093 because there may be a lookup in progress and it's valid to do
2094 cancel/unlockf on it */
2095
David Teiglande7fd4172006-01-18 09:30:29 +00002096static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2097{
David Teiglandef0c2bb2007-03-28 09:56:46 -05002098 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
David Teiglande7fd4172006-01-18 09:30:29 +00002099 int rv = -EINVAL;
2100
David Teiglandef0c2bb2007-03-28 09:56:46 -05002101 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2102 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2103 dlm_print_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00002104 goto out;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002105 }
David Teiglande7fd4172006-01-18 09:30:29 +00002106
David Teiglandef0c2bb2007-03-28 09:56:46 -05002107 /* an lkb may still exist even though the lock is EOL'ed due to a
2108 cancel, unlock or failed noqueue request; an app can't use these
2109 locks; return same error as if the lkid had not been found at all */
2110
2111 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2112 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2113 rv = -ENOENT;
2114 goto out;
2115 }
2116
2117 /* an lkb may be waiting for an rsb lookup to complete where the
2118 lookup was initiated by another lock */
2119
David Teigland42dc1602008-01-09 10:30:45 -06002120 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2121 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
David Teiglandef0c2bb2007-03-28 09:56:46 -05002122 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2123 list_del_init(&lkb->lkb_rsb_lookup);
2124 queue_cast(lkb->lkb_resource, lkb,
2125 args->flags & DLM_LKF_CANCEL ?
2126 -DLM_ECANCEL : -DLM_EUNLOCK);
2127 unhold_lkb(lkb); /* undoes create_lkb() */
David Teiglandef0c2bb2007-03-28 09:56:46 -05002128 }
David Teigland42dc1602008-01-09 10:30:45 -06002129 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2130 rv = -EBUSY;
2131 goto out;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002132 }
2133
2134 /* cancel not allowed with another cancel/unlock in progress */
2135
2136 if (args->flags & DLM_LKF_CANCEL) {
2137 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2138 goto out;
2139
2140 if (is_overlap(lkb))
2141 goto out;
2142
David Teigland3ae1acf2007-05-18 08:59:31 -05002143 /* don't let scand try to do a cancel */
2144 del_timeout(lkb);
2145
David Teiglandef0c2bb2007-03-28 09:56:46 -05002146 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2147 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2148 rv = -EBUSY;
2149 goto out;
2150 }
2151
2152 switch (lkb->lkb_wait_type) {
2153 case DLM_MSG_LOOKUP:
2154 case DLM_MSG_REQUEST:
2155 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2156 rv = -EBUSY;
2157 goto out;
2158 case DLM_MSG_UNLOCK:
2159 case DLM_MSG_CANCEL:
2160 goto out;
2161 }
2162 /* add_to_waiters() will set OVERLAP_CANCEL */
David Teiglande7fd4172006-01-18 09:30:29 +00002163 goto out_ok;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002164 }
David Teiglande7fd4172006-01-18 09:30:29 +00002165
David Teiglandef0c2bb2007-03-28 09:56:46 -05002166 /* do we need to allow a force-unlock if there's a normal unlock
2167 already in progress? in what conditions could the normal unlock
2168 fail such that we'd want to send a force-unlock to be sure? */
David Teiglande7fd4172006-01-18 09:30:29 +00002169
David Teiglandef0c2bb2007-03-28 09:56:46 -05002170 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2171 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2172 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +00002173
David Teiglandef0c2bb2007-03-28 09:56:46 -05002174 if (is_overlap_unlock(lkb))
2175 goto out;
2176
David Teigland3ae1acf2007-05-18 08:59:31 -05002177 /* don't let scand try to do a cancel */
2178 del_timeout(lkb);
2179
David Teiglandef0c2bb2007-03-28 09:56:46 -05002180 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2181 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2182 rv = -EBUSY;
2183 goto out;
2184 }
2185
2186 switch (lkb->lkb_wait_type) {
2187 case DLM_MSG_LOOKUP:
2188 case DLM_MSG_REQUEST:
2189 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2190 rv = -EBUSY;
2191 goto out;
2192 case DLM_MSG_UNLOCK:
2193 goto out;
2194 }
2195 /* add_to_waiters() will set OVERLAP_UNLOCK */
2196 goto out_ok;
2197 }
2198
2199 /* normal unlock not allowed if there's any op in progress */
David Teiglande7fd4172006-01-18 09:30:29 +00002200 rv = -EBUSY;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002201 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
David Teiglande7fd4172006-01-18 09:30:29 +00002202 goto out;
2203
2204 out_ok:
David Teiglandef0c2bb2007-03-28 09:56:46 -05002205 /* an overlapping op shouldn't blow away exflags from other op */
2206 lkb->lkb_exflags |= args->flags;
David Teiglande7fd4172006-01-18 09:30:29 +00002207 lkb->lkb_sbflags = 0;
2208 lkb->lkb_astparam = args->astparam;
David Teiglande7fd4172006-01-18 09:30:29 +00002209 rv = 0;
2210 out:
David Teiglandef0c2bb2007-03-28 09:56:46 -05002211 if (rv)
2212 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2213 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2214 args->flags, lkb->lkb_wait_type,
2215 lkb->lkb_resource->res_name);
David Teiglande7fd4172006-01-18 09:30:29 +00002216 return rv;
2217}
2218
2219/*
2220 * Four stage 4 varieties:
2221 * do_request(), do_convert(), do_unlock(), do_cancel()
2222 * These are called on the master node for the given lock and
2223 * from the central locking logic.
2224 */
2225
2226static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2227{
2228 int error = 0;
2229
David Teiglandc85d65e2007-05-18 09:01:26 -05002230 if (can_be_granted(r, lkb, 1, NULL)) {
David Teiglande7fd4172006-01-18 09:30:29 +00002231 grant_lock(r, lkb);
2232 queue_cast(r, lkb, 0);
2233 goto out;
2234 }
2235
2236 if (can_be_queued(lkb)) {
2237 error = -EINPROGRESS;
2238 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2239 send_blocking_asts(r, lkb);
David Teigland3ae1acf2007-05-18 08:59:31 -05002240 add_timeout(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00002241 goto out;
2242 }
2243
2244 error = -EAGAIN;
2245 if (force_blocking_asts(lkb))
2246 send_blocking_asts_all(r, lkb);
2247 queue_cast(r, lkb, -EAGAIN);
2248
2249 out:
2250 return error;
2251}
2252
2253static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2254{
2255 int error = 0;
David Teiglandc85d65e2007-05-18 09:01:26 -05002256 int deadlk = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00002257
2258 /* changing an existing lock may allow others to be granted */
2259
David Teiglandc85d65e2007-05-18 09:01:26 -05002260 if (can_be_granted(r, lkb, 1, &deadlk)) {
David Teiglande7fd4172006-01-18 09:30:29 +00002261 grant_lock(r, lkb);
2262 queue_cast(r, lkb, 0);
2263 grant_pending_locks(r);
2264 goto out;
2265 }
2266
David Teiglandc85d65e2007-05-18 09:01:26 -05002267 /* can_be_granted() detected that this lock would block in a conversion
2268 deadlock, so we leave it on the granted queue and return EDEADLK in
2269 the ast for the convert. */
2270
2271 if (deadlk) {
2272 /* it's left on the granted queue */
2273 log_debug(r->res_ls, "deadlock %x node %d sts%d g%d r%d %s",
2274 lkb->lkb_id, lkb->lkb_nodeid, lkb->lkb_status,
2275 lkb->lkb_grmode, lkb->lkb_rqmode, r->res_name);
2276 revert_lock(r, lkb);
2277 queue_cast(r, lkb, -EDEADLK);
2278 error = -EDEADLK;
2279 goto out;
2280 }
2281
David Teigland7d3c1fe2007-04-19 10:30:41 -05002282 /* is_demoted() means the can_be_granted() above set the grmode
2283 to NL, and left us on the granted queue. This auto-demotion
2284 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2285 now grantable. We have to try to grant other converting locks
2286 before we try again to grant this one. */
2287
2288 if (is_demoted(lkb)) {
David Teigland36509252007-08-07 09:44:48 -05002289 grant_pending_convert(r, DLM_LOCK_IV, NULL);
David Teigland7d3c1fe2007-04-19 10:30:41 -05002290 if (_can_be_granted(r, lkb, 1)) {
2291 grant_lock(r, lkb);
2292 queue_cast(r, lkb, 0);
David Teiglande7fd4172006-01-18 09:30:29 +00002293 grant_pending_locks(r);
David Teigland7d3c1fe2007-04-19 10:30:41 -05002294 goto out;
2295 }
2296 /* else fall through and move to convert queue */
2297 }
2298
2299 if (can_be_queued(lkb)) {
David Teiglande7fd4172006-01-18 09:30:29 +00002300 error = -EINPROGRESS;
2301 del_lkb(r, lkb);
2302 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2303 send_blocking_asts(r, lkb);
David Teigland3ae1acf2007-05-18 08:59:31 -05002304 add_timeout(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00002305 goto out;
2306 }
2307
2308 error = -EAGAIN;
2309 if (force_blocking_asts(lkb))
2310 send_blocking_asts_all(r, lkb);
2311 queue_cast(r, lkb, -EAGAIN);
2312
2313 out:
2314 return error;
2315}
2316
2317static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2318{
2319 remove_lock(r, lkb);
2320 queue_cast(r, lkb, -DLM_EUNLOCK);
2321 grant_pending_locks(r);
2322 return -DLM_EUNLOCK;
2323}
2324
David Teiglandef0c2bb2007-03-28 09:56:46 -05002325/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
Steven Whitehouse907b9bc2006-09-25 09:26:04 -04002326
David Teiglande7fd4172006-01-18 09:30:29 +00002327static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2328{
David Teiglandef0c2bb2007-03-28 09:56:46 -05002329 int error;
2330
2331 error = revert_lock(r, lkb);
2332 if (error) {
2333 queue_cast(r, lkb, -DLM_ECANCEL);
2334 grant_pending_locks(r);
2335 return -DLM_ECANCEL;
2336 }
2337 return 0;
David Teiglande7fd4172006-01-18 09:30:29 +00002338}
2339
2340/*
2341 * Four stage 3 varieties:
2342 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2343 */
2344
2345/* add a new lkb to a possibly new rsb, called by requesting process */
2346
2347static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2348{
2349 int error;
2350
2351 /* set_master: sets lkb nodeid from r */
2352
2353 error = set_master(r, lkb);
2354 if (error < 0)
2355 goto out;
2356 if (error) {
2357 error = 0;
2358 goto out;
2359 }
2360
2361 if (is_remote(r))
2362 /* receive_request() calls do_request() on remote node */
2363 error = send_request(r, lkb);
2364 else
2365 error = do_request(r, lkb);
2366 out:
2367 return error;
2368}
2369
David Teigland3bcd3682006-02-23 09:56:38 +00002370/* change some property of an existing lkb, e.g. mode */
David Teiglande7fd4172006-01-18 09:30:29 +00002371
2372static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2373{
2374 int error;
2375
2376 if (is_remote(r))
2377 /* receive_convert() calls do_convert() on remote node */
2378 error = send_convert(r, lkb);
2379 else
2380 error = do_convert(r, lkb);
2381
2382 return error;
2383}
2384
2385/* remove an existing lkb from the granted queue */
2386
2387static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2388{
2389 int error;
2390
2391 if (is_remote(r))
2392 /* receive_unlock() calls do_unlock() on remote node */
2393 error = send_unlock(r, lkb);
2394 else
2395 error = do_unlock(r, lkb);
2396
2397 return error;
2398}
2399
2400/* remove an existing lkb from the convert or wait queue */
2401
2402static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2403{
2404 int error;
2405
2406 if (is_remote(r))
2407 /* receive_cancel() calls do_cancel() on remote node */
2408 error = send_cancel(r, lkb);
2409 else
2410 error = do_cancel(r, lkb);
2411
2412 return error;
2413}
2414
2415/*
2416 * Four stage 2 varieties:
2417 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2418 */
2419
2420static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2421 int len, struct dlm_args *args)
2422{
2423 struct dlm_rsb *r;
2424 int error;
2425
2426 error = validate_lock_args(ls, lkb, args);
2427 if (error)
2428 goto out;
2429
2430 error = find_rsb(ls, name, len, R_CREATE, &r);
2431 if (error)
2432 goto out;
2433
2434 lock_rsb(r);
2435
2436 attach_lkb(r, lkb);
2437 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2438
2439 error = _request_lock(r, lkb);
2440
2441 unlock_rsb(r);
2442 put_rsb(r);
2443
2444 out:
2445 return error;
2446}
2447
2448static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2449 struct dlm_args *args)
2450{
2451 struct dlm_rsb *r;
2452 int error;
2453
2454 r = lkb->lkb_resource;
2455
2456 hold_rsb(r);
2457 lock_rsb(r);
2458
2459 error = validate_lock_args(ls, lkb, args);
2460 if (error)
2461 goto out;
2462
2463 error = _convert_lock(r, lkb);
2464 out:
2465 unlock_rsb(r);
2466 put_rsb(r);
2467 return error;
2468}
2469
2470static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2471 struct dlm_args *args)
2472{
2473 struct dlm_rsb *r;
2474 int error;
2475
2476 r = lkb->lkb_resource;
2477
2478 hold_rsb(r);
2479 lock_rsb(r);
2480
2481 error = validate_unlock_args(lkb, args);
2482 if (error)
2483 goto out;
2484
2485 error = _unlock_lock(r, lkb);
2486 out:
2487 unlock_rsb(r);
2488 put_rsb(r);
2489 return error;
2490}
2491
2492static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2493 struct dlm_args *args)
2494{
2495 struct dlm_rsb *r;
2496 int error;
2497
2498 r = lkb->lkb_resource;
2499
2500 hold_rsb(r);
2501 lock_rsb(r);
2502
2503 error = validate_unlock_args(lkb, args);
2504 if (error)
2505 goto out;
2506
2507 error = _cancel_lock(r, lkb);
2508 out:
2509 unlock_rsb(r);
2510 put_rsb(r);
2511 return error;
2512}
2513
2514/*
2515 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2516 */
2517
2518int dlm_lock(dlm_lockspace_t *lockspace,
2519 int mode,
2520 struct dlm_lksb *lksb,
2521 uint32_t flags,
2522 void *name,
2523 unsigned int namelen,
2524 uint32_t parent_lkid,
2525 void (*ast) (void *astarg),
2526 void *astarg,
David Teigland3bcd3682006-02-23 09:56:38 +00002527 void (*bast) (void *astarg, int mode))
David Teiglande7fd4172006-01-18 09:30:29 +00002528{
2529 struct dlm_ls *ls;
2530 struct dlm_lkb *lkb;
2531 struct dlm_args args;
2532 int error, convert = flags & DLM_LKF_CONVERT;
2533
2534 ls = dlm_find_lockspace_local(lockspace);
2535 if (!ls)
2536 return -EINVAL;
2537
David Teigland85e86ed2007-05-18 08:58:15 -05002538 dlm_lock_recovery(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00002539
2540 if (convert)
2541 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2542 else
2543 error = create_lkb(ls, &lkb);
2544
2545 if (error)
2546 goto out;
2547
David Teiglandd7db9232007-05-18 09:00:32 -05002548 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
David Teigland3bcd3682006-02-23 09:56:38 +00002549 astarg, bast, &args);
David Teiglande7fd4172006-01-18 09:30:29 +00002550 if (error)
2551 goto out_put;
2552
2553 if (convert)
2554 error = convert_lock(ls, lkb, &args);
2555 else
2556 error = request_lock(ls, lkb, name, namelen, &args);
2557
2558 if (error == -EINPROGRESS)
2559 error = 0;
2560 out_put:
2561 if (convert || error)
David Teiglandb3f58d82006-02-28 11:16:37 -05002562 __put_lkb(ls, lkb);
David Teiglandc85d65e2007-05-18 09:01:26 -05002563 if (error == -EAGAIN || error == -EDEADLK)
David Teiglande7fd4172006-01-18 09:30:29 +00002564 error = 0;
2565 out:
David Teigland85e86ed2007-05-18 08:58:15 -05002566 dlm_unlock_recovery(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00002567 dlm_put_lockspace(ls);
2568 return error;
2569}
2570
2571int dlm_unlock(dlm_lockspace_t *lockspace,
2572 uint32_t lkid,
2573 uint32_t flags,
2574 struct dlm_lksb *lksb,
2575 void *astarg)
2576{
2577 struct dlm_ls *ls;
2578 struct dlm_lkb *lkb;
2579 struct dlm_args args;
2580 int error;
2581
2582 ls = dlm_find_lockspace_local(lockspace);
2583 if (!ls)
2584 return -EINVAL;
2585
David Teigland85e86ed2007-05-18 08:58:15 -05002586 dlm_lock_recovery(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00002587
2588 error = find_lkb(ls, lkid, &lkb);
2589 if (error)
2590 goto out;
2591
2592 error = set_unlock_args(flags, astarg, &args);
2593 if (error)
2594 goto out_put;
2595
2596 if (flags & DLM_LKF_CANCEL)
2597 error = cancel_lock(ls, lkb, &args);
2598 else
2599 error = unlock_lock(ls, lkb, &args);
2600
2601 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2602 error = 0;
David Teiglandef0c2bb2007-03-28 09:56:46 -05002603 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2604 error = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00002605 out_put:
David Teiglandb3f58d82006-02-28 11:16:37 -05002606 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00002607 out:
David Teigland85e86ed2007-05-18 08:58:15 -05002608 dlm_unlock_recovery(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00002609 dlm_put_lockspace(ls);
2610 return error;
2611}
2612
2613/*
2614 * send/receive routines for remote operations and replies
2615 *
2616 * send_args
2617 * send_common
2618 * send_request receive_request
2619 * send_convert receive_convert
2620 * send_unlock receive_unlock
2621 * send_cancel receive_cancel
2622 * send_grant receive_grant
2623 * send_bast receive_bast
2624 * send_lookup receive_lookup
2625 * send_remove receive_remove
2626 *
2627 * send_common_reply
2628 * receive_request_reply send_request_reply
2629 * receive_convert_reply send_convert_reply
2630 * receive_unlock_reply send_unlock_reply
2631 * receive_cancel_reply send_cancel_reply
2632 * receive_lookup_reply send_lookup_reply
2633 */
2634
David Teigland7e4dac32007-04-02 09:06:41 -05002635static int _create_message(struct dlm_ls *ls, int mb_len,
2636 int to_nodeid, int mstype,
2637 struct dlm_message **ms_ret,
2638 struct dlm_mhandle **mh_ret)
2639{
2640 struct dlm_message *ms;
2641 struct dlm_mhandle *mh;
2642 char *mb;
2643
2644 /* get_buffer gives us a message handle (mh) that we need to
2645 pass into lowcomms_commit and a message buffer (mb) that we
2646 write our data into */
2647
Patrick Caulfield44f487a2007-06-06 09:21:22 -05002648 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
David Teigland7e4dac32007-04-02 09:06:41 -05002649 if (!mh)
2650 return -ENOBUFS;
2651
2652 memset(mb, 0, mb_len);
2653
2654 ms = (struct dlm_message *) mb;
2655
2656 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2657 ms->m_header.h_lockspace = ls->ls_global_id;
2658 ms->m_header.h_nodeid = dlm_our_nodeid();
2659 ms->m_header.h_length = mb_len;
2660 ms->m_header.h_cmd = DLM_MSG;
2661
2662 ms->m_type = mstype;
2663
2664 *mh_ret = mh;
2665 *ms_ret = ms;
2666 return 0;
2667}
2668
David Teiglande7fd4172006-01-18 09:30:29 +00002669static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2670 int to_nodeid, int mstype,
2671 struct dlm_message **ms_ret,
2672 struct dlm_mhandle **mh_ret)
2673{
David Teiglande7fd4172006-01-18 09:30:29 +00002674 int mb_len = sizeof(struct dlm_message);
2675
2676 switch (mstype) {
2677 case DLM_MSG_REQUEST:
2678 case DLM_MSG_LOOKUP:
2679 case DLM_MSG_REMOVE:
2680 mb_len += r->res_length;
2681 break;
2682 case DLM_MSG_CONVERT:
2683 case DLM_MSG_UNLOCK:
2684 case DLM_MSG_REQUEST_REPLY:
2685 case DLM_MSG_CONVERT_REPLY:
2686 case DLM_MSG_GRANT:
2687 if (lkb && lkb->lkb_lvbptr)
2688 mb_len += r->res_ls->ls_lvblen;
2689 break;
2690 }
2691
David Teigland7e4dac32007-04-02 09:06:41 -05002692 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2693 ms_ret, mh_ret);
David Teiglande7fd4172006-01-18 09:30:29 +00002694}
2695
2696/* further lowcomms enhancements or alternate implementations may make
2697 the return value from this function useful at some point */
2698
2699static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2700{
2701 dlm_message_out(ms);
2702 dlm_lowcomms_commit_buffer(mh);
2703 return 0;
2704}
2705
2706static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2707 struct dlm_message *ms)
2708{
2709 ms->m_nodeid = lkb->lkb_nodeid;
2710 ms->m_pid = lkb->lkb_ownpid;
2711 ms->m_lkid = lkb->lkb_id;
2712 ms->m_remid = lkb->lkb_remid;
2713 ms->m_exflags = lkb->lkb_exflags;
2714 ms->m_sbflags = lkb->lkb_sbflags;
2715 ms->m_flags = lkb->lkb_flags;
2716 ms->m_lvbseq = lkb->lkb_lvbseq;
2717 ms->m_status = lkb->lkb_status;
2718 ms->m_grmode = lkb->lkb_grmode;
2719 ms->m_rqmode = lkb->lkb_rqmode;
2720 ms->m_hash = r->res_hash;
2721
2722 /* m_result and m_bastmode are set from function args,
2723 not from lkb fields */
2724
David Teiglande5dae542008-02-06 00:35:45 -06002725 if (lkb->lkb_bastfn)
David Teiglande7fd4172006-01-18 09:30:29 +00002726 ms->m_asts |= AST_BAST;
David Teiglande5dae542008-02-06 00:35:45 -06002727 if (lkb->lkb_astfn)
David Teiglande7fd4172006-01-18 09:30:29 +00002728 ms->m_asts |= AST_COMP;
2729
David Teiglandda49f362006-12-13 10:38:45 -06002730 /* compare with switch in create_message; send_remove() doesn't
2731 use send_args() */
2732
2733 switch (ms->m_type) {
2734 case DLM_MSG_REQUEST:
2735 case DLM_MSG_LOOKUP:
David Teiglande7fd4172006-01-18 09:30:29 +00002736 memcpy(ms->m_extra, r->res_name, r->res_length);
David Teiglandda49f362006-12-13 10:38:45 -06002737 break;
2738 case DLM_MSG_CONVERT:
2739 case DLM_MSG_UNLOCK:
2740 case DLM_MSG_REQUEST_REPLY:
2741 case DLM_MSG_CONVERT_REPLY:
2742 case DLM_MSG_GRANT:
2743 if (!lkb->lkb_lvbptr)
2744 break;
David Teiglande7fd4172006-01-18 09:30:29 +00002745 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
David Teiglandda49f362006-12-13 10:38:45 -06002746 break;
2747 }
David Teiglande7fd4172006-01-18 09:30:29 +00002748}
2749
2750static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2751{
2752 struct dlm_message *ms;
2753 struct dlm_mhandle *mh;
2754 int to_nodeid, error;
2755
David Teiglandef0c2bb2007-03-28 09:56:46 -05002756 error = add_to_waiters(lkb, mstype);
2757 if (error)
2758 return error;
David Teiglande7fd4172006-01-18 09:30:29 +00002759
2760 to_nodeid = r->res_nodeid;
2761
2762 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2763 if (error)
2764 goto fail;
2765
2766 send_args(r, lkb, ms);
2767
2768 error = send_message(mh, ms);
2769 if (error)
2770 goto fail;
2771 return 0;
2772
2773 fail:
David Teiglandef0c2bb2007-03-28 09:56:46 -05002774 remove_from_waiters(lkb, msg_reply_type(mstype));
David Teiglande7fd4172006-01-18 09:30:29 +00002775 return error;
2776}
2777
2778static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2779{
2780 return send_common(r, lkb, DLM_MSG_REQUEST);
2781}
2782
2783static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2784{
2785 int error;
2786
2787 error = send_common(r, lkb, DLM_MSG_CONVERT);
2788
2789 /* down conversions go without a reply from the master */
2790 if (!error && down_conversion(lkb)) {
David Teiglandef0c2bb2007-03-28 09:56:46 -05002791 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
2792 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
David Teiglande7fd4172006-01-18 09:30:29 +00002793 r->res_ls->ls_stub_ms.m_result = 0;
David Teigland32f105a2006-08-23 16:07:31 -04002794 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
David Teiglande7fd4172006-01-18 09:30:29 +00002795 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2796 }
2797
2798 return error;
2799}
2800
2801/* FIXME: if this lkb is the only lock we hold on the rsb, then set
2802 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2803 that the master is still correct. */
2804
2805static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2806{
2807 return send_common(r, lkb, DLM_MSG_UNLOCK);
2808}
2809
2810static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2811{
2812 return send_common(r, lkb, DLM_MSG_CANCEL);
2813}
2814
2815static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2816{
2817 struct dlm_message *ms;
2818 struct dlm_mhandle *mh;
2819 int to_nodeid, error;
2820
2821 to_nodeid = lkb->lkb_nodeid;
2822
2823 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2824 if (error)
2825 goto out;
2826
2827 send_args(r, lkb, ms);
2828
2829 ms->m_result = 0;
2830
2831 error = send_message(mh, ms);
2832 out:
2833 return error;
2834}
2835
2836static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2837{
2838 struct dlm_message *ms;
2839 struct dlm_mhandle *mh;
2840 int to_nodeid, error;
2841
2842 to_nodeid = lkb->lkb_nodeid;
2843
2844 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2845 if (error)
2846 goto out;
2847
2848 send_args(r, lkb, ms);
2849
2850 ms->m_bastmode = mode;
2851
2852 error = send_message(mh, ms);
2853 out:
2854 return error;
2855}
2856
2857static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2858{
2859 struct dlm_message *ms;
2860 struct dlm_mhandle *mh;
2861 int to_nodeid, error;
2862
David Teiglandef0c2bb2007-03-28 09:56:46 -05002863 error = add_to_waiters(lkb, DLM_MSG_LOOKUP);
2864 if (error)
2865 return error;
David Teiglande7fd4172006-01-18 09:30:29 +00002866
2867 to_nodeid = dlm_dir_nodeid(r);
2868
2869 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2870 if (error)
2871 goto fail;
2872
2873 send_args(r, lkb, ms);
2874
2875 error = send_message(mh, ms);
2876 if (error)
2877 goto fail;
2878 return 0;
2879
2880 fail:
David Teiglandef0c2bb2007-03-28 09:56:46 -05002881 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
David Teiglande7fd4172006-01-18 09:30:29 +00002882 return error;
2883}
2884
2885static int send_remove(struct dlm_rsb *r)
2886{
2887 struct dlm_message *ms;
2888 struct dlm_mhandle *mh;
2889 int to_nodeid, error;
2890
2891 to_nodeid = dlm_dir_nodeid(r);
2892
2893 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2894 if (error)
2895 goto out;
2896
2897 memcpy(ms->m_extra, r->res_name, r->res_length);
2898 ms->m_hash = r->res_hash;
2899
2900 error = send_message(mh, ms);
2901 out:
2902 return error;
2903}
2904
2905static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2906 int mstype, int rv)
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, mstype, &ms, &mh);
2915 if (error)
2916 goto out;
2917
2918 send_args(r, lkb, ms);
2919
2920 ms->m_result = rv;
2921
2922 error = send_message(mh, ms);
2923 out:
2924 return error;
2925}
2926
2927static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2928{
2929 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2930}
2931
2932static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2933{
2934 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2935}
2936
2937static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2938{
2939 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2940}
2941
2942static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2943{
2944 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2945}
2946
2947static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2948 int ret_nodeid, int rv)
2949{
2950 struct dlm_rsb *r = &ls->ls_stub_rsb;
2951 struct dlm_message *ms;
2952 struct dlm_mhandle *mh;
2953 int error, nodeid = ms_in->m_header.h_nodeid;
2954
2955 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2956 if (error)
2957 goto out;
2958
2959 ms->m_lkid = ms_in->m_lkid;
2960 ms->m_result = rv;
2961 ms->m_nodeid = ret_nodeid;
2962
2963 error = send_message(mh, ms);
2964 out:
2965 return error;
2966}
2967
2968/* which args we save from a received message depends heavily on the type
2969 of message, unlike the send side where we can safely send everything about
2970 the lkb for any type of message */
2971
2972static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2973{
2974 lkb->lkb_exflags = ms->m_exflags;
David Teigland6f90a8b12006-11-10 14:16:27 -06002975 lkb->lkb_sbflags = ms->m_sbflags;
David Teiglande7fd4172006-01-18 09:30:29 +00002976 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2977 (ms->m_flags & 0x0000FFFF);
2978}
2979
2980static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2981{
2982 lkb->lkb_sbflags = ms->m_sbflags;
2983 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2984 (ms->m_flags & 0x0000FFFF);
2985}
2986
2987static int receive_extralen(struct dlm_message *ms)
2988{
2989 return (ms->m_header.h_length - sizeof(struct dlm_message));
2990}
2991
David Teiglande7fd4172006-01-18 09:30:29 +00002992static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2993 struct dlm_message *ms)
2994{
2995 int len;
2996
2997 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2998 if (!lkb->lkb_lvbptr)
David Teigland52bda2b2007-11-07 09:06:49 -06002999 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00003000 if (!lkb->lkb_lvbptr)
3001 return -ENOMEM;
3002 len = receive_extralen(ms);
Al Viroa9cc9152008-01-26 00:02:29 -05003003 if (len > DLM_RESNAME_MAXLEN)
3004 len = DLM_RESNAME_MAXLEN;
David Teiglande7fd4172006-01-18 09:30:29 +00003005 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3006 }
3007 return 0;
3008}
3009
David Teiglande5dae542008-02-06 00:35:45 -06003010static void fake_bastfn(void *astparam, int mode)
3011{
3012 log_print("fake_bastfn should not be called");
3013}
3014
3015static void fake_astfn(void *astparam)
3016{
3017 log_print("fake_astfn should not be called");
3018}
3019
David Teiglande7fd4172006-01-18 09:30:29 +00003020static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3021 struct dlm_message *ms)
3022{
3023 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3024 lkb->lkb_ownpid = ms->m_pid;
3025 lkb->lkb_remid = ms->m_lkid;
3026 lkb->lkb_grmode = DLM_LOCK_IV;
3027 lkb->lkb_rqmode = ms->m_rqmode;
David Teiglande5dae542008-02-06 00:35:45 -06003028
3029 lkb->lkb_bastfn = (ms->m_asts & AST_BAST) ? &fake_bastfn : NULL;
3030 lkb->lkb_astfn = (ms->m_asts & AST_COMP) ? &fake_astfn : NULL;
David Teiglande7fd4172006-01-18 09:30:29 +00003031
David Teigland8d07fd52006-12-13 10:39:20 -06003032 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3033 /* lkb was just created so there won't be an lvb yet */
David Teigland52bda2b2007-11-07 09:06:49 -06003034 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
David Teigland8d07fd52006-12-13 10:39:20 -06003035 if (!lkb->lkb_lvbptr)
3036 return -ENOMEM;
3037 }
David Teiglande7fd4172006-01-18 09:30:29 +00003038
3039 return 0;
3040}
3041
3042static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3043 struct dlm_message *ms)
3044{
David Teiglande7fd4172006-01-18 09:30:29 +00003045 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3046 return -EBUSY;
3047
David Teiglande7fd4172006-01-18 09:30:29 +00003048 if (receive_lvb(ls, lkb, ms))
3049 return -ENOMEM;
3050
3051 lkb->lkb_rqmode = ms->m_rqmode;
3052 lkb->lkb_lvbseq = ms->m_lvbseq;
3053
3054 return 0;
3055}
3056
3057static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3058 struct dlm_message *ms)
3059{
David Teiglande7fd4172006-01-18 09:30:29 +00003060 if (receive_lvb(ls, lkb, ms))
3061 return -ENOMEM;
3062 return 0;
3063}
3064
3065/* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3066 uses to send a reply and that the remote end uses to process the reply. */
3067
3068static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3069{
3070 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3071 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3072 lkb->lkb_remid = ms->m_lkid;
3073}
3074
David Teiglandc54e04b2008-01-09 09:59:41 -06003075/* This is called after the rsb is locked so that we can safely inspect
3076 fields in the lkb. */
3077
3078static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3079{
3080 int from = ms->m_header.h_nodeid;
3081 int error = 0;
3082
3083 switch (ms->m_type) {
3084 case DLM_MSG_CONVERT:
3085 case DLM_MSG_UNLOCK:
3086 case DLM_MSG_CANCEL:
3087 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3088 error = -EINVAL;
3089 break;
3090
3091 case DLM_MSG_CONVERT_REPLY:
3092 case DLM_MSG_UNLOCK_REPLY:
3093 case DLM_MSG_CANCEL_REPLY:
3094 case DLM_MSG_GRANT:
3095 case DLM_MSG_BAST:
3096 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3097 error = -EINVAL;
3098 break;
3099
3100 case DLM_MSG_REQUEST_REPLY:
3101 if (!is_process_copy(lkb))
3102 error = -EINVAL;
3103 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3104 error = -EINVAL;
3105 break;
3106
3107 default:
3108 error = -EINVAL;
3109 }
3110
3111 if (error)
3112 log_error(lkb->lkb_resource->res_ls,
3113 "ignore invalid message %d from %d %x %x %x %d",
3114 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
3115 lkb->lkb_flags, lkb->lkb_nodeid);
3116 return error;
3117}
3118
David Teiglande7fd4172006-01-18 09:30:29 +00003119static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3120{
3121 struct dlm_lkb *lkb;
3122 struct dlm_rsb *r;
3123 int error, namelen;
3124
3125 error = create_lkb(ls, &lkb);
3126 if (error)
3127 goto fail;
3128
3129 receive_flags(lkb, ms);
3130 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3131 error = receive_request_args(ls, lkb, ms);
3132 if (error) {
David Teiglandb3f58d82006-02-28 11:16:37 -05003133 __put_lkb(ls, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003134 goto fail;
3135 }
3136
3137 namelen = receive_extralen(ms);
3138
3139 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3140 if (error) {
David Teiglandb3f58d82006-02-28 11:16:37 -05003141 __put_lkb(ls, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003142 goto fail;
3143 }
3144
3145 lock_rsb(r);
3146
3147 attach_lkb(r, lkb);
3148 error = do_request(r, lkb);
3149 send_request_reply(r, lkb, error);
3150
3151 unlock_rsb(r);
3152 put_rsb(r);
3153
3154 if (error == -EINPROGRESS)
3155 error = 0;
3156 if (error)
David Teiglandb3f58d82006-02-28 11:16:37 -05003157 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003158 return;
3159
3160 fail:
3161 setup_stub_lkb(ls, ms);
3162 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3163}
3164
3165static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3166{
3167 struct dlm_lkb *lkb;
3168 struct dlm_rsb *r;
David Teigland90135922006-01-20 08:47:07 +00003169 int error, reply = 1;
David Teiglande7fd4172006-01-18 09:30:29 +00003170
3171 error = find_lkb(ls, ms->m_remid, &lkb);
3172 if (error)
3173 goto fail;
3174
3175 r = lkb->lkb_resource;
3176
3177 hold_rsb(r);
3178 lock_rsb(r);
3179
David Teiglandc54e04b2008-01-09 09:59:41 -06003180 error = validate_message(lkb, ms);
3181 if (error)
3182 goto out;
3183
David Teiglande7fd4172006-01-18 09:30:29 +00003184 receive_flags(lkb, ms);
3185 error = receive_convert_args(ls, lkb, ms);
3186 if (error)
David Teiglandc54e04b2008-01-09 09:59:41 -06003187 goto out_reply;
David Teiglande7fd4172006-01-18 09:30:29 +00003188 reply = !down_conversion(lkb);
3189
3190 error = do_convert(r, lkb);
David Teiglandc54e04b2008-01-09 09:59:41 -06003191 out_reply:
David Teiglande7fd4172006-01-18 09:30:29 +00003192 if (reply)
3193 send_convert_reply(r, lkb, error);
David Teiglandc54e04b2008-01-09 09:59:41 -06003194 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003195 unlock_rsb(r);
3196 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003197 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003198 return;
3199
3200 fail:
3201 setup_stub_lkb(ls, ms);
3202 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3203}
3204
3205static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3206{
3207 struct dlm_lkb *lkb;
3208 struct dlm_rsb *r;
3209 int error;
3210
3211 error = find_lkb(ls, ms->m_remid, &lkb);
3212 if (error)
3213 goto fail;
3214
3215 r = lkb->lkb_resource;
3216
3217 hold_rsb(r);
3218 lock_rsb(r);
3219
David Teiglandc54e04b2008-01-09 09:59:41 -06003220 error = validate_message(lkb, ms);
David Teiglande7fd4172006-01-18 09:30:29 +00003221 if (error)
3222 goto out;
3223
David Teiglandc54e04b2008-01-09 09:59:41 -06003224 receive_flags(lkb, ms);
3225 error = receive_unlock_args(ls, lkb, ms);
3226 if (error)
3227 goto out_reply;
David Teiglande7fd4172006-01-18 09:30:29 +00003228
David Teiglandc54e04b2008-01-09 09:59:41 -06003229 error = do_unlock(r, lkb);
3230 out_reply:
3231 send_unlock_reply(r, lkb, error);
3232 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003233 unlock_rsb(r);
3234 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003235 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003236 return;
3237
3238 fail:
3239 setup_stub_lkb(ls, ms);
3240 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3241}
3242
3243static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3244{
3245 struct dlm_lkb *lkb;
3246 struct dlm_rsb *r;
3247 int error;
3248
3249 error = find_lkb(ls, ms->m_remid, &lkb);
3250 if (error)
3251 goto fail;
3252
3253 receive_flags(lkb, ms);
3254
3255 r = lkb->lkb_resource;
3256
3257 hold_rsb(r);
3258 lock_rsb(r);
3259
David Teiglandc54e04b2008-01-09 09:59:41 -06003260 error = validate_message(lkb, ms);
3261 if (error)
3262 goto out;
3263
David Teiglande7fd4172006-01-18 09:30:29 +00003264 error = do_cancel(r, lkb);
3265 send_cancel_reply(r, lkb, error);
David Teiglandc54e04b2008-01-09 09:59:41 -06003266 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003267 unlock_rsb(r);
3268 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003269 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003270 return;
3271
3272 fail:
3273 setup_stub_lkb(ls, ms);
3274 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3275}
3276
3277static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3278{
3279 struct dlm_lkb *lkb;
3280 struct dlm_rsb *r;
3281 int error;
3282
3283 error = find_lkb(ls, ms->m_remid, &lkb);
3284 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003285 log_debug(ls, "receive_grant from %d no lkb %x",
3286 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003287 return;
3288 }
David Teiglande7fd4172006-01-18 09:30:29 +00003289
3290 r = lkb->lkb_resource;
3291
3292 hold_rsb(r);
3293 lock_rsb(r);
3294
David Teiglandc54e04b2008-01-09 09:59:41 -06003295 error = validate_message(lkb, ms);
3296 if (error)
3297 goto out;
3298
David Teiglande7fd4172006-01-18 09:30:29 +00003299 receive_flags_reply(lkb, ms);
David Teigland7d3c1fe2007-04-19 10:30:41 -05003300 if (is_altmode(lkb))
3301 munge_altmode(lkb, ms);
David Teiglande7fd4172006-01-18 09:30:29 +00003302 grant_lock_pc(r, lkb, ms);
3303 queue_cast(r, lkb, 0);
David Teiglandc54e04b2008-01-09 09:59:41 -06003304 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003305 unlock_rsb(r);
3306 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003307 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003308}
3309
3310static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3311{
3312 struct dlm_lkb *lkb;
3313 struct dlm_rsb *r;
3314 int error;
3315
3316 error = find_lkb(ls, ms->m_remid, &lkb);
3317 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003318 log_debug(ls, "receive_bast from %d no lkb %x",
3319 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003320 return;
3321 }
David Teiglande7fd4172006-01-18 09:30:29 +00003322
3323 r = lkb->lkb_resource;
3324
3325 hold_rsb(r);
3326 lock_rsb(r);
3327
David Teiglandc54e04b2008-01-09 09:59:41 -06003328 error = validate_message(lkb, ms);
3329 if (error)
3330 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +00003331
David Teiglandc54e04b2008-01-09 09:59:41 -06003332 queue_bast(r, lkb, ms->m_bastmode);
3333 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003334 unlock_rsb(r);
3335 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003336 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003337}
3338
3339static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3340{
3341 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3342
3343 from_nodeid = ms->m_header.h_nodeid;
3344 our_nodeid = dlm_our_nodeid();
3345
3346 len = receive_extralen(ms);
3347
3348 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3349 if (dir_nodeid != our_nodeid) {
3350 log_error(ls, "lookup dir_nodeid %d from %d",
3351 dir_nodeid, from_nodeid);
3352 error = -EINVAL;
3353 ret_nodeid = -1;
3354 goto out;
3355 }
3356
3357 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3358
3359 /* Optimization: we're master so treat lookup as a request */
3360 if (!error && ret_nodeid == our_nodeid) {
3361 receive_request(ls, ms);
3362 return;
3363 }
3364 out:
3365 send_lookup_reply(ls, ms, ret_nodeid, error);
3366}
3367
3368static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3369{
3370 int len, dir_nodeid, from_nodeid;
3371
3372 from_nodeid = ms->m_header.h_nodeid;
3373
3374 len = receive_extralen(ms);
3375
3376 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3377 if (dir_nodeid != dlm_our_nodeid()) {
3378 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3379 dir_nodeid, from_nodeid);
3380 return;
3381 }
3382
3383 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3384}
3385
David Teigland84991372007-03-30 15:02:40 -05003386static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3387{
3388 do_purge(ls, ms->m_nodeid, ms->m_pid);
3389}
3390
David Teiglande7fd4172006-01-18 09:30:29 +00003391static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3392{
3393 struct dlm_lkb *lkb;
3394 struct dlm_rsb *r;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003395 int error, mstype, result;
David Teiglande7fd4172006-01-18 09:30:29 +00003396
3397 error = find_lkb(ls, ms->m_remid, &lkb);
3398 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003399 log_debug(ls, "receive_request_reply from %d no lkb %x",
3400 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003401 return;
3402 }
David Teiglande7fd4172006-01-18 09:30:29 +00003403
David Teiglande7fd4172006-01-18 09:30:29 +00003404 r = lkb->lkb_resource;
3405 hold_rsb(r);
3406 lock_rsb(r);
3407
David Teiglandc54e04b2008-01-09 09:59:41 -06003408 error = validate_message(lkb, ms);
3409 if (error)
3410 goto out;
3411
David Teiglandef0c2bb2007-03-28 09:56:46 -05003412 mstype = lkb->lkb_wait_type;
3413 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3414 if (error)
3415 goto out;
3416
David Teiglande7fd4172006-01-18 09:30:29 +00003417 /* Optimization: the dir node was also the master, so it took our
3418 lookup as a request and sent request reply instead of lookup reply */
3419 if (mstype == DLM_MSG_LOOKUP) {
3420 r->res_nodeid = ms->m_header.h_nodeid;
3421 lkb->lkb_nodeid = r->res_nodeid;
3422 }
3423
David Teiglandef0c2bb2007-03-28 09:56:46 -05003424 /* this is the value returned from do_request() on the master */
3425 result = ms->m_result;
3426
3427 switch (result) {
David Teiglande7fd4172006-01-18 09:30:29 +00003428 case -EAGAIN:
David Teiglandef0c2bb2007-03-28 09:56:46 -05003429 /* request would block (be queued) on remote master */
David Teiglande7fd4172006-01-18 09:30:29 +00003430 queue_cast(r, lkb, -EAGAIN);
3431 confirm_master(r, -EAGAIN);
David Teiglandef0c2bb2007-03-28 09:56:46 -05003432 unhold_lkb(lkb); /* undoes create_lkb() */
David Teiglande7fd4172006-01-18 09:30:29 +00003433 break;
3434
3435 case -EINPROGRESS:
3436 case 0:
3437 /* request was queued or granted on remote master */
3438 receive_flags_reply(lkb, ms);
3439 lkb->lkb_remid = ms->m_lkid;
David Teigland7d3c1fe2007-04-19 10:30:41 -05003440 if (is_altmode(lkb))
3441 munge_altmode(lkb, ms);
David Teigland3ae1acf2007-05-18 08:59:31 -05003442 if (result) {
David Teiglande7fd4172006-01-18 09:30:29 +00003443 add_lkb(r, lkb, DLM_LKSTS_WAITING);
David Teigland3ae1acf2007-05-18 08:59:31 -05003444 add_timeout(lkb);
3445 } else {
David Teiglande7fd4172006-01-18 09:30:29 +00003446 grant_lock_pc(r, lkb, ms);
3447 queue_cast(r, lkb, 0);
3448 }
David Teiglandef0c2bb2007-03-28 09:56:46 -05003449 confirm_master(r, result);
David Teiglande7fd4172006-01-18 09:30:29 +00003450 break;
3451
David Teigland597d0ca2006-07-12 16:44:04 -05003452 case -EBADR:
David Teiglande7fd4172006-01-18 09:30:29 +00003453 case -ENOTBLK:
3454 /* find_rsb failed to find rsb or rsb wasn't master */
David Teiglandef0c2bb2007-03-28 09:56:46 -05003455 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3456 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
David Teiglande7fd4172006-01-18 09:30:29 +00003457 r->res_nodeid = -1;
3458 lkb->lkb_nodeid = -1;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003459
3460 if (is_overlap(lkb)) {
3461 /* we'll ignore error in cancel/unlock reply */
3462 queue_cast_overlap(r, lkb);
David Teiglandaec64e12008-01-08 15:37:47 -06003463 confirm_master(r, result);
David Teiglandef0c2bb2007-03-28 09:56:46 -05003464 unhold_lkb(lkb); /* undoes create_lkb() */
3465 } else
3466 _request_lock(r, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003467 break;
3468
3469 default:
David Teiglandef0c2bb2007-03-28 09:56:46 -05003470 log_error(ls, "receive_request_reply %x error %d",
3471 lkb->lkb_id, result);
David Teiglande7fd4172006-01-18 09:30:29 +00003472 }
3473
David Teiglandef0c2bb2007-03-28 09:56:46 -05003474 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3475 log_debug(ls, "receive_request_reply %x result %d unlock",
3476 lkb->lkb_id, result);
3477 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3478 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3479 send_unlock(r, lkb);
3480 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3481 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3482 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3483 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3484 send_cancel(r, lkb);
3485 } else {
3486 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3487 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3488 }
3489 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003490 unlock_rsb(r);
3491 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003492 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003493}
3494
3495static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3496 struct dlm_message *ms)
3497{
David Teiglande7fd4172006-01-18 09:30:29 +00003498 /* this is the value returned from do_convert() on the master */
David Teiglandef0c2bb2007-03-28 09:56:46 -05003499 switch (ms->m_result) {
David Teiglande7fd4172006-01-18 09:30:29 +00003500 case -EAGAIN:
3501 /* convert would block (be queued) on remote master */
3502 queue_cast(r, lkb, -EAGAIN);
3503 break;
3504
David Teiglandc85d65e2007-05-18 09:01:26 -05003505 case -EDEADLK:
3506 receive_flags_reply(lkb, ms);
3507 revert_lock_pc(r, lkb);
3508 queue_cast(r, lkb, -EDEADLK);
3509 break;
3510
David Teiglande7fd4172006-01-18 09:30:29 +00003511 case -EINPROGRESS:
3512 /* convert was queued on remote master */
David Teigland7d3c1fe2007-04-19 10:30:41 -05003513 receive_flags_reply(lkb, ms);
3514 if (is_demoted(lkb))
3515 munge_demoted(lkb, ms);
David Teiglande7fd4172006-01-18 09:30:29 +00003516 del_lkb(r, lkb);
3517 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
David Teigland3ae1acf2007-05-18 08:59:31 -05003518 add_timeout(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003519 break;
3520
3521 case 0:
3522 /* convert was granted on remote master */
3523 receive_flags_reply(lkb, ms);
David Teigland7d3c1fe2007-04-19 10:30:41 -05003524 if (is_demoted(lkb))
3525 munge_demoted(lkb, ms);
David Teiglande7fd4172006-01-18 09:30:29 +00003526 grant_lock_pc(r, lkb, ms);
3527 queue_cast(r, lkb, 0);
3528 break;
3529
3530 default:
David Teiglandef0c2bb2007-03-28 09:56:46 -05003531 log_error(r->res_ls, "receive_convert_reply %x error %d",
3532 lkb->lkb_id, ms->m_result);
David Teiglande7fd4172006-01-18 09:30:29 +00003533 }
3534}
3535
3536static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3537{
3538 struct dlm_rsb *r = lkb->lkb_resource;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003539 int error;
David Teiglande7fd4172006-01-18 09:30:29 +00003540
3541 hold_rsb(r);
3542 lock_rsb(r);
3543
David Teiglandc54e04b2008-01-09 09:59:41 -06003544 error = validate_message(lkb, ms);
3545 if (error)
3546 goto out;
3547
David Teiglandef0c2bb2007-03-28 09:56:46 -05003548 /* stub reply can happen with waiters_mutex held */
3549 error = remove_from_waiters_ms(lkb, ms);
3550 if (error)
3551 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +00003552
David Teiglandef0c2bb2007-03-28 09:56:46 -05003553 __receive_convert_reply(r, lkb, ms);
3554 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003555 unlock_rsb(r);
3556 put_rsb(r);
3557}
3558
3559static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3560{
3561 struct dlm_lkb *lkb;
3562 int error;
3563
3564 error = find_lkb(ls, ms->m_remid, &lkb);
3565 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003566 log_debug(ls, "receive_convert_reply from %d no lkb %x",
3567 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003568 return;
3569 }
David Teiglande7fd4172006-01-18 09:30:29 +00003570
David Teiglande7fd4172006-01-18 09:30:29 +00003571 _receive_convert_reply(lkb, ms);
David Teiglandb3f58d82006-02-28 11:16:37 -05003572 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003573}
3574
3575static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3576{
3577 struct dlm_rsb *r = lkb->lkb_resource;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003578 int error;
David Teiglande7fd4172006-01-18 09:30:29 +00003579
3580 hold_rsb(r);
3581 lock_rsb(r);
3582
David Teiglandc54e04b2008-01-09 09:59:41 -06003583 error = validate_message(lkb, ms);
3584 if (error)
3585 goto out;
3586
David Teiglandef0c2bb2007-03-28 09:56:46 -05003587 /* stub reply can happen with waiters_mutex held */
3588 error = remove_from_waiters_ms(lkb, ms);
3589 if (error)
3590 goto out;
3591
David Teiglande7fd4172006-01-18 09:30:29 +00003592 /* this is the value returned from do_unlock() on the master */
3593
David Teiglandef0c2bb2007-03-28 09:56:46 -05003594 switch (ms->m_result) {
David Teiglande7fd4172006-01-18 09:30:29 +00003595 case -DLM_EUNLOCK:
3596 receive_flags_reply(lkb, ms);
3597 remove_lock_pc(r, lkb);
3598 queue_cast(r, lkb, -DLM_EUNLOCK);
3599 break;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003600 case -ENOENT:
3601 break;
David Teiglande7fd4172006-01-18 09:30:29 +00003602 default:
David Teiglandef0c2bb2007-03-28 09:56:46 -05003603 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3604 lkb->lkb_id, ms->m_result);
David Teiglande7fd4172006-01-18 09:30:29 +00003605 }
David Teiglandef0c2bb2007-03-28 09:56:46 -05003606 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003607 unlock_rsb(r);
3608 put_rsb(r);
3609}
3610
3611static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3612{
3613 struct dlm_lkb *lkb;
3614 int error;
3615
3616 error = find_lkb(ls, ms->m_remid, &lkb);
3617 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003618 log_debug(ls, "receive_unlock_reply from %d no lkb %x",
3619 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003620 return;
3621 }
David Teiglande7fd4172006-01-18 09:30:29 +00003622
David Teiglande7fd4172006-01-18 09:30:29 +00003623 _receive_unlock_reply(lkb, ms);
David Teiglandb3f58d82006-02-28 11:16:37 -05003624 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003625}
3626
3627static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3628{
3629 struct dlm_rsb *r = lkb->lkb_resource;
David Teiglandef0c2bb2007-03-28 09:56:46 -05003630 int error;
David Teiglande7fd4172006-01-18 09:30:29 +00003631
3632 hold_rsb(r);
3633 lock_rsb(r);
3634
David Teiglandc54e04b2008-01-09 09:59:41 -06003635 error = validate_message(lkb, ms);
3636 if (error)
3637 goto out;
3638
David Teiglandef0c2bb2007-03-28 09:56:46 -05003639 /* stub reply can happen with waiters_mutex held */
3640 error = remove_from_waiters_ms(lkb, ms);
3641 if (error)
3642 goto out;
3643
David Teiglande7fd4172006-01-18 09:30:29 +00003644 /* this is the value returned from do_cancel() on the master */
3645
David Teiglandef0c2bb2007-03-28 09:56:46 -05003646 switch (ms->m_result) {
David Teiglande7fd4172006-01-18 09:30:29 +00003647 case -DLM_ECANCEL:
3648 receive_flags_reply(lkb, ms);
3649 revert_lock_pc(r, lkb);
David Teigland84d8cd62007-05-29 08:44:23 -05003650 queue_cast(r, lkb, -DLM_ECANCEL);
David Teiglandef0c2bb2007-03-28 09:56:46 -05003651 break;
3652 case 0:
David Teiglande7fd4172006-01-18 09:30:29 +00003653 break;
3654 default:
David Teiglandef0c2bb2007-03-28 09:56:46 -05003655 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3656 lkb->lkb_id, ms->m_result);
David Teiglande7fd4172006-01-18 09:30:29 +00003657 }
David Teiglandef0c2bb2007-03-28 09:56:46 -05003658 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003659 unlock_rsb(r);
3660 put_rsb(r);
3661}
3662
3663static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3664{
3665 struct dlm_lkb *lkb;
3666 int error;
3667
3668 error = find_lkb(ls, ms->m_remid, &lkb);
3669 if (error) {
David Teiglandc54e04b2008-01-09 09:59:41 -06003670 log_debug(ls, "receive_cancel_reply from %d no lkb %x",
3671 ms->m_header.h_nodeid, ms->m_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00003672 return;
3673 }
David Teiglande7fd4172006-01-18 09:30:29 +00003674
David Teiglande7fd4172006-01-18 09:30:29 +00003675 _receive_cancel_reply(lkb, ms);
David Teiglandb3f58d82006-02-28 11:16:37 -05003676 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003677}
3678
3679static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3680{
3681 struct dlm_lkb *lkb;
3682 struct dlm_rsb *r;
3683 int error, ret_nodeid;
3684
3685 error = find_lkb(ls, ms->m_lkid, &lkb);
3686 if (error) {
3687 log_error(ls, "receive_lookup_reply no lkb");
3688 return;
3689 }
3690
David Teiglandef0c2bb2007-03-28 09:56:46 -05003691 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
David Teiglande7fd4172006-01-18 09:30:29 +00003692 FIXME: will a non-zero error ever be returned? */
David Teiglande7fd4172006-01-18 09:30:29 +00003693
3694 r = lkb->lkb_resource;
3695 hold_rsb(r);
3696 lock_rsb(r);
3697
David Teiglandef0c2bb2007-03-28 09:56:46 -05003698 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3699 if (error)
3700 goto out;
3701
David Teiglande7fd4172006-01-18 09:30:29 +00003702 ret_nodeid = ms->m_nodeid;
3703 if (ret_nodeid == dlm_our_nodeid()) {
3704 r->res_nodeid = 0;
3705 ret_nodeid = 0;
3706 r->res_first_lkid = 0;
3707 } else {
3708 /* set_master() will copy res_nodeid to lkb_nodeid */
3709 r->res_nodeid = ret_nodeid;
3710 }
3711
David Teiglandef0c2bb2007-03-28 09:56:46 -05003712 if (is_overlap(lkb)) {
3713 log_debug(ls, "receive_lookup_reply %x unlock %x",
3714 lkb->lkb_id, lkb->lkb_flags);
3715 queue_cast_overlap(r, lkb);
3716 unhold_lkb(lkb); /* undoes create_lkb() */
3717 goto out_list;
3718 }
3719
David Teiglande7fd4172006-01-18 09:30:29 +00003720 _request_lock(r, lkb);
3721
David Teiglandef0c2bb2007-03-28 09:56:46 -05003722 out_list:
David Teiglande7fd4172006-01-18 09:30:29 +00003723 if (!ret_nodeid)
3724 process_lookup_list(r);
David Teiglandef0c2bb2007-03-28 09:56:46 -05003725 out:
David Teiglande7fd4172006-01-18 09:30:29 +00003726 unlock_rsb(r);
3727 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05003728 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00003729}
3730
David Teiglandc36258b2007-09-27 15:53:38 -05003731static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
David Teiglande7fd4172006-01-18 09:30:29 +00003732{
David Teigland46b43ee2008-01-08 16:24:00 -06003733 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
3734 log_debug(ls, "ignore non-member message %d from %d %x %x %d",
3735 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
3736 ms->m_remid, ms->m_result);
3737 return;
3738 }
3739
David Teiglande7fd4172006-01-18 09:30:29 +00003740 switch (ms->m_type) {
3741
3742 /* messages sent to a master node */
3743
3744 case DLM_MSG_REQUEST:
3745 receive_request(ls, ms);
3746 break;
3747
3748 case DLM_MSG_CONVERT:
3749 receive_convert(ls, ms);
3750 break;
3751
3752 case DLM_MSG_UNLOCK:
3753 receive_unlock(ls, ms);
3754 break;
3755
3756 case DLM_MSG_CANCEL:
3757 receive_cancel(ls, ms);
3758 break;
3759
3760 /* messages sent from a master node (replies to above) */
3761
3762 case DLM_MSG_REQUEST_REPLY:
3763 receive_request_reply(ls, ms);
3764 break;
3765
3766 case DLM_MSG_CONVERT_REPLY:
3767 receive_convert_reply(ls, ms);
3768 break;
3769
3770 case DLM_MSG_UNLOCK_REPLY:
3771 receive_unlock_reply(ls, ms);
3772 break;
3773
3774 case DLM_MSG_CANCEL_REPLY:
3775 receive_cancel_reply(ls, ms);
3776 break;
3777
3778 /* messages sent from a master node (only two types of async msg) */
3779
3780 case DLM_MSG_GRANT:
3781 receive_grant(ls, ms);
3782 break;
3783
3784 case DLM_MSG_BAST:
3785 receive_bast(ls, ms);
3786 break;
3787
3788 /* messages sent to a dir node */
3789
3790 case DLM_MSG_LOOKUP:
3791 receive_lookup(ls, ms);
3792 break;
3793
3794 case DLM_MSG_REMOVE:
3795 receive_remove(ls, ms);
3796 break;
3797
3798 /* messages sent from a dir node (remove has no reply) */
3799
3800 case DLM_MSG_LOOKUP_REPLY:
3801 receive_lookup_reply(ls, ms);
3802 break;
3803
David Teigland84991372007-03-30 15:02:40 -05003804 /* other messages */
3805
3806 case DLM_MSG_PURGE:
3807 receive_purge(ls, ms);
3808 break;
3809
David Teiglande7fd4172006-01-18 09:30:29 +00003810 default:
3811 log_error(ls, "unknown message type %d", ms->m_type);
3812 }
3813
David Teiglande7fd4172006-01-18 09:30:29 +00003814 dlm_astd_wake();
David Teiglande7fd4172006-01-18 09:30:29 +00003815}
3816
David Teiglandc36258b2007-09-27 15:53:38 -05003817/* If the lockspace is in recovery mode (locking stopped), then normal
3818 messages are saved on the requestqueue for processing after recovery is
3819 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3820 messages off the requestqueue before we process new ones. This occurs right
3821 after recovery completes when we transition from saving all messages on
3822 requestqueue, to processing all the saved messages, to processing new
3823 messages as they arrive. */
David Teiglande7fd4172006-01-18 09:30:29 +00003824
David Teiglandc36258b2007-09-27 15:53:38 -05003825static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
3826 int nodeid)
3827{
3828 if (dlm_locking_stopped(ls)) {
Al Viro8b0d8e02008-01-25 00:28:28 -05003829 dlm_add_requestqueue(ls, nodeid, ms);
David Teiglandc36258b2007-09-27 15:53:38 -05003830 } else {
3831 dlm_wait_requestqueue(ls);
3832 _receive_message(ls, ms);
3833 }
3834}
3835
3836/* This is called by dlm_recoverd to process messages that were saved on
3837 the requestqueue. */
3838
3839void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
3840{
3841 _receive_message(ls, ms);
3842}
3843
3844/* This is called by the midcomms layer when something is received for
3845 the lockspace. It could be either a MSG (normal message sent as part of
3846 standard locking activity) or an RCOM (recovery message sent as part of
3847 lockspace recovery). */
3848
Al Viroeef7d732008-01-25 00:58:46 -05003849void dlm_receive_buffer(union dlm_packet *p, int nodeid)
David Teiglandc36258b2007-09-27 15:53:38 -05003850{
Al Viroeef7d732008-01-25 00:58:46 -05003851 struct dlm_header *hd = &p->header;
David Teiglandc36258b2007-09-27 15:53:38 -05003852 struct dlm_ls *ls;
3853 int type = 0;
3854
3855 switch (hd->h_cmd) {
3856 case DLM_MSG:
Al Viroeef7d732008-01-25 00:58:46 -05003857 dlm_message_in(&p->message);
3858 type = p->message.m_type;
David Teiglandc36258b2007-09-27 15:53:38 -05003859 break;
3860 case DLM_RCOM:
Al Viroeef7d732008-01-25 00:58:46 -05003861 dlm_rcom_in(&p->rcom);
3862 type = p->rcom.rc_type;
David Teiglandc36258b2007-09-27 15:53:38 -05003863 break;
3864 default:
3865 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
3866 return;
3867 }
3868
3869 if (hd->h_nodeid != nodeid) {
3870 log_print("invalid h_nodeid %d from %d lockspace %x",
3871 hd->h_nodeid, nodeid, hd->h_lockspace);
3872 return;
3873 }
3874
3875 ls = dlm_find_lockspace_global(hd->h_lockspace);
3876 if (!ls) {
David Teigland594199e2008-01-16 11:03:41 -06003877 if (dlm_config.ci_log_debug)
3878 log_print("invalid lockspace %x from %d cmd %d type %d",
3879 hd->h_lockspace, nodeid, hd->h_cmd, type);
David Teiglandc36258b2007-09-27 15:53:38 -05003880
3881 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
Al Viroeef7d732008-01-25 00:58:46 -05003882 dlm_send_ls_not_ready(nodeid, &p->rcom);
David Teiglandc36258b2007-09-27 15:53:38 -05003883 return;
3884 }
3885
3886 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3887 be inactive (in this ls) before transitioning to recovery mode */
3888
3889 down_read(&ls->ls_recv_active);
3890 if (hd->h_cmd == DLM_MSG)
Al Viroeef7d732008-01-25 00:58:46 -05003891 dlm_receive_message(ls, &p->message, nodeid);
David Teiglandc36258b2007-09-27 15:53:38 -05003892 else
Al Viroeef7d732008-01-25 00:58:46 -05003893 dlm_receive_rcom(ls, &p->rcom, nodeid);
David Teiglandc36258b2007-09-27 15:53:38 -05003894 up_read(&ls->ls_recv_active);
3895
3896 dlm_put_lockspace(ls);
3897}
David Teiglande7fd4172006-01-18 09:30:29 +00003898
3899static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3900{
3901 if (middle_conversion(lkb)) {
3902 hold_lkb(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05003903 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
David Teiglande7fd4172006-01-18 09:30:29 +00003904 ls->ls_stub_ms.m_result = -EINPROGRESS;
David Teigland075529b2006-12-13 10:40:26 -06003905 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
David Teiglandc54e04b2008-01-09 09:59:41 -06003906 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
David Teiglande7fd4172006-01-18 09:30:29 +00003907 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3908
3909 /* Same special case as in receive_rcom_lock_args() */
3910 lkb->lkb_grmode = DLM_LOCK_IV;
3911 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3912 unhold_lkb(lkb);
3913
3914 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3915 lkb->lkb_flags |= DLM_IFL_RESEND;
3916 }
3917
3918 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3919 conversions are async; there's no reply from the remote master */
3920}
3921
3922/* A waiting lkb needs recovery if the master node has failed, or
3923 the master node is changing (only when no directory is used) */
3924
3925static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3926{
3927 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3928 return 1;
3929
3930 if (!dlm_no_directory(ls))
3931 return 0;
3932
3933 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3934 return 1;
3935
3936 return 0;
3937}
3938
3939/* Recovery for locks that are waiting for replies from nodes that are now
3940 gone. We can just complete unlocks and cancels by faking a reply from the
3941 dead node. Requests and up-conversions we flag to be resent after
3942 recovery. Down-conversions can just be completed with a fake reply like
3943 unlocks. Conversions between PR and CW need special attention. */
3944
3945void dlm_recover_waiters_pre(struct dlm_ls *ls)
3946{
3947 struct dlm_lkb *lkb, *safe;
David Teigland601342c2008-01-07 16:15:05 -06003948 int wait_type, stub_unlock_result, stub_cancel_result;
David Teiglande7fd4172006-01-18 09:30:29 +00003949
David Teigland90135922006-01-20 08:47:07 +00003950 mutex_lock(&ls->ls_waiters_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +00003951
3952 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3953 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3954 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3955
3956 /* all outstanding lookups, regardless of destination will be
3957 resent after recovery is done */
3958
3959 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3960 lkb->lkb_flags |= DLM_IFL_RESEND;
3961 continue;
3962 }
3963
3964 if (!waiter_needs_recovery(ls, lkb))
3965 continue;
3966
David Teigland601342c2008-01-07 16:15:05 -06003967 wait_type = lkb->lkb_wait_type;
3968 stub_unlock_result = -DLM_EUNLOCK;
3969 stub_cancel_result = -DLM_ECANCEL;
3970
3971 /* Main reply may have been received leaving a zero wait_type,
3972 but a reply for the overlapping op may not have been
3973 received. In that case we need to fake the appropriate
3974 reply for the overlap op. */
3975
3976 if (!wait_type) {
3977 if (is_overlap_cancel(lkb)) {
3978 wait_type = DLM_MSG_CANCEL;
3979 if (lkb->lkb_grmode == DLM_LOCK_IV)
3980 stub_cancel_result = 0;
3981 }
3982 if (is_overlap_unlock(lkb)) {
3983 wait_type = DLM_MSG_UNLOCK;
3984 if (lkb->lkb_grmode == DLM_LOCK_IV)
3985 stub_unlock_result = -ENOENT;
3986 }
3987
3988 log_debug(ls, "rwpre overlap %x %x %d %d %d",
3989 lkb->lkb_id, lkb->lkb_flags, wait_type,
3990 stub_cancel_result, stub_unlock_result);
3991 }
3992
3993 switch (wait_type) {
David Teiglande7fd4172006-01-18 09:30:29 +00003994
3995 case DLM_MSG_REQUEST:
3996 lkb->lkb_flags |= DLM_IFL_RESEND;
3997 break;
3998
3999 case DLM_MSG_CONVERT:
4000 recover_convert_waiter(ls, lkb);
4001 break;
4002
4003 case DLM_MSG_UNLOCK:
4004 hold_lkb(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004005 ls->ls_stub_ms.m_type = DLM_MSG_UNLOCK_REPLY;
David Teigland601342c2008-01-07 16:15:05 -06004006 ls->ls_stub_ms.m_result = stub_unlock_result;
David Teigland075529b2006-12-13 10:40:26 -06004007 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
David Teiglandc54e04b2008-01-09 09:59:41 -06004008 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
David Teiglande7fd4172006-01-18 09:30:29 +00004009 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
David Teiglandb3f58d82006-02-28 11:16:37 -05004010 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004011 break;
4012
4013 case DLM_MSG_CANCEL:
4014 hold_lkb(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004015 ls->ls_stub_ms.m_type = DLM_MSG_CANCEL_REPLY;
David Teigland601342c2008-01-07 16:15:05 -06004016 ls->ls_stub_ms.m_result = stub_cancel_result;
David Teigland075529b2006-12-13 10:40:26 -06004017 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
David Teiglandc54e04b2008-01-09 09:59:41 -06004018 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
David Teiglande7fd4172006-01-18 09:30:29 +00004019 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
David Teiglandb3f58d82006-02-28 11:16:37 -05004020 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004021 break;
4022
4023 default:
David Teigland601342c2008-01-07 16:15:05 -06004024 log_error(ls, "invalid lkb wait_type %d %d",
4025 lkb->lkb_wait_type, wait_type);
David Teiglande7fd4172006-01-18 09:30:29 +00004026 }
David Teigland81456802006-07-25 14:05:09 -05004027 schedule();
David Teiglande7fd4172006-01-18 09:30:29 +00004028 }
David Teigland90135922006-01-20 08:47:07 +00004029 mutex_unlock(&ls->ls_waiters_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +00004030}
4031
David Teiglandef0c2bb2007-03-28 09:56:46 -05004032static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
David Teiglande7fd4172006-01-18 09:30:29 +00004033{
4034 struct dlm_lkb *lkb;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004035 int found = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00004036
David Teigland90135922006-01-20 08:47:07 +00004037 mutex_lock(&ls->ls_waiters_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +00004038 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
4039 if (lkb->lkb_flags & DLM_IFL_RESEND) {
David Teiglandef0c2bb2007-03-28 09:56:46 -05004040 hold_lkb(lkb);
4041 found = 1;
David Teiglande7fd4172006-01-18 09:30:29 +00004042 break;
4043 }
4044 }
David Teigland90135922006-01-20 08:47:07 +00004045 mutex_unlock(&ls->ls_waiters_mutex);
David Teiglande7fd4172006-01-18 09:30:29 +00004046
David Teiglandef0c2bb2007-03-28 09:56:46 -05004047 if (!found)
David Teiglande7fd4172006-01-18 09:30:29 +00004048 lkb = NULL;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004049 return lkb;
David Teiglande7fd4172006-01-18 09:30:29 +00004050}
4051
4052/* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4053 master or dir-node for r. Processing the lkb may result in it being placed
4054 back on waiters. */
4055
David Teiglandef0c2bb2007-03-28 09:56:46 -05004056/* We do this after normal locking has been enabled and any saved messages
4057 (in requestqueue) have been processed. We should be confident that at
4058 this point we won't get or process a reply to any of these waiting
4059 operations. But, new ops may be coming in on the rsbs/locks here from
4060 userspace or remotely. */
4061
4062/* there may have been an overlap unlock/cancel prior to recovery or after
4063 recovery. if before, the lkb may still have a pos wait_count; if after, the
4064 overlap flag would just have been set and nothing new sent. we can be
4065 confident here than any replies to either the initial op or overlap ops
4066 prior to recovery have been received. */
4067
David Teiglande7fd4172006-01-18 09:30:29 +00004068int dlm_recover_waiters_post(struct dlm_ls *ls)
4069{
4070 struct dlm_lkb *lkb;
4071 struct dlm_rsb *r;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004072 int error = 0, mstype, err, oc, ou;
David Teiglande7fd4172006-01-18 09:30:29 +00004073
4074 while (1) {
4075 if (dlm_locking_stopped(ls)) {
4076 log_debug(ls, "recover_waiters_post aborted");
4077 error = -EINTR;
4078 break;
4079 }
4080
David Teiglandef0c2bb2007-03-28 09:56:46 -05004081 lkb = find_resend_waiter(ls);
4082 if (!lkb)
David Teiglande7fd4172006-01-18 09:30:29 +00004083 break;
4084
4085 r = lkb->lkb_resource;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004086 hold_rsb(r);
4087 lock_rsb(r);
4088
4089 mstype = lkb->lkb_wait_type;
4090 oc = is_overlap_cancel(lkb);
4091 ou = is_overlap_unlock(lkb);
4092 err = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00004093
4094 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
4095 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
4096
David Teiglandef0c2bb2007-03-28 09:56:46 -05004097 /* At this point we assume that we won't get a reply to any
4098 previous op or overlap op on this lock. First, do a big
4099 remove_from_waiters() for all previous ops. */
David Teiglande7fd4172006-01-18 09:30:29 +00004100
David Teiglandef0c2bb2007-03-28 09:56:46 -05004101 lkb->lkb_flags &= ~DLM_IFL_RESEND;
4102 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4103 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4104 lkb->lkb_wait_type = 0;
4105 lkb->lkb_wait_count = 0;
4106 mutex_lock(&ls->ls_waiters_mutex);
4107 list_del_init(&lkb->lkb_wait_reply);
4108 mutex_unlock(&ls->ls_waiters_mutex);
4109 unhold_lkb(lkb); /* for waiters list */
David Teiglande7fd4172006-01-18 09:30:29 +00004110
David Teiglandef0c2bb2007-03-28 09:56:46 -05004111 if (oc || ou) {
4112 /* do an unlock or cancel instead of resending */
4113 switch (mstype) {
4114 case DLM_MSG_LOOKUP:
4115 case DLM_MSG_REQUEST:
4116 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
4117 -DLM_ECANCEL);
4118 unhold_lkb(lkb); /* undoes create_lkb() */
4119 break;
4120 case DLM_MSG_CONVERT:
4121 if (oc) {
4122 queue_cast(r, lkb, -DLM_ECANCEL);
4123 } else {
4124 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
4125 _unlock_lock(r, lkb);
4126 }
4127 break;
4128 default:
4129 err = 1;
4130 }
4131 } else {
4132 switch (mstype) {
4133 case DLM_MSG_LOOKUP:
4134 case DLM_MSG_REQUEST:
4135 _request_lock(r, lkb);
4136 if (is_master(r))
4137 confirm_master(r, 0);
4138 break;
4139 case DLM_MSG_CONVERT:
4140 _convert_lock(r, lkb);
4141 break;
4142 default:
4143 err = 1;
4144 }
David Teiglande7fd4172006-01-18 09:30:29 +00004145 }
David Teiglandef0c2bb2007-03-28 09:56:46 -05004146
4147 if (err)
4148 log_error(ls, "recover_waiters_post %x %d %x %d %d",
4149 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
4150 unlock_rsb(r);
4151 put_rsb(r);
4152 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004153 }
4154
4155 return error;
4156}
4157
4158static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
4159 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
4160{
4161 struct dlm_ls *ls = r->res_ls;
4162 struct dlm_lkb *lkb, *safe;
4163
4164 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
4165 if (test(ls, lkb)) {
David Teigland97a35d12006-05-02 13:34:03 -04004166 rsb_set_flag(r, RSB_LOCKS_PURGED);
David Teiglande7fd4172006-01-18 09:30:29 +00004167 del_lkb(r, lkb);
4168 /* this put should free the lkb */
David Teiglandb3f58d82006-02-28 11:16:37 -05004169 if (!dlm_put_lkb(lkb))
David Teiglande7fd4172006-01-18 09:30:29 +00004170 log_error(ls, "purged lkb not released");
4171 }
4172 }
4173}
4174
4175static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4176{
4177 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
4178}
4179
4180static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4181{
4182 return is_master_copy(lkb);
4183}
4184
4185static void purge_dead_locks(struct dlm_rsb *r)
4186{
4187 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
4188 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
4189 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
4190}
4191
4192void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4193{
4194 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4195 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4196 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4197}
4198
4199/* Get rid of locks held by nodes that are gone. */
4200
4201int dlm_purge_locks(struct dlm_ls *ls)
4202{
4203 struct dlm_rsb *r;
4204
4205 log_debug(ls, "dlm_purge_locks");
4206
4207 down_write(&ls->ls_root_sem);
4208 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4209 hold_rsb(r);
4210 lock_rsb(r);
4211 if (is_master(r))
4212 purge_dead_locks(r);
4213 unlock_rsb(r);
4214 unhold_rsb(r);
4215
4216 schedule();
4217 }
4218 up_write(&ls->ls_root_sem);
4219
4220 return 0;
4221}
4222
David Teigland97a35d12006-05-02 13:34:03 -04004223static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4224{
4225 struct dlm_rsb *r, *r_ret = NULL;
4226
David Teiglandc7be7612009-01-07 16:50:41 -06004227 spin_lock(&ls->ls_rsbtbl[bucket].lock);
David Teigland97a35d12006-05-02 13:34:03 -04004228 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
4229 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4230 continue;
4231 hold_rsb(r);
4232 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4233 r_ret = r;
4234 break;
4235 }
David Teiglandc7be7612009-01-07 16:50:41 -06004236 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
David Teigland97a35d12006-05-02 13:34:03 -04004237 return r_ret;
4238}
4239
4240void dlm_grant_after_purge(struct dlm_ls *ls)
David Teiglande7fd4172006-01-18 09:30:29 +00004241{
4242 struct dlm_rsb *r;
David Teigland2b4e9262006-07-25 13:59:48 -05004243 int bucket = 0;
David Teiglande7fd4172006-01-18 09:30:29 +00004244
David Teigland2b4e9262006-07-25 13:59:48 -05004245 while (1) {
4246 r = find_purged_rsb(ls, bucket);
4247 if (!r) {
4248 if (bucket == ls->ls_rsbtbl_size - 1)
4249 break;
4250 bucket++;
David Teigland97a35d12006-05-02 13:34:03 -04004251 continue;
David Teigland2b4e9262006-07-25 13:59:48 -05004252 }
David Teigland97a35d12006-05-02 13:34:03 -04004253 lock_rsb(r);
4254 if (is_master(r)) {
4255 grant_pending_locks(r);
4256 confirm_master(r, 0);
David Teiglande7fd4172006-01-18 09:30:29 +00004257 }
David Teigland97a35d12006-05-02 13:34:03 -04004258 unlock_rsb(r);
4259 put_rsb(r);
David Teigland2b4e9262006-07-25 13:59:48 -05004260 schedule();
David Teiglande7fd4172006-01-18 09:30:29 +00004261 }
David Teiglande7fd4172006-01-18 09:30:29 +00004262}
4263
4264static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4265 uint32_t remid)
4266{
4267 struct dlm_lkb *lkb;
4268
4269 list_for_each_entry(lkb, head, lkb_statequeue) {
4270 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4271 return lkb;
4272 }
4273 return NULL;
4274}
4275
4276static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4277 uint32_t remid)
4278{
4279 struct dlm_lkb *lkb;
4280
4281 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4282 if (lkb)
4283 return lkb;
4284 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4285 if (lkb)
4286 return lkb;
4287 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4288 if (lkb)
4289 return lkb;
4290 return NULL;
4291}
4292
Al Viroae773d02008-01-25 19:55:09 -05004293/* needs at least dlm_rcom + rcom_lock */
David Teiglande7fd4172006-01-18 09:30:29 +00004294static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4295 struct dlm_rsb *r, struct dlm_rcom *rc)
4296{
4297 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
David Teiglande7fd4172006-01-18 09:30:29 +00004298
4299 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
Al Viro163a1852008-01-25 02:08:26 -05004300 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
4301 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
4302 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
4303 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
David Teiglande7fd4172006-01-18 09:30:29 +00004304 lkb->lkb_flags |= DLM_IFL_MSTCPY;
Al Viro163a1852008-01-25 02:08:26 -05004305 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
David Teiglande7fd4172006-01-18 09:30:29 +00004306 lkb->lkb_rqmode = rl->rl_rqmode;
4307 lkb->lkb_grmode = rl->rl_grmode;
4308 /* don't set lkb_status because add_lkb wants to itself */
4309
David Teiglande5dae542008-02-06 00:35:45 -06004310 lkb->lkb_bastfn = (rl->rl_asts & AST_BAST) ? &fake_bastfn : NULL;
4311 lkb->lkb_astfn = (rl->rl_asts & AST_COMP) ? &fake_astfn : NULL;
David Teiglande7fd4172006-01-18 09:30:29 +00004312
David Teiglande7fd4172006-01-18 09:30:29 +00004313 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
Al Viroa5dd0632008-01-25 20:22:22 -05004314 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4315 sizeof(struct rcom_lock);
4316 if (lvblen > ls->ls_lvblen)
4317 return -EINVAL;
David Teigland52bda2b2007-11-07 09:06:49 -06004318 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
David Teiglande7fd4172006-01-18 09:30:29 +00004319 if (!lkb->lkb_lvbptr)
4320 return -ENOMEM;
David Teiglande7fd4172006-01-18 09:30:29 +00004321 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4322 }
4323
4324 /* Conversions between PR and CW (middle modes) need special handling.
4325 The real granted mode of these converting locks cannot be determined
4326 until all locks have been rebuilt on the rsb (recover_conversion) */
4327
Al Viro163a1852008-01-25 02:08:26 -05004328 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
4329 middle_conversion(lkb)) {
David Teiglande7fd4172006-01-18 09:30:29 +00004330 rl->rl_status = DLM_LKSTS_CONVERT;
4331 lkb->lkb_grmode = DLM_LOCK_IV;
4332 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4333 }
4334
4335 return 0;
4336}
4337
4338/* This lkb may have been recovered in a previous aborted recovery so we need
4339 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4340 If so we just send back a standard reply. If not, we create a new lkb with
4341 the given values and send back our lkid. We send back our lkid by sending
4342 back the rcom_lock struct we got but with the remid field filled in. */
4343
Al Viroae773d02008-01-25 19:55:09 -05004344/* needs at least dlm_rcom + rcom_lock */
David Teiglande7fd4172006-01-18 09:30:29 +00004345int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4346{
4347 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4348 struct dlm_rsb *r;
4349 struct dlm_lkb *lkb;
4350 int error;
4351
4352 if (rl->rl_parent_lkid) {
4353 error = -EOPNOTSUPP;
4354 goto out;
4355 }
4356
Al Viro163a1852008-01-25 02:08:26 -05004357 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
4358 R_MASTER, &r);
David Teiglande7fd4172006-01-18 09:30:29 +00004359 if (error)
4360 goto out;
4361
4362 lock_rsb(r);
4363
Al Viro163a1852008-01-25 02:08:26 -05004364 lkb = search_remid(r, rc->rc_header.h_nodeid, le32_to_cpu(rl->rl_lkid));
David Teiglande7fd4172006-01-18 09:30:29 +00004365 if (lkb) {
4366 error = -EEXIST;
4367 goto out_remid;
4368 }
4369
4370 error = create_lkb(ls, &lkb);
4371 if (error)
4372 goto out_unlock;
4373
4374 error = receive_rcom_lock_args(ls, lkb, r, rc);
4375 if (error) {
David Teiglandb3f58d82006-02-28 11:16:37 -05004376 __put_lkb(ls, lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004377 goto out_unlock;
4378 }
4379
4380 attach_lkb(r, lkb);
4381 add_lkb(r, lkb, rl->rl_status);
4382 error = 0;
4383
4384 out_remid:
4385 /* this is the new value returned to the lock holder for
4386 saving in its process-copy lkb */
Al Viro163a1852008-01-25 02:08:26 -05004387 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
David Teiglande7fd4172006-01-18 09:30:29 +00004388
4389 out_unlock:
4390 unlock_rsb(r);
4391 put_rsb(r);
4392 out:
4393 if (error)
Al Viro163a1852008-01-25 02:08:26 -05004394 log_debug(ls, "recover_master_copy %d %x", error,
4395 le32_to_cpu(rl->rl_lkid));
4396 rl->rl_result = cpu_to_le32(error);
David Teiglande7fd4172006-01-18 09:30:29 +00004397 return error;
4398}
4399
Al Viroae773d02008-01-25 19:55:09 -05004400/* needs at least dlm_rcom + rcom_lock */
David Teiglande7fd4172006-01-18 09:30:29 +00004401int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4402{
4403 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4404 struct dlm_rsb *r;
4405 struct dlm_lkb *lkb;
4406 int error;
4407
Al Viro163a1852008-01-25 02:08:26 -05004408 error = find_lkb(ls, le32_to_cpu(rl->rl_lkid), &lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004409 if (error) {
Al Viro163a1852008-01-25 02:08:26 -05004410 log_error(ls, "recover_process_copy no lkid %x",
4411 le32_to_cpu(rl->rl_lkid));
David Teiglande7fd4172006-01-18 09:30:29 +00004412 return error;
4413 }
4414
4415 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4416
Al Viro163a1852008-01-25 02:08:26 -05004417 error = le32_to_cpu(rl->rl_result);
David Teiglande7fd4172006-01-18 09:30:29 +00004418
4419 r = lkb->lkb_resource;
4420 hold_rsb(r);
4421 lock_rsb(r);
4422
4423 switch (error) {
David Teiglanddc200a82006-12-13 10:36:37 -06004424 case -EBADR:
4425 /* There's a chance the new master received our lock before
4426 dlm_recover_master_reply(), this wouldn't happen if we did
4427 a barrier between recover_masters and recover_locks. */
4428 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4429 (unsigned long)r, r->res_name);
4430 dlm_send_rcom_lock(r, lkb);
4431 goto out;
David Teiglande7fd4172006-01-18 09:30:29 +00004432 case -EEXIST:
4433 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4434 /* fall through */
4435 case 0:
Al Viro163a1852008-01-25 02:08:26 -05004436 lkb->lkb_remid = le32_to_cpu(rl->rl_remid);
David Teiglande7fd4172006-01-18 09:30:29 +00004437 break;
4438 default:
4439 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4440 error, lkb->lkb_id);
4441 }
4442
4443 /* an ack for dlm_recover_locks() which waits for replies from
4444 all the locks it sends to new masters */
4445 dlm_recovered_lock(r);
David Teiglanddc200a82006-12-13 10:36:37 -06004446 out:
David Teiglande7fd4172006-01-18 09:30:29 +00004447 unlock_rsb(r);
4448 put_rsb(r);
David Teiglandb3f58d82006-02-28 11:16:37 -05004449 dlm_put_lkb(lkb);
David Teiglande7fd4172006-01-18 09:30:29 +00004450
4451 return 0;
4452}
4453
David Teigland597d0ca2006-07-12 16:44:04 -05004454int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4455 int mode, uint32_t flags, void *name, unsigned int namelen,
David Teiglandd7db9232007-05-18 09:00:32 -05004456 unsigned long timeout_cs)
David Teigland597d0ca2006-07-12 16:44:04 -05004457{
4458 struct dlm_lkb *lkb;
4459 struct dlm_args args;
4460 int error;
4461
David Teigland85e86ed2007-05-18 08:58:15 -05004462 dlm_lock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004463
4464 error = create_lkb(ls, &lkb);
4465 if (error) {
4466 kfree(ua);
4467 goto out;
4468 }
4469
4470 if (flags & DLM_LKF_VALBLK) {
David Teigland62a0f622007-01-31 13:25:00 -06004471 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
David Teigland597d0ca2006-07-12 16:44:04 -05004472 if (!ua->lksb.sb_lvbptr) {
4473 kfree(ua);
4474 __put_lkb(ls, lkb);
4475 error = -ENOMEM;
4476 goto out;
4477 }
4478 }
4479
David Teigland52bda2b2007-11-07 09:06:49 -06004480 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
David Teigland597d0ca2006-07-12 16:44:04 -05004481 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4482 lock and that lkb_astparam is the dlm_user_args structure. */
4483
David Teiglandd7db9232007-05-18 09:00:32 -05004484 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
David Teiglande5dae542008-02-06 00:35:45 -06004485 fake_astfn, ua, fake_bastfn, &args);
David Teigland597d0ca2006-07-12 16:44:04 -05004486 lkb->lkb_flags |= DLM_IFL_USER;
4487 ua->old_mode = DLM_LOCK_IV;
4488
4489 if (error) {
4490 __put_lkb(ls, lkb);
4491 goto out;
4492 }
4493
4494 error = request_lock(ls, lkb, name, namelen, &args);
4495
4496 switch (error) {
4497 case 0:
4498 break;
4499 case -EINPROGRESS:
4500 error = 0;
4501 break;
4502 case -EAGAIN:
4503 error = 0;
4504 /* fall through */
4505 default:
4506 __put_lkb(ls, lkb);
4507 goto out;
4508 }
4509
4510 /* add this new lkb to the per-process list of locks */
4511 spin_lock(&ua->proc->locks_spin);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004512 hold_lkb(lkb);
David Teigland597d0ca2006-07-12 16:44:04 -05004513 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4514 spin_unlock(&ua->proc->locks_spin);
4515 out:
David Teigland85e86ed2007-05-18 08:58:15 -05004516 dlm_unlock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004517 return error;
4518}
4519
4520int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
David Teiglandd7db9232007-05-18 09:00:32 -05004521 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4522 unsigned long timeout_cs)
David Teigland597d0ca2006-07-12 16:44:04 -05004523{
4524 struct dlm_lkb *lkb;
4525 struct dlm_args args;
4526 struct dlm_user_args *ua;
4527 int error;
4528
David Teigland85e86ed2007-05-18 08:58:15 -05004529 dlm_lock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004530
4531 error = find_lkb(ls, lkid, &lkb);
4532 if (error)
4533 goto out;
4534
4535 /* user can change the params on its lock when it converts it, or
4536 add an lvb that didn't exist before */
4537
David Teiglandd292c0c2008-02-06 23:27:04 -06004538 ua = lkb->lkb_ua;
David Teigland597d0ca2006-07-12 16:44:04 -05004539
4540 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
David Teigland62a0f622007-01-31 13:25:00 -06004541 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
David Teigland597d0ca2006-07-12 16:44:04 -05004542 if (!ua->lksb.sb_lvbptr) {
4543 error = -ENOMEM;
4544 goto out_put;
4545 }
4546 }
4547 if (lvb_in && ua->lksb.sb_lvbptr)
4548 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4549
David Teiglandd7db9232007-05-18 09:00:32 -05004550 ua->xid = ua_tmp->xid;
David Teigland597d0ca2006-07-12 16:44:04 -05004551 ua->castparam = ua_tmp->castparam;
4552 ua->castaddr = ua_tmp->castaddr;
4553 ua->bastparam = ua_tmp->bastparam;
4554 ua->bastaddr = ua_tmp->bastaddr;
Patrick Caulfield10948eb2006-08-23 09:49:31 +01004555 ua->user_lksb = ua_tmp->user_lksb;
David Teigland597d0ca2006-07-12 16:44:04 -05004556 ua->old_mode = lkb->lkb_grmode;
4557
David Teiglandd7db9232007-05-18 09:00:32 -05004558 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
David Teiglande5dae542008-02-06 00:35:45 -06004559 fake_astfn, ua, fake_bastfn, &args);
David Teigland597d0ca2006-07-12 16:44:04 -05004560 if (error)
4561 goto out_put;
4562
4563 error = convert_lock(ls, lkb, &args);
4564
David Teiglandc85d65e2007-05-18 09:01:26 -05004565 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
David Teigland597d0ca2006-07-12 16:44:04 -05004566 error = 0;
4567 out_put:
4568 dlm_put_lkb(lkb);
4569 out:
David Teigland85e86ed2007-05-18 08:58:15 -05004570 dlm_unlock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004571 kfree(ua_tmp);
4572 return error;
4573}
4574
4575int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4576 uint32_t flags, uint32_t lkid, char *lvb_in)
4577{
4578 struct dlm_lkb *lkb;
4579 struct dlm_args args;
4580 struct dlm_user_args *ua;
4581 int error;
4582
David Teigland85e86ed2007-05-18 08:58:15 -05004583 dlm_lock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004584
4585 error = find_lkb(ls, lkid, &lkb);
4586 if (error)
4587 goto out;
4588
David Teiglandd292c0c2008-02-06 23:27:04 -06004589 ua = lkb->lkb_ua;
David Teigland597d0ca2006-07-12 16:44:04 -05004590
4591 if (lvb_in && ua->lksb.sb_lvbptr)
4592 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
Patrick Caulfieldb434eda2007-10-01 15:28:42 +01004593 if (ua_tmp->castparam)
4594 ua->castparam = ua_tmp->castparam;
Patrick Caulfieldcc346d52006-08-08 10:34:40 -04004595 ua->user_lksb = ua_tmp->user_lksb;
David Teigland597d0ca2006-07-12 16:44:04 -05004596
4597 error = set_unlock_args(flags, ua, &args);
4598 if (error)
4599 goto out_put;
4600
4601 error = unlock_lock(ls, lkb, &args);
4602
4603 if (error == -DLM_EUNLOCK)
4604 error = 0;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004605 /* from validate_unlock_args() */
4606 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4607 error = 0;
David Teigland597d0ca2006-07-12 16:44:04 -05004608 if (error)
4609 goto out_put;
4610
4611 spin_lock(&ua->proc->locks_spin);
David Teiglanda1bc86e2007-01-15 10:34:52 -06004612 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4613 if (!list_empty(&lkb->lkb_ownqueue))
4614 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
David Teigland597d0ca2006-07-12 16:44:04 -05004615 spin_unlock(&ua->proc->locks_spin);
David Teigland597d0ca2006-07-12 16:44:04 -05004616 out_put:
4617 dlm_put_lkb(lkb);
4618 out:
David Teigland85e86ed2007-05-18 08:58:15 -05004619 dlm_unlock_recovery(ls);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004620 kfree(ua_tmp);
David Teigland597d0ca2006-07-12 16:44:04 -05004621 return error;
4622}
4623
4624int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4625 uint32_t flags, uint32_t lkid)
4626{
4627 struct dlm_lkb *lkb;
4628 struct dlm_args args;
4629 struct dlm_user_args *ua;
4630 int error;
4631
David Teigland85e86ed2007-05-18 08:58:15 -05004632 dlm_lock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004633
4634 error = find_lkb(ls, lkid, &lkb);
4635 if (error)
4636 goto out;
4637
David Teiglandd292c0c2008-02-06 23:27:04 -06004638 ua = lkb->lkb_ua;
Patrick Caulfieldb434eda2007-10-01 15:28:42 +01004639 if (ua_tmp->castparam)
4640 ua->castparam = ua_tmp->castparam;
Patrick Caulfieldc059f702006-08-23 10:24:03 +01004641 ua->user_lksb = ua_tmp->user_lksb;
David Teigland597d0ca2006-07-12 16:44:04 -05004642
4643 error = set_unlock_args(flags, ua, &args);
4644 if (error)
4645 goto out_put;
4646
4647 error = cancel_lock(ls, lkb, &args);
4648
4649 if (error == -DLM_ECANCEL)
4650 error = 0;
David Teiglandef0c2bb2007-03-28 09:56:46 -05004651 /* from validate_unlock_args() */
4652 if (error == -EBUSY)
4653 error = 0;
David Teigland597d0ca2006-07-12 16:44:04 -05004654 out_put:
4655 dlm_put_lkb(lkb);
4656 out:
David Teigland85e86ed2007-05-18 08:58:15 -05004657 dlm_unlock_recovery(ls);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004658 kfree(ua_tmp);
David Teigland597d0ca2006-07-12 16:44:04 -05004659 return error;
4660}
4661
David Teigland8b4021f2007-05-29 08:46:00 -05004662int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4663{
4664 struct dlm_lkb *lkb;
4665 struct dlm_args args;
4666 struct dlm_user_args *ua;
4667 struct dlm_rsb *r;
4668 int error;
4669
4670 dlm_lock_recovery(ls);
4671
4672 error = find_lkb(ls, lkid, &lkb);
4673 if (error)
4674 goto out;
4675
David Teiglandd292c0c2008-02-06 23:27:04 -06004676 ua = lkb->lkb_ua;
David Teigland8b4021f2007-05-29 08:46:00 -05004677
4678 error = set_unlock_args(flags, ua, &args);
4679 if (error)
4680 goto out_put;
4681
4682 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4683
4684 r = lkb->lkb_resource;
4685 hold_rsb(r);
4686 lock_rsb(r);
4687
4688 error = validate_unlock_args(lkb, &args);
4689 if (error)
4690 goto out_r;
4691 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4692
4693 error = _cancel_lock(r, lkb);
4694 out_r:
4695 unlock_rsb(r);
4696 put_rsb(r);
4697
4698 if (error == -DLM_ECANCEL)
4699 error = 0;
4700 /* from validate_unlock_args() */
4701 if (error == -EBUSY)
4702 error = 0;
4703 out_put:
4704 dlm_put_lkb(lkb);
4705 out:
4706 dlm_unlock_recovery(ls);
4707 return error;
4708}
4709
David Teiglandef0c2bb2007-03-28 09:56:46 -05004710/* lkb's that are removed from the waiters list by revert are just left on the
4711 orphans list with the granted orphan locks, to be freed by purge */
4712
David Teigland597d0ca2006-07-12 16:44:04 -05004713static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4714{
David Teiglandef0c2bb2007-03-28 09:56:46 -05004715 struct dlm_args args;
4716 int error;
David Teigland597d0ca2006-07-12 16:44:04 -05004717
David Teiglandef0c2bb2007-03-28 09:56:46 -05004718 hold_lkb(lkb);
4719 mutex_lock(&ls->ls_orphans_mutex);
4720 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
4721 mutex_unlock(&ls->ls_orphans_mutex);
David Teigland597d0ca2006-07-12 16:44:04 -05004722
David Teiglandd292c0c2008-02-06 23:27:04 -06004723 set_unlock_args(0, lkb->lkb_ua, &args);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004724
4725 error = cancel_lock(ls, lkb, &args);
4726 if (error == -DLM_ECANCEL)
4727 error = 0;
4728 return error;
David Teigland597d0ca2006-07-12 16:44:04 -05004729}
4730
4731/* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4732 Regardless of what rsb queue the lock is on, it's removed and freed. */
4733
4734static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4735{
David Teigland597d0ca2006-07-12 16:44:04 -05004736 struct dlm_args args;
4737 int error;
4738
David Teiglandd292c0c2008-02-06 23:27:04 -06004739 set_unlock_args(DLM_LKF_FORCEUNLOCK, lkb->lkb_ua, &args);
David Teigland597d0ca2006-07-12 16:44:04 -05004740
4741 error = unlock_lock(ls, lkb, &args);
4742 if (error == -DLM_EUNLOCK)
4743 error = 0;
4744 return error;
4745}
4746
David Teiglandef0c2bb2007-03-28 09:56:46 -05004747/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4748 (which does lock_rsb) due to deadlock with receiving a message that does
4749 lock_rsb followed by dlm_user_add_ast() */
4750
4751static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
4752 struct dlm_user_proc *proc)
4753{
4754 struct dlm_lkb *lkb = NULL;
4755
4756 mutex_lock(&ls->ls_clear_proc_locks);
4757 if (list_empty(&proc->locks))
4758 goto out;
4759
4760 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
4761 list_del_init(&lkb->lkb_ownqueue);
4762
4763 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4764 lkb->lkb_flags |= DLM_IFL_ORPHAN;
4765 else
4766 lkb->lkb_flags |= DLM_IFL_DEAD;
4767 out:
4768 mutex_unlock(&ls->ls_clear_proc_locks);
4769 return lkb;
4770}
4771
David Teigland597d0ca2006-07-12 16:44:04 -05004772/* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4773 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4774 which we clear here. */
4775
4776/* proc CLOSING flag is set so no more device_reads should look at proc->asts
4777 list, and no more device_writes should add lkb's to proc->locks list; so we
4778 shouldn't need to take asts_spin or locks_spin here. this assumes that
4779 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4780 them ourself. */
4781
4782void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4783{
4784 struct dlm_lkb *lkb, *safe;
4785
David Teigland85e86ed2007-05-18 08:58:15 -05004786 dlm_lock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004787
David Teiglandef0c2bb2007-03-28 09:56:46 -05004788 while (1) {
4789 lkb = del_proc_lock(ls, proc);
4790 if (!lkb)
4791 break;
David Teigland84d8cd62007-05-29 08:44:23 -05004792 del_timeout(lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004793 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
David Teigland597d0ca2006-07-12 16:44:04 -05004794 orphan_proc_lock(ls, lkb);
David Teiglandef0c2bb2007-03-28 09:56:46 -05004795 else
David Teigland597d0ca2006-07-12 16:44:04 -05004796 unlock_proc_lock(ls, lkb);
David Teigland597d0ca2006-07-12 16:44:04 -05004797
4798 /* this removes the reference for the proc->locks list
4799 added by dlm_user_request, it may result in the lkb
4800 being freed */
4801
4802 dlm_put_lkb(lkb);
4803 }
David Teiglanda1bc86e2007-01-15 10:34:52 -06004804
David Teiglandef0c2bb2007-03-28 09:56:46 -05004805 mutex_lock(&ls->ls_clear_proc_locks);
4806
David Teiglanda1bc86e2007-01-15 10:34:52 -06004807 /* in-progress unlocks */
4808 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4809 list_del_init(&lkb->lkb_ownqueue);
4810 lkb->lkb_flags |= DLM_IFL_DEAD;
4811 dlm_put_lkb(lkb);
4812 }
4813
4814 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
David Teigland8a358ca2008-01-07 15:55:18 -06004815 lkb->lkb_ast_type = 0;
David Teiglanda1bc86e2007-01-15 10:34:52 -06004816 list_del(&lkb->lkb_astqueue);
4817 dlm_put_lkb(lkb);
4818 }
4819
David Teigland597d0ca2006-07-12 16:44:04 -05004820 mutex_unlock(&ls->ls_clear_proc_locks);
David Teigland85e86ed2007-05-18 08:58:15 -05004821 dlm_unlock_recovery(ls);
David Teigland597d0ca2006-07-12 16:44:04 -05004822}
David Teiglanda1bc86e2007-01-15 10:34:52 -06004823
David Teigland84991372007-03-30 15:02:40 -05004824static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4825{
4826 struct dlm_lkb *lkb, *safe;
4827
4828 while (1) {
4829 lkb = NULL;
4830 spin_lock(&proc->locks_spin);
4831 if (!list_empty(&proc->locks)) {
4832 lkb = list_entry(proc->locks.next, struct dlm_lkb,
4833 lkb_ownqueue);
4834 list_del_init(&lkb->lkb_ownqueue);
4835 }
4836 spin_unlock(&proc->locks_spin);
4837
4838 if (!lkb)
4839 break;
4840
4841 lkb->lkb_flags |= DLM_IFL_DEAD;
4842 unlock_proc_lock(ls, lkb);
4843 dlm_put_lkb(lkb); /* ref from proc->locks list */
4844 }
4845
4846 spin_lock(&proc->locks_spin);
4847 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4848 list_del_init(&lkb->lkb_ownqueue);
4849 lkb->lkb_flags |= DLM_IFL_DEAD;
4850 dlm_put_lkb(lkb);
4851 }
4852 spin_unlock(&proc->locks_spin);
4853
4854 spin_lock(&proc->asts_spin);
4855 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4856 list_del(&lkb->lkb_astqueue);
4857 dlm_put_lkb(lkb);
4858 }
4859 spin_unlock(&proc->asts_spin);
4860}
4861
4862/* pid of 0 means purge all orphans */
4863
4864static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
4865{
4866 struct dlm_lkb *lkb, *safe;
4867
4868 mutex_lock(&ls->ls_orphans_mutex);
4869 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
4870 if (pid && lkb->lkb_ownpid != pid)
4871 continue;
4872 unlock_proc_lock(ls, lkb);
4873 list_del_init(&lkb->lkb_ownqueue);
4874 dlm_put_lkb(lkb);
4875 }
4876 mutex_unlock(&ls->ls_orphans_mutex);
4877}
4878
4879static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
4880{
4881 struct dlm_message *ms;
4882 struct dlm_mhandle *mh;
4883 int error;
4884
4885 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
4886 DLM_MSG_PURGE, &ms, &mh);
4887 if (error)
4888 return error;
4889 ms->m_nodeid = nodeid;
4890 ms->m_pid = pid;
4891
4892 return send_message(mh, ms);
4893}
4894
4895int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
4896 int nodeid, int pid)
4897{
4898 int error = 0;
4899
4900 if (nodeid != dlm_our_nodeid()) {
4901 error = send_purge(ls, nodeid, pid);
4902 } else {
David Teigland85e86ed2007-05-18 08:58:15 -05004903 dlm_lock_recovery(ls);
David Teigland84991372007-03-30 15:02:40 -05004904 if (pid == current->pid)
4905 purge_proc_locks(ls, proc);
4906 else
4907 do_purge(ls, nodeid, pid);
David Teigland85e86ed2007-05-18 08:58:15 -05004908 dlm_unlock_recovery(ls);
David Teigland84991372007-03-30 15:02:40 -05004909 }
4910 return error;
4911}
4912