// 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] = …; /* * 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] = …; #define modes_compat(gr, rq) … int dlm_modes_compat(int mode1, int mode2) { … } /* * 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] = …; void dlm_print_lkb(struct dlm_lkb *lkb) { … } static void dlm_print_rsb(struct dlm_rsb *r) { … } void dlm_dump_rsb(struct dlm_rsb *r) { … } /* Threads cannot use the lockspace while it's being recovered */ void dlm_lock_recovery(struct dlm_ls *ls) { … } void dlm_unlock_recovery(struct dlm_ls *ls) { … } int dlm_lock_recovery_try(struct dlm_ls *ls) { … } static inline int can_be_queued(struct dlm_lkb *lkb) { … } static inline int force_blocking_asts(struct dlm_lkb *lkb) { … } static inline int is_demoted(struct dlm_lkb *lkb) { … } static inline int is_altmode(struct dlm_lkb *lkb) { … } static inline int is_granted(struct dlm_lkb *lkb) { … } static inline int is_remote(struct dlm_rsb *r) { … } static inline int is_process_copy(struct dlm_lkb *lkb) { … } static inline int is_master_copy(struct dlm_lkb *lkb) { … } static inline int middle_conversion(struct dlm_lkb *lkb) { … } static inline int down_conversion(struct dlm_lkb *lkb) { … } static inline int is_overlap_unlock(struct dlm_lkb *lkb) { … } static inline int is_overlap_cancel(struct dlm_lkb *lkb) { … } static inline int is_overlap(struct dlm_lkb *lkb) { … } static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { … } static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode) { … } /* * Basic operations on rsb's and lkb's */ static inline unsigned long rsb_toss_jiffies(void) { … } /* 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) { … } void dlm_hold_rsb(struct dlm_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) { … } /* 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) { … } static void put_rsb(struct dlm_rsb *r) { … } void dlm_put_rsb(struct dlm_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) { … } /* 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) { … } /* ls_rsbtbl_lock must be held */ static void del_scan(struct dlm_ls *ls, struct dlm_rsb *r) { … } static void add_scan(struct dlm_ls *ls, struct dlm_rsb *r) { … } /* 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 … /* Called by lockspace scan_timer to free unused rsb's. */ void dlm_rsb_scan(struct timer_list *timer) { … } /* 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) { … } int dlm_search_rsb_tree(struct rhashtable *rhash, const void *name, int len, struct dlm_rsb **r_ret) { … } static int rsb_insert(struct dlm_rsb *rsb, struct rhashtable *rhash) { … } /* * 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) { … } /* 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) { … } /* * 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) { … } /* 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) { … } 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) { … } /* * 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) { … } int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, const char *name, int len, unsigned int flags, int *r_nodeid, int *result) { … } static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash) { … } void dlm_dump_rsb_name(struct dlm_ls *ls, const char *name, int len) { … } static void deactivate_rsb(struct kref *kref) { … } /* See comment for unhold_lkb */ static void unhold_rsb(struct dlm_rsb *r) { … } void free_inactive_rsb(struct dlm_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) { … } static void detach_lkb(struct dlm_lkb *lkb) { … } static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret, unsigned long start, unsigned long end) { … } static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret) { … } static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret) { … } static void kill_lkb(struct kref *kref) { … } /* __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) { … } int dlm_put_lkb(struct dlm_lkb *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) { … } static void unhold_lkb_assert(struct kref *kref) { … } /* 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) { … } static void lkb_add_ordered(struct list_head *new, struct list_head *head, int mode) { … } /* add/remove lkb to rsb's grant/convert/wait queue */ static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status) { … } static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts) { … } static int msg_reply_type(int mstype) { … } /* 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) { … } /* 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) { … } static int remove_from_waiters(struct dlm_lkb *lkb, int mstype) { … } /* 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) { … } /* lkb is master or local copy */ static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* lkb is process copy (pc) */ static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } /* 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) { … } static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *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) { … } static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } /* 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) { … } /* 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) { … } static void munge_altmode(struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head) { … } /* Check if the given lkb conflicts with another lkb on the queue. */ static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb) { … } /* * "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) { … } /* * 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) { … } static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now, int recover, int *err) { … } /* 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) { … } static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw, unsigned int *count) { … } /* 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) { … } static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count) { … } static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq) { … } static void send_bast_queue(struct dlm_rsb *r, struct list_head *head, struct dlm_lkb *lkb) { … } static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *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) { … } static void process_lookup_list(struct dlm_rsb *r) { … } /* confirm_master -- confirm (or deny) an rsb's master nodeid */ static void confirm_master(struct dlm_rsb *r, int 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) { … } static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args) { … } static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_args *args) { … } /* 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) { … } /* * 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) { … } static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, int error) { … } static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, int error) { … } static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, int error) { … } /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */ static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, int error) { … } /* * Four stage 3 varieties: * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock() */ /* add a new lkb to a possibly new rsb, called by requesting process */ static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* change some property of an existing lkb, e.g. mode */ static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* remove an existing lkb from the granted queue */ static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* remove an existing lkb from the convert or wait queue */ static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* * Four stage 2 varieties: * request_lock(), convert_lock(), unlock_lock(), cancel_lock() */ static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, const void *name, int len, struct dlm_args *args) { … } static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_args *args) { … } static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_args *args) { … } static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_args *args) { … } /* * Two stage 1 varieties: dlm_lock() and dlm_unlock() */ int dlm_lock(dlm_lockspace_t *lockspace, int mode, struct dlm_lksb *lksb, uint32_t flags, const void *name, unsigned int namelen, uint32_t parent_lkid, void (*ast) (void *astarg), void *astarg, void (*bast) (void *astarg, int mode)) { … } int dlm_unlock(dlm_lockspace_t *lockspace, uint32_t lkid, uint32_t flags, struct dlm_lksb *lksb, void *astarg) { … } /* * send/receive routines for remote operations and replies * * send_args * send_common * send_request receive_request * send_convert receive_convert * send_unlock receive_unlock * send_cancel receive_cancel * send_grant receive_grant * send_bast receive_bast * send_lookup receive_lookup * send_remove receive_remove * * send_common_reply * receive_request_reply send_request_reply * receive_convert_reply send_convert_reply * receive_unlock_reply send_unlock_reply * receive_cancel_reply send_cancel_reply * receive_lookup_reply send_lookup_reply */ static int _create_message(struct dlm_ls *ls, int mb_len, int to_nodeid, int mstype, struct dlm_message **ms_ret, struct dlm_mhandle **mh_ret) { … } static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb, int to_nodeid, int mstype, struct dlm_message **ms_ret, struct dlm_mhandle **mh_ret) { … } /* further lowcomms enhancements or alternate implementations may make the return value from this function useful at some point */ static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms, const void *name, int namelen) { … } static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb, struct dlm_message *ms) { … } static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype) { … } static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) { … } /* FIXME: if this lkb is the only lock we hold on the rsb, then set MASTER_UNCERTAIN to force the next request on the rsb to confirm that the master is still correct. */ 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 send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype, int rv) { … } static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { … } static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { … } static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { … } static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) { … } static int send_lookup_reply(struct dlm_ls *ls, const struct dlm_message *ms_in, int ret_nodeid, int rv) { … } /* which args we save from a received message depends heavily on the type of message, unlike the send side where we can safely send everything about the lkb for any type of message */ static void receive_flags(struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static void receive_flags_reply(struct dlm_lkb *lkb, const struct dlm_message *ms, bool local) { … } static int receive_extralen(const struct dlm_message *ms) { … } static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static void fake_bastfn(void *astparam, int mode) { … } static void fake_astfn(void *astparam) { … } static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, const struct dlm_message *ms) { … } /* We fill in the local-lkb fields with the info that send_xxxx_reply() uses to send a reply and that the remote end uses to process the reply. */ static void setup_local_lkb(struct dlm_ls *ls, const struct dlm_message *ms) { … } /* This is called after the rsb is locked so that we can safely inspect fields in the lkb. */ static int validate_message(struct dlm_lkb *lkb, const struct dlm_message *ms) { … } static int receive_request(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_convert(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_unlock(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_cancel(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_grant(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_bast(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void receive_lookup(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void receive_remove(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void receive_purge(struct dlm_ls *ls, const struct dlm_message *ms) { … } static int receive_request_reply(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, const struct dlm_message *ms, bool local) { … } static void _receive_convert_reply(struct dlm_lkb *lkb, const struct dlm_message *ms, bool local) { … } static int receive_convert_reply(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void _receive_unlock_reply(struct dlm_lkb *lkb, const struct dlm_message *ms, bool local) { … } static int receive_unlock_reply(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void _receive_cancel_reply(struct dlm_lkb *lkb, const struct dlm_message *ms, bool local) { … } static int receive_cancel_reply(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void receive_lookup_reply(struct dlm_ls *ls, const struct dlm_message *ms) { … } static void _receive_message(struct dlm_ls *ls, const struct dlm_message *ms, uint32_t saved_seq) { … } /* If the lockspace is in recovery mode (locking stopped), then normal messages are saved on the requestqueue for processing after recovery is done. When not in recovery mode, we wait for dlm_recoverd to drain saved messages off the requestqueue before we process new ones. This occurs right after recovery completes when we transition from saving all messages on requestqueue, to processing all the saved messages, to processing new messages as they arrive. */ static void dlm_receive_message(struct dlm_ls *ls, const struct dlm_message *ms, int nodeid) { … } /* This is called by dlm_recoverd to process messages that were saved on the requestqueue. */ void dlm_receive_message_saved(struct dlm_ls *ls, const struct dlm_message *ms, uint32_t saved_seq) { … } /* This is called by the midcomms layer when something is received for the lockspace. It could be either a MSG (normal message sent as part of standard locking activity) or an RCOM (recovery message sent as part of lockspace recovery). */ void dlm_receive_buffer(const union dlm_packet *p, int nodeid) { … } static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_message *ms_local) { … } /* A waiting lkb needs recovery if the master node has failed, or the master node is changing (only when no directory is used) */ static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb, int dir_nodeid) { … } /* Recovery for locks that are waiting for replies from nodes that are now gone. We can just complete unlocks and cancels by faking a reply from the dead node. Requests and up-conversions we flag to be resent after recovery. Down-conversions can just be completed with a fake reply like unlocks. Conversions between PR and CW need special attention. */ void dlm_recover_waiters_pre(struct dlm_ls *ls) { … } static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls) { … } /* * Forced state reset for locks that were in the middle of remote operations * when recovery happened (i.e. lkbs that were on the waiters list, waiting * for a reply from a remote operation.) The lkbs remaining on the waiters * list need to be reevaluated; some may need resending to a different node * than previously, and some may now need local handling rather than remote. * * First, the lkb state for the voided remote operation is forcibly reset, * equivalent to what remove_from_waiters() would normally do: * . lkb removed from ls_waiters list * . lkb wait_type cleared * . lkb waiters_count cleared * . lkb ref count decremented for each waiters_count (almost always 1, * but possibly 2 in case of cancel/unlock overlapping, which means * two remote replies were being expected for the lkb.) * * Second, the lkb is reprocessed like an original operation would be, * by passing it to _request_lock or _convert_lock, which will either * process the lkb operation locally, or send it to a remote node again * and put the lkb back onto the waiters list. * * When reprocessing the lkb, we may find that it's flagged for an overlapping * force-unlock or cancel, either from before recovery began, or after recovery * finished. If this is the case, the unlock/cancel is done directly, and the * original operation is not initiated again (no _request_lock/_convert_lock.) */ int dlm_recover_waiters_post(struct dlm_ls *ls) { … } static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r, struct list_head *list) { … } void dlm_purge_mstcpy_locks(struct dlm_rsb *r) { … } static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r, struct list_head *list, int nodeid_gone, unsigned int *count) { … } /* Get rid of locks held by nodes that are gone. */ void dlm_recover_purge(struct dlm_ls *ls, const struct list_head *root_list) { … } static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls) { … } /* * Attempt to grant locks on resources that we are the master of. * Locks may have become grantable during recovery because locks * from departed nodes have been purged (or not rebuilt), allowing * previously blocked locks to now be granted. The subset of rsb's * we are interested in are those with lkb's on either the convert or * waiting queues. * * Simplest would be to go through each master rsb and check for non-empty * convert or waiting queues, and attempt to grant on those rsbs. * Checking the queues requires lock_rsb, though, for which we'd need * to release the rsbtbl lock. This would make iterating through all * rsb's very inefficient. So, we rely on earlier recovery routines * to set RECOVER_GRANT on any rsb's that we should attempt to grant * locks for. */ void dlm_recover_grant(struct dlm_ls *ls) { … } static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid, uint32_t remid) { … } static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid, uint32_t remid) { … } /* needs at least dlm_rcom + rcom_lock */ static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_rsb *r, const struct dlm_rcom *rc) { … } /* This lkb may have been recovered in a previous aborted recovery so we need to check if the rsb already has an lkb with the given remote nodeid/lkid. If so we just send back a standard reply. If not, we create a new lkb with the given values and send back our lkid. We send back our lkid by sending back the rcom_lock struct we got but with the remid field filled in. */ /* needs at least dlm_rcom + rcom_lock */ int dlm_recover_master_copy(struct dlm_ls *ls, const struct dlm_rcom *rc, __le32 *rl_remid, __le32 *rl_result) { … } /* needs at least dlm_rcom + rcom_lock */ int dlm_recover_process_copy(struct dlm_ls *ls, const struct dlm_rcom *rc, uint64_t seq) { … } int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua, int mode, uint32_t flags, void *name, unsigned int namelen) { … } int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, int mode, uint32_t flags, uint32_t lkid, char *lvb_in) { … } /* * The caller asks for an orphan lock on a given resource with a given mode. * If a matching lock exists, it's moved to the owner's list of locks and * the lkid is returned. */ int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, int mode, uint32_t flags, void *name, unsigned int namelen, uint32_t *lkid) { … } int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, uint32_t flags, uint32_t lkid, char *lvb_in) { … } int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, uint32_t flags, uint32_t lkid) { … } int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid) { … } /* lkb's that are removed from the waiters list by revert are just left on the orphans list with the granted orphan locks, to be freed by purge */ static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb) { … } /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't granted. Regardless of what rsb queue the lock is on, it's removed and freed. The IVVALBLK flag causes the lvb on the resource to be invalidated if our lock is PW/EX (it's ignored if our granted mode is smaller.) */ static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb) { … } /* We have to release clear_proc_locks mutex before calling unlock_proc_lock() (which does lock_rsb) due to deadlock with receiving a message that does lock_rsb followed by dlm_user_add_cb() */ static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls, struct dlm_user_proc *proc) { … } /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts, which we clear here. */ /* proc CLOSING flag is set so no more device_reads should look at proc->asts list, and no more device_writes should add lkb's to proc->locks list; so we shouldn't need to take asts_spin or locks_spin here. this assumes that device reads/writes/closes are serialized -- FIXME: we may need to serialize them ourself. */ void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc) { … } static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc) { … } /* pid of 0 means purge all orphans */ static void do_purge(struct dlm_ls *ls, int nodeid, int pid) { … } static int send_purge(struct dlm_ls *ls, int nodeid, int pid) { … } int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc, int nodeid, int pid) { … } /* debug functionality */ int dlm_debug_add_lkb(struct dlm_ls *ls, uint32_t lkb_id, char *name, int len, int lkb_nodeid, unsigned int lkb_dflags, int lkb_status) { … } int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id, int mstype, int to_nodeid) { … }
Codebase Browser
linux