// SPDX-License-Identifier: GPL-2.0-only/***************************************************************************************************************************************************************** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.*****************************************************************************************************************************************************************/ /* Central locking logic has four stages: dlm_lock() dlm_unlock() request_lock(ls, lkb) convert_lock(ls, lkb) unlock_lock(ls, lkb) cancel_lock(ls, lkb) _request_lock(r, lkb) _convert_lock(r, lkb) _unlock_lock(r, lkb) _cancel_lock(r, lkb) do_request(r, lkb) do_convert(r, lkb) do_unlock(r, lkb) do_cancel(r, lkb) Stage 1 (lock, unlock) is mainly about checking input args and splitting into one of the four main operations: dlm_lock = request_lock dlm_lock+CONVERT = convert_lock dlm_unlock = unlock_lock dlm_unlock+CANCEL = cancel_lock Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is provided to the next stage. Stage 3, _xxxx_lock(), determines if the operation is local or remote. When remote, it calls send_xxxx(), when local it calls do_xxxx(). Stage 4, do_xxxx(), is the guts of the operation. It manipulates the given rsb and lkb and queues callbacks. For remote operations, send_xxxx() results in the corresponding do_xxxx() function being executed on the remote node. The connecting send/receive calls on local (L) and remote (R) nodes: L: send_xxxx() -> R: receive_xxxx() R: do_xxxx() L: receive_xxxx_reply() <- R: send_xxxx_reply()*/#include <trace/events/dlm.h> #include <linux/types.h>#include <linux/rbtree.h>#include <linux/slab.h>#include "dlm_internal.h"#include <linux/dlm_device.h>#include "memory.h"#include "midcomms.h"#include "requestqueue.h"#include "util.h"#include "dir.h"#include "member.h"#include "lockspace.h"#include "ast.h"#include "lock.h"#include "rcom.h"#include "recover.h"#include "lvb_table.h"#include "user.h"#include "config.h" static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);static int send_remove(struct dlm_rsb *r);static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, const struct dlm_message *ms, bool local);static int receive_extralen(const struct dlm_message *ms);static void do_purge(struct dlm_ls *ls, int nodeid, int pid);static void deactivate_rsb(struct kref *kref); /* * Lock compatibilty matrix - thanks Steve * UN = Unlocked state. Not really a state, used as a flag * PD = Padding. Used to make the matrix a nice power of two in size * Other states are the same as the VMS DLM. * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same) */ static const int __dlm_compat_matrix[8][8] = { /* UN NL CR CW PR PW EX PD */ {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */ {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */ {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */ {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */ {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */ {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */ {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */ {0, 0, 0, 0, 0, 0, 0, 0} /* PD */}; /* * This defines the direction of transfer of LVB data. * Granted mode is the row; requested mode is the column. * Usage: matrix[grmode+1][rqmode+1] * 1 = LVB is returned to the caller * 0 = LVB is written to the resource * -1 = nothing happens to the LVB */ const int dlm_lvb_operations[8][8] = { /* UN NL CR CW PR PW EX PD*/ { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */ { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */ { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */ { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */ { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */ { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */ { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */ { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */}; #define modes_compat(gr, rq) \ __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1] int dlm_modes_compat(int mode1, int mode2){ return __dlm_compat_matrix[mode1 + 1][mode2 + 1];} /* * Compatibility matrix for conversions with QUECVT set. * Granted mode is the row; requested mode is the column. * Usage: matrix[grmode+1][rqmode+1] */ static const int __quecvt_compat_matrix[8][8] = { /* UN NL CR CW PR PW EX PD */ {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */ {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */ {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */ {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */ {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */ {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */ {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */ {0, 0, 0, 0, 0, 0, 0, 0} /* PD */}; void dlm_print_lkb(struct dlm_lkb *lkb){ printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x " "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n", lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags, dlm_iflags_val(lkb), lkb->lkb_status, lkb->lkb_rqmode, lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid, (unsigned long long)lkb->lkb_recover_seq);} static void dlm_print_rsb(struct dlm_rsb *r){ printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x " "rlc %d name %s\n", r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid, r->res_flags, r->res_first_lkid, r->res_recover_locks_count, r->res_name);} void dlm_dump_rsb(struct dlm_rsb *r){ struct dlm_lkb *lkb; dlm_print_rsb(r); printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n", list_empty(&r->res_root_list), list_empty(&r->res_recover_list)); printk(KERN_ERR "rsb lookup list\n"); list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup) dlm_print_lkb(lkb); printk(KERN_ERR "rsb grant queue:\n"); list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) dlm_print_lkb(lkb); printk(KERN_ERR "rsb convert queue:\n"); list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) dlm_print_lkb(lkb); printk(KERN_ERR "rsb wait queue:\n"); list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue) dlm_print_lkb(lkb);} /* Threads cannot use the lockspace while it's being recovered */ void dlm_lock_recovery(struct dlm_ls *ls){ down_read(&ls->ls_in_recovery);} void dlm_unlock_recovery(struct dlm_ls *ls){ up_read(&ls->ls_in_recovery);} int dlm_lock_recovery_try(struct dlm_ls *ls){ return down_read_trylock(&ls->ls_in_recovery);} static inline int can_be_queued(struct dlm_lkb *lkb){ return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);} static inline int force_blocking_asts(struct dlm_lkb *lkb){ return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);} static inline int is_demoted(struct dlm_lkb *lkb){ return test_bit(DLM_SBF_DEMOTED_BIT, &lkb->lkb_sbflags);} static inline int is_altmode(struct dlm_lkb *lkb){ return test_bit(DLM_SBF_ALTMODE_BIT, &lkb->lkb_sbflags);} static inline int is_granted(struct dlm_lkb *lkb){ return (lkb->lkb_status == DLM_LKSTS_GRANTED);} static inline int is_remote(struct dlm_rsb *r){ DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r);); return !!r->res_nodeid;} static inline int is_process_copy(struct dlm_lkb *lkb){ return lkb->lkb_nodeid && !test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);} static inline int is_master_copy(struct dlm_lkb *lkb){ return test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);} static inline int middle_conversion(struct dlm_lkb *lkb){ if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) || (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW)) return 1; return 0;} static inline int down_conversion(struct dlm_lkb *lkb){ return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);} static inline int is_overlap_unlock(struct dlm_lkb *lkb){ return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);} static inline int is_overlap_cancel(struct dlm_lkb *lkb){ return test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);} static inline int is_overlap(struct dlm_lkb *lkb){ return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags) || test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);} static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv){ if (is_master_copy(lkb)) return; DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb);); if (rv == -DLM_ECANCEL && test_and_clear_bit(DLM_IFL_DEADLOCK_CANCEL_BIT, &lkb->lkb_iflags)) rv = -EDEADLK; dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, dlm_sbflags_val(lkb));} static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb){ queue_cast(r, lkb, is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);} static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode){ if (is_master_copy(lkb)) { send_bast(r, lkb, rqmode); } else { dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0); }} /* * Basic operations on rsb's and lkb's */ static inline unsigned long rsb_toss_jiffies(void){ return jiffies + (READ_ONCE(dlm_config.ci_toss_secs) * HZ);} /* This is only called to add a reference when the code already holds a valid reference to the rsb, so there's no need for locking. */ static inline void hold_rsb(struct dlm_rsb *r){ /* inactive rsbs are not ref counted */ WARN_ON(rsb_flag(r, RSB_INACTIVE)); kref_get(&r->res_ref);} void dlm_hold_rsb(struct dlm_rsb *r){ hold_rsb(r);} /* TODO move this to lib/refcount.c */static __must_check booldlm_refcount_dec_and_write_lock_bh(refcount_t *r, rwlock_t *lock) __cond_acquires(true, lock){ if (refcount_dec_not_one(r)) return false; write_lock_bh(lock); if (!refcount_dec_and_test(r)) { write_unlock_bh(lock); return false; } return true;} /* TODO move this to include/linux/kref.h */static inline int dlm_kref_put_write_lock_bh(struct kref *kref, void (*release)(struct kref *kref), rwlock_t *lock){ if (dlm_refcount_dec_and_write_lock_bh(&kref->refcount, lock)) { release(kref); return 1; } return 0;} static void put_rsb(struct dlm_rsb *r){ struct dlm_ls *ls = r->res_ls; int rv; rv = dlm_kref_put_write_lock_bh(&r->res_ref, deactivate_rsb, &ls->ls_rsbtbl_lock); if (rv) write_unlock_bh(&ls->ls_rsbtbl_lock);} void dlm_put_rsb(struct dlm_rsb *r){ put_rsb(r);} /* connected with timer_delete_sync() in dlm_ls_stop() to stop * new timers when recovery is triggered and don't run them * again until a resume_scan_timer() tries it again. */static void enable_scan_timer(struct dlm_ls *ls, unsigned long jiffies){ if (!dlm_locking_stopped(ls)) mod_timer(&ls->ls_scan_timer, jiffies);} /* This function tries to resume the timer callback if a rsb * is on the scan list and no timer is pending. It might that * the first entry is on currently executed as timer callback * but we don't care if a timer queued up again and does * nothing. Should be a rare case. */void resume_scan_timer(struct dlm_ls *ls){ struct dlm_rsb *r; spin_lock_bh(&ls->ls_scan_lock); r = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); if (r && !timer_pending(&ls->ls_scan_timer)) enable_scan_timer(ls, r->res_toss_time); spin_unlock_bh(&ls->ls_scan_lock);} /* ls_rsbtbl_lock must be held */ static void del_scan(struct dlm_ls *ls, struct dlm_rsb *r){ struct dlm_rsb *first; /* active rsbs should never be on the scan list */ WARN_ON(!rsb_flag(r, RSB_INACTIVE)); spin_lock_bh(&ls->ls_scan_lock); r->res_toss_time = 0; /* if the rsb is not queued do nothing */ if (list_empty(&r->res_scan_list)) goto out; /* get the first element before delete */ first = list_first_entry(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); list_del_init(&r->res_scan_list); /* check if the first element was the rsb we deleted */ if (first == r) { /* try to get the new first element, if the list * is empty now try to delete the timer, if we are * too late we don't care. * * if the list isn't empty and a new first element got * in place, set the new timer expire time. */ first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); if (!first) timer_delete(&ls->ls_scan_timer); else enable_scan_timer(ls, first->res_toss_time); } out: spin_unlock_bh(&ls->ls_scan_lock);} static void add_scan(struct dlm_ls *ls, struct dlm_rsb *r){ int our_nodeid = dlm_our_nodeid(); struct dlm_rsb *first; /* A dir record for a remote master rsb should never be on the scan list. */ WARN_ON(!dlm_no_directory(ls) && (r->res_master_nodeid != our_nodeid) && (dlm_dir_nodeid(r) == our_nodeid)); /* An active rsb should never be on the scan list. */ WARN_ON(!rsb_flag(r, RSB_INACTIVE)); /* An rsb should not already be on the scan list. */ WARN_ON(!list_empty(&r->res_scan_list)); spin_lock_bh(&ls->ls_scan_lock); /* set the new rsb absolute expire time in the rsb */ r->res_toss_time = rsb_toss_jiffies(); if (list_empty(&ls->ls_scan_list)) { /* if the queue is empty add the element and it's * our new expire time */ list_add_tail(&r->res_scan_list, &ls->ls_scan_list); enable_scan_timer(ls, r->res_toss_time); } else { /* try to get the maybe new first element and then add * to this rsb with the oldest expire time to the end * of the queue. If the list was empty before this * rsb expire time is our next expiration if it wasn't * the now new first elemet is our new expiration time */ first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); list_add_tail(&r->res_scan_list, &ls->ls_scan_list); if (!first) enable_scan_timer(ls, r->res_toss_time); else enable_scan_timer(ls, first->res_toss_time); } spin_unlock_bh(&ls->ls_scan_lock);} Lines 1–500 of 6338. Load more
// SPDX-License-Identifier: GPL-2.0-only /****************************************************************************** ******************************************************************************* ** ** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved. ** ** ******************************************************************************* ******************************************************************************/ /* Central locking logic has four stages: dlm_lock() dlm_unlock() request_lock(ls, lkb) convert_lock(ls, lkb) unlock_lock(ls, lkb) cancel_lock(ls, lkb) _request_lock(r, lkb) _convert_lock(r, lkb) _unlock_lock(r, lkb) _cancel_lock(r, lkb) do_request(r, lkb) do_convert(r, lkb) do_unlock(r, lkb) do_cancel(r, lkb) Stage 1 (lock, unlock) is mainly about checking input args and splitting into one of the four main operations: dlm_lock = request_lock dlm_lock+CONVERT = convert_lock dlm_unlock = unlock_lock dlm_unlock+CANCEL = cancel_lock Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is provided to the next stage. Stage 3, _xxxx_lock(), determines if the operation is local or remote. When remote, it calls send_xxxx(), when local it calls do_xxxx(). Stage 4, do_xxxx(), is the guts of the operation. It manipulates the given rsb and lkb and queues callbacks. For remote operations, send_xxxx() results in the corresponding do_xxxx() function being executed on the remote node. The connecting send/receive calls on local (L) and remote (R) nodes: L: send_xxxx() -> R: receive_xxxx() R: do_xxxx() L: receive_xxxx_reply() <- R: send_xxxx_reply() */ #include <trace/events/dlm.h> #include <linux/types.h> #include <linux/rbtree.h> #include <linux/slab.h> #include "dlm_internal.h" #include <linux/dlm_device.h> #include "memory.h" #include "midcomms.h" #include "requestqueue.h" #include "util.h" #include "dir.h" #include "member.h" #include "lockspace.h" #include "ast.h" #include "lock.h" #include "rcom.h" #include "recover.h" #include "lvb_table.h" #include "user.h" #include "config.h" static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode); static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb); static int send_remove(struct dlm_rsb *r); static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb); static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb); static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, const struct dlm_message *ms, bool local); static int receive_extralen(const struct dlm_message *ms); static void do_purge(struct dlm_ls *ls, int nodeid, int pid); static void deactivate_rsb(struct kref *kref); /* * Lock compatibilty matrix - thanks Steve * UN = Unlocked state. Not really a state, used as a flag * PD = Padding. Used to make the matrix a nice power of two in size * Other states are the same as the VMS DLM. * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same) */ static const int __dlm_compat_matrix[8][8] = { /* UN NL CR CW PR PW EX PD */ {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */ {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */ {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */ {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */ {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */ {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */ {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */ {0, 0, 0, 0, 0, 0, 0, 0} /* PD */ }; /* * This defines the direction of transfer of LVB data. * Granted mode is the row; requested mode is the column. * Usage: matrix[grmode+1][rqmode+1] * 1 = LVB is returned to the caller * 0 = LVB is written to the resource * -1 = nothing happens to the LVB */ const int dlm_lvb_operations[8][8] = { /* UN NL CR CW PR PW EX PD*/ { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */ { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */ { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */ { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */ { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */ { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */ { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */ { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */ }; #define modes_compat(gr, rq) \ __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1] int dlm_modes_compat(int mode1, int mode2) { return __dlm_compat_matrix[mode1 + 1][mode2 + 1]; } /* * Compatibility matrix for conversions with QUECVT set. * Granted mode is the row; requested mode is the column. * Usage: matrix[grmode+1][rqmode+1] */ static const int __quecvt_compat_matrix[8][8] = { /* UN NL CR CW PR PW EX PD */ {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */ {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */ {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */ {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */ {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */ {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */ {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */ {0, 0, 0, 0, 0, 0, 0, 0} /* PD */ }; void dlm_print_lkb(struct dlm_lkb *lkb) { printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x " "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n", lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags, dlm_iflags_val(lkb), lkb->lkb_status, lkb->lkb_rqmode, lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid, (unsigned long long)lkb->lkb_recover_seq); } static void dlm_print_rsb(struct dlm_rsb *r) { printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x " "rlc %d name %s\n", r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid, r->res_flags, r->res_first_lkid, r->res_recover_locks_count, r->res_name); } void dlm_dump_rsb(struct dlm_rsb *r) { struct dlm_lkb *lkb; dlm_print_rsb(r); printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n", list_empty(&r->res_root_list), list_empty(&r->res_recover_list)); printk(KERN_ERR "rsb lookup list\n"); list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup) dlm_print_lkb(lkb); printk(KERN_ERR "rsb grant queue:\n"); list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) dlm_print_lkb(lkb); printk(KERN_ERR "rsb convert queue:\n"); list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) dlm_print_lkb(lkb); printk(KERN_ERR "rsb wait queue:\n"); list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue) dlm_print_lkb(lkb); } /* Threads cannot use the lockspace while it's being recovered */ void dlm_lock_recovery(struct dlm_ls *ls) { down_read(&ls->ls_in_recovery); } void dlm_unlock_recovery(struct dlm_ls *ls) { up_read(&ls->ls_in_recovery); } int dlm_lock_recovery_try(struct dlm_ls *ls) { return down_read_trylock(&ls->ls_in_recovery); } static inline int can_be_queued(struct dlm_lkb *lkb) { return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE); } static inline int force_blocking_asts(struct dlm_lkb *lkb) { return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST); } static inline int is_demoted(struct dlm_lkb *lkb) { return test_bit(DLM_SBF_DEMOTED_BIT, &lkb->lkb_sbflags); } static inline int is_altmode(struct dlm_lkb *lkb) { return test_bit(DLM_SBF_ALTMODE_BIT, &lkb->lkb_sbflags); } static inline int is_granted(struct dlm_lkb *lkb) { return (lkb->lkb_status == DLM_LKSTS_GRANTED); } static inline int is_remote(struct dlm_rsb *r) { DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r);); return !!r->res_nodeid; } static inline int is_process_copy(struct dlm_lkb *lkb) { return lkb->lkb_nodeid && !test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags); } static inline int is_master_copy(struct dlm_lkb *lkb) { return test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags); } static inline int middle_conversion(struct dlm_lkb *lkb) { if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) || (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW)) return 1; return 0; } static inline int down_conversion(struct dlm_lkb *lkb) { return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode); } static inline int is_overlap_unlock(struct dlm_lkb *lkb) { return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags); } static inline int is_overlap_cancel(struct dlm_lkb *lkb) { return test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags); } static inline int is_overlap(struct dlm_lkb *lkb) { return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags) || test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags); } static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { if (is_master_copy(lkb)) return; DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb);); if (rv == -DLM_ECANCEL && test_and_clear_bit(DLM_IFL_DEADLOCK_CANCEL_BIT, &lkb->lkb_iflags)) rv = -EDEADLK; dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, dlm_sbflags_val(lkb)); } static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb) { queue_cast(r, lkb, is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL); } static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode) { if (is_master_copy(lkb)) { send_bast(r, lkb, rqmode); } else { dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0); } } /* * Basic operations on rsb's and lkb's */ static inline unsigned long rsb_toss_jiffies(void) { return jiffies + (READ_ONCE(dlm_config.ci_toss_secs) * HZ); } /* This is only called to add a reference when the code already holds a valid reference to the rsb, so there's no need for locking. */ static inline void hold_rsb(struct dlm_rsb *r) { /* inactive rsbs are not ref counted */ WARN_ON(rsb_flag(r, RSB_INACTIVE)); kref_get(&r->res_ref); } void dlm_hold_rsb(struct dlm_rsb *r) { hold_rsb(r); } /* TODO move this to lib/refcount.c */ static __must_check bool dlm_refcount_dec_and_write_lock_bh(refcount_t *r, rwlock_t *lock) __cond_acquires(true, lock) { if (refcount_dec_not_one(r)) return false; write_lock_bh(lock); if (!refcount_dec_and_test(r)) { write_unlock_bh(lock); return false; } return true; } /* TODO move this to include/linux/kref.h */ static inline int dlm_kref_put_write_lock_bh(struct kref *kref, void (*release)(struct kref *kref), rwlock_t *lock) { if (dlm_refcount_dec_and_write_lock_bh(&kref->refcount, lock)) { release(kref); return 1; } return 0; } static void put_rsb(struct dlm_rsb *r) { struct dlm_ls *ls = r->res_ls; int rv; rv = dlm_kref_put_write_lock_bh(&r->res_ref, deactivate_rsb, &ls->ls_rsbtbl_lock); if (rv) write_unlock_bh(&ls->ls_rsbtbl_lock); } void dlm_put_rsb(struct dlm_rsb *r) { put_rsb(r); } /* connected with timer_delete_sync() in dlm_ls_stop() to stop * new timers when recovery is triggered and don't run them * again until a resume_scan_timer() tries it again. */ static void enable_scan_timer(struct dlm_ls *ls, unsigned long jiffies) { if (!dlm_locking_stopped(ls)) mod_timer(&ls->ls_scan_timer, jiffies); } /* This function tries to resume the timer callback if a rsb * is on the scan list and no timer is pending. It might that * the first entry is on currently executed as timer callback * but we don't care if a timer queued up again and does * nothing. Should be a rare case. */ void resume_scan_timer(struct dlm_ls *ls) { struct dlm_rsb *r; spin_lock_bh(&ls->ls_scan_lock); r = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); if (r && !timer_pending(&ls->ls_scan_timer)) enable_scan_timer(ls, r->res_toss_time); spin_unlock_bh(&ls->ls_scan_lock); } /* ls_rsbtbl_lock must be held */ static void del_scan(struct dlm_ls *ls, struct dlm_rsb *r) { struct dlm_rsb *first; /* active rsbs should never be on the scan list */ WARN_ON(!rsb_flag(r, RSB_INACTIVE)); spin_lock_bh(&ls->ls_scan_lock); r->res_toss_time = 0; /* if the rsb is not queued do nothing */ if (list_empty(&r->res_scan_list)) goto out; /* get the first element before delete */ first = list_first_entry(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); list_del_init(&r->res_scan_list); /* check if the first element was the rsb we deleted */ if (first == r) { /* try to get the new first element, if the list * is empty now try to delete the timer, if we are * too late we don't care. * * if the list isn't empty and a new first element got * in place, set the new timer expire time. */ first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); if (!first) timer_delete(&ls->ls_scan_timer); else enable_scan_timer(ls, first->res_toss_time); } out: spin_unlock_bh(&ls->ls_scan_lock); } static void add_scan(struct dlm_ls *ls, struct dlm_rsb *r) { int our_nodeid = dlm_our_nodeid(); struct dlm_rsb *first; /* A dir record for a remote master rsb should never be on the scan list. */ WARN_ON(!dlm_no_directory(ls) && (r->res_master_nodeid != our_nodeid) && (dlm_dir_nodeid(r) == our_nodeid)); /* An active rsb should never be on the scan list. */ WARN_ON(!rsb_flag(r, RSB_INACTIVE)); /* An rsb should not already be on the scan list. */ WARN_ON(!list_empty(&r->res_scan_list)); spin_lock_bh(&ls->ls_scan_lock); /* set the new rsb absolute expire time in the rsb */ r->res_toss_time = rsb_toss_jiffies(); if (list_empty(&ls->ls_scan_list)) { /* if the queue is empty add the element and it's * our new expire time */ list_add_tail(&r->res_scan_list, &ls->ls_scan_list); enable_scan_timer(ls, r->res_toss_time); } else { /* try to get the maybe new first element and then add * to this rsb with the oldest expire time to the end * of the queue. If the list was empty before this * rsb expire time is our next expiration if it wasn't * the now new first elemet is our new expiration time */ first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb, res_scan_list); list_add_tail(&r->res_scan_list, &ls->ls_scan_list); if (!first) enable_scan_timer(ls, r->res_toss_time); else enable_scan_timer(ls, first->res_toss_time); } spin_unlock_bh(&ls->ls_scan_lock); }