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 "midcomms.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
76static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
77static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
78static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
79static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
80static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
81static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
82static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
83static int send_remove(struct dlm_rsb *r);
84static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
85static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
86static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
87 struct dlm_message *ms);
88static int receive_extralen(struct dlm_message *ms);
89static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
90static void del_timeout(struct dlm_lkb *lkb);
91static void toss_rsb(struct kref *kref);
92
93
94
95
96
97
98
99
100
101static 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
122const 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
137int 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
148static 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
160void 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
170static 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
179void 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
203static inline void dlm_lock_recovery(struct dlm_ls *ls)
204{
205 down_read(&ls->ls_in_recovery);
206}
207
208void dlm_unlock_recovery(struct dlm_ls *ls)
209{
210 up_read(&ls->ls_in_recovery);
211}
212
213int dlm_lock_recovery_try(struct dlm_ls *ls)
214{
215 return down_read_trylock(&ls->ls_in_recovery);
216}
217
218static inline int can_be_queued(struct dlm_lkb *lkb)
219{
220 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
221}
222
223static inline int force_blocking_asts(struct dlm_lkb *lkb)
224{
225 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
226}
227
228static inline int is_demoted(struct dlm_lkb *lkb)
229{
230 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
231}
232
233static inline int is_altmode(struct dlm_lkb *lkb)
234{
235 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
236}
237
238static inline int is_granted(struct dlm_lkb *lkb)
239{
240 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
241}
242
243static 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
249static inline int is_process_copy(struct dlm_lkb *lkb)
250{
251 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
252}
253
254static inline int is_master_copy(struct dlm_lkb *lkb)
255{
256 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
257}
258
259static 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
267static inline int down_conversion(struct dlm_lkb *lkb)
268{
269 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
270}
271
272static inline int is_overlap_unlock(struct dlm_lkb *lkb)
273{
274 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
275}
276
277static inline int is_overlap_cancel(struct dlm_lkb *lkb)
278{
279 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
280}
281
282static 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
288static 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
312static 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
318static 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
334static inline void hold_rsb(struct dlm_rsb *r)
335{
336 kref_get(&r->res_ref);
337}
338
339void dlm_hold_rsb(struct dlm_rsb *r)
340{
341 hold_rsb(r);
342}
343
344
345
346
347static 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
357void dlm_put_rsb(struct dlm_rsb *r)
358{
359 put_rsb(r);
360}
361
362static 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
398static 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
436static 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
445int 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
470static 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
543static 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
722static 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
816static 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
841static 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
910int 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
1083static 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
1100void 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
1123static 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
1142static 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
1149static 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
1167static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1168{
1169 hold_rsb(r);
1170 lkb->lkb_resource = r;
1171}
1172
1173static 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
1181static 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
1218static 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
1232static 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
1245static 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
1267int 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
1281static inline void hold_lkb(struct dlm_lkb *lkb)
1282{
1283 kref_get(&lkb->lkb_ref);
1284}
1285
1286
1287
1288
1289
1290
1291static 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
1298static 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
1312static 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
1346static 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
1353static 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
1361static 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
1378static 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
1393void 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
1450static 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
1508static 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
1593static 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
1607static 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
1625static 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
1648static 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
1800void 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
1812static 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
1836static 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
1854void 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
1921void 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
1945static 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
1996static 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
2025static 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
2056static 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
2065static 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
2071static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2072{
2073 _remove_lock(r, lkb);
2074}
2075
2076
2077
2078
2079
2080static 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
2107static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2108{
2109 return revert_lock(r, lkb);
2110}
2111
2112static 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
2126static 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
2132static 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
2143static 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
2160static 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
2171static 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
2190static 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
2202static 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
2256static 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
2295static 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
2441static 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
2503static 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
2572static 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
2597static 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
2611static 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
2647static 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
2661static 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
2677static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2678{
2679 send_bast_queue(r, &r->res_grantqueue, lkb);
2680}
2681
2682static 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
2707static 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
2756static 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
2769static 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
2806static 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
2872static 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
2886static 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
2937static 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
3074static 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
3097static 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
3111static 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
3166static 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
3184static 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
3191static 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
3199static 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
3211static 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
3225static 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
3254static 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
3273static 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
3292static 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
3314static 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
3340static 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
3362static 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
3384static 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
3410int 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
3463int 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
3527static 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_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
3541 if (!mh)
3542 return -ENOBUFS;
3543
3544 ms = (struct dlm_message *) mb;
3545
3546 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3547 ms->m_header.u.h_lockspace = ls->ls_global_id;
3548 ms->m_header.h_nodeid = dlm_our_nodeid();
3549 ms->m_header.h_length = mb_len;
3550 ms->m_header.h_cmd = DLM_MSG;
3551
3552 ms->m_type = mstype;
3553
3554 *mh_ret = mh;
3555 *ms_ret = ms;
3556 return 0;
3557}
3558
3559static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3560 int to_nodeid, int mstype,
3561 struct dlm_message **ms_ret,
3562 struct dlm_mhandle **mh_ret)
3563{
3564 int mb_len = sizeof(struct dlm_message);
3565
3566 switch (mstype) {
3567 case DLM_MSG_REQUEST:
3568 case DLM_MSG_LOOKUP:
3569 case DLM_MSG_REMOVE:
3570 mb_len += r->res_length;
3571 break;
3572 case DLM_MSG_CONVERT:
3573 case DLM_MSG_UNLOCK:
3574 case DLM_MSG_REQUEST_REPLY:
3575 case DLM_MSG_CONVERT_REPLY:
3576 case DLM_MSG_GRANT:
3577 if (lkb && lkb->lkb_lvbptr)
3578 mb_len += r->res_ls->ls_lvblen;
3579 break;
3580 }
3581
3582 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3583 ms_ret, mh_ret);
3584}
3585
3586
3587
3588
3589static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3590{
3591 dlm_message_out(ms);
3592 dlm_midcomms_commit_mhandle(mh);
3593 return 0;
3594}
3595
3596static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3597 struct dlm_message *ms)
3598{
3599 ms->m_nodeid = lkb->lkb_nodeid;
3600 ms->m_pid = lkb->lkb_ownpid;
3601 ms->m_lkid = lkb->lkb_id;
3602 ms->m_remid = lkb->lkb_remid;
3603 ms->m_exflags = lkb->lkb_exflags;
3604 ms->m_sbflags = lkb->lkb_sbflags;
3605 ms->m_flags = lkb->lkb_flags;
3606 ms->m_lvbseq = lkb->lkb_lvbseq;
3607 ms->m_status = lkb->lkb_status;
3608 ms->m_grmode = lkb->lkb_grmode;
3609 ms->m_rqmode = lkb->lkb_rqmode;
3610 ms->m_hash = r->res_hash;
3611
3612
3613
3614
3615 if (lkb->lkb_bastfn)
3616 ms->m_asts |= DLM_CB_BAST;
3617 if (lkb->lkb_astfn)
3618 ms->m_asts |= DLM_CB_CAST;
3619
3620
3621
3622
3623 switch (ms->m_type) {
3624 case DLM_MSG_REQUEST:
3625 case DLM_MSG_LOOKUP:
3626 memcpy(ms->m_extra, r->res_name, r->res_length);
3627 break;
3628 case DLM_MSG_CONVERT:
3629 case DLM_MSG_UNLOCK:
3630 case DLM_MSG_REQUEST_REPLY:
3631 case DLM_MSG_CONVERT_REPLY:
3632 case DLM_MSG_GRANT:
3633 if (!lkb->lkb_lvbptr)
3634 break;
3635 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3636 break;
3637 }
3638}
3639
3640static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3641{
3642 struct dlm_message *ms;
3643 struct dlm_mhandle *mh;
3644 int to_nodeid, error;
3645
3646 to_nodeid = r->res_nodeid;
3647
3648 error = add_to_waiters(lkb, mstype, to_nodeid);
3649 if (error)
3650 return error;
3651
3652 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3653 if (error)
3654 goto fail;
3655
3656 send_args(r, lkb, ms);
3657
3658 error = send_message(mh, ms);
3659 if (error)
3660 goto fail;
3661 return 0;
3662
3663 fail:
3664 remove_from_waiters(lkb, msg_reply_type(mstype));
3665 return error;
3666}
3667
3668static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3669{
3670 return send_common(r, lkb, DLM_MSG_REQUEST);
3671}
3672
3673static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3674{
3675 int error;
3676
3677 error = send_common(r, lkb, DLM_MSG_CONVERT);
3678
3679
3680 if (!error && down_conversion(lkb)) {
3681 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3682 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3683 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3684 r->res_ls->ls_stub_ms.m_result = 0;
3685 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3686 }
3687
3688 return error;
3689}
3690
3691
3692
3693
3694
3695static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3696{
3697 return send_common(r, lkb, DLM_MSG_UNLOCK);
3698}
3699
3700static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3701{
3702 return send_common(r, lkb, DLM_MSG_CANCEL);
3703}
3704
3705static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3706{
3707 struct dlm_message *ms;
3708 struct dlm_mhandle *mh;
3709 int to_nodeid, error;
3710
3711 to_nodeid = lkb->lkb_nodeid;
3712
3713 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3714 if (error)
3715 goto out;
3716
3717 send_args(r, lkb, ms);
3718
3719 ms->m_result = 0;
3720
3721 error = send_message(mh, ms);
3722 out:
3723 return error;
3724}
3725
3726static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3727{
3728 struct dlm_message *ms;
3729 struct dlm_mhandle *mh;
3730 int to_nodeid, error;
3731
3732 to_nodeid = lkb->lkb_nodeid;
3733
3734 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3735 if (error)
3736 goto out;
3737
3738 send_args(r, lkb, ms);
3739
3740 ms->m_bastmode = mode;
3741
3742 error = send_message(mh, ms);
3743 out:
3744 return error;
3745}
3746
3747static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3748{
3749 struct dlm_message *ms;
3750 struct dlm_mhandle *mh;
3751 int to_nodeid, error;
3752
3753 to_nodeid = dlm_dir_nodeid(r);
3754
3755 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3756 if (error)
3757 return error;
3758
3759 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3760 if (error)
3761 goto fail;
3762
3763 send_args(r, lkb, ms);
3764
3765 error = send_message(mh, ms);
3766 if (error)
3767 goto fail;
3768 return 0;
3769
3770 fail:
3771 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3772 return error;
3773}
3774
3775static int send_remove(struct dlm_rsb *r)
3776{
3777 struct dlm_message *ms;
3778 struct dlm_mhandle *mh;
3779 int to_nodeid, error;
3780
3781 to_nodeid = dlm_dir_nodeid(r);
3782
3783 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3784 if (error)
3785 goto out;
3786
3787 memcpy(ms->m_extra, r->res_name, r->res_length);
3788 ms->m_hash = r->res_hash;
3789
3790 error = send_message(mh, ms);
3791 out:
3792 return error;
3793}
3794
3795static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3796 int mstype, int rv)
3797{
3798 struct dlm_message *ms;
3799 struct dlm_mhandle *mh;
3800 int to_nodeid, error;
3801
3802 to_nodeid = lkb->lkb_nodeid;
3803
3804 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3805 if (error)
3806 goto out;
3807
3808 send_args(r, lkb, ms);
3809
3810 ms->m_result = rv;
3811
3812 error = send_message(mh, ms);
3813 out:
3814 return error;
3815}
3816
3817static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3818{
3819 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3820}
3821
3822static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3823{
3824 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3825}
3826
3827static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3828{
3829 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3830}
3831
3832static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3833{
3834 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3835}
3836
3837static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3838 int ret_nodeid, int rv)
3839{
3840 struct dlm_rsb *r = &ls->ls_stub_rsb;
3841 struct dlm_message *ms;
3842 struct dlm_mhandle *mh;
3843 int error, nodeid = ms_in->m_header.h_nodeid;
3844
3845 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3846 if (error)
3847 goto out;
3848
3849 ms->m_lkid = ms_in->m_lkid;
3850 ms->m_result = rv;
3851 ms->m_nodeid = ret_nodeid;
3852
3853 error = send_message(mh, ms);
3854 out:
3855 return error;
3856}
3857
3858
3859
3860
3861
3862static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3863{
3864 lkb->lkb_exflags = ms->m_exflags;
3865 lkb->lkb_sbflags = ms->m_sbflags;
3866 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3867 (ms->m_flags & 0x0000FFFF);
3868}
3869
3870static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3871{
3872 if (ms->m_flags == DLM_IFL_STUB_MS)
3873 return;
3874
3875 lkb->lkb_sbflags = ms->m_sbflags;
3876 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3877 (ms->m_flags & 0x0000FFFF);
3878}
3879
3880static int receive_extralen(struct dlm_message *ms)
3881{
3882 return (ms->m_header.h_length - sizeof(struct dlm_message));
3883}
3884
3885static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3886 struct dlm_message *ms)
3887{
3888 int len;
3889
3890 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3891 if (!lkb->lkb_lvbptr)
3892 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3893 if (!lkb->lkb_lvbptr)
3894 return -ENOMEM;
3895 len = receive_extralen(ms);
3896 if (len > ls->ls_lvblen)
3897 len = ls->ls_lvblen;
3898 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3899 }
3900 return 0;
3901}
3902
3903static void fake_bastfn(void *astparam, int mode)
3904{
3905 log_print("fake_bastfn should not be called");
3906}
3907
3908static void fake_astfn(void *astparam)
3909{
3910 log_print("fake_astfn should not be called");
3911}
3912
3913static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3914 struct dlm_message *ms)
3915{
3916 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3917 lkb->lkb_ownpid = ms->m_pid;
3918 lkb->lkb_remid = ms->m_lkid;
3919 lkb->lkb_grmode = DLM_LOCK_IV;
3920 lkb->lkb_rqmode = ms->m_rqmode;
3921
3922 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3923 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3924
3925 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3926
3927 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3928 if (!lkb->lkb_lvbptr)
3929 return -ENOMEM;
3930 }
3931
3932 return 0;
3933}
3934
3935static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3936 struct dlm_message *ms)
3937{
3938 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3939 return -EBUSY;
3940
3941 if (receive_lvb(ls, lkb, ms))
3942 return -ENOMEM;
3943
3944 lkb->lkb_rqmode = ms->m_rqmode;
3945 lkb->lkb_lvbseq = ms->m_lvbseq;
3946
3947 return 0;
3948}
3949
3950static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3951 struct dlm_message *ms)
3952{
3953 if (receive_lvb(ls, lkb, ms))
3954 return -ENOMEM;
3955 return 0;
3956}
3957
3958
3959
3960
3961static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3962{
3963 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3964 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3965 lkb->lkb_remid = ms->m_lkid;
3966}
3967
3968
3969
3970
3971static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3972{
3973 int from = ms->m_header.h_nodeid;
3974 int error = 0;
3975
3976 switch (ms->m_type) {
3977 case DLM_MSG_CONVERT:
3978 case DLM_MSG_UNLOCK:
3979 case DLM_MSG_CANCEL:
3980 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3981 error = -EINVAL;
3982 break;
3983
3984 case DLM_MSG_CONVERT_REPLY:
3985 case DLM_MSG_UNLOCK_REPLY:
3986 case DLM_MSG_CANCEL_REPLY:
3987 case DLM_MSG_GRANT:
3988 case DLM_MSG_BAST:
3989 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3990 error = -EINVAL;
3991 break;
3992
3993 case DLM_MSG_REQUEST_REPLY:
3994 if (!is_process_copy(lkb))
3995 error = -EINVAL;
3996 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3997 error = -EINVAL;
3998 break;
3999
4000 default:
4001 error = -EINVAL;
4002 }
4003
4004 if (error)
4005 log_error(lkb->lkb_resource->res_ls,
4006 "ignore invalid message %d from %d %x %x %x %d",
4007 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
4008 lkb->lkb_flags, lkb->lkb_nodeid);
4009 return error;
4010}
4011
4012static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4013{
4014 char name[DLM_RESNAME_MAXLEN + 1];
4015 struct dlm_message *ms;
4016 struct dlm_mhandle *mh;
4017 struct dlm_rsb *r;
4018 uint32_t hash, b;
4019 int rv, dir_nodeid;
4020
4021 memset(name, 0, sizeof(name));
4022 memcpy(name, ms_name, len);
4023
4024 hash = jhash(name, len, 0);
4025 b = hash & (ls->ls_rsbtbl_size - 1);
4026
4027 dir_nodeid = dlm_hash2nodeid(ls, hash);
4028
4029 log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4030
4031 spin_lock(&ls->ls_rsbtbl[b].lock);
4032 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4033 if (!rv) {
4034 spin_unlock(&ls->ls_rsbtbl[b].lock);
4035 log_error(ls, "repeat_remove on keep %s", name);
4036 return;
4037 }
4038
4039 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4040 if (!rv) {
4041 spin_unlock(&ls->ls_rsbtbl[b].lock);
4042 log_error(ls, "repeat_remove on toss %s", name);
4043 return;
4044 }
4045
4046
4047
4048 spin_lock(&ls->ls_remove_spin);
4049 ls->ls_remove_len = len;
4050 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4051 spin_unlock(&ls->ls_remove_spin);
4052 spin_unlock(&ls->ls_rsbtbl[b].lock);
4053
4054 rv = _create_message(ls, sizeof(struct dlm_message) + len,
4055 dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4056 if (rv)
4057 return;
4058
4059 memcpy(ms->m_extra, name, len);
4060 ms->m_hash = hash;
4061
4062 send_message(mh, ms);
4063
4064 spin_lock(&ls->ls_remove_spin);
4065 ls->ls_remove_len = 0;
4066 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4067 spin_unlock(&ls->ls_remove_spin);
4068}
4069
4070static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4071{
4072 struct dlm_lkb *lkb;
4073 struct dlm_rsb *r;
4074 int from_nodeid;
4075 int error, namelen = 0;
4076
4077 from_nodeid = ms->m_header.h_nodeid;
4078
4079 error = create_lkb(ls, &lkb);
4080 if (error)
4081 goto fail;
4082
4083 receive_flags(lkb, ms);
4084 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4085 error = receive_request_args(ls, lkb, ms);
4086 if (error) {
4087 __put_lkb(ls, lkb);
4088 goto fail;
4089 }
4090
4091
4092
4093
4094
4095
4096
4097 namelen = receive_extralen(ms);
4098
4099 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4100 R_RECEIVE_REQUEST, &r);
4101 if (error) {
4102 __put_lkb(ls, lkb);
4103 goto fail;
4104 }
4105
4106 lock_rsb(r);
4107
4108 if (r->res_master_nodeid != dlm_our_nodeid()) {
4109 error = validate_master_nodeid(ls, r, from_nodeid);
4110 if (error) {
4111 unlock_rsb(r);
4112 put_rsb(r);
4113 __put_lkb(ls, lkb);
4114 goto fail;
4115 }
4116 }
4117
4118 attach_lkb(r, lkb);
4119 error = do_request(r, lkb);
4120 send_request_reply(r, lkb, error);
4121 do_request_effects(r, lkb, error);
4122
4123 unlock_rsb(r);
4124 put_rsb(r);
4125
4126 if (error == -EINPROGRESS)
4127 error = 0;
4128 if (error)
4129 dlm_put_lkb(lkb);
4130 return 0;
4131
4132 fail:
4133
4134
4135
4136
4137
4138
4139
4140
4141
4142
4143
4144
4145
4146
4147
4148 if (error != -ENOTBLK) {
4149 log_limit(ls, "receive_request %x from %d %d",
4150 ms->m_lkid, from_nodeid, error);
4151 }
4152
4153 if (namelen && error == -EBADR) {
4154 send_repeat_remove(ls, ms->m_extra, namelen);
4155 msleep(1000);
4156 }
4157
4158 setup_stub_lkb(ls, ms);
4159 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4160 return error;
4161}
4162
4163static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4164{
4165 struct dlm_lkb *lkb;
4166 struct dlm_rsb *r;
4167 int error, reply = 1;
4168
4169 error = find_lkb(ls, ms->m_remid, &lkb);
4170 if (error)
4171 goto fail;
4172
4173 if (lkb->lkb_remid != ms->m_lkid) {
4174 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4175 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4176 (unsigned long long)lkb->lkb_recover_seq,
4177 ms->m_header.h_nodeid, ms->m_lkid);
4178 error = -ENOENT;
4179 dlm_put_lkb(lkb);
4180 goto fail;
4181 }
4182
4183 r = lkb->lkb_resource;
4184
4185 hold_rsb(r);
4186 lock_rsb(r);
4187
4188 error = validate_message(lkb, ms);
4189 if (error)
4190 goto out;
4191
4192 receive_flags(lkb, ms);
4193
4194 error = receive_convert_args(ls, lkb, ms);
4195 if (error) {
4196 send_convert_reply(r, lkb, error);
4197 goto out;
4198 }
4199
4200 reply = !down_conversion(lkb);
4201
4202 error = do_convert(r, lkb);
4203 if (reply)
4204 send_convert_reply(r, lkb, error);
4205 do_convert_effects(r, lkb, error);
4206 out:
4207 unlock_rsb(r);
4208 put_rsb(r);
4209 dlm_put_lkb(lkb);
4210 return 0;
4211
4212 fail:
4213 setup_stub_lkb(ls, ms);
4214 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4215 return error;
4216}
4217
4218static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4219{
4220 struct dlm_lkb *lkb;
4221 struct dlm_rsb *r;
4222 int error;
4223
4224 error = find_lkb(ls, ms->m_remid, &lkb);
4225 if (error)
4226 goto fail;
4227
4228 if (lkb->lkb_remid != ms->m_lkid) {
4229 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4230 lkb->lkb_id, lkb->lkb_remid,
4231 ms->m_header.h_nodeid, ms->m_lkid);
4232 error = -ENOENT;
4233 dlm_put_lkb(lkb);
4234 goto fail;
4235 }
4236
4237 r = lkb->lkb_resource;
4238
4239 hold_rsb(r);
4240 lock_rsb(r);
4241
4242 error = validate_message(lkb, ms);
4243 if (error)
4244 goto out;
4245
4246 receive_flags(lkb, ms);
4247
4248 error = receive_unlock_args(ls, lkb, ms);
4249 if (error) {
4250 send_unlock_reply(r, lkb, error);
4251 goto out;
4252 }
4253
4254 error = do_unlock(r, lkb);
4255 send_unlock_reply(r, lkb, error);
4256 do_unlock_effects(r, lkb, error);
4257 out:
4258 unlock_rsb(r);
4259 put_rsb(r);
4260 dlm_put_lkb(lkb);
4261 return 0;
4262
4263 fail:
4264 setup_stub_lkb(ls, ms);
4265 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4266 return error;
4267}
4268
4269static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4270{
4271 struct dlm_lkb *lkb;
4272 struct dlm_rsb *r;
4273 int error;
4274
4275 error = find_lkb(ls, ms->m_remid, &lkb);
4276 if (error)
4277 goto fail;
4278
4279 receive_flags(lkb, ms);
4280
4281 r = lkb->lkb_resource;
4282
4283 hold_rsb(r);
4284 lock_rsb(r);
4285
4286 error = validate_message(lkb, ms);
4287 if (error)
4288 goto out;
4289
4290 error = do_cancel(r, lkb);
4291 send_cancel_reply(r, lkb, error);
4292 do_cancel_effects(r, lkb, error);
4293 out:
4294 unlock_rsb(r);
4295 put_rsb(r);
4296 dlm_put_lkb(lkb);
4297 return 0;
4298
4299 fail:
4300 setup_stub_lkb(ls, ms);
4301 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4302 return error;
4303}
4304
4305static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4306{
4307 struct dlm_lkb *lkb;
4308 struct dlm_rsb *r;
4309 int error;
4310
4311 error = find_lkb(ls, ms->m_remid, &lkb);
4312 if (error)
4313 return error;
4314
4315 r = lkb->lkb_resource;
4316
4317 hold_rsb(r);
4318 lock_rsb(r);
4319
4320 error = validate_message(lkb, ms);
4321 if (error)
4322 goto out;
4323
4324 receive_flags_reply(lkb, ms);
4325 if (is_altmode(lkb))
4326 munge_altmode(lkb, ms);
4327 grant_lock_pc(r, lkb, ms);
4328 queue_cast(r, lkb, 0);
4329 out:
4330 unlock_rsb(r);
4331 put_rsb(r);
4332 dlm_put_lkb(lkb);
4333 return 0;
4334}
4335
4336static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4337{
4338 struct dlm_lkb *lkb;
4339 struct dlm_rsb *r;
4340 int error;
4341
4342 error = find_lkb(ls, ms->m_remid, &lkb);
4343 if (error)
4344 return error;
4345
4346 r = lkb->lkb_resource;
4347
4348 hold_rsb(r);
4349 lock_rsb(r);
4350
4351 error = validate_message(lkb, ms);
4352 if (error)
4353 goto out;
4354
4355 queue_bast(r, lkb, ms->m_bastmode);
4356 lkb->lkb_highbast = ms->m_bastmode;
4357 out:
4358 unlock_rsb(r);
4359 put_rsb(r);
4360 dlm_put_lkb(lkb);
4361 return 0;
4362}
4363
4364static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4365{
4366 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4367
4368 from_nodeid = ms->m_header.h_nodeid;
4369 our_nodeid = dlm_our_nodeid();
4370
4371 len = receive_extralen(ms);
4372
4373 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4374 &ret_nodeid, NULL);
4375
4376
4377 if (!error && ret_nodeid == our_nodeid) {
4378 receive_request(ls, ms);
4379 return;
4380 }
4381 send_lookup_reply(ls, ms, ret_nodeid, error);
4382}
4383
4384static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4385{
4386 char name[DLM_RESNAME_MAXLEN+1];
4387 struct dlm_rsb *r;
4388 uint32_t hash, b;
4389 int rv, len, dir_nodeid, from_nodeid;
4390
4391 from_nodeid = ms->m_header.h_nodeid;
4392
4393 len = receive_extralen(ms);
4394
4395 if (len > DLM_RESNAME_MAXLEN) {
4396 log_error(ls, "receive_remove from %d bad len %d",
4397 from_nodeid, len);
4398 return;
4399 }
4400
4401 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4402 if (dir_nodeid != dlm_our_nodeid()) {
4403 log_error(ls, "receive_remove from %d bad nodeid %d",
4404 from_nodeid, dir_nodeid);
4405 return;
4406 }
4407
4408
4409
4410
4411
4412
4413
4414
4415
4416
4417 memset(name, 0, sizeof(name));
4418 memcpy(name, ms->m_extra, len);
4419
4420 hash = jhash(name, len, 0);
4421 b = hash & (ls->ls_rsbtbl_size - 1);
4422
4423 spin_lock(&ls->ls_rsbtbl[b].lock);
4424
4425 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4426 if (rv) {
4427
4428 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4429 if (rv) {
4430
4431 log_error(ls, "receive_remove from %d not found %s",
4432 from_nodeid, name);
4433 spin_unlock(&ls->ls_rsbtbl[b].lock);
4434 return;
4435 }
4436 if (r->res_master_nodeid != from_nodeid) {
4437
4438 log_error(ls, "receive_remove keep from %d master %d",
4439 from_nodeid, r->res_master_nodeid);
4440 dlm_print_rsb(r);
4441 spin_unlock(&ls->ls_rsbtbl[b].lock);
4442 return;
4443 }
4444
4445 log_debug(ls, "receive_remove from %d master %d first %x %s",
4446 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4447 name);
4448 spin_unlock(&ls->ls_rsbtbl[b].lock);
4449 return;
4450 }
4451
4452 if (r->res_master_nodeid != from_nodeid) {
4453 log_error(ls, "receive_remove toss from %d master %d",
4454 from_nodeid, r->res_master_nodeid);
4455 dlm_print_rsb(r);
4456 spin_unlock(&ls->ls_rsbtbl[b].lock);
4457 return;
4458 }
4459
4460 if (kref_put(&r->res_ref, kill_rsb)) {
4461 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4462 spin_unlock(&ls->ls_rsbtbl[b].lock);
4463 dlm_free_rsb(r);
4464 } else {
4465 log_error(ls, "receive_remove from %d rsb ref error",
4466 from_nodeid);
4467 dlm_print_rsb(r);
4468 spin_unlock(&ls->ls_rsbtbl[b].lock);
4469 }
4470}
4471
4472static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4473{
4474 do_purge(ls, ms->m_nodeid, ms->m_pid);
4475}
4476
4477static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4478{
4479 struct dlm_lkb *lkb;
4480 struct dlm_rsb *r;
4481 int error, mstype, result;
4482 int from_nodeid = ms->m_header.h_nodeid;
4483
4484 error = find_lkb(ls, ms->m_remid, &lkb);
4485 if (error)
4486 return error;
4487
4488 r = lkb->lkb_resource;
4489 hold_rsb(r);
4490 lock_rsb(r);
4491
4492 error = validate_message(lkb, ms);
4493 if (error)
4494 goto out;
4495
4496 mstype = lkb->lkb_wait_type;
4497 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4498 if (error) {
4499 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4500 lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4501 dlm_dump_rsb(r);
4502 goto out;
4503 }
4504
4505
4506
4507 if (mstype == DLM_MSG_LOOKUP) {
4508 r->res_master_nodeid = from_nodeid;
4509 r->res_nodeid = from_nodeid;
4510 lkb->lkb_nodeid = from_nodeid;
4511 }
4512
4513
4514 result = ms->m_result;
4515
4516 switch (result) {
4517 case -EAGAIN:
4518
4519 queue_cast(r, lkb, -EAGAIN);
4520 confirm_master(r, -EAGAIN);
4521 unhold_lkb(lkb);
4522 break;
4523
4524 case -EINPROGRESS:
4525 case 0:
4526
4527 receive_flags_reply(lkb, ms);
4528 lkb->lkb_remid = ms->m_lkid;
4529 if (is_altmode(lkb))
4530 munge_altmode(lkb, ms);
4531 if (result) {
4532 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4533 add_timeout(lkb);
4534 } else {
4535 grant_lock_pc(r, lkb, ms);
4536 queue_cast(r, lkb, 0);
4537 }
4538 confirm_master(r, result);
4539 break;
4540
4541 case -EBADR:
4542 case -ENOTBLK:
4543
4544 log_limit(ls, "receive_request_reply %x from %d %d "
4545 "master %d dir %d first %x %s", lkb->lkb_id,
4546 from_nodeid, result, r->res_master_nodeid,
4547 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4548
4549 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4550 r->res_master_nodeid != dlm_our_nodeid()) {
4551
4552 r->res_master_nodeid = 0;
4553 r->res_nodeid = -1;
4554 lkb->lkb_nodeid = -1;
4555 }
4556
4557 if (is_overlap(lkb)) {
4558
4559 queue_cast_overlap(r, lkb);
4560 confirm_master(r, result);
4561 unhold_lkb(lkb);
4562 } else {
4563 _request_lock(r, lkb);
4564
4565 if (r->res_master_nodeid == dlm_our_nodeid())
4566 confirm_master(r, 0);
4567 }
4568 break;
4569
4570 default:
4571 log_error(ls, "receive_request_reply %x error %d",
4572 lkb->lkb_id, result);
4573 }
4574
4575 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4576 log_debug(ls, "receive_request_reply %x result %d unlock",
4577 lkb->lkb_id, result);
4578 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4579 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4580 send_unlock(r, lkb);
4581 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4582 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4583 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4584 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4585 send_cancel(r, lkb);
4586 } else {
4587 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4588 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4589 }
4590 out:
4591 unlock_rsb(r);
4592 put_rsb(r);
4593 dlm_put_lkb(lkb);
4594 return 0;
4595}
4596
4597static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4598 struct dlm_message *ms)
4599{
4600
4601 switch (ms->m_result) {
4602 case -EAGAIN:
4603
4604 queue_cast(r, lkb, -EAGAIN);
4605 break;
4606
4607 case -EDEADLK:
4608 receive_flags_reply(lkb, ms);
4609 revert_lock_pc(r, lkb);
4610 queue_cast(r, lkb, -EDEADLK);
4611 break;
4612
4613 case -EINPROGRESS:
4614
4615 receive_flags_reply(lkb, ms);
4616 if (is_demoted(lkb))
4617 munge_demoted(lkb);
4618 del_lkb(r, lkb);
4619 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4620 add_timeout(lkb);
4621 break;
4622
4623 case 0:
4624
4625 receive_flags_reply(lkb, ms);
4626 if (is_demoted(lkb))
4627 munge_demoted(lkb);
4628 grant_lock_pc(r, lkb, ms);
4629 queue_cast(r, lkb, 0);
4630 break;
4631
4632 default:
4633 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4634 lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4635 ms->m_result);
4636 dlm_print_rsb(r);
4637 dlm_print_lkb(lkb);
4638 }
4639}
4640
4641static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4642{
4643 struct dlm_rsb *r = lkb->lkb_resource;
4644 int error;
4645
4646 hold_rsb(r);
4647 lock_rsb(r);
4648
4649 error = validate_message(lkb, ms);
4650 if (error)
4651 goto out;
4652
4653
4654 error = remove_from_waiters_ms(lkb, ms);
4655 if (error)
4656 goto out;
4657
4658 __receive_convert_reply(r, lkb, ms);
4659 out:
4660 unlock_rsb(r);
4661 put_rsb(r);
4662}
4663
4664static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4665{
4666 struct dlm_lkb *lkb;
4667 int error;
4668
4669 error = find_lkb(ls, ms->m_remid, &lkb);
4670 if (error)
4671 return error;
4672
4673 _receive_convert_reply(lkb, ms);
4674 dlm_put_lkb(lkb);
4675 return 0;
4676}
4677
4678static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4679{
4680 struct dlm_rsb *r = lkb->lkb_resource;
4681 int error;
4682
4683 hold_rsb(r);
4684 lock_rsb(r);
4685
4686 error = validate_message(lkb, ms);
4687 if (error)
4688 goto out;
4689
4690
4691 error = remove_from_waiters_ms(lkb, ms);
4692 if (error)
4693 goto out;
4694
4695
4696
4697 switch (ms->m_result) {
4698 case -DLM_EUNLOCK:
4699 receive_flags_reply(lkb, ms);
4700 remove_lock_pc(r, lkb);
4701 queue_cast(r, lkb, -DLM_EUNLOCK);
4702 break;
4703 case -ENOENT:
4704 break;
4705 default:
4706 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4707 lkb->lkb_id, ms->m_result);
4708 }
4709 out:
4710 unlock_rsb(r);
4711 put_rsb(r);
4712}
4713
4714static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4715{
4716 struct dlm_lkb *lkb;
4717 int error;
4718
4719 error = find_lkb(ls, ms->m_remid, &lkb);
4720 if (error)
4721 return error;
4722
4723 _receive_unlock_reply(lkb, ms);
4724 dlm_put_lkb(lkb);
4725 return 0;
4726}
4727
4728static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4729{
4730 struct dlm_rsb *r = lkb->lkb_resource;
4731 int error;
4732
4733 hold_rsb(r);
4734 lock_rsb(r);
4735
4736 error = validate_message(lkb, ms);
4737 if (error)
4738 goto out;
4739
4740
4741 error = remove_from_waiters_ms(lkb, ms);
4742 if (error)
4743 goto out;
4744
4745
4746
4747 switch (ms->m_result) {
4748 case -DLM_ECANCEL:
4749 receive_flags_reply(lkb, ms);
4750 revert_lock_pc(r, lkb);
4751 queue_cast(r, lkb, -DLM_ECANCEL);
4752 break;
4753 case 0:
4754 break;
4755 default:
4756 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4757 lkb->lkb_id, ms->m_result);
4758 }
4759 out:
4760 unlock_rsb(r);
4761 put_rsb(r);
4762}
4763
4764static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4765{
4766 struct dlm_lkb *lkb;
4767 int error;
4768
4769 error = find_lkb(ls, ms->m_remid, &lkb);
4770 if (error)
4771 return error;
4772
4773 _receive_cancel_reply(lkb, ms);
4774 dlm_put_lkb(lkb);
4775 return 0;
4776}
4777
4778static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4779{
4780 struct dlm_lkb *lkb;
4781 struct dlm_rsb *r;
4782 int error, ret_nodeid;
4783 int do_lookup_list = 0;
4784
4785 error = find_lkb(ls, ms->m_lkid, &lkb);
4786 if (error) {
4787 log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4788 return;
4789 }
4790
4791
4792
4793
4794 r = lkb->lkb_resource;
4795 hold_rsb(r);
4796 lock_rsb(r);
4797
4798 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4799 if (error)
4800 goto out;
4801
4802 ret_nodeid = ms->m_nodeid;
4803
4804
4805
4806
4807
4808
4809
4810 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4811
4812 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4813 "master %d dir %d our %d first %x %s",
4814 lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4815 r->res_master_nodeid, r->res_dir_nodeid,
4816 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4817 }
4818
4819 if (ret_nodeid == dlm_our_nodeid()) {
4820 r->res_master_nodeid = ret_nodeid;
4821 r->res_nodeid = 0;
4822 do_lookup_list = 1;
4823 r->res_first_lkid = 0;
4824 } else if (ret_nodeid == -1) {
4825
4826 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4827 lkb->lkb_id, ms->m_header.h_nodeid);
4828 r->res_master_nodeid = 0;
4829 r->res_nodeid = -1;
4830 lkb->lkb_nodeid = -1;
4831 } else {
4832
4833 r->res_master_nodeid = ret_nodeid;
4834 r->res_nodeid = ret_nodeid;
4835 }
4836
4837 if (is_overlap(lkb)) {
4838 log_debug(ls, "receive_lookup_reply %x unlock %x",
4839 lkb->lkb_id, lkb->lkb_flags);
4840 queue_cast_overlap(r, lkb);
4841 unhold_lkb(lkb);
4842 goto out_list;
4843 }
4844
4845 _request_lock(r, lkb);
4846
4847 out_list:
4848 if (do_lookup_list)
4849 process_lookup_list(r);
4850 out:
4851 unlock_rsb(r);
4852 put_rsb(r);
4853 dlm_put_lkb(lkb);
4854}
4855
4856static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4857 uint32_t saved_seq)
4858{
4859 int error = 0, noent = 0;
4860
4861 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4862 log_limit(ls, "receive %d from non-member %d %x %x %d",
4863 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4864 ms->m_remid, ms->m_result);
4865 return;
4866 }
4867
4868 switch (ms->m_type) {
4869
4870
4871
4872 case DLM_MSG_REQUEST:
4873 error = receive_request(ls, ms);
4874 break;
4875
4876 case DLM_MSG_CONVERT:
4877 error = receive_convert(ls, ms);
4878 break;
4879
4880 case DLM_MSG_UNLOCK:
4881 error = receive_unlock(ls, ms);
4882 break;
4883
4884 case DLM_MSG_CANCEL:
4885 noent = 1;
4886 error = receive_cancel(ls, ms);
4887 break;
4888
4889
4890
4891 case DLM_MSG_REQUEST_REPLY:
4892 error = receive_request_reply(ls, ms);
4893 break;
4894
4895 case DLM_MSG_CONVERT_REPLY:
4896 error = receive_convert_reply(ls, ms);
4897 break;
4898
4899 case DLM_MSG_UNLOCK_REPLY:
4900 error = receive_unlock_reply(ls, ms);
4901 break;
4902
4903 case DLM_MSG_CANCEL_REPLY:
4904 error = receive_cancel_reply(ls, ms);
4905 break;
4906
4907
4908
4909 case DLM_MSG_GRANT:
4910 noent = 1;
4911 error = receive_grant(ls, ms);
4912 break;
4913
4914 case DLM_MSG_BAST:
4915 noent = 1;
4916 error = receive_bast(ls, ms);
4917 break;
4918
4919
4920
4921 case DLM_MSG_LOOKUP:
4922 receive_lookup(ls, ms);
4923 break;
4924
4925 case DLM_MSG_REMOVE:
4926 receive_remove(ls, ms);
4927 break;
4928
4929
4930
4931 case DLM_MSG_LOOKUP_REPLY:
4932 receive_lookup_reply(ls, ms);
4933 break;
4934
4935
4936
4937 case DLM_MSG_PURGE:
4938 receive_purge(ls, ms);
4939 break;
4940
4941 default:
4942 log_error(ls, "unknown message type %d", ms->m_type);
4943 }
4944
4945
4946
4947
4948
4949
4950
4951
4952
4953
4954
4955
4956 if (error == -ENOENT && noent) {
4957 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4958 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4959 ms->m_lkid, saved_seq);
4960 } else if (error == -ENOENT) {
4961 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4962 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4963 ms->m_lkid, saved_seq);
4964
4965 if (ms->m_type == DLM_MSG_CONVERT)
4966 dlm_dump_rsb_hash(ls, ms->m_hash);
4967 }
4968
4969 if (error == -EINVAL) {
4970 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4971 "saved_seq %u",
4972 ms->m_type, ms->m_header.h_nodeid,
4973 ms->m_lkid, ms->m_remid, saved_seq);
4974 }
4975}
4976
4977
4978
4979
4980
4981
4982
4983
4984
4985static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4986 int nodeid)
4987{
4988 if (dlm_locking_stopped(ls)) {
4989
4990
4991
4992 if (!ls->ls_generation) {
4993 log_limit(ls, "receive %d from %d ignore old gen",
4994 ms->m_type, nodeid);
4995 return;
4996 }
4997
4998 dlm_add_requestqueue(ls, nodeid, ms);
4999 } else {
5000 dlm_wait_requestqueue(ls);
5001 _receive_message(ls, ms, 0);
5002 }
5003}
5004
5005
5006
5007
5008void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5009 uint32_t saved_seq)
5010{
5011 _receive_message(ls, ms, saved_seq);
5012}
5013
5014
5015
5016
5017
5018
5019void dlm_receive_buffer(union dlm_packet *p, int nodeid)
5020{
5021 struct dlm_header *hd = &p->header;
5022 struct dlm_ls *ls;
5023 int type = 0;
5024
5025 switch (hd->h_cmd) {
5026 case DLM_MSG:
5027 dlm_message_in(&p->message);
5028 type = p->message.m_type;
5029 break;
5030 case DLM_RCOM:
5031 dlm_rcom_in(&p->rcom);
5032 type = p->rcom.rc_type;
5033 break;
5034 default:
5035 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5036 return;
5037 }
5038
5039 if (hd->h_nodeid != nodeid) {
5040 log_print("invalid h_nodeid %d from %d lockspace %x",
5041 hd->h_nodeid, nodeid, hd->u.h_lockspace);
5042 return;
5043 }
5044
5045 ls = dlm_find_lockspace_global(hd->u.h_lockspace);
5046 if (!ls) {
5047 if (dlm_config.ci_log_debug) {
5048 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5049 "%u from %d cmd %d type %d\n",
5050 hd->u.h_lockspace, nodeid, hd->h_cmd, type);
5051 }
5052
5053 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5054 dlm_send_ls_not_ready(nodeid, &p->rcom);
5055 return;
5056 }
5057
5058
5059
5060
5061 down_read(&ls->ls_recv_active);
5062 if (hd->h_cmd == DLM_MSG)
5063 dlm_receive_message(ls, &p->message, nodeid);
5064 else
5065 dlm_receive_rcom(ls, &p->rcom, nodeid);
5066 up_read(&ls->ls_recv_active);
5067
5068 dlm_put_lockspace(ls);
5069}
5070
5071static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5072 struct dlm_message *ms_stub)
5073{
5074 if (middle_conversion(lkb)) {
5075 hold_lkb(lkb);
5076 memset(ms_stub, 0, sizeof(struct dlm_message));
5077 ms_stub->m_flags = DLM_IFL_STUB_MS;
5078 ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
5079 ms_stub->m_result = -EINPROGRESS;
5080 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5081 _receive_convert_reply(lkb, ms_stub);
5082
5083
5084 lkb->lkb_grmode = DLM_LOCK_IV;
5085 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5086 unhold_lkb(lkb);
5087
5088 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5089 lkb->lkb_flags |= DLM_IFL_RESEND;
5090 }
5091
5092
5093
5094}
5095
5096
5097
5098
5099static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5100 int dir_nodeid)
5101{
5102 if (dlm_no_directory(ls))
5103 return 1;
5104
5105 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5106 return 1;
5107
5108 return 0;
5109}
5110
5111
5112
5113
5114
5115
5116
5117void dlm_recover_waiters_pre(struct dlm_ls *ls)
5118{
5119 struct dlm_lkb *lkb, *safe;
5120 struct dlm_message *ms_stub;
5121 int wait_type, stub_unlock_result, stub_cancel_result;
5122 int dir_nodeid;
5123
5124 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5125 if (!ms_stub)
5126 return;
5127
5128 mutex_lock(&ls->ls_waiters_mutex);
5129
5130 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5131
5132 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5133
5134
5135
5136
5137 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5138 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5139 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5140 lkb->lkb_id,
5141 lkb->lkb_remid,
5142 lkb->lkb_wait_type,
5143 lkb->lkb_resource->res_nodeid,
5144 lkb->lkb_nodeid,
5145 lkb->lkb_wait_nodeid,
5146 dir_nodeid);
5147 }
5148
5149
5150
5151
5152 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5153 lkb->lkb_flags |= DLM_IFL_RESEND;
5154 continue;
5155 }
5156
5157 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5158 continue;
5159
5160 wait_type = lkb->lkb_wait_type;
5161 stub_unlock_result = -DLM_EUNLOCK;
5162 stub_cancel_result = -DLM_ECANCEL;
5163
5164
5165
5166
5167
5168
5169 if (!wait_type) {
5170 if (is_overlap_cancel(lkb)) {
5171 wait_type = DLM_MSG_CANCEL;
5172 if (lkb->lkb_grmode == DLM_LOCK_IV)
5173 stub_cancel_result = 0;
5174 }
5175 if (is_overlap_unlock(lkb)) {
5176 wait_type = DLM_MSG_UNLOCK;
5177 if (lkb->lkb_grmode == DLM_LOCK_IV)
5178 stub_unlock_result = -ENOENT;
5179 }
5180
5181 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5182 lkb->lkb_id, lkb->lkb_flags, wait_type,
5183 stub_cancel_result, stub_unlock_result);
5184 }
5185
5186 switch (wait_type) {
5187
5188 case DLM_MSG_REQUEST:
5189 lkb->lkb_flags |= DLM_IFL_RESEND;
5190 break;
5191
5192 case DLM_MSG_CONVERT:
5193 recover_convert_waiter(ls, lkb, ms_stub);
5194 break;
5195
5196 case DLM_MSG_UNLOCK:
5197 hold_lkb(lkb);
5198 memset(ms_stub, 0, sizeof(struct dlm_message));
5199 ms_stub->m_flags = DLM_IFL_STUB_MS;
5200 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
5201 ms_stub->m_result = stub_unlock_result;
5202 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5203 _receive_unlock_reply(lkb, ms_stub);
5204 dlm_put_lkb(lkb);
5205 break;
5206
5207 case DLM_MSG_CANCEL:
5208 hold_lkb(lkb);
5209 memset(ms_stub, 0, sizeof(struct dlm_message));
5210 ms_stub->m_flags = DLM_IFL_STUB_MS;
5211 ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
5212 ms_stub->m_result = stub_cancel_result;
5213 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5214 _receive_cancel_reply(lkb, ms_stub);
5215 dlm_put_lkb(lkb);
5216 break;
5217
5218 default:
5219 log_error(ls, "invalid lkb wait_type %d %d",
5220 lkb->lkb_wait_type, wait_type);
5221 }
5222 schedule();
5223 }
5224 mutex_unlock(&ls->ls_waiters_mutex);
5225 kfree(ms_stub);
5226}
5227
5228static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5229{
5230 struct dlm_lkb *lkb;
5231 int found = 0;
5232
5233 mutex_lock(&ls->ls_waiters_mutex);
5234 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5235 if (lkb->lkb_flags & DLM_IFL_RESEND) {
5236 hold_lkb(lkb);
5237 found = 1;
5238 break;
5239 }
5240 }
5241 mutex_unlock(&ls->ls_waiters_mutex);
5242
5243 if (!found)
5244 lkb = NULL;
5245 return lkb;
5246}
5247
5248
5249
5250
5251
5252
5253
5254
5255
5256
5257
5258
5259
5260
5261
5262
5263
5264int dlm_recover_waiters_post(struct dlm_ls *ls)
5265{
5266 struct dlm_lkb *lkb;
5267 struct dlm_rsb *r;
5268 int error = 0, mstype, err, oc, ou;
5269
5270 while (1) {
5271 if (dlm_locking_stopped(ls)) {
5272 log_debug(ls, "recover_waiters_post aborted");
5273 error = -EINTR;
5274 break;
5275 }
5276
5277 lkb = find_resend_waiter(ls);
5278 if (!lkb)
5279 break;
5280
5281 r = lkb->lkb_resource;
5282 hold_rsb(r);
5283 lock_rsb(r);
5284
5285 mstype = lkb->lkb_wait_type;
5286 oc = is_overlap_cancel(lkb);
5287 ou = is_overlap_unlock(lkb);
5288 err = 0;
5289
5290 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5291 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5292 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5293 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5294 dlm_dir_nodeid(r), oc, ou);
5295
5296
5297
5298
5299
5300 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5301 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5302 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5303 lkb->lkb_wait_type = 0;
5304 lkb->lkb_wait_count = 0;
5305 mutex_lock(&ls->ls_waiters_mutex);
5306 list_del_init(&lkb->lkb_wait_reply);
5307 mutex_unlock(&ls->ls_waiters_mutex);
5308 unhold_lkb(lkb);
5309
5310 if (oc || ou) {
5311
5312 switch (mstype) {
5313 case DLM_MSG_LOOKUP:
5314 case DLM_MSG_REQUEST:
5315 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5316 -DLM_ECANCEL);
5317 unhold_lkb(lkb);
5318 break;
5319 case DLM_MSG_CONVERT:
5320 if (oc) {
5321 queue_cast(r, lkb, -DLM_ECANCEL);
5322 } else {
5323 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5324 _unlock_lock(r, lkb);
5325 }
5326 break;
5327 default:
5328 err = 1;
5329 }
5330 } else {
5331 switch (mstype) {
5332 case DLM_MSG_LOOKUP:
5333 case DLM_MSG_REQUEST:
5334 _request_lock(r, lkb);
5335 if (is_master(r))
5336 confirm_master(r, 0);
5337 break;
5338 case DLM_MSG_CONVERT:
5339 _convert_lock(r, lkb);
5340 break;
5341 default:
5342 err = 1;
5343 }
5344 }
5345
5346 if (err) {
5347 log_error(ls, "waiter %x msg %d r_nodeid %d "
5348 "dir_nodeid %d overlap %d %d",
5349 lkb->lkb_id, mstype, r->res_nodeid,
5350 dlm_dir_nodeid(r), oc, ou);
5351 }
5352 unlock_rsb(r);
5353 put_rsb(r);
5354 dlm_put_lkb(lkb);
5355 }
5356
5357 return error;
5358}
5359
5360static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5361 struct list_head *list)
5362{
5363 struct dlm_lkb *lkb, *safe;
5364
5365 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5366 if (!is_master_copy(lkb))
5367 continue;
5368
5369
5370
5371
5372 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5373 continue;
5374
5375 del_lkb(r, lkb);
5376
5377
5378 if (!dlm_put_lkb(lkb))
5379 log_error(ls, "purged mstcpy lkb not released");
5380 }
5381}
5382
5383void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5384{
5385 struct dlm_ls *ls = r->res_ls;
5386
5387 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5388 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5389 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5390}
5391
5392static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5393 struct list_head *list,
5394 int nodeid_gone, unsigned int *count)
5395{
5396 struct dlm_lkb *lkb, *safe;
5397
5398 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5399 if (!is_master_copy(lkb))
5400 continue;
5401
5402 if ((lkb->lkb_nodeid == nodeid_gone) ||
5403 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5404
5405
5406
5407 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5408 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5409 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5410 }
5411
5412 del_lkb(r, lkb);
5413
5414
5415 if (!dlm_put_lkb(lkb))
5416 log_error(ls, "purged dead lkb not released");
5417
5418 rsb_set_flag(r, RSB_RECOVER_GRANT);
5419
5420 (*count)++;
5421 }
5422 }
5423}
5424
5425
5426
5427void dlm_recover_purge(struct dlm_ls *ls)
5428{
5429 struct dlm_rsb *r;
5430 struct dlm_member *memb;
5431 int nodes_count = 0;
5432 int nodeid_gone = 0;
5433 unsigned int lkb_count = 0;
5434
5435
5436
5437
5438 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5439 nodes_count++;
5440 nodeid_gone = memb->nodeid;
5441 }
5442
5443 if (!nodes_count)
5444 return;
5445
5446 down_write(&ls->ls_root_sem);
5447 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5448 hold_rsb(r);
5449 lock_rsb(r);
5450 if (is_master(r)) {
5451 purge_dead_list(ls, r, &r->res_grantqueue,
5452 nodeid_gone, &lkb_count);
5453 purge_dead_list(ls, r, &r->res_convertqueue,
5454 nodeid_gone, &lkb_count);
5455 purge_dead_list(ls, r, &r->res_waitqueue,
5456 nodeid_gone, &lkb_count);
5457 }
5458 unlock_rsb(r);
5459 unhold_rsb(r);
5460 cond_resched();
5461 }
5462 up_write(&ls->ls_root_sem);
5463
5464 if (lkb_count)
5465 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5466 lkb_count, nodes_count);
5467}
5468
5469static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5470{
5471 struct rb_node *n;
5472 struct dlm_rsb *r;
5473
5474 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5475 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5476 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5477
5478 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5479 continue;
5480 if (!is_master(r)) {
5481 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5482 continue;
5483 }
5484 hold_rsb(r);
5485 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5486 return r;
5487 }
5488 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5489 return NULL;
5490}
5491
5492
5493
5494
5495
5496
5497
5498
5499
5500
5501
5502
5503
5504
5505
5506
5507
5508
5509void dlm_recover_grant(struct dlm_ls *ls)
5510{
5511 struct dlm_rsb *r;
5512 int bucket = 0;
5513 unsigned int count = 0;
5514 unsigned int rsb_count = 0;
5515 unsigned int lkb_count = 0;
5516
5517 while (1) {
5518 r = find_grant_rsb(ls, bucket);
5519 if (!r) {
5520 if (bucket == ls->ls_rsbtbl_size - 1)
5521 break;
5522 bucket++;
5523 continue;
5524 }
5525 rsb_count++;
5526 count = 0;
5527 lock_rsb(r);
5528
5529 grant_pending_locks(r, &count);
5530 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5531 lkb_count += count;
5532 confirm_master(r, 0);
5533 unlock_rsb(r);
5534 put_rsb(r);
5535 cond_resched();
5536 }
5537
5538 if (lkb_count)
5539 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5540 lkb_count, rsb_count);
5541}
5542
5543static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5544 uint32_t remid)
5545{
5546 struct dlm_lkb *lkb;
5547
5548 list_for_each_entry(lkb, head, lkb_statequeue) {
5549 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5550 return lkb;
5551 }
5552 return NULL;
5553}
5554
5555static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5556 uint32_t remid)
5557{
5558 struct dlm_lkb *lkb;
5559
5560 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5561 if (lkb)
5562 return lkb;
5563 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5564 if (lkb)
5565 return lkb;
5566 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5567 if (lkb)
5568 return lkb;
5569 return NULL;
5570}
5571
5572
5573static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5574 struct dlm_rsb *r, struct dlm_rcom *rc)
5575{
5576 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5577
5578 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
5579 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5580 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5581 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5582 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5583 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5584 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5585 lkb->lkb_rqmode = rl->rl_rqmode;
5586 lkb->lkb_grmode = rl->rl_grmode;
5587
5588
5589 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5590 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5591
5592 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5593 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
5594 sizeof(struct rcom_lock);
5595 if (lvblen > ls->ls_lvblen)
5596 return -EINVAL;
5597 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5598 if (!lkb->lkb_lvbptr)
5599 return -ENOMEM;
5600 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5601 }
5602
5603
5604
5605
5606
5607 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5608 middle_conversion(lkb)) {
5609 rl->rl_status = DLM_LKSTS_CONVERT;
5610 lkb->lkb_grmode = DLM_LOCK_IV;
5611 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5612 }
5613
5614 return 0;
5615}
5616
5617
5618
5619
5620
5621
5622
5623
5624int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5625{
5626 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5627 struct dlm_rsb *r;
5628 struct dlm_lkb *lkb;
5629 uint32_t remid = 0;
5630 int from_nodeid = rc->rc_header.h_nodeid;
5631 int error;
5632
5633 if (rl->rl_parent_lkid) {
5634 error = -EOPNOTSUPP;
5635 goto out;
5636 }
5637
5638 remid = le32_to_cpu(rl->rl_lkid);
5639
5640
5641
5642
5643
5644
5645
5646
5647
5648 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5649 from_nodeid, R_RECEIVE_RECOVER, &r);
5650 if (error)
5651 goto out;
5652
5653 lock_rsb(r);
5654
5655 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5656 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5657 from_nodeid, remid);
5658 error = -EBADR;
5659 goto out_unlock;
5660 }
5661
5662 lkb = search_remid(r, from_nodeid, remid);
5663 if (lkb) {
5664 error = -EEXIST;
5665 goto out_remid;
5666 }
5667
5668 error = create_lkb(ls, &lkb);
5669 if (error)
5670 goto out_unlock;
5671
5672 error = receive_rcom_lock_args(ls, lkb, r, rc);
5673 if (error) {
5674 __put_lkb(ls, lkb);
5675 goto out_unlock;
5676 }
5677
5678 attach_lkb(r, lkb);
5679 add_lkb(r, lkb, rl->rl_status);
5680 error = 0;
5681 ls->ls_recover_locks_in++;
5682
5683 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5684 rsb_set_flag(r, RSB_RECOVER_GRANT);
5685
5686 out_remid:
5687
5688
5689 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5690
5691 lkb->lkb_recover_seq = ls->ls_recover_seq;
5692
5693 out_unlock:
5694 unlock_rsb(r);
5695 put_rsb(r);
5696 out:
5697 if (error && error != -EEXIST)
5698 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5699 from_nodeid, remid, error);
5700 rl->rl_result = cpu_to_le32(error);
5701 return error;
5702}
5703
5704
5705int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5706{
5707 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5708 struct dlm_rsb *r;
5709 struct dlm_lkb *lkb;
5710 uint32_t lkid, remid;
5711 int error, result;
5712
5713 lkid = le32_to_cpu(rl->rl_lkid);
5714 remid = le32_to_cpu(rl->rl_remid);
5715 result = le32_to_cpu(rl->rl_result);
5716
5717 error = find_lkb(ls, lkid, &lkb);
5718 if (error) {
5719 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5720 lkid, rc->rc_header.h_nodeid, remid, result);
5721 return error;
5722 }
5723
5724 r = lkb->lkb_resource;
5725 hold_rsb(r);
5726 lock_rsb(r);
5727
5728 if (!is_process_copy(lkb)) {
5729 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5730 lkid, rc->rc_header.h_nodeid, remid, result);
5731 dlm_dump_rsb(r);
5732 unlock_rsb(r);
5733 put_rsb(r);
5734 dlm_put_lkb(lkb);
5735 return -EINVAL;
5736 }
5737
5738 switch (result) {
5739 case -EBADR:
5740
5741
5742
5743
5744 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5745 lkid, rc->rc_header.h_nodeid, remid, result);
5746
5747 dlm_send_rcom_lock(r, lkb);
5748 goto out;
5749 case -EEXIST:
5750 case 0:
5751 lkb->lkb_remid = remid;
5752 break;
5753 default:
5754 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5755 lkid, rc->rc_header.h_nodeid, remid, result);
5756 }
5757
5758
5759
5760 dlm_recovered_lock(r);
5761 out:
5762 unlock_rsb(r);
5763 put_rsb(r);
5764 dlm_put_lkb(lkb);
5765
5766 return 0;
5767}
5768
5769int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5770 int mode, uint32_t flags, void *name, unsigned int namelen,
5771 unsigned long timeout_cs)
5772{
5773 struct dlm_lkb *lkb;
5774 struct dlm_args args;
5775 int error;
5776
5777 dlm_lock_recovery(ls);
5778
5779 error = create_lkb(ls, &lkb);
5780 if (error) {
5781 kfree(ua);
5782 goto out;
5783 }
5784
5785 if (flags & DLM_LKF_VALBLK) {
5786 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5787 if (!ua->lksb.sb_lvbptr) {
5788 kfree(ua);
5789 __put_lkb(ls, lkb);
5790 error = -ENOMEM;
5791 goto out;
5792 }
5793 }
5794 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5795 fake_astfn, ua, fake_bastfn, &args);
5796 if (error) {
5797 kfree(ua->lksb.sb_lvbptr);
5798 ua->lksb.sb_lvbptr = NULL;
5799 kfree(ua);
5800 __put_lkb(ls, lkb);
5801 goto out;
5802 }
5803
5804
5805
5806
5807 lkb->lkb_flags |= DLM_IFL_USER;
5808 error = request_lock(ls, lkb, name, namelen, &args);
5809
5810 switch (error) {
5811 case 0:
5812 break;
5813 case -EINPROGRESS:
5814 error = 0;
5815 break;
5816 case -EAGAIN:
5817 error = 0;
5818 fallthrough;
5819 default:
5820 __put_lkb(ls, lkb);
5821 goto out;
5822 }
5823
5824
5825 spin_lock(&ua->proc->locks_spin);
5826 hold_lkb(lkb);
5827 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5828 spin_unlock(&ua->proc->locks_spin);
5829 out:
5830 dlm_unlock_recovery(ls);
5831 return error;
5832}
5833
5834int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5835 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5836 unsigned long timeout_cs)
5837{
5838 struct dlm_lkb *lkb;
5839 struct dlm_args args;
5840 struct dlm_user_args *ua;
5841 int error;
5842
5843 dlm_lock_recovery(ls);
5844
5845 error = find_lkb(ls, lkid, &lkb);
5846 if (error)
5847 goto out;
5848
5849
5850
5851
5852 ua = lkb->lkb_ua;
5853
5854 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5855 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5856 if (!ua->lksb.sb_lvbptr) {
5857 error = -ENOMEM;
5858 goto out_put;
5859 }
5860 }
5861 if (lvb_in && ua->lksb.sb_lvbptr)
5862 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5863
5864 ua->xid = ua_tmp->xid;
5865 ua->castparam = ua_tmp->castparam;
5866 ua->castaddr = ua_tmp->castaddr;
5867 ua->bastparam = ua_tmp->bastparam;
5868 ua->bastaddr = ua_tmp->bastaddr;
5869 ua->user_lksb = ua_tmp->user_lksb;
5870
5871 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5872 fake_astfn, ua, fake_bastfn, &args);
5873 if (error)
5874 goto out_put;
5875
5876 error = convert_lock(ls, lkb, &args);
5877
5878 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5879 error = 0;
5880 out_put:
5881 dlm_put_lkb(lkb);
5882 out:
5883 dlm_unlock_recovery(ls);
5884 kfree(ua_tmp);
5885 return error;
5886}
5887
5888
5889
5890
5891
5892
5893
5894int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5895 int mode, uint32_t flags, void *name, unsigned int namelen,
5896 unsigned long timeout_cs, uint32_t *lkid)
5897{
5898 struct dlm_lkb *lkb;
5899 struct dlm_user_args *ua;
5900 int found_other_mode = 0;
5901 int found = 0;
5902 int rv = 0;
5903
5904 mutex_lock(&ls->ls_orphans_mutex);
5905 list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5906 if (lkb->lkb_resource->res_length != namelen)
5907 continue;
5908 if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5909 continue;
5910 if (lkb->lkb_grmode != mode) {
5911 found_other_mode = 1;
5912 continue;
5913 }
5914
5915 found = 1;
5916 list_del_init(&lkb->lkb_ownqueue);
5917 lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5918 *lkid = lkb->lkb_id;
5919 break;
5920 }
5921 mutex_unlock(&ls->ls_orphans_mutex);
5922
5923 if (!found && found_other_mode) {
5924 rv = -EAGAIN;
5925 goto out;
5926 }
5927
5928 if (!found) {
5929 rv = -ENOENT;
5930 goto out;
5931 }
5932
5933 lkb->lkb_exflags = flags;
5934 lkb->lkb_ownpid = (int) current->pid;
5935
5936 ua = lkb->lkb_ua;
5937
5938 ua->proc = ua_tmp->proc;
5939 ua->xid = ua_tmp->xid;
5940 ua->castparam = ua_tmp->castparam;
5941 ua->castaddr = ua_tmp->castaddr;
5942 ua->bastparam = ua_tmp->bastparam;
5943 ua->bastaddr = ua_tmp->bastaddr;
5944 ua->user_lksb = ua_tmp->user_lksb;
5945
5946
5947
5948
5949
5950
5951
5952 spin_lock(&ua->proc->locks_spin);
5953 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5954 spin_unlock(&ua->proc->locks_spin);
5955 out:
5956 kfree(ua_tmp);
5957 return rv;
5958}
5959
5960int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5961 uint32_t flags, uint32_t lkid, char *lvb_in)
5962{
5963 struct dlm_lkb *lkb;
5964 struct dlm_args args;
5965 struct dlm_user_args *ua;
5966 int error;
5967
5968 dlm_lock_recovery(ls);
5969
5970 error = find_lkb(ls, lkid, &lkb);
5971 if (error)
5972 goto out;
5973
5974 ua = lkb->lkb_ua;
5975
5976 if (lvb_in && ua->lksb.sb_lvbptr)
5977 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5978 if (ua_tmp->castparam)
5979 ua->castparam = ua_tmp->castparam;
5980 ua->user_lksb = ua_tmp->user_lksb;
5981
5982 error = set_unlock_args(flags, ua, &args);
5983 if (error)
5984 goto out_put;
5985
5986 error = unlock_lock(ls, lkb, &args);
5987
5988 if (error == -DLM_EUNLOCK)
5989 error = 0;
5990
5991 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
5992 error = 0;
5993 if (error)
5994 goto out_put;
5995
5996 spin_lock(&ua->proc->locks_spin);
5997
5998 if (!list_empty(&lkb->lkb_ownqueue))
5999 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6000 spin_unlock(&ua->proc->locks_spin);
6001 out_put:
6002 dlm_put_lkb(lkb);
6003 out:
6004 dlm_unlock_recovery(ls);
6005 kfree(ua_tmp);
6006 return error;
6007}
6008
6009int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6010 uint32_t flags, uint32_t lkid)
6011{
6012 struct dlm_lkb *lkb;
6013 struct dlm_args args;
6014 struct dlm_user_args *ua;
6015 int error;
6016
6017 dlm_lock_recovery(ls);
6018
6019 error = find_lkb(ls, lkid, &lkb);
6020 if (error)
6021 goto out;
6022
6023 ua = lkb->lkb_ua;
6024 if (ua_tmp->castparam)
6025 ua->castparam = ua_tmp->castparam;
6026 ua->user_lksb = ua_tmp->user_lksb;
6027
6028 error = set_unlock_args(flags, ua, &args);
6029 if (error)
6030 goto out_put;
6031
6032 error = cancel_lock(ls, lkb, &args);
6033
6034 if (error == -DLM_ECANCEL)
6035 error = 0;
6036
6037 if (error == -EBUSY)
6038 error = 0;
6039 out_put:
6040 dlm_put_lkb(lkb);
6041 out:
6042 dlm_unlock_recovery(ls);
6043 kfree(ua_tmp);
6044 return error;
6045}
6046
6047int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6048{
6049 struct dlm_lkb *lkb;
6050 struct dlm_args args;
6051 struct dlm_user_args *ua;
6052 struct dlm_rsb *r;
6053 int error;
6054
6055 dlm_lock_recovery(ls);
6056
6057 error = find_lkb(ls, lkid, &lkb);
6058 if (error)
6059 goto out;
6060
6061 ua = lkb->lkb_ua;
6062
6063 error = set_unlock_args(flags, ua, &args);
6064 if (error)
6065 goto out_put;
6066
6067
6068
6069 r = lkb->lkb_resource;
6070 hold_rsb(r);
6071 lock_rsb(r);
6072
6073 error = validate_unlock_args(lkb, &args);
6074 if (error)
6075 goto out_r;
6076 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6077
6078 error = _cancel_lock(r, lkb);
6079 out_r:
6080 unlock_rsb(r);
6081 put_rsb(r);
6082
6083 if (error == -DLM_ECANCEL)
6084 error = 0;
6085
6086 if (error == -EBUSY)
6087 error = 0;
6088 out_put:
6089 dlm_put_lkb(lkb);
6090 out:
6091 dlm_unlock_recovery(ls);
6092 return error;
6093}
6094
6095
6096
6097
6098static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6099{
6100 struct dlm_args args;
6101 int error;
6102
6103 hold_lkb(lkb);
6104 mutex_lock(&ls->ls_orphans_mutex);
6105 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6106 mutex_unlock(&ls->ls_orphans_mutex);
6107
6108 set_unlock_args(0, lkb->lkb_ua, &args);
6109
6110 error = cancel_lock(ls, lkb, &args);
6111 if (error == -DLM_ECANCEL)
6112 error = 0;
6113 return error;
6114}
6115
6116
6117
6118
6119
6120
6121static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6122{
6123 struct dlm_args args;
6124 int error;
6125
6126 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6127 lkb->lkb_ua, &args);
6128
6129 error = unlock_lock(ls, lkb, &args);
6130 if (error == -DLM_EUNLOCK)
6131 error = 0;
6132 return error;
6133}
6134
6135
6136
6137
6138
6139static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6140 struct dlm_user_proc *proc)
6141{
6142 struct dlm_lkb *lkb = NULL;
6143
6144 mutex_lock(&ls->ls_clear_proc_locks);
6145 if (list_empty(&proc->locks))
6146 goto out;
6147
6148 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6149 list_del_init(&lkb->lkb_ownqueue);
6150
6151 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6152 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6153 else
6154 lkb->lkb_flags |= DLM_IFL_DEAD;
6155 out:
6156 mutex_unlock(&ls->ls_clear_proc_locks);
6157 return lkb;
6158}
6159
6160
6161
6162
6163
6164
6165
6166
6167
6168
6169
6170void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6171{
6172 struct dlm_lkb *lkb, *safe;
6173
6174 dlm_lock_recovery(ls);
6175
6176 while (1) {
6177 lkb = del_proc_lock(ls, proc);
6178 if (!lkb)
6179 break;
6180 del_timeout(lkb);
6181 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6182 orphan_proc_lock(ls, lkb);
6183 else
6184 unlock_proc_lock(ls, lkb);
6185
6186
6187
6188
6189
6190 dlm_put_lkb(lkb);
6191 }
6192
6193 mutex_lock(&ls->ls_clear_proc_locks);
6194
6195
6196 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6197 list_del_init(&lkb->lkb_ownqueue);
6198 lkb->lkb_flags |= DLM_IFL_DEAD;
6199 dlm_put_lkb(lkb);
6200 }
6201
6202 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6203 memset(&lkb->lkb_callbacks, 0,
6204 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6205 list_del_init(&lkb->lkb_cb_list);
6206 dlm_put_lkb(lkb);
6207 }
6208
6209 mutex_unlock(&ls->ls_clear_proc_locks);
6210 dlm_unlock_recovery(ls);
6211}
6212
6213static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6214{
6215 struct dlm_lkb *lkb, *safe;
6216
6217 while (1) {
6218 lkb = NULL;
6219 spin_lock(&proc->locks_spin);
6220 if (!list_empty(&proc->locks)) {
6221 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6222 lkb_ownqueue);
6223 list_del_init(&lkb->lkb_ownqueue);
6224 }
6225 spin_unlock(&proc->locks_spin);
6226
6227 if (!lkb)
6228 break;
6229
6230 lkb->lkb_flags |= DLM_IFL_DEAD;
6231 unlock_proc_lock(ls, lkb);
6232 dlm_put_lkb(lkb);
6233 }
6234
6235 spin_lock(&proc->locks_spin);
6236 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6237 list_del_init(&lkb->lkb_ownqueue);
6238 lkb->lkb_flags |= DLM_IFL_DEAD;
6239 dlm_put_lkb(lkb);
6240 }
6241 spin_unlock(&proc->locks_spin);
6242
6243 spin_lock(&proc->asts_spin);
6244 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6245 memset(&lkb->lkb_callbacks, 0,
6246 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6247 list_del_init(&lkb->lkb_cb_list);
6248 dlm_put_lkb(lkb);
6249 }
6250 spin_unlock(&proc->asts_spin);
6251}
6252
6253
6254
6255static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6256{
6257 struct dlm_lkb *lkb, *safe;
6258
6259 mutex_lock(&ls->ls_orphans_mutex);
6260 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6261 if (pid && lkb->lkb_ownpid != pid)
6262 continue;
6263 unlock_proc_lock(ls, lkb);
6264 list_del_init(&lkb->lkb_ownqueue);
6265 dlm_put_lkb(lkb);
6266 }
6267 mutex_unlock(&ls->ls_orphans_mutex);
6268}
6269
6270static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6271{
6272 struct dlm_message *ms;
6273 struct dlm_mhandle *mh;
6274 int error;
6275
6276 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6277 DLM_MSG_PURGE, &ms, &mh);
6278 if (error)
6279 return error;
6280 ms->m_nodeid = nodeid;
6281 ms->m_pid = pid;
6282
6283 return send_message(mh, ms);
6284}
6285
6286int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6287 int nodeid, int pid)
6288{
6289 int error = 0;
6290
6291 if (nodeid && (nodeid != dlm_our_nodeid())) {
6292 error = send_purge(ls, nodeid, pid);
6293 } else {
6294 dlm_lock_recovery(ls);
6295 if (pid == current->pid)
6296 purge_proc_locks(ls, proc);
6297 else
6298 do_purge(ls, nodeid, pid);
6299 dlm_unlock_recovery(ls);
6300 }
6301 return error;
6302}
6303
6304