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