blob: 8bee4f444afdc02485f9680585c0ebe6f0ec0cc1 [file] [log] [blame]
// 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(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);
}
/* if we hit contention we do in 250 ms a retry to trylock.
* if there is any other mod_timer in between we don't care
* about that it expires earlier again this is only for the
* unlikely case nothing happened in this time.
*/
#define DLM_TOSS_TIMER_RETRY (jiffies + msecs_to_jiffies(250))
/* Called by lockspace scan_timer to free unused rsb's. */
void dlm_rsb_scan(struct timer_list *timer)
{
struct dlm_ls *ls = from_timer(ls, timer, ls_scan_timer);
int our_nodeid = dlm_our_nodeid();
struct dlm_rsb *r;
int rv;
while (1) {
/* interrupting point to leave iteration when
* recovery waits for timer_delete_sync(), recovery
* will take care to delete everything in scan list.
*/
if (dlm_locking_stopped(ls))
break;
rv = spin_trylock(&ls->ls_scan_lock);
if (!rv) {
/* rearm again try timer */
enable_scan_timer(ls, DLM_TOSS_TIMER_RETRY);
break;
}
r = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb,
res_scan_list);
if (!r) {
/* the next add_scan will enable the timer again */
spin_unlock(&ls->ls_scan_lock);
break;
}
/*
* If the first rsb is not yet expired, then stop because the
* list is sorted with nearest expiration first.
*/
if (time_before(jiffies, r->res_toss_time)) {
/* rearm with the next rsb to expire in the future */
enable_scan_timer(ls, r->res_toss_time);
spin_unlock(&ls->ls_scan_lock);
break;
}
/* in find_rsb_dir/nodir there is a reverse order of this
* lock, however this is only a trylock if we hit some
* possible contention we try it again.
*/
rv = write_trylock(&ls->ls_rsbtbl_lock);
if (!rv) {
spin_unlock(&ls->ls_scan_lock);
/* rearm again try timer */
enable_scan_timer(ls, DLM_TOSS_TIMER_RETRY);
break;
}
list_del(&r->res_slow_list);
rhashtable_remove_fast(&ls->ls_rsbtbl, &r->res_node,
dlm_rhash_rsb_params);
rsb_clear_flag(r, RSB_HASHED);
/* ls_rsbtbl_lock is not needed when calling send_remove() */
write_unlock(&ls->ls_rsbtbl_lock);
list_del_init(&r->res_scan_list);
spin_unlock(&ls->ls_scan_lock);
/* An rsb that is a dir record for a remote master rsb
* cannot be removed, and should not have a timer enabled.
*/
WARN_ON(!dlm_no_directory(ls) &&
(r->res_master_nodeid != our_nodeid) &&
(dlm_dir_nodeid(r) == our_nodeid));
/* We're the master of this rsb but we're not
* the directory record, so we need to tell the
* dir node to remove the dir record
*/
if (!dlm_no_directory(ls) &&
(r->res_master_nodeid == our_nodeid) &&
(dlm_dir_nodeid(r) != our_nodeid))
send_remove(r);
free_inactive_rsb(r);
}
}
/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
unlock any spinlocks, go back and call pre_rsb_struct again.
Otherwise, take an rsb off the list and return it. */
static int get_rsb_struct(struct dlm_ls *ls, const void *name, int len,
struct dlm_rsb **r_ret)
{
struct dlm_rsb *r;
r = dlm_allocate_rsb(ls);
if (!r)
return -ENOMEM;
r->res_ls = ls;
r->res_length = len;
memcpy(r->res_name, name, len);
spin_lock_init(&r->res_lock);
INIT_LIST_HEAD(&r->res_lookup);
INIT_LIST_HEAD(&r->res_grantqueue);
INIT_LIST_HEAD(&r->res_convertqueue);
INIT_LIST_HEAD(&r->res_waitqueue);
INIT_LIST_HEAD(&r->res_root_list);
INIT_LIST_HEAD(&r->res_scan_list);
INIT_LIST_HEAD(&r->res_recover_list);
INIT_LIST_HEAD(&r->res_masters_list);
*r_ret = r;
return 0;
}
int dlm_search_rsb_tree(struct rhashtable *rhash, const void *name, int len,
struct dlm_rsb **r_ret)
{
char key[DLM_RESNAME_MAXLEN] = {};
memcpy(key, name, len);
*r_ret = rhashtable_lookup_fast(rhash, &key, dlm_rhash_rsb_params);
if (*r_ret)
return 0;
return -EBADR;
}
static int rsb_insert(struct dlm_rsb *rsb, struct rhashtable *rhash)
{
int rv;
rv = rhashtable_insert_fast(rhash, &rsb->res_node,
dlm_rhash_rsb_params);
if (!rv)
rsb_set_flag(rsb, RSB_HASHED);
return rv;
}
/*
* Find rsb in rsbtbl and potentially create/add one
*
* Delaying the release of rsb's has a similar benefit to applications keeping
* NL locks on an rsb, but without the guarantee that the cached master value
* will still be valid when the rsb is reused. Apps aren't always smart enough
* to keep NL locks on an rsb that they may lock again shortly; this can lead
* to excessive master lookups and removals if we don't delay the release.
*
* Searching for an rsb means looking through both the normal list and toss
* list. When found on the toss list the rsb is moved to the normal list with
* ref count of 1; when found on normal list the ref count is incremented.
*
* rsb's on the keep list are being used locally and refcounted.
* rsb's on the toss list are not being used locally, and are not refcounted.
*
* The toss list rsb's were either
* - previously used locally but not any more (were on keep list, then
* moved to toss list when last refcount dropped)
* - created and put on toss list as a directory record for a lookup
* (we are the dir node for the res, but are not using the res right now,
* but some other node is)
*
* The purpose of find_rsb() is to return a refcounted rsb for local use.
* So, if the given rsb is on the toss list, it is moved to the keep list
* before being returned.
*
* deactivate_rsb() happens when all local usage of the rsb is done, i.e. no
* more refcounts exist, so the rsb is moved from the keep list to the
* toss list.
*
* rsb's on both keep and toss lists are used for doing a name to master
* lookups. rsb's that are in use locally (and being refcounted) are on
* the keep list, rsb's that are not in use locally (not refcounted) and
* only exist for name/master lookups are on the toss list.
*
* rsb's on the toss list who's dir_nodeid is not local can have stale
* name/master mappings. So, remote requests on such rsb's can potentially
* return with an error, which means the mapping is stale and needs to
* be updated with a new lookup. (The idea behind MASTER UNCERTAIN and
* first_lkid is to keep only a single outstanding request on an rsb
* while that rsb has a potentially stale master.)
*/
static int find_rsb_dir(struct dlm_ls *ls, const void *name, int len,
uint32_t hash, int dir_nodeid, int from_nodeid,
unsigned int flags, struct dlm_rsb **r_ret)
{
struct dlm_rsb *r = NULL;
int our_nodeid = dlm_our_nodeid();
int from_local = 0;
int from_other = 0;
int from_dir = 0;
int create = 0;
int error;
if (flags & R_RECEIVE_REQUEST) {
if (from_nodeid == dir_nodeid)
from_dir = 1;
else
from_other = 1;
} else if (flags & R_REQUEST) {
from_local = 1;
}
/*
* flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
* from_nodeid has sent us a lock in dlm_recover_locks, believing
* we're the new master. Our local recovery may not have set
* res_master_nodeid to our_nodeid yet, so allow either. Don't
* create the rsb; dlm_recover_process_copy() will handle EBADR
* by resending.
*
* If someone sends us a request, we are the dir node, and we do
* not find the rsb anywhere, then recreate it. This happens if
* someone sends us a request after we have removed/freed an rsb.
* (They sent a request instead of lookup because they are using
* an rsb taken from their scan list.)
*/
if (from_local || from_dir ||
(from_other && (dir_nodeid == our_nodeid))) {
create = 1;
}
retry:
/* check if the rsb is active under read lock - likely path */
read_lock_bh(&ls->ls_rsbtbl_lock);
error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
if (error) {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_new;
}
/*
* rsb is active, so we can't check master_nodeid without lock_rsb.
*/
if (rsb_flag(r, RSB_INACTIVE)) {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_inactive;
}
kref_get(&r->res_ref);
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto out;
do_inactive:
write_lock_bh(&ls->ls_rsbtbl_lock);
/*
* The expectation here is that the rsb will have HASHED and
* INACTIVE flags set, and that the rsb can be moved from
* inactive back to active again. However, between releasing
* the read lock and acquiring the write lock, this rsb could
* have been removed from rsbtbl, and had HASHED cleared, to
* be freed. To deal with this case, we would normally need
* to repeat dlm_search_rsb_tree while holding the write lock,
* but rcu allows us to simply check the HASHED flag, because
* the rcu read lock means the rsb will not be freed yet.
* If the HASHED flag is not set, then the rsb is being freed,
* so we add a new rsb struct. If the HASHED flag is set,
* and INACTIVE is not set, it means another thread has
* made the rsb active, as we're expecting to do here, and
* we just repeat the lookup (this will be very unlikely.)
*/
if (rsb_flag(r, RSB_HASHED)) {
if (!rsb_flag(r, RSB_INACTIVE)) {
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto retry;
}
} else {
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_new;
}
/*
* rsb found inactive (master_nodeid may be out of date unless
* we are the dir_nodeid or were the master) No other thread
* is using this rsb because it's inactive, so we can
* look at or update res_master_nodeid without lock_rsb.
*/
if ((r->res_master_nodeid != our_nodeid) && from_other) {
/* our rsb was not master, and another node (not the dir node)
has sent us a request */
log_debug(ls, "find_rsb inactive from_other %d master %d dir %d %s",
from_nodeid, r->res_master_nodeid, dir_nodeid,
r->res_name);
write_unlock_bh(&ls->ls_rsbtbl_lock);
error = -ENOTBLK;
goto out;
}
if ((r->res_master_nodeid != our_nodeid) && from_dir) {
/* don't think this should ever happen */
log_error(ls, "find_rsb inactive from_dir %d master %d",
from_nodeid, r->res_master_nodeid);
dlm_print_rsb(r);
/* fix it and go on */
r->res_master_nodeid = our_nodeid;
r->res_nodeid = 0;
rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
r->res_first_lkid = 0;
}
if (from_local && (r->res_master_nodeid != our_nodeid)) {
/* Because we have held no locks on this rsb,
res_master_nodeid could have become stale. */
rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
r->res_first_lkid = 0;
}
/* A dir record will not be on the scan list. */
if (r->res_dir_nodeid != our_nodeid)
del_scan(ls, r);
list_move(&r->res_slow_list, &ls->ls_slow_active);
rsb_clear_flag(r, RSB_INACTIVE);
kref_init(&r->res_ref); /* ref is now used in active state */
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto out;
do_new:
/*
* rsb not found
*/
if (error == -EBADR && !create)
goto out;
error = get_rsb_struct(ls, name, len, &r);
if (WARN_ON_ONCE(error))
goto out;
r->res_hash = hash;
r->res_dir_nodeid = dir_nodeid;
kref_init(&r->res_ref);
if (from_dir) {
/* want to see how often this happens */
log_debug(ls, "find_rsb new from_dir %d recreate %s",
from_nodeid, r->res_name);
r->res_master_nodeid = our_nodeid;
r->res_nodeid = 0;
goto out_add;
}
if (from_other && (dir_nodeid != our_nodeid)) {
/* should never happen */
log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
from_nodeid, dir_nodeid, our_nodeid, r->res_name);
dlm_free_rsb(r);
r = NULL;
error = -ENOTBLK;
goto out;
}
if (from_other) {
log_debug(ls, "find_rsb new from_other %d dir %d %s",
from_nodeid, dir_nodeid, r->res_name);
}
if (dir_nodeid == our_nodeid) {
/* When we are the dir nodeid, we can set the master
node immediately */
r->res_master_nodeid = our_nodeid;
r->res_nodeid = 0;
} else {
/* set_master will send_lookup to dir_nodeid */
r->res_master_nodeid = 0;
r->res_nodeid = -1;
}
out_add:
write_lock_bh(&ls->ls_rsbtbl_lock);
error = rsb_insert(r, &ls->ls_rsbtbl);
if (error == -EEXIST) {
/* somebody else was faster and it seems the
* rsb exists now, we do a whole relookup
*/
write_unlock_bh(&ls->ls_rsbtbl_lock);
dlm_free_rsb(r);
goto retry;
} else if (!error) {
list_add(&r->res_slow_list, &ls->ls_slow_active);
}
write_unlock_bh(&ls->ls_rsbtbl_lock);
out:
*r_ret = r;
return error;
}
/* During recovery, other nodes can send us new MSTCPY locks (from
dlm_recover_locks) before we've made ourself master (in
dlm_recover_masters). */
static int find_rsb_nodir(struct dlm_ls *ls, const void *name, int len,
uint32_t hash, int dir_nodeid, int from_nodeid,
unsigned int flags, struct dlm_rsb **r_ret)
{
struct dlm_rsb *r = NULL;
int our_nodeid = dlm_our_nodeid();
int recover = (flags & R_RECEIVE_RECOVER);
int error;
retry:
/* check if the rsb is in active state under read lock - likely path */
read_lock_bh(&ls->ls_rsbtbl_lock);
error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
if (error) {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_new;
}
if (rsb_flag(r, RSB_INACTIVE)) {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_inactive;
}
/*
* rsb is active, so we can't check master_nodeid without lock_rsb.
*/
kref_get(&r->res_ref);
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto out;
do_inactive:
write_lock_bh(&ls->ls_rsbtbl_lock);
/* See comment in find_rsb_dir. */
if (rsb_flag(r, RSB_HASHED)) {
if (!rsb_flag(r, RSB_INACTIVE)) {
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto retry;
}
} else {
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_new;
}
/*
* rsb found inactive. No other thread is using this rsb because
* it's inactive, so we can look at or update res_master_nodeid
* without lock_rsb.
*/
if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
/* our rsb is not master, and another node has sent us a
request; this should never happen */
log_error(ls, "find_rsb inactive from_nodeid %d master %d dir %d",
from_nodeid, r->res_master_nodeid, dir_nodeid);
dlm_print_rsb(r);
write_unlock_bh(&ls->ls_rsbtbl_lock);
error = -ENOTBLK;
goto out;
}
if (!recover && (r->res_master_nodeid != our_nodeid) &&
(dir_nodeid == our_nodeid)) {
/* our rsb is not master, and we are dir; may as well fix it;
this should never happen */
log_error(ls, "find_rsb inactive our %d master %d dir %d",
our_nodeid, r->res_master_nodeid, dir_nodeid);
dlm_print_rsb(r);
r->res_master_nodeid = our_nodeid;
r->res_nodeid = 0;
}
list_move(&r->res_slow_list, &ls->ls_slow_active);
rsb_clear_flag(r, RSB_INACTIVE);
kref_init(&r->res_ref);
del_scan(ls, r);
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto out;
do_new:
/*
* rsb not found
*/
error = get_rsb_struct(ls, name, len, &r);
if (WARN_ON_ONCE(error))
goto out;
r->res_hash = hash;
r->res_dir_nodeid = dir_nodeid;
r->res_master_nodeid = dir_nodeid;
r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
kref_init(&r->res_ref);
write_lock_bh(&ls->ls_rsbtbl_lock);
error = rsb_insert(r, &ls->ls_rsbtbl);
if (error == -EEXIST) {
/* somebody else was faster and it seems the
* rsb exists now, we do a whole relookup
*/
write_unlock_bh(&ls->ls_rsbtbl_lock);
dlm_free_rsb(r);
goto retry;
} else if (!error) {
list_add(&r->res_slow_list, &ls->ls_slow_active);
}
write_unlock_bh(&ls->ls_rsbtbl_lock);
out:
*r_ret = r;
return error;
}
/*
* rsb rcu usage
*
* While rcu read lock is held, the rsb cannot be freed,
* which allows a lookup optimization.
*
* Two threads are accessing the same rsb concurrently,
* the first (A) is trying to use the rsb, the second (B)
* is trying to free the rsb.
*
* thread A thread B
* (trying to use rsb) (trying to free rsb)
*
* A1. rcu read lock
* A2. rsbtbl read lock
* A3. look up rsb in rsbtbl
* A4. rsbtbl read unlock
* B1. rsbtbl write lock
* B2. look up rsb in rsbtbl
* B3. remove rsb from rsbtbl
* B4. clear rsb HASHED flag
* B5. rsbtbl write unlock
* B6. begin freeing rsb using rcu...
*
* (rsb is inactive, so try to make it active again)
* A5. read rsb HASHED flag (safe because rsb is not freed yet)
* A6. the rsb HASHED flag is not set, which it means the rsb
* is being removed from rsbtbl and freed, so don't use it.
* A7. rcu read unlock
*
* B7. ...finish freeing rsb using rcu
* A8. create a new rsb
*
* Without the rcu optimization, steps A5-8 would need to do
* an extra rsbtbl lookup:
* A5. rsbtbl write lock
* A6. look up rsb in rsbtbl, not found
* A7. rsbtbl write unlock
* A8. create a new rsb
*/
static int find_rsb(struct dlm_ls *ls, const void *name, int len,
int from_nodeid, unsigned int flags,
struct dlm_rsb **r_ret)
{
int dir_nodeid;
uint32_t hash;
int rv;
if (len > DLM_RESNAME_MAXLEN)
return -EINVAL;
hash = jhash(name, len, 0);
dir_nodeid = dlm_hash2nodeid(ls, hash);
rcu_read_lock();
if (dlm_no_directory(ls))
rv = find_rsb_nodir(ls, name, len, hash, dir_nodeid,
from_nodeid, flags, r_ret);
else
rv = find_rsb_dir(ls, name, len, hash, dir_nodeid,
from_nodeid, flags, r_ret);
rcu_read_unlock();
return rv;
}
/* we have received a request and found that res_master_nodeid != our_nodeid,
so we need to return an error or make ourself the master */
static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
int from_nodeid)
{
if (dlm_no_directory(ls)) {
log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
from_nodeid, r->res_master_nodeid,
r->res_dir_nodeid);
dlm_print_rsb(r);
return -ENOTBLK;
}
if (from_nodeid != r->res_dir_nodeid) {
/* our rsb is not master, and another node (not the dir node)
has sent us a request. this is much more common when our
master_nodeid is zero, so limit debug to non-zero. */
if (r->res_master_nodeid) {
log_debug(ls, "validate master from_other %d master %d "
"dir %d first %x %s", from_nodeid,
r->res_master_nodeid, r->res_dir_nodeid,
r->res_first_lkid, r->res_name);
}
return -ENOTBLK;
} else {
/* our rsb is not master, but the dir nodeid has sent us a
request; this could happen with master 0 / res_nodeid -1 */
if (r->res_master_nodeid) {
log_error(ls, "validate master from_dir %d master %d "
"first %x %s",
from_nodeid, r->res_master_nodeid,
r->res_first_lkid, r->res_name);
}
r->res_master_nodeid = dlm_our_nodeid();
r->res_nodeid = 0;
return 0;
}
}
static void __dlm_master_lookup(struct dlm_ls *ls, struct dlm_rsb *r, int our_nodeid,
int from_nodeid, bool is_inactive, unsigned int flags,
int *r_nodeid, int *result)
{
int fix_master = (flags & DLM_LU_RECOVER_MASTER);
int from_master = (flags & DLM_LU_RECOVER_DIR);
if (r->res_dir_nodeid != our_nodeid) {
/* should not happen, but may as well fix it and carry on */
log_error(ls, "%s res_dir %d our %d %s", __func__,
r->res_dir_nodeid, our_nodeid, r->res_name);
r->res_dir_nodeid = our_nodeid;
}
if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
/* Recovery uses this function to set a new master when
* the previous master failed. Setting NEW_MASTER will
* force dlm_recover_masters to call recover_master on this
* rsb even though the res_nodeid is no longer removed.
*/
r->res_master_nodeid = from_nodeid;
r->res_nodeid = from_nodeid;
rsb_set_flag(r, RSB_NEW_MASTER);
if (is_inactive) {
/* I don't think we should ever find it inactive. */
log_error(ls, "%s fix_master inactive", __func__);
dlm_dump_rsb(r);
}
}
if (from_master && (r->res_master_nodeid != from_nodeid)) {
/* this will happen if from_nodeid became master during
* a previous recovery cycle, and we aborted the previous
* cycle before recovering this master value
*/
log_limit(ls, "%s from_master %d master_nodeid %d res_nodeid %d first %x %s",
__func__, from_nodeid, r->res_master_nodeid,
r->res_nodeid, r->res_first_lkid, r->res_name);
if (r->res_master_nodeid == our_nodeid) {
log_error(ls, "from_master %d our_master", from_nodeid);
dlm_dump_rsb(r);
goto ret_assign;
}
r->res_master_nodeid = from_nodeid;
r->res_nodeid = from_nodeid;
rsb_set_flag(r, RSB_NEW_MASTER);
}
if (!r->res_master_nodeid) {
/* this will happen if recovery happens while we're looking
* up the master for this rsb
*/
log_debug(ls, "%s master 0 to %d first %x %s", __func__,
from_nodeid, r->res_first_lkid, r->res_name);
r->res_master_nodeid = from_nodeid;
r->res_nodeid = from_nodeid;
}
if (!from_master && !fix_master &&
(r->res_master_nodeid == from_nodeid)) {
/* this can happen when the master sends remove, the dir node
* finds the rsb on the active list and ignores the remove,
* and the former master sends a lookup
*/
log_limit(ls, "%s from master %d flags %x first %x %s",
__func__, from_nodeid, flags, r->res_first_lkid,
r->res_name);
}
ret_assign:
*r_nodeid = r->res_master_nodeid;
if (result)
*result = DLM_LU_MATCH;
}
/*
* We're the dir node for this res and another node wants to know the
* master nodeid. During normal operation (non recovery) this is only
* called from receive_lookup(); master lookups when the local node is
* the dir node are done by find_rsb().
*
* normal operation, we are the dir node for a resource
* . _request_lock
* . set_master
* . send_lookup
* . receive_lookup
* . dlm_master_lookup flags 0
*
* recover directory, we are rebuilding dir for all resources
* . dlm_recover_directory
* . dlm_rcom_names
* remote node sends back the rsb names it is master of and we are dir of
* . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
* we either create new rsb setting remote node as master, or find existing
* rsb and set master to be the remote node.
*
* recover masters, we are finding the new master for resources
* . dlm_recover_masters
* . recover_master
* . dlm_send_rcom_lookup
* . receive_rcom_lookup
* . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
*/
static int _dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, const char *name,
int len, unsigned int flags, int *r_nodeid, int *result)
{
struct dlm_rsb *r = NULL;
uint32_t hash;
int our_nodeid = dlm_our_nodeid();
int dir_nodeid, error;
if (len > DLM_RESNAME_MAXLEN)
return -EINVAL;
if (from_nodeid == our_nodeid) {
log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
our_nodeid, flags);
return -EINVAL;
}
hash = jhash(name, len, 0);
dir_nodeid = dlm_hash2nodeid(ls, hash);
if (dir_nodeid != our_nodeid) {
log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
from_nodeid, dir_nodeid, our_nodeid, hash,
ls->ls_num_nodes);
*r_nodeid = -1;
return -EINVAL;
}
retry:
/* check if the rsb is active under read lock - likely path */
read_lock_bh(&ls->ls_rsbtbl_lock);
error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
if (!error) {
if (rsb_flag(r, RSB_INACTIVE)) {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto do_inactive;
}
/* because the rsb is active, we need to lock_rsb before
* checking/changing re_master_nodeid
*/
hold_rsb(r);
read_unlock_bh(&ls->ls_rsbtbl_lock);
lock_rsb(r);
__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, false,
flags, r_nodeid, result);
/* the rsb was active */
unlock_rsb(r);
put_rsb(r);
return 0;
} else {
read_unlock_bh(&ls->ls_rsbtbl_lock);
goto not_found;
}
do_inactive:
/* unlikely path - relookup under write */
write_lock_bh(&ls->ls_rsbtbl_lock);
error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
if (!error) {
if (!rsb_flag(r, RSB_INACTIVE)) {
write_unlock_bh(&ls->ls_rsbtbl_lock);
/* something as changed, very unlikely but
* try again
*/
goto retry;
}
} else {
write_unlock_bh(&ls->ls_rsbtbl_lock);
goto not_found;
}
/* because the rsb is inactive, it's not refcounted and lock_rsb
is not used, but is protected by the rsbtbl lock */
__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, true, flags,
r_nodeid, result);
/* A dir record rsb should never be on scan list. */
/* Try to fix this with del_scan? */
WARN_ON(!list_empty(&r->res_scan_list));
write_unlock_bh(&ls->ls_rsbtbl_lock);
return 0;
not_found:
error = get_rsb_struct(ls, name, len, &r);
if (WARN_ON_ONCE(error))
goto out;
r->res_hash = hash;
r->res_dir_nodeid = our_nodeid;
r->res_master_nodeid = from_nodeid;
r->res_nodeid = from_nodeid;
rsb_set_flag(r, RSB_INACTIVE);
write_lock_bh(&ls->ls_rsbtbl_lock);
error = rsb_insert(r, &ls->ls_rsbtbl);
if (error == -EEXIST) {
/* somebody else was faster and it seems the
* rsb exists now, we do a whole relookup
*/
write_unlock_bh(&ls->ls_rsbtbl_lock);
dlm_free_rsb(r);
goto retry;
} else if (error) {
write_unlock_bh(&ls->ls_rsbtbl_lock);
/* should never happen */
dlm_free_rsb(r);
goto retry;
}
list_add(&r->res_slow_list, &ls->ls_slow_inactive);
write_unlock_bh(&ls->ls_rsbtbl_lock);
if (result)
*result = DLM_LU_ADD;
*r_nodeid = from_nodeid;
out:
return error;
}
int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, const char *name,
int len, unsigned int flags, int *r_nodeid, int *result)
{
int rv;
rcu_read_lock();
rv = _dlm_master_lookup(ls, from_nodeid, name, len, flags, r_nodeid, result);
rcu_read_unlock();
return rv;
}
static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
{
struct dlm_rsb *r;
read_lock_bh(&ls->ls_rsbtbl_lock);
list_for_each_entry(r, &ls->ls_slow_active, res_slow_list) {
if (r->res_hash == hash)
dlm_dump_rsb(r);
}
read_unlock_bh(&ls->ls_rsbtbl_lock);
}
void dlm_dump_rsb_name(struct dlm_ls *ls, const char *name, int len)
{
struct dlm_rsb *r = NULL;
int error;
read_lock_bh(&ls->ls_rsbtbl_lock);
error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
if (!error)
goto out;
dlm_dump_rsb(r);
out:
read_unlock_bh(&ls->ls_rsbtbl_lock);
}
static void deactivate_rsb(struct kref *kref)
{
struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
struct dlm_ls *ls = r->res_ls;
int our_nodeid = dlm_our_nodeid();
DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
rsb_set_flag(r, RSB_INACTIVE);
list_move(&r->res_slow_list, &ls->ls_slow_inactive);
/*
* When the rsb becomes unused:
* - If it's not a dir record for a remote master rsb,
* then it is put on the scan list to be freed.
* - If it's a dir record for a remote master rsb,
* then it is kept in the inactive state until
* receive_remove() from the master node.
*/
if (!dlm_no_directory(ls) &&
(r->res_master_nodeid != our_nodeid) &&
(dlm_dir_nodeid(r) != our_nodeid))
add_scan(ls, r);
if (r->res_lvbptr) {
dlm_free_lvb(r->res_lvbptr);
r->res_lvbptr = NULL;
}
}
/* See comment for unhold_lkb */
static void unhold_rsb(struct dlm_rsb *r)
{
int rv;
/* inactive rsbs are not ref counted */
WARN_ON(rsb_flag(r, RSB_INACTIVE));
rv = kref_put(&r->res_ref, deactivate_rsb);
DLM_ASSERT(!rv, dlm_dump_rsb(r););
}
void free_inactive_rsb(struct dlm_rsb *r)
{
WARN_ON_ONCE(!rsb_flag(r, RSB_INACTIVE));
DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_scan_list), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
DLM_ASSERT(list_empty(&r->res_masters_list), dlm_dump_rsb(r););
dlm_free_rsb(r);
}
/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
The rsb must exist as long as any lkb's for it do. */
static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
hold_rsb(r);
lkb->lkb_resource = r;
}
static void detach_lkb(struct dlm_lkb *lkb)
{
if (lkb->lkb_resource) {
put_rsb(lkb->lkb_resource);
lkb->lkb_resource = NULL;
}
}
static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
unsigned long start, unsigned long end)
{
struct xa_limit limit;
struct dlm_lkb *lkb;
int rv;
limit.max = end;
limit.min = start;
lkb = dlm_allocate_lkb(ls);
if (!lkb)
return -ENOMEM;
lkb->lkb_last_bast_cb_mode = DLM_LOCK_IV;
lkb->lkb_last_cast_cb_mode = DLM_LOCK_IV;
lkb->lkb_last_cb_mode = DLM_LOCK_IV;
lkb->lkb_nodeid = -1;
lkb->lkb_grmode = DLM_LOCK_IV;
kref_init(&lkb->lkb_ref);
INIT_LIST_HEAD(&lkb->lkb_ownqueue);
INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
write_lock_bh(&ls->ls_lkbxa_lock);
rv = xa_alloc(&ls->ls_lkbxa, &lkb->lkb_id, lkb, limit, GFP_ATOMIC);
write_unlock_bh(&ls->ls_lkbxa_lock);
if (rv < 0) {
log_error(ls, "create_lkb xa error %d", rv);
dlm_free_lkb(lkb);
return rv;
}
*lkb_ret = lkb;
return 0;
}
static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
{
return _create_lkb(ls, lkb_ret, 1, ULONG_MAX);
}
static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
{
struct dlm_lkb *lkb;
read_lock_bh(&ls->ls_lkbxa_lock);
lkb = xa_load(&ls->ls_lkbxa, lkid);
if (lkb)
kref_get(&lkb->lkb_ref);
read_unlock_bh(&ls->ls_lkbxa_lock);
*lkb_ret = lkb;
return lkb ? 0 : -ENOENT;
}
static void kill_lkb(struct kref *kref)
{
struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
/* All work is done after the return from kref_put() so we
can release the write_lock before the detach_lkb */
DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
}
/* __put_lkb() is used when an lkb may not have an rsb attached to
it so we need to provide the lockspace explicitly */
static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
{
uint32_t lkid = lkb->lkb_id;
int rv;
rv = dlm_kref_put_write_lock_bh(&lkb->lkb_ref, kill_lkb,
&ls->ls_lkbxa_lock);
if (rv) {
xa_erase(&ls->ls_lkbxa, lkid);
write_unlock_bh(&ls->ls_lkbxa_lock);
detach_lkb(lkb);
/* for local/process lkbs, lvbptr points to caller's lksb */
if (lkb->lkb_lvbptr && is_master_copy(lkb))
dlm_free_lvb(lkb->lkb_lvbptr);
dlm_free_lkb(lkb);
}
return rv;
}
int dlm_put_lkb(struct dlm_lkb *lkb)
{
struct dlm_ls *ls;
DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
ls = lkb->lkb_resource->res_ls;
return __put_lkb(ls, lkb);
}
/* This is only called to add a reference when the code already holds
a valid reference to the lkb, so there's no need for locking. */
static inline void hold_lkb(struct dlm_lkb *lkb)
{
kref_get(&lkb->lkb_ref);
}
static void unhold_lkb_assert(struct kref *kref)
{
struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
DLM_ASSERT(false, dlm_print_lkb(lkb););
}
/* This is called when we need to remove a reference and are certain
it's not the last ref. e.g. del_lkb is always called between a
find_lkb/put_lkb and is always the inverse of a previous add_lkb.
put_lkb would work fine, but would involve unnecessary locking */
static inline void unhold_lkb(struct dlm_lkb *lkb)
{
kref_put(&lkb->lkb_ref, unhold_lkb_assert);
}
static void lkb_add_ordered(struct list_head *new, struct list_head *head,
int mode)
{
struct dlm_lkb *lkb = NULL, *iter;
list_for_each_entry(iter, head, lkb_statequeue)
if (iter->lkb_rqmode < mode) {
lkb = iter;
list_add_tail(new, &iter->lkb_statequeue);
break;
}
if (!lkb)
list_add_tail(new, head);
}
/* add/remove lkb to rsb's grant/convert/wait queue */
static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
{
kref_get(&lkb->lkb_ref);
DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
lkb->lkb_timestamp = ktime_get();
lkb->lkb_status = status;
switch (status) {
case DLM_LKSTS_WAITING:
if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
else
list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
break;
case DLM_LKSTS_GRANTED:
/* convention says granted locks kept in order of grmode */
lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
lkb->lkb_grmode);
break;
case DLM_LKSTS_CONVERT:
if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
else
list_add_tail(&lkb->lkb_statequeue,
&r->res_convertqueue);
break;
default:
DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
}
}
static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
lkb->lkb_status = 0;
list_del(&lkb->lkb_statequeue);
unhold_lkb(lkb);
}
static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
{
hold_lkb(lkb);
del_lkb(r, lkb);
add_lkb(r, lkb, sts);
unhold_lkb(lkb);
}
static int msg_reply_type(int mstype)
{
switch (mstype) {
case DLM_MSG_REQUEST:
return DLM_MSG_REQUEST_REPLY;
case DLM_MSG_CONVERT:
return DLM_MSG_CONVERT_REPLY;
case DLM_MSG_UNLOCK:
return DLM_MSG_UNLOCK_REPLY;
case DLM_MSG_CANCEL:
return DLM_MSG_CANCEL_REPLY;
case DLM_MSG_LOOKUP:
return DLM_MSG_LOOKUP_REPLY;
}
return -1;
}
/* add/remove lkb from global waiters list of lkb's waiting for
a reply from a remote node */
static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
int error = 0;
spin_lock_bh(&ls->ls_waiters_lock);
if (is_overlap_unlock(lkb) ||
(is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
error = -EINVAL;
goto out;
}
if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
switch (mstype) {
case DLM_MSG_UNLOCK:
set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
break;
case DLM_MSG_CANCEL:
set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
break;
default:
error = -EBUSY;
goto out;
}
lkb->lkb_wait_count++;
hold_lkb(lkb);
log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
lkb->lkb_id, lkb->lkb_wait_type, mstype,
lkb->lkb_wait_count, dlm_iflags_val(lkb));
goto out;
}
DLM_ASSERT(!lkb->lkb_wait_count,
dlm_print_lkb(lkb);
printk("wait_count %d\n", lkb->lkb_wait_count););
lkb->lkb_wait_count++;
lkb->lkb_wait_type = mstype;
lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
hold_lkb(lkb);
list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
out:
if (error)
log_error(ls, "addwait error %x %d flags %x %d %d %s",
lkb->lkb_id, error, dlm_iflags_val(lkb), mstype,
lkb->lkb_wait_type, lkb->lkb_resource->res_name);
spin_unlock_bh(&ls->ls_waiters_lock);
return error;
}
/* We clear the RESEND flag because we might be taking an lkb off the waiters
list as part of process_requestqueue (e.g. a lookup that has an optimized
request reply on the requestqueue) between dlm_recover_waiters_pre() which
set RESEND and dlm_recover_waiters_post() */
static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
const struct dlm_message *ms)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
int overlap_done = 0;
if (mstype == DLM_MSG_UNLOCK_REPLY &&
test_and_clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags)) {
log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
overlap_done = 1;
goto out_del;
}
if (mstype == DLM_MSG_CANCEL_REPLY &&
test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags)) {
log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
overlap_done = 1;
goto out_del;
}
/* Cancel state was preemptively cleared by a successful convert,
see next comment, nothing to do. */
if ((mstype == DLM_MSG_CANCEL_REPLY) &&
(lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
log_debug(ls, "remwait %x cancel_reply wait_type %d",
lkb->lkb_id, lkb->lkb_wait_type);
return -1;
}
/* Remove for the convert reply, and premptively remove for the
cancel reply. A convert has been granted while there's still
an outstanding cancel on it (the cancel is moot and the result
in the cancel reply should be 0). We preempt the cancel reply
because the app gets the convert result and then can follow up
with another op, like convert. This subsequent op would see the
lingering state of the cancel and fail with -EBUSY. */
if ((mstype == DLM_MSG_CONVERT_REPLY) &&
(lkb->lkb_wait_type == DLM_MSG_CONVERT) && ms && !ms->m_result &&
test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags)) {
log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
lkb->lkb_id);
lkb->lkb_wait_type = 0;
lkb->lkb_wait_count--;
unhold_lkb(lkb);
goto out_del;
}
/* N.B. type of reply may not always correspond to type of original
msg due to lookup->request optimization, verify others? */
if (lkb->lkb_wait_type) {
lkb->lkb_wait_type = 0;
goto out_del;
}
log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0,
lkb->lkb_remid, mstype, dlm_iflags_val(lkb));
return -1;
out_del:
/* the force-unlock/cancel has completed and we haven't recvd a reply
to the op that was in progress prior to the unlock/cancel; we
give up on any reply to the earlier op. FIXME: not sure when/how
this would happen */
if (overlap_done && lkb->lkb_wait_type) {
log_error(ls, "remwait error %x reply %d wait_type %d overlap",
lkb->lkb_id, mstype, lkb->lkb_wait_type);
lkb->lkb_wait_count--;
unhold_lkb(lkb);
lkb->lkb_wait_type = 0;
}
DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
clear_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
lkb->lkb_wait_count--;
if (!lkb->lkb_wait_count)
list_del_init(&lkb->lkb_wait_reply);
unhold_lkb(lkb);
return 0;
}
static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
int error;
spin_lock_bh(&ls->ls_waiters_lock);
error = _remove_from_waiters(lkb, mstype, NULL);
spin_unlock_bh(&ls->ls_waiters_lock);
return error;
}
/* Handles situations where we might be processing a "fake" or "local" reply in
* the recovery context which stops any locking activity. Only debugfs might
* change the lockspace waiters but they will held the recovery lock to ensure
* remove_from_waiters_ms() in local case will be the only user manipulating the
* lockspace waiters in recovery context.
*/
static int remove_from_waiters_ms(struct dlm_lkb *lkb,
const struct dlm_message *ms, bool local)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
int error;
if (!local)
spin_lock_bh(&ls->ls_waiters_lock);
else
WARN_ON_ONCE(!rwsem_is_locked(&ls->ls_in_recovery) ||
!dlm_locking_stopped(ls));
error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms);
if (!local)
spin_unlock_bh(&ls->ls_waiters_lock);
return error;
}
/* lkb is master or local copy */
static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
int b, len = r->res_ls->ls_lvblen;
/* b=1 lvb returned to caller
b=0 lvb written to rsb or invalidated
b=-1 do nothing */
b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
if (b == 1) {
if (!lkb->lkb_lvbptr)
return;
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
return;
if (!r->res_lvbptr)
return;
memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
lkb->lkb_lvbseq = r->res_lvbseq;
} else if (b == 0) {
if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
rsb_set_flag(r, RSB_VALNOTVALID);
return;
}
if (!lkb->lkb_lvbptr)
return;
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
return;
if (!r->res_lvbptr)
r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
if (!r->res_lvbptr)
return;
memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
r->res_lvbseq++;
lkb->lkb_lvbseq = r->res_lvbseq;
rsb_clear_flag(r, RSB_VALNOTVALID);
}
if (rsb_flag(r, RSB_VALNOTVALID))
set_bit(DLM_SBF_VALNOTVALID_BIT, &lkb->lkb_sbflags);
}
static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
if (lkb->lkb_grmode < DLM_LOCK_PW)
return;
if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
rsb_set_flag(r, RSB_VALNOTVALID);
return;
}
if (!lkb->lkb_lvbptr)
return;
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
return;
if (!r->res_lvbptr)
r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
if (!r->res_lvbptr)
return;
memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
r->res_lvbseq++;
rsb_clear_flag(r, RSB_VALNOTVALID);
}
/* lkb is process copy (pc) */
static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
const struct dlm_message *ms)
{
int b;
if (!lkb->lkb_lvbptr)
return;
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
return;
b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
if (b == 1) {
int len = receive_extralen(ms);
if (len > r->res_ls->ls_lvblen)
len = r->res_ls->ls_lvblen;
memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
}
}
/* Manipulate lkb's on rsb's convert/granted/waiting queues
remove_lock -- used for unlock, removes lkb from granted
revert_lock -- used for cancel, moves lkb from convert to granted
grant_lock -- used for request and convert, adds lkb to granted or
moves lkb from convert or waiting to granted
Each of these is used for master or local copy lkb's. There is
also a _pc() variation used to make the corresponding change on
a process copy (pc) lkb. */
static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
del_lkb(r, lkb);
lkb->lkb_grmode = DLM_LOCK_IV;
/* this unhold undoes the original ref from create_lkb()
so this leads to the lkb being freed */
unhold_lkb(lkb);
}
static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
set_lvb_unlock(r, lkb);
_remove_lock(r, lkb);
}
static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
_remove_lock(r, lkb);
}
/* returns: 0 did nothing
1 moved lock to granted
-1 removed lock */
static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
int rv = 0;
lkb->lkb_rqmode = DLM_LOCK_IV;
switch (lkb->lkb_status) {
case DLM_LKSTS_GRANTED:
break;
case DLM_LKSTS_CONVERT:
move_lkb(r, lkb, DLM_LKSTS_GRANTED);
rv = 1;
break;
case DLM_LKSTS_WAITING:
del_lkb(r, lkb);
lkb->lkb_grmode = DLM_LOCK_IV;
/* this unhold undoes the original ref from create_lkb()
so this leads to the lkb being freed */
unhold_lkb(lkb);
rv = -1;
break;
default:
log_print("invalid status for revert %d", lkb->lkb_status);
}
return rv;
}
static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
return revert_lock(r, lkb);
}
static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
if (lkb->lkb_grmode != lkb->lkb_rqmode) {
lkb->lkb_grmode = lkb->lkb_rqmode;
if (lkb->lkb_status)
move_lkb(r, lkb, DLM_LKSTS_GRANTED);
else
add_lkb(r, lkb, DLM_LKSTS_GRANTED);
}
lkb->lkb_rqmode = DLM_LOCK_IV;
lkb->lkb_highbast = 0;
}
static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
set_lvb_lock(r, lkb);
_grant_lock(r, lkb);
}
static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
const struct dlm_message *ms)
{
set_lvb_lock_pc(r, lkb, ms);
_grant_lock(r, lkb);
}
/* called by grant_pending_locks() which means an async grant message must
be sent to the requesting node in addition to granting the lock if the
lkb belongs to a remote node. */
static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
grant_lock(r, lkb);
if (is_master_copy(lkb))
send_grant(r, lkb);
else
queue_cast(r, lkb, 0);
}
/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
change the granted/requested modes. We're munging things accordingly in
the process copy.
CONVDEADLK: our grmode may have been forced down to NL to resolve a
conversion deadlock
ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
compatible with other granted locks */
static void munge_demoted(struct dlm_lkb *lkb)
{
if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
log_print("munge_demoted %x invalid modes gr %d rq %d",
lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
return;
}
lkb->lkb_grmode = DLM_LOCK_NL;
}
static void munge_altmode(struct dlm_lkb *lkb, const struct dlm_message *ms)
{
if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) &&
ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) {
log_print("munge_altmode %x invalid reply type %d",
lkb->lkb_id, le32_to_cpu(ms->m_type));
return;
}
if (lkb->lkb_exflags & DLM_LKF_ALTPR)
lkb->lkb_rqmode = DLM_LOCK_PR;
else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
lkb->lkb_rqmode = DLM_LOCK_CW;
else {
log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
dlm_print_lkb(lkb);
}
}
static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
{
struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
lkb_statequeue);
if (lkb->lkb_id == first->lkb_id)
return 1;
return 0;
}
/* Check if the given lkb conflicts with another lkb on the queue. */
static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
{
struct dlm_lkb *this;
list_for_each_entry(this, head, lkb_statequeue) {
if (this == lkb)
continue;
if (!modes_compat(this, lkb))
return 1;
}
return 0;
}
/*
* "A conversion deadlock arises with a pair of lock requests in the converting
* queue for one resource. The granted mode of each lock blocks the requested
* mode of the other lock."
*
* Part 2: if the granted mode of lkb is preventing an earlier lkb in the
* convert queue from being granted, then deadlk/demote lkb.
*
* Example:
* Granted Queue: empty
* Convert Queue: NL->EX (first lock)
* PR->EX (second lock)
*
* The first lock can't be granted because of the granted mode of the second
* lock and the second lock can't be granted because it's not first in the
* list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
* demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
* flag set and return DEMOTED in the lksb flags.
*
* Originally, this function detected conv-deadlk in a more limited scope:
* - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
* - if lkb1 was the first entry in the queue (not just earlier), and was
* blocked by the granted mode of lkb2, and there was nothing on the
* granted queue preventing lkb1 from being granted immediately, i.e.
* lkb2 was the only thing preventing lkb1 from being granted.
*
* That second condition meant we'd only say there was conv-deadlk if
* resolving it (by demotion) would lead to the first lock on the convert
* queue being granted right away. It allowed conversion deadlocks to exist
* between locks on the convert queue while they couldn't be granted anyway.
*
* Now, we detect and take action on conversion deadlocks immediately when
* they're created, even if they may not be immediately consequential. If
* lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
* mode that would prevent lkb1's conversion from being granted, we do a
* deadlk/demote on lkb2 right away and don't let it onto the convert queue.
* I think this means that the lkb_is_ahead condition below should always
* be zero, i.e. there will never be conv-deadlk between two locks that are
* both already on the convert queue.
*/
static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
{
struct dlm_lkb *lkb1;
int lkb_is_ahead = 0;
list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
if (lkb1 == lkb2) {
lkb_is_ahead = 1;
continue;
}
if (!lkb_is_ahead) {
if (!modes_compat(lkb2, lkb1))
return 1;
} else {
if (!modes_compat(lkb2, lkb1) &&
!modes_compat(lkb1, lkb2))
return 1;
}
}
return 0;
}
/*
* Return 1 if the lock can be granted, 0 otherwise.
* Also detect and resolve conversion deadlocks.
*
* lkb is the lock to be granted
*
* now is 1 if the function is being called in the context of the
* immediate request, it is 0 if called later, after the lock has been
* queued.
*
* recover is 1 if dlm_recover_grant() is trying to grant conversions
* after recovery.
*
* References are from chapter 6 of "VAXcluster Principles" by Roy Davis
*/
static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
int recover)
{
int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
/*
* 6-10: Version 5.4 introduced an option to address the phenomenon of
* a new request for a NL mode lock being blocked.
*
* 6-11: If the optional EXPEDITE flag is used with the new NL mode
* request, then it would be granted. In essence, the use of this flag
* tells the Lock Manager to expedite theis request by not considering
* what may be in the CONVERTING or WAITING queues... As of this
* writing, the EXPEDITE flag can be used only with new requests for NL
* mode locks. This flag is not valid for conversion requests.
*
* A shortcut. Earlier checks return an error if EXPEDITE is used in a
* conversion or used with a non-NL requested mode. We also know an
* EXPEDITE request is always granted immediately, so now must always
* be 1. The full condition to grant an expedite request: (now &&
* !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
* therefore be shortened to just checking the flag.
*/
if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
return 1;
/*
* A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
* added to the remaining conditions.
*/
if (queue_conflict(&r->res_grantqueue, lkb))
return 0;
/*
* 6-3: By default, a conversion request is immediately granted if the
* requested mode is compatible with the modes of all other granted
* locks
*/
if (queue_conflict(&r->res_convertqueue, lkb))
return 0;
/*
* The RECOVER_GRANT flag means dlm_recover_grant() is granting
* locks for a recovered rsb, on which lkb's have been rebuilt.
* The lkb's may have been rebuilt on the queues in a different
* order than they were in on the previous master. So, granting
* queued conversions in order after recovery doesn't make sense
* since the order hasn't been preserved anyway. The new order
* could also have created a new "in place" conversion deadlock.
* (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
* After recovery, there would be no granted locks, and possibly
* NL->EX, PR->EX, an in-place conversion deadlock.) So, after
* recovery, grant conversions without considering order.
*/
if (conv && recover)
return 1;
/*
* 6-5: But the default algorithm for deciding whether to grant or
* queue conversion requests does not by itself guarantee that such
* requests are serviced on a "first come first serve" basis. This, in
* turn, can lead to a phenomenon known as "indefinate postponement".
*
* 6-7: This issue is dealt with by using the optional QUECVT flag with
* the system service employed to request a lock conversion. This flag
* forces certain conversion requests to be queued, even if they are
* compatible with the granted modes of other locks on the same
* resource. Thus, the use of this flag results in conversion requests
* being ordered on a "first come first servce" basis.
*
* DCT: This condition is all about new conversions being able to occur
* "in place" while the lock remains on the granted queue (assuming
* nothing else conflicts.) IOW if QUECVT isn't set, a conversion
* doesn't _have_ to go onto the convert queue where it's processed in
* order. The "now" variable is necessary to distinguish converts
* being received and processed for the first time now, because once a
* convert is moved to the conversion queue the condition below applies
* requiring fifo granting.
*/
if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
return 1;
/*
* Even if the convert is compat with all granted locks,
* QUECVT forces it behind other locks on the convert queue.
*/
if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
if (list_empty(&r->res_convertqueue))
return 1;
else
return 0;
}
/*
* The NOORDER flag is set to avoid the standard vms rules on grant
* order.
*/
if (lkb->lkb_exflags & DLM_LKF_NOORDER)
return 1;
/*
* 6-3: Once in that queue [CONVERTING], a conversion request cannot be
* granted until all other conversion requests ahead of it are granted
* and/or canceled.
*/
if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
return 1;
/*
* 6-4: By default, a new request is immediately granted only if all
* three of the following conditions are satisfied when the request is
* issued:
* - The queue of ungranted conversion requests for the resource is
* empty.
* - The queue of ungranted new requests for the resource is empty.
* - The mode of the new request is compatible with the most
* restrictive mode of all granted locks on the resource.
*/
if (now && !conv && list_empty(&r->res_convertqueue) &&
list_empty(&r->res_waitqueue))
return 1;
/*
* 6-4: Once a lock request is in the queue of ungranted new requests,
* it cannot be granted until the queue of ungranted conversion
* requests is empty, all ungranted new requests ahead of it are
* granted and/or canceled, and it is compatible with the granted mode
* of the most restrictive lock granted on the resource.
*/
if (!now && !conv && list_empty(&r->res_convertqueue) &&
first_in_list(lkb, &r->res_waitqueue))
return 1;
return 0;
}
static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
int recover, int *err)
{
int rv;
int8_t alt = 0, rqmode = lkb->lkb_rqmode;
int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
if (err)
*err = 0;
rv = _can_be_granted(r, lkb, now, recover);
if (rv)
goto out;
/*
* The CONVDEADLK flag is non-standard and tells the dlm to resolve
* conversion deadlocks by demoting grmode to NL, otherwise the dlm
* cancels one of the locks.
*/
if (is_convert && can_be_queued(lkb) &&
conversion_deadlock_detect(r, lkb)) {
if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
lkb->lkb_grmode = DLM_LOCK_NL;
set_bit(DLM_SBF_DEMOTED_BIT, &lkb->lkb_sbflags);
} else if (err) {
*err = -EDEADLK;
} else {
log_print("can_be_granted deadlock %x now %d",
lkb->lkb_id, now);
dlm_dump_rsb(r);
}
goto out;
}
/*
* The ALTPR and ALTCW flags are non-standard and tell the dlm to try
* to grant a request in a mode other than the normal rqmode. It's a
* simple way to provide a big optimization to applications that can
* use them.
*/
if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
alt = DLM_LOCK_PR;
else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
alt = DLM_LOCK_CW;
if (alt) {
lkb->lkb_rqmode = alt;
rv = _can_be_granted(r, lkb, now, 0);
if (rv)
set_bit(DLM_SBF_ALTMODE_BIT, &lkb->lkb_sbflags);
else
lkb->lkb_rqmode = rqmode;
}
out:
return rv;
}
/* Returns the highest requested mode of all blocked conversions; sets
cw if there's a blocked conversion to DLM_LOCK_CW. */
static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
unsigned int *count)
{
struct dlm_lkb *lkb, *s;
int recover = rsb_flag(r, RSB_RECOVER_GRANT);
int hi, demoted, quit, grant_restart, demote_restart;
int deadlk;
quit = 0;
restart:
grant_restart = 0;
demote_restart = 0;
hi = DLM_LOCK_IV;
list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
demoted = is_demoted(lkb);
deadlk = 0;
if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
grant_lock_pending(r, lkb);
grant_restart = 1;
if (count)
(*count)++;
continue;
}
if (!demoted && is_demoted(lkb)) {
log_print("WARN: pending demoted %x node %d %s",
lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
demote_restart = 1;
continue;
}
if (deadlk) {
/*
* If DLM_LKB_NODLKWT flag is set and conversion
* deadlock is detected, we request blocking AST and
* down (or cancel) conversion.
*/
if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
if (lkb->lkb_highbast < lkb->lkb_rqmode) {
queue_bast(r, lkb, lkb->lkb_rqmode);
lkb->lkb_highbast = lkb->lkb_rqmode;
}
} else {
log_print("WARN: pending deadlock %x node %d %s",
lkb->lkb_id, lkb->lkb_nodeid,
r->res_name);
dlm_dump_rsb(r);
}
continue;
}
hi = max_t(int, lkb->lkb_rqmode, hi);
if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
*cw = 1;
}
if (grant_restart)
goto restart;
if (demote_restart && !quit) {
quit = 1;
goto restart;
}
return max_t(int, high, hi);
}
static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
unsigned int *count)
{
struct dlm_lkb *lkb, *s;
list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
if (can_be_granted(r, lkb, 0, 0, NULL)) {
grant_lock_pending(r, lkb);
if (count)
(*count)++;
} else {
high = max_t(int, lkb->lkb_rqmode, high);
if (lkb->lkb_rqmode == DLM_LOCK_CW)
*cw = 1;
}
}
return high;
}
/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
on either the convert or waiting queue.
high is the largest rqmode of all locks blocked on the convert or
waiting queue. */
static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
{
if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
if (gr->lkb_highbast < DLM_LOCK_EX)
return 1;
return 0;
}
if (gr->lkb_highbast < high &&
!__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
return 1;
return 0;
}
static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
{
struct dlm_lkb *lkb, *s;
int high = DLM_LOCK_IV;
int cw = 0;
if (!is_master(r)) {
log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
dlm_dump_rsb(r);
return;
}
high = grant_pending_convert(r, high, &cw, count);
high = grant_pending_wait(r, high, &cw, count);
if (high == DLM_LOCK_IV)
return;
/*
* If there are locks left on the wait/convert queue then send blocking
* ASTs to granted locks based on the largest requested mode (high)
* found above.
*/
list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
if (cw && high == DLM_LOCK_PR &&
lkb->lkb_grmode == DLM_LOCK_PR)
queue_bast(r, lkb, DLM_LOCK_CW);
else
queue_bast(r, lkb, high);
lkb->lkb_highbast = high;
}
}
}
static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
{
if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
(gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
if (gr->lkb_highbast < DLM_LOCK_EX)
return 1;
return 0;
}
if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
return 1;
return 0;
}
static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
struct dlm_lkb *lkb)
{
struct dlm_lkb *gr;
list_for_each_entry(gr, head, lkb_statequeue) {
/* skip self when sending basts to convertqueue */
if (gr == lkb)
continue;
if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
queue_bast(r, gr, lkb->lkb_rqmode);
gr->lkb_highbast = lkb->lkb_rqmode;
}
}
}
static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
send_bast_queue(r, &r->res_grantqueue, lkb);
}
static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
send_bast_queue(r, &r->res_grantqueue, lkb);
send_bast_queue(r, &r->res_convertqueue, lkb);
}
/* set_master(r, lkb) -- set the master nodeid of a resource
The purpose of this function is to set the nodeid field in the given
lkb using the nodeid field in the given rsb. If the rsb's nodeid is
known, it can just be copied to the lkb and the function will return
0. If the rsb's nodeid is _not_ known, it needs to be looked up
before it can be copied to the lkb.
When the rsb nodeid is being looked up remotely, the initial lkb
causing the lookup is kept on the ls_waiters list waiting for the
lookup reply. Other lkb's waiting for the same rsb lookup are kept
on the rsb's res_lookup list until the master is verified.
Return values:
0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1: the rsb master is not available and the lkb has been placed on
a wait queue
*/
static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
int our_nodeid = dlm_our_nodeid();
if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
r->res_first_lkid = lkb->lkb_id;
lkb->lkb_nodeid = r->res_nodeid;
return 0;
}
if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
return 1;
}
if (r->res_master_nodeid == our_nodeid) {
lkb->lkb_nodeid = 0;
return 0;
}
if (r->res_master_nodeid) {
lkb->lkb_nodeid = r->res_master_nodeid;
return 0;
}
if (dlm_dir_nodeid(r) == our_nodeid) {
/* This is a somewhat unusual case; find_rsb will usually
have set res_master_nodeid when dir nodeid is local, but
there are cases where we become the dir node after we've
past find_rsb and go through _request_lock again.
confirm_master() or process_lookup_list() needs to be
called after this. */
log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
r->res_name);
r->res_master_nodeid = our_nodeid;
r->res_nodeid = 0;
lkb->lkb_nodeid = 0;
return 0;
}
r->res_first_lkid = lkb->lkb_id;
send_lookup(r, lkb);
return 1;
}
static void process_lookup_list(struct dlm_rsb *r)
{
struct dlm_lkb *lkb, *safe;
list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
list_del_init(&lkb->lkb_rsb_lookup);
_request_lock(r, lkb);
}
}
/* confirm_master -- confirm (or deny) an rsb's master nodeid */
static void confirm_master(struct dlm_rsb *r, int error)
{
struct dlm_lkb *lkb;
if (!r->res_first_lkid)
return;
switch (error) {
case 0:
case -EINPROGRESS:
r->res_first_lkid = 0;
process_lookup_list(r);
break;
case -EAGAIN:
case -EBADR:
case -ENOTBLK:
/* the remote request failed and won't be retried (it was
a NOQUEUE, or has been canceled/unlocked); make a waiting
lkb the first_lkid */
r->res_first_lkid = 0;
if (!list_empty(&r->res_lookup)) {
lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
lkb_rsb_lookup);
list_del_init(&lkb->lkb_rsb_lookup);
r->res_first_lkid = lkb->lkb_id;
_request_lock(r, lkb);
}
break;
default:
log_error(r->res_ls, "confirm_master unknown error %d", error);
}
}
static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
int namelen, void (*ast)(void *astparam),
void *astparam,
void (*bast)(void *astparam, int mode),
struct dlm_args *args)
{
int rv = -EINVAL;
/* check for invalid arg usage */
if (mode < 0 || mode > DLM_LOCK_EX)
goto out;
if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
goto out;
if (flags & DLM_LKF_CANCEL)
goto out;
if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
goto out;
if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
goto out;
if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
goto out;
if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
goto out;
if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
goto out;
if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
goto out;
if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
goto out;
if (!ast || !lksb)
goto out;
if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
goto out;
if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
goto out;
/* these args will be copied to the lkb in validate_lock_args,
it cannot be done now because when converting locks, fields in
an active lkb cannot be modified before locking the rsb */
args->flags = flags;
args->astfn = ast;
args->astparam = astparam;
args->bastfn = bast;
args->mode = mode;
args->lksb = lksb;
rv = 0;
out:
return rv;
}
static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
{
if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
DLM_LKF_FORCEUNLOCK))
return -EINVAL;
if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
return -EINVAL;
args->flags = flags;
args->astparam = astarg;
return 0;
}
static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
struct dlm_args *args)
{
int rv = -EBUSY;
if (args->flags & DLM_LKF_CONVERT) {
if (lkb->lkb_status != DLM_LKSTS_GRANTED)
goto out;
/* lock not allowed if there's any op in progress */
if (lkb->lkb_wait_type || lkb->lkb_wait_count)
goto out;
if (is_overlap(lkb))
goto out;
rv = -EINVAL;
if (test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
goto out;
if (args->flags & DLM_LKF_QUECVT &&
!__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
goto out;
}
lkb->lkb_exflags = args->flags;
dlm_set_sbflags_val(lkb, 0);
lkb->lkb_astfn = args->astfn;
lkb->lkb_astparam = args->astparam;
lkb->lkb_bastfn = args->bastfn;
lkb->lkb_rqmode = args->mode;
lkb->lkb_lksb = args->lksb;
lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
lkb->lkb_ownpid = (int) current->pid;
rv = 0;
out:
switch (rv) {
case 0:
break;
case -EINVAL:
/* annoy the user because dlm usage is wrong */
WARN_ON(1);
log_error(ls, "%s %d %x %x %x %d %d %s", __func__,
rv, lkb->lkb_id, dlm_iflags_val(lkb), args->flags,
lkb->lkb_status, lkb->lkb_wait_type,
lkb->lkb_resource->res_name);
break;
default:
log_debug(ls, "%s %d %x %x %x %d %d %s", __func__,
rv, lkb->lkb_id, dlm_iflags_val(lkb), args->flags,
lkb->lkb_status, lkb->lkb_wait_type,
lkb->lkb_resource->res_name);
break;
}
return rv;
}
/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
for success */
/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
because there may be a lookup in progress and it's valid to do
cancel/unlockf on it */
static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
int rv = -EBUSY;
/* normal unlock not allowed if there's any op in progress */
if (!(args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) &&
(lkb->lkb_wait_type || lkb->lkb_wait_count))
goto out;
/* an lkb may be waiting for an rsb lookup to complete where the
lookup was initiated by another lock */
if (!list_empty(&lkb->lkb_rsb_lookup)) {
if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
list_del_init(&lkb->lkb_rsb_lookup);
queue_cast(lkb->lkb_resource, lkb,
args->flags & DLM_LKF_CANCEL ?
-DLM_ECANCEL : -DLM_EUNLOCK);
unhold_lkb(lkb); /* undoes create_lkb() */
}
/* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
goto out;
}
rv = -EINVAL;
if (test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags)) {
log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
dlm_print_lkb(lkb);
goto out;
}
/* an lkb may still exist even though the lock is EOL'ed due to a
* cancel, unlock or failed noqueue request; an app can't use these
* locks; return same error as if the lkid had not been found at all
*/
if (test_bit(DLM_IFL_ENDOFLIFE_BIT, &lkb->lkb_iflags)) {
log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
rv = -ENOENT;
goto out;
}
/* cancel not allowed with another cancel/unlock in progress */
if (args->flags & DLM_LKF_CANCEL) {
if (lkb->lkb_exflags & DLM_LKF_CANCEL)
goto out;
if (is_overlap(lkb))
goto out;
if (test_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags)) {
set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
rv = -EBUSY;
goto out;
}
/* there's nothing to cancel */
if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
!lkb->lkb_wait_type) {
rv = -EBUSY;
goto out;
}
switch (lkb->lkb_wait_type) {
case DLM_MSG_LOOKUP:
case DLM_MSG_REQUEST:
set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
rv = -EBUSY;
goto out;
case DLM_MSG_UNLOCK:
case DLM_MSG_CANCEL:
goto out;
}
/* add_to_waiters() will set OVERLAP_CANCEL */
goto out_ok;
}
/* do we need to allow a force-unlock if there's a normal unlock
already in progress? in what conditions could the normal unlock
fail such that we'd want to send a force-unlock to be sure? */
if (args->flags & DLM_LKF_FORCEUNLOCK) {
if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
goto out;
if (is_overlap_unlock(lkb))
goto out;
if (test_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags)) {
set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
rv = -EBUSY;
goto out;
}
switch (lkb->lkb_wait_type) {
case DLM_MSG_LOOKUP:
case DLM_MSG_REQUEST:
set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
rv = -EBUSY;
goto out;
case DLM_MSG_UNLOCK:
goto out;
}
/* add_to_waiters() will set OVERLAP_UNLOCK */
}
out_ok:
/* an overlapping op shouldn't blow away exflags from other op */
lkb->lkb_exflags |= args->flags;
dlm_set_sbflags_val(lkb, 0);
lkb->lkb_astparam = args->astparam;
rv = 0;
out:
switch (rv) {
case 0:
break;
case -EINVAL:
/* annoy the user because dlm usage is wrong */
WARN_ON(1);
log_error(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
lkb->lkb_id, dlm_iflags_val(lkb), lkb->lkb_exflags,
args->flags, lkb->lkb_wait_type,
lkb->lkb_resource->res_name);
break;
default:
log_debug(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
lkb->lkb_id, dlm_iflags_val(lkb), lkb->lkb_exflags,
args->flags, lkb->lkb_wait_type,
lkb->lkb_resource->res_name);
break;
}
return rv;
}
/*
* Four stage 4 varieties:
* do_request(), do_convert(), do_unlock(), do_cancel()
* These are called on the master node for the given lock and
* from the central locking logic.
*/
static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
int error = 0;
if (can_be_granted(r, lkb, 1, 0, NULL)) {
grant_lock(r, lkb);
queue_cast(r, lkb, 0);
goto out;
}
if (can_be_queued(lkb)) {
error = -EINPROGRESS;
add_lkb(r, lkb, DLM_LKSTS_WAITING);
goto out;
}
error = -EAGAIN;
queue_cast(r, lkb, -EAGAIN);
out:
return error;
}
static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
int error)
{
switch (error) {
case -EAGAIN:
if (force_blocking_asts(lkb))
send_blocking_asts_all(r, lkb);
break;
case -EINPROGRESS:
send_blocking_asts(r, lkb);
break;
}
}
static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
int error = 0;
int deadlk = 0;
/* changing an existing lock may allow others to be granted */
if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
grant_lock(r, lkb);
queue_cast(r, lkb, 0);
goto out;
}
/* can_be_granted() detected that this lock would block in a conversion
deadlock, so we leave it on the granted queue and return EDEADLK in
the ast for the convert. */
if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
/* it's left on the granted queue */
revert_lock(r, lkb);
queue_cast(r, lkb, -EDEADLK);
error = -EDEADLK;
goto out;
}
/* is_demoted() means the can_be_granted() above set the grmode
to NL, and left us on the granted queue. This auto-demotion
(due to CONVDEADLK) might mean other locks, and/or this lock, are
now grantable. We have to try to grant other converting locks
before we try again to grant this one. */
if (is_demoted(lkb)) {
grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
if (_can_be_granted(r, lkb, 1, 0)) {
grant_lock(r, lkb);
queue_cast(r, lkb, 0);
goto out;
}
/* else fall through and move to convert queue */
}
if (can_be_queued(lkb)) {
error = -EINPROGRESS;
del_lkb(r, lkb);
add_lkb(r, lkb, DLM_LKSTS_CONVERT);
goto out;
}
error = -EAGAIN;
queue_cast(r, lkb, -EAGAIN);
out:
return error;
}
static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
int error)
{
switch (error) {
case 0:
grant_pending_locks(r, NULL);
/* grant_pending_locks also sends basts */
break;
case -EAGAIN:
if (force_blocking_asts(lkb))
send_blocking_asts_all(r, lkb);
break;
case -EINPROGRESS:
send_blocking_asts(r, lkb);
break;
}
}
static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
remove_lock(r, lkb);
queue_cast(r, lkb, -DLM_EUNLOCK);
return -DLM_EUNLOCK;
}
static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
int error)
{
grant_pending_locks(r, NULL);
}
/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
{