diff options
author | Alexander Aring <aahringo@redhat.com> | 2024-06-03 17:55:56 -0400 |
---|---|---|
committer | David Teigland <teigland@redhat.com> | 2024-06-11 13:13:00 -0500 |
commit | 68bde2a67a6d6e5a2b8e1a64bad51dd8c3975256 (patch) | |
tree | b79223b504283b3914c490d733f984c0633d59d3 /fs/dlm | |
parent | f328a26eeb5380bc74e58cb9c3280a4908452df7 (diff) |
dlm: implement LSFL_SOFTIRQ_SAFE
When a lockspace user allows it, run callback functions directly from
softirq context, instead of queueing callbacks to be run from the
dlm_callback workqueue context.
Signed-off-by: Alexander Aring <aahringo@redhat.com>
Signed-off-by: David Teigland <teigland@redhat.com>
Diffstat (limited to 'fs/dlm')
-rw-r--r-- | fs/dlm/ast.c | 157 | ||||
-rw-r--r-- | fs/dlm/ast.h | 11 | ||||
-rw-r--r-- | fs/dlm/dlm_internal.h | 1 | ||||
-rw-r--r-- | fs/dlm/lockspace.c | 3 | ||||
-rw-r--r-- | fs/dlm/user.c | 38 |
5 files changed, 126 insertions, 84 deletions
diff --git a/fs/dlm/ast.c b/fs/dlm/ast.c index 52ce27031314..742b30b61c19 100644 --- a/fs/dlm/ast.c +++ b/fs/dlm/ast.c @@ -18,35 +18,52 @@ #include "user.h" #include "ast.h" -static void dlm_callback_work(struct work_struct *work) +static void dlm_run_callback(uint32_t ls_id, uint32_t lkb_id, int8_t mode, + uint32_t flags, uint8_t sb_flags, int sb_status, + struct dlm_lksb *lksb, + void (*astfn)(void *astparam), + void (*bastfn)(void *astparam, int mode), + void *astparam, const char *res_name, + size_t res_length) { - struct dlm_callback *cb = container_of(work, struct dlm_callback, work); - - if (cb->flags & DLM_CB_BAST) { - trace_dlm_bast(cb->ls_id, cb->lkb_id, cb->mode, cb->res_name, - cb->res_length); - cb->bastfn(cb->astparam, cb->mode); - } else if (cb->flags & DLM_CB_CAST) { - trace_dlm_ast(cb->ls_id, cb->lkb_id, cb->sb_status, - cb->sb_flags, cb->res_name, cb->res_length); - cb->lkb_lksb->sb_status = cb->sb_status; - cb->lkb_lksb->sb_flags = cb->sb_flags; - cb->astfn(cb->astparam); + if (flags & DLM_CB_BAST) { + trace_dlm_bast(ls_id, lkb_id, mode, res_name, res_length); + bastfn(astparam, mode); + } else if (flags & DLM_CB_CAST) { + trace_dlm_ast(ls_id, lkb_id, sb_status, sb_flags, res_name, + res_length); + lksb->sb_status = sb_status; + lksb->sb_flags = sb_flags; + astfn(astparam); } +} +static void dlm_do_callback(struct dlm_callback *cb) +{ + dlm_run_callback(cb->ls_id, cb->lkb_id, cb->mode, cb->flags, + cb->sb_flags, cb->sb_status, cb->lkb_lksb, + cb->astfn, cb->bastfn, cb->astparam, + cb->res_name, cb->res_length); dlm_free_cb(cb); } -int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, - int status, uint32_t sbflags, - struct dlm_callback **cb) +static void dlm_callback_work(struct work_struct *work) +{ + struct dlm_callback *cb = container_of(work, struct dlm_callback, work); + + dlm_do_callback(cb); +} + +bool dlm_may_skip_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, + int status, uint32_t sbflags, int *copy_lvb) { struct dlm_rsb *rsb = lkb->lkb_resource; - int rv = DLM_ENQUEUE_CALLBACK_SUCCESS; struct dlm_ls *ls = rsb->res_ls; - int copy_lvb = 0; int prev_mode; + if (copy_lvb) + *copy_lvb = 0; + if (flags & DLM_CB_BAST) { /* if cb is a bast, it should be skipped if the blocking mode is * compatible with the last granted mode @@ -56,7 +73,7 @@ int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, log_debug(ls, "skip %x bast mode %d for cast mode %d", lkb->lkb_id, mode, lkb->lkb_last_cast_cb_mode); - goto out; + return true; } } @@ -74,7 +91,7 @@ int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, (prev_mode > mode && prev_mode > DLM_LOCK_PR)) { log_debug(ls, "skip %x add bast mode %d for bast mode %d", lkb->lkb_id, mode, prev_mode); - goto out; + return true; } } @@ -85,8 +102,10 @@ int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, prev_mode = lkb->lkb_last_cast_cb_mode; if (!status && lkb->lkb_lksb->sb_lvbptr && - dlm_lvb_operations[prev_mode + 1][mode + 1]) - copy_lvb = 1; + dlm_lvb_operations[prev_mode + 1][mode + 1]) { + if (copy_lvb) + *copy_lvb = 1; + } } lkb->lkb_last_cast_cb_mode = mode; @@ -96,11 +115,19 @@ int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, lkb->lkb_last_cb_mode = mode; lkb->lkb_last_cb_flags = flags; + return false; +} + +int dlm_get_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, + int status, uint32_t sbflags, + struct dlm_callback **cb) +{ + struct dlm_rsb *rsb = lkb->lkb_resource; + struct dlm_ls *ls = rsb->res_ls; + *cb = dlm_allocate_cb(); - if (!*cb) { - rv = DLM_ENQUEUE_CALLBACK_FAILURE; - goto out; - } + if (WARN_ON_ONCE(!*cb)) + return -ENOMEM; /* for tracing */ (*cb)->lkb_id = lkb->lkb_id; @@ -112,19 +139,34 @@ int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, (*cb)->mode = mode; (*cb)->sb_status = status; (*cb)->sb_flags = (sbflags & 0x000000FF); - (*cb)->copy_lvb = copy_lvb; (*cb)->lkb_lksb = lkb->lkb_lksb; - rv = DLM_ENQUEUE_CALLBACK_NEED_SCHED; + return 0; +} + +static int dlm_get_queue_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, + int status, uint32_t sbflags, + struct dlm_callback **cb) +{ + int rv; -out: - return rv; + rv = dlm_get_cb(lkb, flags, mode, status, sbflags, cb); + if (rv) + return rv; + + (*cb)->astfn = lkb->lkb_astfn; + (*cb)->bastfn = lkb->lkb_bastfn; + (*cb)->astparam = lkb->lkb_astparam; + INIT_WORK(&(*cb)->work, dlm_callback_work); + + return 0; } void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status, - uint32_t sbflags) + uint32_t sbflags) { - struct dlm_ls *ls = lkb->lkb_resource->res_ls; + struct dlm_rsb *rsb = lkb->lkb_resource; + struct dlm_ls *ls = rsb->res_ls; struct dlm_callback *cb; int rv; @@ -133,35 +175,34 @@ void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status, return; } - rv = dlm_queue_lkb_callback(lkb, flags, mode, status, sbflags, - &cb); - switch (rv) { - case DLM_ENQUEUE_CALLBACK_NEED_SCHED: - cb->astfn = lkb->lkb_astfn; - cb->bastfn = lkb->lkb_bastfn; - cb->astparam = lkb->lkb_astparam; - INIT_WORK(&cb->work, dlm_callback_work); - - spin_lock_bh(&ls->ls_cb_lock); - if (test_bit(LSFL_CB_DELAY, &ls->ls_flags)) + if (dlm_may_skip_callback(lkb, flags, mode, status, sbflags, NULL)) + return; + + spin_lock_bh(&ls->ls_cb_lock); + if (test_bit(LSFL_CB_DELAY, &ls->ls_flags)) { + rv = dlm_get_queue_cb(lkb, flags, mode, status, sbflags, &cb); + if (!rv) list_add(&cb->list, &ls->ls_cb_delay); - else - queue_work(ls->ls_callback_wq, &cb->work); - spin_unlock_bh(&ls->ls_cb_lock); - break; - case DLM_ENQUEUE_CALLBACK_SUCCESS: - break; - case DLM_ENQUEUE_CALLBACK_FAILURE: - fallthrough; - default: - WARN_ON_ONCE(1); - break; + } else { + if (test_bit(LSFL_SOFTIRQ, &ls->ls_flags)) { + dlm_run_callback(ls->ls_global_id, lkb->lkb_id, mode, flags, + sbflags, status, lkb->lkb_lksb, + lkb->lkb_astfn, lkb->lkb_bastfn, + lkb->lkb_astparam, rsb->res_name, + rsb->res_length); + } else { + rv = dlm_get_queue_cb(lkb, flags, mode, status, sbflags, &cb); + if (!rv) + queue_work(ls->ls_callback_wq, &cb->work); + } } + spin_unlock_bh(&ls->ls_cb_lock); } int dlm_callback_start(struct dlm_ls *ls) { - if (!test_bit(LSFL_FS, &ls->ls_flags)) + if (!test_bit(LSFL_FS, &ls->ls_flags) || + test_bit(LSFL_SOFTIRQ, &ls->ls_flags)) return 0; ls->ls_callback_wq = alloc_ordered_workqueue("dlm_callback", @@ -207,7 +248,11 @@ more: spin_lock_bh(&ls->ls_cb_lock); list_for_each_entry_safe(cb, safe, &ls->ls_cb_delay, list) { list_del(&cb->list); - queue_work(ls->ls_callback_wq, &cb->work); + if (test_bit(LSFL_SOFTIRQ, &ls->ls_flags)) + dlm_do_callback(cb); + else + queue_work(ls->ls_callback_wq, &cb->work); + count++; if (count == MAX_CB_QUEUE) break; diff --git a/fs/dlm/ast.h b/fs/dlm/ast.h index 9093ff043bee..e2b86845d331 100644 --- a/fs/dlm/ast.h +++ b/fs/dlm/ast.h @@ -11,12 +11,11 @@ #ifndef __ASTD_DOT_H__ #define __ASTD_DOT_H__ -#define DLM_ENQUEUE_CALLBACK_NEED_SCHED 1 -#define DLM_ENQUEUE_CALLBACK_SUCCESS 0 -#define DLM_ENQUEUE_CALLBACK_FAILURE -1 -int dlm_queue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, - int status, uint32_t sbflags, - struct dlm_callback **cb); +bool dlm_may_skip_callback(struct dlm_lkb *lkb, uint32_t flags, int mode, + int status, uint32_t sbflags, int *copy_lvb); +int dlm_get_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, + int status, uint32_t sbflags, + struct dlm_callback **cb); void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status, uint32_t sbflags); diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index c6baf25b9cae..32d98e63d25e 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -699,6 +699,7 @@ struct dlm_ls { #define LSFL_NODIR 10 #define LSFL_RECV_MSG_BLOCKED 11 #define LSFL_FS 12 +#define LSFL_SOFTIRQ 13 #define DLM_PROC_FLAGS_CLOSING 1 #define DLM_PROC_FLAGS_COMPAT 2 diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c index a7ac0fcb4ef3..7c4f45ad2245 100644 --- a/fs/dlm/lockspace.c +++ b/fs/dlm/lockspace.c @@ -407,6 +407,9 @@ static int new_lockspace(const char *name, const char *cluster, ls->ls_ops_arg = ops_arg; } + if (flags & DLM_LSFL_SOFTIRQ) + set_bit(LSFL_SOFTIRQ, &ls->ls_flags); + /* ls_exflags are forced to match among nodes, and we don't * need to require all nodes to have some flags set */ diff --git a/fs/dlm/user.c b/fs/dlm/user.c index f6635a5314f4..5cb3896be826 100644 --- a/fs/dlm/user.c +++ b/fs/dlm/user.c @@ -182,7 +182,7 @@ void dlm_user_add_ast(struct dlm_lkb *lkb, uint32_t flags, int mode, struct dlm_user_args *ua; struct dlm_user_proc *proc; struct dlm_callback *cb; - int rv; + int rv, copy_lvb; if (test_bit(DLM_DFL_ORPHAN_BIT, &lkb->lkb_dflags) || test_bit(DLM_IFL_DEAD_BIT, &lkb->lkb_iflags)) @@ -213,28 +213,22 @@ void dlm_user_add_ast(struct dlm_lkb *lkb, uint32_t flags, int mode, spin_lock_bh(&proc->asts_spin); - rv = dlm_queue_lkb_callback(lkb, flags, mode, status, sbflags, &cb); - switch (rv) { - case DLM_ENQUEUE_CALLBACK_NEED_SCHED: - cb->ua = *ua; - cb->lkb_lksb = &cb->ua.lksb; - if (cb->copy_lvb) { - memcpy(cb->lvbptr, ua->lksb.sb_lvbptr, - DLM_USER_LVB_LEN); - cb->lkb_lksb->sb_lvbptr = cb->lvbptr; + if (!dlm_may_skip_callback(lkb, flags, mode, status, sbflags, + ©_lvb)) { + rv = dlm_get_cb(lkb, flags, mode, status, sbflags, &cb); + if (!rv) { + cb->copy_lvb = copy_lvb; + cb->ua = *ua; + cb->lkb_lksb = &cb->ua.lksb; + if (copy_lvb) { + memcpy(cb->lvbptr, ua->lksb.sb_lvbptr, + DLM_USER_LVB_LEN); + cb->lkb_lksb->sb_lvbptr = cb->lvbptr; + } + + list_add_tail(&cb->list, &proc->asts); + wake_up_interruptible(&proc->wait); } - - list_add_tail(&cb->list, &proc->asts); - wake_up_interruptible(&proc->wait); - break; - case DLM_ENQUEUE_CALLBACK_SUCCESS: - break; - case DLM_ENQUEUE_CALLBACK_FAILURE: - fallthrough; - default: - spin_unlock_bh(&proc->asts_spin); - WARN_ON_ONCE(1); - goto out; } spin_unlock_bh(&proc->asts_spin); |