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