UbiquitousOS
/
OpenCloudOS-Kernel
mirror of https://gitee.com/OpenCloudOS/OpenCloudOS-Kernel.git
11
0
Fork 0
Code Issues Projects Releases Wiki Activity

dlm for 6.2 

These patches include the usual cleanups and minor fixes, the removal of
 code that is no longer needed due to recent improvements, and
 improvements to processing large volumes of messages during heavy
 locking activity.
 
 - Misc code cleanup.
 
 - Fix a couple socket handling bugs: a double release on an error path
   and a data-ready race in an accept loop.
 
 - Remove code for resending dir-remove messages. This code is no longer
   needed since the midcomms layer now ensures the messages are resent if
   needed.
 
 - Add tracepoints for dlm messages.
 
 - Improve callback queueing by replacing the fixed array with a list.
 
 - Simplify the handling of a remove message followed by a lookup
   message by sending both without releasing a spinlock in between.
 
 - Improve the concurrency of sending and receiving messages by holding
   locks for a shorter time, and changing how workqueues are used.
 
 - Remove old code for shutting down sockets, which is no longer needed
   with the reliable connection handling that was recently added.
 -----BEGIN PGP SIGNATURE-----
 
 iQIcBAABAgAGBQJjl2TfAAoJEDgbc8f8gGmqvj8P/ibVRzW00DeKJJU8Gz5qlcQo
 YziuUxR2xYffDwif/VB2FFW1dE9lY9bTTYHz6ctiIKip51XsVTVj9oaXEebSt5Xu
 /jEMJXUnMCUPcXz6T5FoaXAt+oMKqo7QhitCD4kqdM8AUZQLXKu/YM/FC1EyIMgl
 zGmHch99ojGPQmmPrFdJq8JQSF75+r28RfH9qmy7dzctX0Iok12J8Gsgd1J9Hwti
 PITmNBs0YauT7AdYwR3UhuGC9g73RISJ4QTjL+wpj7+P2F6pNYb8XXUXsEZ2wZYH
 sgo325HbIXZ8AbxIw0Tbx2NNmIQaTELZRCOSxgEjcN7AMhn1xkXYPE3mfMCU1bDZ
 TtfN86HvWYFRgNotGpDUrNnd9XJoyeHtCWBDikEepYRWEiyYkzipj5mFGRzINlvL
 Orou1tKBchvtDgQzGg6XwNkxnxR6l/xfhg1AYizYXCgOCxfEzJgIqCUttzMCPkec
 HXi/y17IDIvNvIhtmZ0EYh/6W40UYlnncREszerg5IR3ssuDmF5aA6MHdiSt0dn/
 PkjZ5DRbZaEbNl37mjShJHxztP1QKW85c3isgEBe4jvncu8VY3WDCbQRw5gjZvC4
 LnQF4hcGLm0y51fU5DXv+CBlQGRWaqYan8bWaIdVoA2UgDbM0x8eWC8IUUYBO5ak
 rCBOpWJudzZFH6ynyfwD
 =TJvP
 -----END PGP SIGNATURE-----

Merge tag 'dlm-6.2' of git://git.kernel.org/pub/scm/linux/kernel/git/teigland/linux-dlm

Pull dlm updates from David Teigland:
 "These patches include the usual cleanups and minor fixes, the removal
  of code that is no longer needed due to recent improvements, and
  improvements to processing large volumes of messages during heavy
  locking activity.

  Summary:

   - Misc code cleanup

   - Fix a couple of socket handling bugs: a double release on an error
     path and a data-ready race in an accept loop

   - Remove code for resending dir-remove messages. This code is no
     longer needed since the midcomms layer now ensures the messages are
     resent if needed

   - Add tracepoints for dlm messages

   - Improve callback queueing by replacing the fixed array with a list

   - Simplify the handling of a remove message followed by a lookup
     message by sending both without releasing a spinlock in between

   - Improve the concurrency of sending and receiving messages by
     holding locks for a shorter time, and changing how workqueues are
     used

   - Remove old code for shutting down sockets, which is no longer
     needed with the reliable connection handling that was recently
     added"

* tag 'dlm-6.2' of git://git.kernel.org/pub/scm/linux/kernel/git/teigland/linux-dlm: (37 commits)
  fs: dlm: fix building without lockdep
  fs: dlm: parallelize lowcomms socket handling
  fs: dlm: don't init error value
  fs: dlm: use saved sk_error_report()
  fs: dlm: use sock2con without checking null
  fs: dlm: remove dlm_node_addrs lookup list
  fs: dlm: don't put dlm_local_addrs on heap
  fs: dlm: cleanup listen sock handling
  fs: dlm: remove socket shutdown handling
  fs: dlm: use listen sock as dlm running indicator
  fs: dlm: use list_first_entry_or_null
  fs: dlm: remove twice INIT_WORK
  fs: dlm: add midcomms init/start functions
  fs: dlm: add dst nodeid for msg tracing
  fs: dlm: rename seq to h_seq for msg tracing
  fs: dlm: rename DLM_IFL_NEED_SCHED to DLM_IFL_CB_PENDING
  fs: dlm: ast do WARN_ON_ONCE() on hotpath
  fs: dlm: drop lkb ref in bug case
  fs: dlm: avoid false-positive checker warning
  fs: dlm: use WARN_ON_ONCE() instead of WARN_ON()
  ...
This commit is contained in:
Linus Torvalds 2022-12-12 20:41:50 -08:00
parent 56c003e4db 7a5e9f1f83
commit 97971df811
20 changed files with 1466 additions and 1254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

324
fs/dlm/ast.c
View File

@ -12,55 +12,67 @@
#include <trace/events/dlm.h>
#include "dlm_internal.h"
#include "memory.h"
#include "lock.h"
#include "user.h"
#include "ast.h"
static uint64_t dlm_cb_seq;
static DEFINE_SPINLOCK(dlm_cb_seq_spin);
static void dlm_dump_lkb_callbacks(struct dlm_lkb *lkb)
void dlm_release_callback(struct kref *ref)
{
int i;
struct dlm_callback *cb = container_of(ref, struct dlm_callback, ref);
log_print("last_bast %x %llu flags %x mode %d sb %d %x",
lkb->lkb_id,
(unsigned long long)lkb->lkb_last_bast.seq,
lkb->lkb_last_bast.flags,
lkb->lkb_last_bast.mode,
lkb->lkb_last_bast.sb_status,
lkb->lkb_last_bast.sb_flags);
log_print("last_cast %x %llu flags %x mode %d sb %d %x",
lkb->lkb_id,
(unsigned long long)lkb->lkb_last_cast.seq,
lkb->lkb_last_cast.flags,
lkb->lkb_last_cast.mode,
lkb->lkb_last_cast.sb_status,
lkb->lkb_last_cast.sb_flags);
for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
log_print("cb %x %llu flags %x mode %d sb %d %x",
lkb->lkb_id,
(unsigned long long)lkb->lkb_callbacks[i].seq,
lkb->lkb_callbacks[i].flags,
lkb->lkb_callbacks[i].mode,
lkb->lkb_callbacks[i].sb_status,
lkb->lkb_callbacks[i].sb_flags);
}
dlm_free_cb(cb);
}
int dlm_add_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags, uint64_t seq)
void dlm_callback_set_last_ptr(struct dlm_callback **from,
struct dlm_callback *to)
{
if (*from)
kref_put(&(*from)->ref, dlm_release_callback);
if (to)
kref_get(&to->ref);
*from = to;
}
void dlm_purge_lkb_callbacks(struct dlm_lkb *lkb)
{
struct dlm_callback *cb, *safe;
list_for_each_entry_safe(cb, safe, &lkb->lkb_callbacks, list) {
list_del(&cb->list);
kref_put(&cb->ref, dlm_release_callback);
}
lkb->lkb_flags &= ~DLM_IFL_CB_PENDING;
/* invalidate */
dlm_callback_set_last_ptr(&lkb->lkb_last_cast, NULL);
dlm_callback_set_last_ptr(&lkb->lkb_last_cb, NULL);
lkb->lkb_last_bast_mode = -1;
}
int dlm_enqueue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
uint64_t prev_seq;
int rv = DLM_ENQUEUE_CALLBACK_SUCCESS;
struct dlm_callback *cb;
int prev_mode;
int i, rv;
for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
if (lkb->lkb_callbacks[i].seq)
continue;
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
*/
if (lkb->lkb_last_cast) {
if (dlm_modes_compat(mode, lkb->lkb_last_cast->mode)) {
log_debug(ls, "skip %x bast mode %d for cast mode %d",
lkb->lkb_id, mode,
lkb->lkb_last_cast->mode);
goto out;
}
}
/*
* Suppress some redundant basts here, do more on removal.
@ -68,148 +80,95 @@ int dlm_add_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
* is a bast for the same mode or a more restrictive mode.
* (the addional > PR check is needed for PR/CW inversion)
*/
if ((i > 0) && (flags & DLM_CB_BAST) &&
(lkb->lkb_callbacks[i-1].flags & DLM_CB_BAST)) {
prev_seq = lkb->lkb_callbacks[i-1].seq;
prev_mode = lkb->lkb_callbacks[i-1].mode;
if (lkb->lkb_last_cb && lkb->lkb_last_cb->flags & DLM_CB_BAST) {
prev_mode = lkb->lkb_last_cb->mode;
if ((prev_mode == mode) ||
(prev_mode > mode && prev_mode > DLM_LOCK_PR)) {
log_debug(ls, "skip %x add bast %llu mode %d "
"for bast %llu mode %d",
lkb->lkb_id,
(unsigned long long)seq,
mode,
(unsigned long long)prev_seq,
prev_mode);
rv = 0;
log_debug(ls, "skip %x add bast mode %d for bast mode %d",
lkb->lkb_id, mode, prev_mode);
goto out;
}
}
lkb->lkb_callbacks[i].seq = seq;
lkb->lkb_callbacks[i].flags = flags;
lkb->lkb_callbacks[i].mode = mode;
lkb->lkb_callbacks[i].sb_status = status;
lkb->lkb_callbacks[i].sb_flags = (sbflags & 0x000000FF);
rv = 0;
break;
}
if (i == DLM_CALLBACKS_SIZE) {
log_error(ls, "no callbacks %x %llu flags %x mode %d sb %d %x",
lkb->lkb_id, (unsigned long long)seq,
flags, mode, status, sbflags);
dlm_dump_lkb_callbacks(lkb);
rv = -1;
cb = dlm_allocate_cb();
if (!cb) {
rv = DLM_ENQUEUE_CALLBACK_FAILURE;
goto out;
}
cb->flags = flags;
cb->mode = mode;
cb->sb_status = status;
cb->sb_flags = (sbflags & 0x000000FF);
kref_init(&cb->ref);
if (!(lkb->lkb_flags & DLM_IFL_CB_PENDING)) {
lkb->lkb_flags |= DLM_IFL_CB_PENDING;
rv = DLM_ENQUEUE_CALLBACK_NEED_SCHED;
}
list_add_tail(&cb->list, &lkb->lkb_callbacks);
if (flags & DLM_CB_CAST)
dlm_callback_set_last_ptr(&lkb->lkb_last_cast, cb);
dlm_callback_set_last_ptr(&lkb->lkb_last_cb, cb);
out:
return rv;
}
int dlm_rem_lkb_callback(struct dlm_ls *ls, struct dlm_lkb *lkb,
struct dlm_callback *cb, int *resid)
int dlm_dequeue_lkb_callback(struct dlm_lkb *lkb, struct dlm_callback **cb)
{
int i, rv;
/* oldest undelivered cb is callbacks first entry */
*cb = list_first_entry_or_null(&lkb->lkb_callbacks,
struct dlm_callback, list);
if (!*cb)
return DLM_DEQUEUE_CALLBACK_EMPTY;
*resid = 0;
/* remove it from callbacks so shift others down */
list_del(&(*cb)->list);
if (list_empty(&lkb->lkb_callbacks))
return DLM_DEQUEUE_CALLBACK_LAST;
if (!lkb->lkb_callbacks[0].seq) {
rv = -ENOENT;
goto out;
}
/* oldest undelivered cb is callbacks[0] */
memcpy(cb, &lkb->lkb_callbacks[0], sizeof(struct dlm_callback));
memset(&lkb->lkb_callbacks[0], 0, sizeof(struct dlm_callback));
/* shift others down */
for (i = 1; i < DLM_CALLBACKS_SIZE; i++) {
if (!lkb->lkb_callbacks[i].seq)
break;
memcpy(&lkb->lkb_callbacks[i-1], &lkb->lkb_callbacks[i],
sizeof(struct dlm_callback));
memset(&lkb->lkb_callbacks[i], 0, sizeof(struct dlm_callback));
(*resid)++;
}
/* if cb is a bast, it should be skipped if the blocking mode is
compatible with the last granted mode */
if ((cb->flags & DLM_CB_BAST) && lkb->lkb_last_cast.seq) {
if (dlm_modes_compat(cb->mode, lkb->lkb_last_cast.mode)) {
cb->flags |= DLM_CB_SKIP;
log_debug(ls, "skip %x bast %llu mode %d "
"for cast %llu mode %d",
lkb->lkb_id,
(unsigned long long)cb->seq,
cb->mode,
(unsigned long long)lkb->lkb_last_cast.seq,
lkb->lkb_last_cast.mode);
rv = 0;
goto out;
}
}
if (cb->flags & DLM_CB_CAST) {
memcpy(&lkb->lkb_last_cast, cb, sizeof(struct dlm_callback));
lkb->lkb_last_cast_time = ktime_get();
}
if (cb->flags & DLM_CB_BAST) {
memcpy(&lkb->lkb_last_bast, cb, sizeof(struct dlm_callback));
lkb->lkb_last_bast_time = ktime_get();
}
rv = 0;
out:
return rv;
return DLM_DEQUEUE_CALLBACK_SUCCESS;
}
void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status,
uint32_t sbflags)
{
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
uint64_t new_seq, prev_seq;
int rv;
spin_lock(&dlm_cb_seq_spin);
new_seq = ++dlm_cb_seq;
if (!dlm_cb_seq)
new_seq = ++dlm_cb_seq;
spin_unlock(&dlm_cb_seq_spin);
if (lkb->lkb_flags & DLM_IFL_USER) {
dlm_user_add_ast(lkb, flags, mode, status, sbflags, new_seq);
dlm_user_add_ast(lkb, flags, mode, status, sbflags);
return;
}
mutex_lock(&lkb->lkb_cb_mutex);
prev_seq = lkb->lkb_callbacks[0].seq;
rv = dlm_add_lkb_callback(lkb, flags, mode, status, sbflags, new_seq);
if (rv < 0)
goto out;
if (!prev_seq) {
spin_lock(&lkb->lkb_cb_lock);
rv = dlm_enqueue_lkb_callback(lkb, flags, mode, status, sbflags);
switch (rv) {
case DLM_ENQUEUE_CALLBACK_NEED_SCHED:
kref_get(&lkb->lkb_ref);
mutex_lock(&ls->ls_cb_mutex);
spin_lock(&ls->ls_cb_lock);
if (test_bit(LSFL_CB_DELAY, &ls->ls_flags)) {
list_add(&lkb->lkb_cb_list, &ls->ls_cb_delay);
} else {
queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work);
}
mutex_unlock(&ls->ls_cb_mutex);
spin_unlock(&ls->ls_cb_lock);
break;
case DLM_ENQUEUE_CALLBACK_FAILURE:
WARN_ON_ONCE(1);
break;
case DLM_ENQUEUE_CALLBACK_SUCCESS:
break;
default:
WARN_ON_ONCE(1);
break;
}
out:
mutex_unlock(&lkb->lkb_cb_mutex);
spin_unlock(&lkb->lkb_cb_lock);
}
void dlm_callback_work(struct work_struct *work)
@ -218,53 +177,46 @@ void dlm_callback_work(struct work_struct *work)
struct dlm_ls *ls = lkb->lkb_resource->res_ls;
void (*castfn) (void *astparam);
void (*bastfn) (void *astparam, int mode);
struct dlm_callback callbacks[DLM_CALLBACKS_SIZE];
int i, rv, resid;
struct dlm_callback *cb;
int rv;
memset(&callbacks, 0, sizeof(callbacks));
spin_lock(&lkb->lkb_cb_lock);
rv = dlm_dequeue_lkb_callback(lkb, &cb);
spin_unlock(&lkb->lkb_cb_lock);
mutex_lock(&lkb->lkb_cb_mutex);
if (!lkb->lkb_callbacks[0].seq) {
/* no callback work exists, shouldn't happen */
log_error(ls, "dlm_callback_work %x no work", lkb->lkb_id);
dlm_print_lkb(lkb);
dlm_dump_lkb_callbacks(lkb);
}
if (WARN_ON_ONCE(rv == DLM_DEQUEUE_CALLBACK_EMPTY))
goto out;
for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
rv = dlm_rem_lkb_callback(ls, lkb, &callbacks[i], &resid);
if (rv < 0)
break;
}
for (;;) {
castfn = lkb->lkb_astfn;
bastfn = lkb->lkb_bastfn;
if (resid) {
/* cbs remain, loop should have removed all, shouldn't happen */
log_error(ls, "dlm_callback_work %x resid %d", lkb->lkb_id,
resid);
dlm_print_lkb(lkb);
dlm_dump_lkb_callbacks(lkb);
}
mutex_unlock(&lkb->lkb_cb_mutex);
castfn = lkb->lkb_astfn;
bastfn = lkb->lkb_bastfn;
for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
if (!callbacks[i].seq)
break;
if (callbacks[i].flags & DLM_CB_SKIP) {
continue;
} else if (callbacks[i].flags & DLM_CB_BAST) {
trace_dlm_bast(ls, lkb, callbacks[i].mode);
bastfn(lkb->lkb_astparam, callbacks[i].mode);
} else if (callbacks[i].flags & DLM_CB_CAST) {
lkb->lkb_lksb->sb_status = callbacks[i].sb_status;
lkb->lkb_lksb->sb_flags = callbacks[i].sb_flags;
if (cb->flags & DLM_CB_BAST) {
trace_dlm_bast(ls, lkb, cb->mode);
lkb->lkb_last_bast_time = ktime_get();
lkb->lkb_last_bast_mode = cb->mode;
bastfn(lkb->lkb_astparam, cb->mode);
} else if (cb->flags & DLM_CB_CAST) {
lkb->lkb_lksb->sb_status = cb->sb_status;
lkb->lkb_lksb->sb_flags = cb->sb_flags;
trace_dlm_ast(ls, lkb);
lkb->lkb_last_cast_time = ktime_get();
castfn(lkb->lkb_astparam);
}
kref_put(&cb->ref, dlm_release_callback);
spin_lock(&lkb->lkb_cb_lock);
rv = dlm_dequeue_lkb_callback(lkb, &cb);
if (rv == DLM_DEQUEUE_CALLBACK_EMPTY) {
lkb->lkb_flags &= ~DLM_IFL_CB_PENDING;
spin_unlock(&lkb->lkb_cb_lock);
break;
}
spin_unlock(&lkb->lkb_cb_lock);
}
out:
/* undo kref_get from dlm_add_callback, may cause lkb to be freed */
dlm_put_lkb(lkb);
}
@ -289,9 +241,9 @@ void dlm_callback_stop(struct dlm_ls *ls)
void dlm_callback_suspend(struct dlm_ls *ls)
{
if (ls->ls_callback_wq) {
mutex_lock(&ls->ls_cb_mutex);
spin_lock(&ls->ls_cb_lock);
set_bit(LSFL_CB_DELAY, &ls->ls_flags);
mutex_unlock(&ls->ls_cb_mutex);
spin_unlock(&ls->ls_cb_lock);
flush_workqueue(ls->ls_callback_wq);
}
@ -308,10 +260,8 @@ void dlm_callback_resume(struct dlm_ls *ls)
if (!ls->ls_callback_wq)
return;
clear_bit(LSFL_CB_DELAY, &ls->ls_flags);
more:
mutex_lock(&ls->ls_cb_mutex);
spin_lock(&ls->ls_cb_lock);
list_for_each_entry_safe(lkb, safe, &ls->ls_cb_delay, lkb_cb_list) {
list_del_init(&lkb->lkb_cb_list);
queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work);
@ -320,7 +270,9 @@ more:
break;
}
empty = list_empty(&ls->ls_cb_delay);
mutex_unlock(&ls->ls_cb_mutex);
if (empty)
clear_bit(LSFL_CB_DELAY, &ls->ls_flags);
spin_unlock(&ls->ls_cb_lock);
sum += count;
if (!empty) {

17
fs/dlm/ast.h
View File

@ -11,13 +11,22 @@
#ifndef __ASTD_DOT_H__
#define __ASTD_DOT_H__
int dlm_add_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags, uint64_t seq);
int dlm_rem_lkb_callback(struct dlm_ls *ls, struct dlm_lkb *lkb,
struct dlm_callback *cb, int *resid);
#define DLM_ENQUEUE_CALLBACK_NEED_SCHED 1
#define DLM_ENQUEUE_CALLBACK_SUCCESS 0
#define DLM_ENQUEUE_CALLBACK_FAILURE -1
int dlm_enqueue_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags);
#define DLM_DEQUEUE_CALLBACK_EMPTY 2
#define DLM_DEQUEUE_CALLBACK_LAST 1
#define DLM_DEQUEUE_CALLBACK_SUCCESS 0
int dlm_dequeue_lkb_callback(struct dlm_lkb *lkb, struct dlm_callback **cb);
void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status,
uint32_t sbflags);
void dlm_callback_set_last_ptr(struct dlm_callback **from,
struct dlm_callback *to);
void dlm_release_callback(struct kref *ref);
void dlm_purge_lkb_callbacks(struct dlm_lkb *lkb);
void dlm_callback_work(struct work_struct *work);
int dlm_callback_start(struct dlm_ls *ls);
void dlm_callback_stop(struct dlm_ls *ls);

4
fs/dlm/config.c
View File

@ -183,7 +183,7 @@ static int dlm_check_protocol_and_dlm_running(unsigned int x)
return -EINVAL;
}
if (dlm_allow_conn)
if (dlm_lowcomms_is_running())
return -EBUSY;
return 0;
@ -194,7 +194,7 @@ static int dlm_check_zero_and_dlm_running(unsigned int x)
if (!x)
return -EINVAL;
if (dlm_allow_conn)
if (dlm_lowcomms_is_running())
return -EBUSY;
return 0;

2
fs/dlm/debug_fs.c
View File

@ -246,7 +246,7 @@ static void print_format3_lock(struct seq_file *s, struct dlm_lkb *lkb,
lkb->lkb_status,
lkb->lkb_grmode,
lkb->lkb_rqmode,
lkb->lkb_last_bast.mode,
lkb->lkb_last_bast_mode,
rsb_lookup,
lkb->lkb_wait_type,
lkb->lkb_lvbseq,

25
fs/dlm/dlm_internal.h
View File

@ -211,6 +211,7 @@ struct dlm_args {
#endif
#define DLM_IFL_DEADLOCK_CANCEL 0x01000000
#define DLM_IFL_STUB_MS 0x02000000 /* magic number for m_flags */
#define DLM_IFL_CB_PENDING 0x04000000
/* least significant 2 bytes are message changed, they are full transmitted
* but at receive side only the 2 bytes LSB will be set.
*
@ -222,18 +223,17 @@ struct dlm_args {
#define DLM_IFL_USER 0x00000001
#define DLM_IFL_ORPHAN 0x00000002
#define DLM_CALLBACKS_SIZE 6
#define DLM_CB_CAST 0x00000001
#define DLM_CB_BAST 0x00000002
#define DLM_CB_SKIP 0x00000004
struct dlm_callback {
uint64_t seq;
uint32_t flags; /* DLM_CBF_ */
int sb_status; /* copy to lksb status */
uint8_t sb_flags; /* copy to lksb flags */
int8_t mode; /* rq mode of bast, gr mode of cast */
struct list_head list;
struct kref ref;
};
struct dlm_lkb {
@ -268,12 +268,13 @@ struct dlm_lkb {
unsigned long lkb_timeout_cs;
#endif
struct mutex lkb_cb_mutex;
spinlock_t lkb_cb_lock;
struct work_struct lkb_cb_work;
struct list_head lkb_cb_list; /* for ls_cb_delay or proc->asts */
struct dlm_callback lkb_callbacks[DLM_CALLBACKS_SIZE];
struct dlm_callback lkb_last_cast;
struct dlm_callback lkb_last_bast;
struct list_head lkb_callbacks;
struct dlm_callback *lkb_last_cast;
struct dlm_callback *lkb_last_cb;
int lkb_last_bast_mode;
ktime_t lkb_last_cast_time; /* for debugging */
ktime_t lkb_last_bast_time; /* for debugging */
@ -591,11 +592,7 @@ struct dlm_ls {
int ls_new_rsb_count;
struct list_head ls_new_rsb; /* new rsb structs */
spinlock_t ls_remove_spin;
wait_queue_head_t ls_remove_wait;
char ls_remove_name[DLM_RESNAME_MAXLEN+1];
char *ls_remove_names[DLM_REMOVE_NAMES_MAX];
int ls_remove_len;
int ls_remove_lens[DLM_REMOVE_NAMES_MAX];
struct list_head ls_nodes; /* current nodes in ls */
@ -631,7 +628,7 @@ struct dlm_ls {
/* recovery related */
struct mutex ls_cb_mutex;
spinlock_t ls_cb_lock;
struct list_head ls_cb_delay; /* save for queue_work later */
struct timer_list ls_timer;
struct task_struct *ls_recoverd_task;
@ -670,7 +667,7 @@ struct dlm_ls {
void *ls_ops_arg;
int ls_namelen;
char ls_name[1];
char ls_name[DLM_LOCKSPACE_LEN + 1];
};
/*

190
fs/dlm/lock.c
View File

@ -1209,6 +1209,7 @@ static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
if (!lkb)
return -ENOMEM;
lkb->lkb_last_bast_mode = -1;
lkb->lkb_nodeid = -1;
lkb->lkb_grmode = DLM_LOCK_IV;
kref_init(&lkb->lkb_ref);
@ -1218,7 +1219,8 @@ static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
INIT_LIST_HEAD(&lkb->lkb_time_list);
#endif
INIT_LIST_HEAD(&lkb->lkb_cb_list);
mutex_init(&lkb->lkb_cb_mutex);
INIT_LIST_HEAD(&lkb->lkb_callbacks);
spin_lock_init(&lkb->lkb_cb_lock);
INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
idr_preload(GFP_NOFS);
@ -1587,37 +1589,6 @@ static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
return error;
}
/* If there's an rsb for the same resource being removed, ensure
* that the remove message is sent before the new lookup message.
*/
#define DLM_WAIT_PENDING_COND(ls, r) \
(ls->ls_remove_len && \
!rsb_cmp(r, ls->ls_remove_name, \
ls->ls_remove_len))
static void wait_pending_remove(struct dlm_rsb *r)
{
struct dlm_ls *ls = r->res_ls;
restart:
spin_lock(&ls->ls_remove_spin);
if (DLM_WAIT_PENDING_COND(ls, r)) {
log_debug(ls, "delay lookup for remove dir %d %s",
r->res_dir_nodeid, r->res_name);
spin_unlock(&ls->ls_remove_spin);
wait_event(ls->ls_remove_wait, !DLM_WAIT_PENDING_COND(ls, r));
goto restart;
}
spin_unlock(&ls->ls_remove_spin);
}
/*
* ls_remove_spin protects ls_remove_name and ls_remove_len which are
* read by other threads in wait_pending_remove. ls_remove_names
* and ls_remove_lens are only used by the scan thread, so they do
* not need protection.
*/
static void shrink_bucket(struct dlm_ls *ls, int b)
{
struct rb_node *n, *next;
@ -1699,11 +1670,6 @@ static void shrink_bucket(struct dlm_ls *ls, int b)
* list and sending the removal. Keeping this gap small is
* important to keep us (the master node) from being out of sync
* with the remote dir node for very long.
*
* From the time the rsb is removed from toss until just after
* send_remove, the rsb name is saved in ls_remove_name. A new
* lookup checks this to ensure that a new lookup message for the
* same resource name is not sent just before the remove message.
*/
for (i = 0; i < remote_count; i++) {
@ -1750,22 +1716,8 @@ static void shrink_bucket(struct dlm_ls *ls, int b)
}
rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
/* block lookup of same name until we've sent remove */
spin_lock(&ls->ls_remove_spin);
ls->ls_remove_len = len;
memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
spin_unlock(&ls->ls_remove_spin);
spin_unlock(&ls->ls_rsbtbl[b].lock);
send_remove(r);
/* allow lookup of name again */
spin_lock(&ls->ls_remove_spin);
ls->ls_remove_len = 0;
memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
spin_unlock(&ls->ls_remove_spin);
wake_up(&ls->ls_remove_wait);
spin_unlock(&ls->ls_rsbtbl[b].lock);
dlm_free_rsb(r);
}
@ -2716,8 +2668,6 @@ static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
return 0;
}
wait_pending_remove(r);
r->res_first_lkid = lkb->lkb_id;
send_lookup(r, lkb);
return 1;
@ -3552,7 +3502,8 @@ int dlm_unlock(dlm_lockspace_t *lockspace,
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)
struct dlm_mhandle **mh_ret,
gfp_t allocation)
{
struct dlm_message *ms;
struct dlm_mhandle *mh;
@ -3562,7 +3513,7 @@ static int _create_message(struct dlm_ls *ls, int mb_len,
pass into midcomms_commit and a message buffer (mb) that we
write our data into */
mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, allocation, &mb);
if (!mh)
return -ENOBUFS;
@ -3584,7 +3535,8 @@ static int _create_message(struct dlm_ls *ls, int mb_len,
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)
struct dlm_mhandle **mh_ret,
gfp_t allocation)
{
int mb_len = sizeof(struct dlm_message);
@ -3605,15 +3557,16 @@ static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
}
return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
ms_ret, mh_ret);
ms_ret, mh_ret, allocation);
}
/* 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)
static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms,
const void *name, int namelen)
{
dlm_midcomms_commit_mhandle(mh);
dlm_midcomms_commit_mhandle(mh, name, namelen);
return 0;
}
@ -3673,13 +3626,13 @@ static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
if (error)
return error;
error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh, GFP_NOFS);
if (error)
goto fail;
send_args(r, lkb, ms);
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
if (error)
goto fail;
return 0;
@ -3734,7 +3687,8 @@ static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
to_nodeid = lkb->lkb_nodeid;
error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh,
GFP_NOFS);
if (error)
goto out;
@ -3742,7 +3696,7 @@ static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
ms->m_result = 0;
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
out:
return error;
}
@ -3755,7 +3709,8 @@ static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
to_nodeid = lkb->lkb_nodeid;
error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh,
GFP_NOFS);
if (error)
goto out;
@ -3763,7 +3718,7 @@ static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
ms->m_bastmode = cpu_to_le32(mode);
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
out:
return error;
}
@ -3780,13 +3735,14 @@ static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
if (error)
return error;
error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh,
GFP_NOFS);
if (error)
goto fail;
send_args(r, lkb, ms);
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
if (error)
goto fail;
return 0;
@ -3804,14 +3760,15 @@ static int send_remove(struct dlm_rsb *r)
to_nodeid = dlm_dir_nodeid(r);
error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh,
GFP_ATOMIC);
if (error)
goto out;
memcpy(ms->m_extra, r->res_name, r->res_length);
ms->m_hash = cpu_to_le32(r->res_hash);
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
out:
return error;
}
@ -3825,7 +3782,7 @@ static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
to_nodeid = lkb->lkb_nodeid;
error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh, GFP_NOFS);
if (error)
goto out;
@ -3833,7 +3790,7 @@ static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
ms->m_result = cpu_to_le32(to_dlm_errno(rv));
error = send_message(mh, ms);
error = send_message(mh, ms, r->res_name, r->res_length);
out:
return error;
}
@ -3866,7 +3823,8 @@ static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
struct dlm_mhandle *mh;
int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh,
GFP_NOFS);
if (error)
goto out;
@ -3874,7 +3832,7 @@ static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
ms->m_result = cpu_to_le32(to_dlm_errno(rv));
ms->m_nodeid = cpu_to_le32(ret_nodeid);
error = send_message(mh, ms);
error = send_message(mh, ms, ms_in->m_extra, receive_extralen(ms_in));
out:
return error;
}
@ -4044,66 +4002,6 @@ out:
return error;
}
static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
{
char name[DLM_RESNAME_MAXLEN + 1];
struct dlm_message *ms;
struct dlm_mhandle *mh;
struct dlm_rsb *r;
uint32_t hash, b;
int rv, dir_nodeid;
memset(name, 0, sizeof(name));
memcpy(name, ms_name, len);
hash = jhash(name, len, 0);
b = hash & (ls->ls_rsbtbl_size - 1);
dir_nodeid = dlm_hash2nodeid(ls, hash);
log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
spin_lock(&ls->ls_rsbtbl[b].lock);
rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
if (!rv) {
spin_unlock(&ls->ls_rsbtbl[b].lock);
log_error(ls, "repeat_remove on keep %s", name);
return;
}
rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
if (!rv) {
spin_unlock(&ls->ls_rsbtbl[b].lock);
log_error(ls, "repeat_remove on toss %s", name);
return;
}
/* use ls->remove_name2 to avoid conflict with shrink? */
spin_lock(&ls->ls_remove_spin);
ls->ls_remove_len = len;
memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
spin_unlock(&ls->ls_remove_spin);
spin_unlock(&ls->ls_rsbtbl[b].lock);
rv = _create_message(ls, sizeof(struct dlm_message) + len,
dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
if (rv)
goto out;
memcpy(ms->m_extra, name, len);
ms->m_hash = cpu_to_le32(hash);
send_message(mh, ms);
out:
spin_lock(&ls->ls_remove_spin);
ls->ls_remove_len = 0;
memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
spin_unlock(&ls->ls_remove_spin);
wake_up(&ls->ls_remove_wait);
}
static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
{
struct dlm_lkb *lkb;
@ -4173,25 +4071,11 @@ static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
ENOTBLK request failures when the lookup reply designating us
as master is delayed. */
/* We could repeatedly return -EBADR here if our send_remove() is
delayed in being sent/arriving/being processed on the dir node.
Another node would repeatedly lookup up the master, and the dir
node would continue returning our nodeid until our send_remove
took effect.
We send another remove message in case our previous send_remove
was lost/ignored/missed somehow. */
if (error != -ENOTBLK) {
log_limit(ls, "receive_request %x from %d %d",
le32_to_cpu(ms->m_lkid), from_nodeid, error);
}
if (namelen && error == -EBADR) {
send_repeat_remove(ls, ms->m_extra, namelen);
msleep(1000);
}
setup_stub_lkb(ls, ms);
send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
return error;
@ -6294,8 +6178,7 @@ void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
}
list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
memset(&lkb->lkb_callbacks, 0,
sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
dlm_purge_lkb_callbacks(lkb);
list_del_init(&lkb->lkb_cb_list);
dlm_put_lkb(lkb);
}
@ -6336,8 +6219,7 @@ static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
spin_lock(&proc->asts_spin);
list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
memset(&lkb->lkb_callbacks, 0,
sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
dlm_purge_lkb_callbacks(lkb);
list_del_init(&lkb->lkb_cb_list);
dlm_put_lkb(lkb);
}
@ -6368,13 +6250,13 @@ static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
int error;
error = _create_message(ls, sizeof(struct dlm_message), nodeid,
DLM_MSG_PURGE, &ms, &mh);
DLM_MSG_PURGE, &ms, &mh, GFP_NOFS);
if (error)
return error;
ms->m_nodeid = cpu_to_le32(nodeid);
ms->m_pid = cpu_to_le32(pid);
return send_message(mh, ms);
return send_message(mh, ms, NULL, 0);
}
int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,

14
fs/dlm/lockspace.c
View File

@ -17,7 +17,6 @@
#include "recoverd.h"
#include "dir.h"
#include "midcomms.h"
#include "lowcomms.h"
#include "config.h"
#include "memory.h"
#include "lock.h"
@ -391,7 +390,7 @@ static int threads_start(void)
/* Thread for sending/receiving messages for all lockspace's */
error = dlm_midcomms_start();
if (error) {
log_print("cannot start dlm lowcomms %d", error);
log_print("cannot start dlm midcomms %d", error);
goto scand_fail;
}
@ -473,7 +472,7 @@ static int new_lockspace(const char *name, const char *cluster,
error = -ENOMEM;
ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS);
ls = kzalloc(sizeof(*ls), GFP_NOFS);
if (!ls)
goto out;
memcpy(ls->ls_name, name, namelen);
@ -524,9 +523,6 @@ static int new_lockspace(const char *name, const char *cluster,
spin_lock_init(&ls->ls_rsbtbl[i].lock);
}
spin_lock_init(&ls->ls_remove_spin);
init_waitqueue_head(&ls->ls_remove_wait);
for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1,
GFP_KERNEL);
@ -567,7 +563,7 @@ static int new_lockspace(const char *name, const char *cluster,
init_completion(&ls->ls_recovery_done);
ls->ls_recovery_result = -1;
mutex_init(&ls->ls_cb_mutex);
spin_lock_init(&ls->ls_cb_lock);
INIT_LIST_HEAD(&ls->ls_cb_delay);
ls->ls_recoverd_task = NULL;
@ -726,7 +722,7 @@ static int __dlm_new_lockspace(const char *name, const char *cluster,
if (!ls_count) {
dlm_scand_stop();
dlm_midcomms_shutdown();
dlm_lowcomms_stop();
dlm_midcomms_stop();
}
out:
mutex_unlock(&ls_lock);
@ -929,7 +925,7 @@ int dlm_release_lockspace(void *lockspace, int force)
if (!error)
ls_count--;
if (!ls_count)
dlm_lowcomms_stop();
dlm_midcomms_stop();
mutex_unlock(&ls_lock);
return error;

1558
fs/dlm/lowcomms.c
View File

File diff suppressed because it is too large Load Diff

6
fs/dlm/lowcomms.h
View File

@ -29,12 +29,14 @@ static inline int nodeid_hash(int nodeid)
return nodeid & (CONN_HASH_SIZE-1);
}
/* switch to check if dlm is running */
extern int dlm_allow_conn;
/* check if dlm is running */
bool dlm_lowcomms_is_running(void);
int dlm_lowcomms_start(void);
void dlm_lowcomms_shutdown(void);
void dlm_lowcomms_shutdown_node(int nodeid, bool force);
void dlm_lowcomms_stop(void);
void dlm_lowcomms_init(void);
void dlm_lowcomms_exit(void);
int dlm_lowcomms_close(int nodeid);
struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation,

7
fs/dlm/main.c
View File

@ -17,7 +17,7 @@
#include "user.h"
#include "memory.h"
#include "config.h"
#include "lowcomms.h"
#include "midcomms.h"
#define CREATE_TRACE_POINTS
#include <trace/events/dlm.h>
@ -30,6 +30,8 @@ static int __init init_dlm(void)
if (error)
goto out;
dlm_midcomms_init();
error = dlm_lockspace_init();
if (error)
goto out_mem;
@ -66,6 +68,7 @@ static int __init init_dlm(void)
out_lockspace:
dlm_lockspace_exit();
out_mem:
dlm_midcomms_exit();
dlm_memory_exit();
out:
return error;
@ -79,7 +82,7 @@ static void __exit exit_dlm(void)
dlm_config_exit();
dlm_memory_exit();
dlm_lockspace_exit();
dlm_lowcomms_exit();
dlm_midcomms_exit();
dlm_unregister_debugfs();
}

5
fs/dlm/member.c
View File

@ -573,7 +573,10 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
node = &rv->nodes[i];
if (dlm_is_member(ls, node->nodeid))
continue;
dlm_add_member(ls, node);
error = dlm_add_member(ls, node);
if (error)
return error;
log_rinfo(ls, "add member %d", node->nodeid);
}

30
fs/dlm/memory.c
View File

@ -14,12 +14,14 @@
#include "lowcomms.h"
#include "config.h"
#include "memory.h"
#include "ast.h"
static struct kmem_cache *writequeue_cache;
static struct kmem_cache *mhandle_cache;
static struct kmem_cache *msg_cache;
static struct kmem_cache *lkb_cache;
static struct kmem_cache *rsb_cache;
static struct kmem_cache *cb_cache;
int __init dlm_memory_init(void)
@ -46,8 +48,16 @@ int __init dlm_memory_init(void)
if (!rsb_cache)
goto rsb;
cb_cache = kmem_cache_create("dlm_cb", sizeof(struct dlm_callback),
__alignof__(struct dlm_callback), 0,
NULL);
if (!rsb_cache)
goto cb;
return 0;
cb:
kmem_cache_destroy(rsb_cache);
rsb:
kmem_cache_destroy(msg_cache);
msg:
@ -67,6 +77,7 @@ void dlm_memory_exit(void)
kmem_cache_destroy(msg_cache);
kmem_cache_destroy(lkb_cache);
kmem_cache_destroy(rsb_cache);
kmem_cache_destroy(cb_cache);
}
char *dlm_allocate_lvb(struct dlm_ls *ls)
@ -115,12 +126,17 @@ void dlm_free_lkb(struct dlm_lkb *lkb)
kfree(ua);
}
}
/* drop references if they are set */
dlm_callback_set_last_ptr(&lkb->lkb_last_cast, NULL);
dlm_callback_set_last_ptr(&lkb->lkb_last_cb, NULL);
kmem_cache_free(lkb_cache, lkb);
}
struct dlm_mhandle *dlm_allocate_mhandle(void)
struct dlm_mhandle *dlm_allocate_mhandle(gfp_t allocation)
{
return kmem_cache_alloc(mhandle_cache, GFP_NOFS);
return kmem_cache_alloc(mhandle_cache, allocation);
}
void dlm_free_mhandle(struct dlm_mhandle *mhandle)
@ -147,3 +163,13 @@ void dlm_free_msg(struct dlm_msg *msg)
{
kmem_cache_free(msg_cache, msg);
}
struct dlm_callback *dlm_allocate_cb(void)
{
return kmem_cache_alloc(cb_cache, GFP_ATOMIC);
}
void dlm_free_cb(struct dlm_callback *cb)
{
kmem_cache_free(cb_cache, cb);
}

4
fs/dlm/memory.h
View File

@ -20,12 +20,14 @@ struct dlm_lkb *dlm_allocate_lkb(struct dlm_ls *ls);
void dlm_free_lkb(struct dlm_lkb *l);
char *dlm_allocate_lvb(struct dlm_ls *ls);
void dlm_free_lvb(char *l);
struct dlm_mhandle *dlm_allocate_mhandle(void);
struct dlm_mhandle *dlm_allocate_mhandle(gfp_t allocation);
void dlm_free_mhandle(struct dlm_mhandle *mhandle);
struct writequeue_entry *dlm_allocate_writequeue(void);
void dlm_free_writequeue(struct writequeue_entry *writequeue);
struct dlm_msg *dlm_allocate_msg(gfp_t allocation);
void dlm_free_msg(struct dlm_msg *msg);
struct dlm_callback *dlm_allocate_cb(void);
void dlm_free_cb(struct dlm_callback *cb);
#endif /* __MEMORY_DOT_H__ */

141
fs/dlm/midcomms.c
View File

@ -132,6 +132,7 @@
*/
#define DLM_DEBUG_FENCE_TERMINATION 0
#include <trace/events/dlm.h>
#include <net/tcp.h>
#include "dlm_internal.h"
@ -194,7 +195,7 @@ struct midcomms_node {
};
struct dlm_mhandle {
const struct dlm_header *inner_hd;
const union dlm_packet *inner_p;
struct midcomms_node *node;
struct dlm_opts *opts;
struct dlm_msg *msg;
@ -305,11 +306,11 @@ static void dlm_send_queue_flush(struct midcomms_node *node)
pr_debug("flush midcomms send queue of node %d\n", node->nodeid);
rcu_read_lock();
spin_lock(&node->send_queue_lock);
spin_lock_bh(&node->send_queue_lock);
list_for_each_entry_rcu(mh, &node->send_queue, list) {
dlm_mhandle_delete(node, mh);
}
spin_unlock(&node->send_queue_lock);
spin_unlock_bh(&node->send_queue_lock);
rcu_read_unlock();
}
@ -415,7 +416,7 @@ static int dlm_send_fin(struct midcomms_node *node,
m_header->h_cmd = DLM_FIN;
pr_debug("sending fin msg to node %d\n", node->nodeid);
dlm_midcomms_commit_mhandle(mh);
dlm_midcomms_commit_mhandle(mh, NULL, 0);
set_bit(DLM_NODE_FLAG_STOP_TX, &node->flags);
return 0;
@ -436,7 +437,7 @@ static void dlm_receive_ack(struct midcomms_node *node, uint32_t seq)
}
}
spin_lock(&node->send_queue_lock);
spin_lock_bh(&node->send_queue_lock);
list_for_each_entry_rcu(mh, &node->send_queue, list) {
if (before(mh->seq, seq)) {
dlm_mhandle_delete(node, mh);
@ -445,7 +446,7 @@ static void dlm_receive_ack(struct midcomms_node *node, uint32_t seq)
break;
}
}
spin_unlock(&node->send_queue_lock);
spin_unlock_bh(&node->send_queue_lock);
rcu_read_unlock();
}
@ -468,12 +469,26 @@ static void dlm_pas_fin_ack_rcv(struct midcomms_node *node)
spin_unlock(&node->state_lock);
log_print("%s: unexpected state: %d\n",
__func__, node->state);
WARN_ON(1);
WARN_ON_ONCE(1);
return;
}
spin_unlock(&node->state_lock);
}
static void dlm_receive_buffer_3_2_trace(uint32_t seq, union dlm_packet *p)
{
switch (p->header.h_cmd) {
case DLM_MSG:
trace_dlm_recv_message(dlm_our_nodeid(), seq, &p->message);
break;
case DLM_RCOM:
trace_dlm_recv_rcom(dlm_our_nodeid(), seq, &p->rcom);
break;
default:
break;
}
}
static void dlm_midcomms_receive_buffer(union dlm_packet *p,
struct midcomms_node *node,
uint32_t seq)
@ -525,7 +540,7 @@ static void dlm_midcomms_receive_buffer(union dlm_packet *p,
spin_unlock(&node->state_lock);
log_print("%s: unexpected state: %d\n",
__func__, node->state);
WARN_ON(1);
WARN_ON_ONCE(1);
return;
}
spin_unlock(&node->state_lock);
@ -533,7 +548,8 @@ static void dlm_midcomms_receive_buffer(union dlm_packet *p,
set_bit(DLM_NODE_FLAG_STOP_RX, &node->flags);
break;
default:
WARN_ON(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags));
WARN_ON_ONCE(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags));
dlm_receive_buffer_3_2_trace(seq, p);
dlm_receive_buffer(p, node->nodeid);
set_bit(DLM_NODE_ULP_DELIVERED, &node->flags);
break;
@ -754,7 +770,7 @@ static void dlm_midcomms_receive_buffer_3_2(union dlm_packet *p, int nodeid)
goto out;
}
WARN_ON(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags));
WARN_ON_ONCE(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags));
dlm_receive_buffer(p, nodeid);
break;
case DLM_OPTS:
@ -874,12 +890,7 @@ static void dlm_midcomms_receive_buffer_3_1(union dlm_packet *p, int nodeid)
dlm_receive_buffer(p, nodeid);
}
/*
* Called from the low-level comms layer to process a buffer of
* commands.
*/
int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len)
int dlm_validate_incoming_buffer(int nodeid, unsigned char *buf, int len)
{
const unsigned char *ptr = buf;
const struct dlm_header *hd;
@ -914,6 +925,32 @@ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len)
if (msglen > len)
break;
ret += msglen;
len -= msglen;
ptr += msglen;
}
return ret;
}
/*
* Called from the low-level comms layer to process a buffer of
* commands.
*/
int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len)
{
const unsigned char *ptr = buf;
const struct dlm_header *hd;
uint16_t msglen;
int ret = 0;
while (len >= sizeof(struct dlm_header)) {
hd = (struct dlm_header *)ptr;
msglen = le16_to_cpu(hd->h_length);
if (msglen > len)
break;
switch (hd->h_version) {
case cpu_to_le32(DLM_VERSION_3_1):
dlm_midcomms_receive_buffer_3_1((union dlm_packet *)ptr, nodeid);
@ -1030,9 +1067,9 @@ static void midcomms_new_msg_cb(void *data)
atomic_inc(&mh->node->send_queue_cnt);
spin_lock(&mh->node->send_queue_lock);
spin_lock_bh(&mh->node->send_queue_lock);
list_add_tail_rcu(&mh->list, &mh->node->send_queue);
spin_unlock(&mh->node->send_queue_lock);
spin_unlock_bh(&mh->node->send_queue_lock);
mh->seq = mh->node->seq_send++;
}
@ -1055,7 +1092,7 @@ static struct dlm_msg *dlm_midcomms_get_msg_3_2(struct dlm_mhandle *mh, int node
dlm_fill_opts_header(opts, len, mh->seq);
*ppc += sizeof(*opts);
mh->inner_hd = (const struct dlm_header *)*ppc;
mh->inner_p = (const union dlm_packet *)*ppc;
return msg;
}
@ -1079,9 +1116,9 @@ struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len,
}
/* this is a bug, however we going on and hope it will be resolved */
WARN_ON(test_bit(DLM_NODE_FLAG_STOP_TX, &node->flags));
WARN_ON_ONCE(test_bit(DLM_NODE_FLAG_STOP_TX, &node->flags));
mh = dlm_allocate_mhandle();
mh = dlm_allocate_mhandle(allocation);
if (!mh)
goto err;
@ -1111,7 +1148,7 @@ struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len,
break;
default:
dlm_free_mhandle(mh);
WARN_ON(1);
WARN_ON_ONCE(1);
goto err;
}
@ -1130,11 +1167,32 @@ err:
}
#endif
static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh)
static void dlm_midcomms_commit_msg_3_2_trace(const struct dlm_mhandle *mh,
const void *name, int namelen)
{
switch (mh->inner_p->header.h_cmd) {
case DLM_MSG:
trace_dlm_send_message(mh->node->nodeid, mh->seq,
&mh->inner_p->message,
name, namelen);
break;
case DLM_RCOM:
trace_dlm_send_rcom(mh->node->nodeid, mh->seq,
&mh->inner_p->rcom);
break;
default:
/* nothing to trace */
break;
}
}
static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh,
const void *name, int namelen)
{
/* nexthdr chain for fast lookup */
mh->opts->o_nextcmd = mh->inner_hd->h_cmd;
mh->opts->o_nextcmd = mh->inner_p->header.h_cmd;
mh->committed = true;
dlm_midcomms_commit_msg_3_2_trace(mh, name, namelen);
dlm_lowcomms_commit_msg(mh->msg);
}
@ -1142,8 +1200,10 @@ static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh)
* dlm_midcomms_get_mhandle
*/
#ifndef __CHECKER__
void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh)
void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh,
const void *name, int namelen)
{
switch (mh->node->version) {
case DLM_VERSION_3_1:
srcu_read_unlock(&nodes_srcu, mh->idx);
@ -1154,25 +1214,40 @@ void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh)
dlm_free_mhandle(mh);
break;
case DLM_VERSION_3_2:
dlm_midcomms_commit_msg_3_2(mh);
dlm_midcomms_commit_msg_3_2(mh, name, namelen);
srcu_read_unlock(&nodes_srcu, mh->idx);
break;
default:
srcu_read_unlock(&nodes_srcu, mh->idx);
WARN_ON(1);
WARN_ON_ONCE(1);
break;
}
}
#endif
int dlm_midcomms_start(void)
{
return dlm_lowcomms_start();
}
void dlm_midcomms_stop(void)
{
dlm_lowcomms_stop();
}
void dlm_midcomms_init(void)
{
int i;
for (i = 0; i < CONN_HASH_SIZE; i++)
INIT_HLIST_HEAD(&node_hash[i]);
return dlm_lowcomms_start();
dlm_lowcomms_init();
}
void dlm_midcomms_exit(void)
{
dlm_lowcomms_exit();
}
static void dlm_act_fin_ack_rcv(struct midcomms_node *node)
@ -1201,7 +1276,7 @@ static void dlm_act_fin_ack_rcv(struct midcomms_node *node)
spin_unlock(&node->state_lock);
log_print("%s: unexpected state: %d\n",
__func__, node->state);
WARN_ON(1);
WARN_ON_ONCE(1);
return;
}
spin_unlock(&node->state_lock);
@ -1319,7 +1394,7 @@ static void midcomms_node_release(struct rcu_head *rcu)
{
struct midcomms_node *node = container_of(rcu, struct midcomms_node, rcu);
WARN_ON(atomic_read(&node->send_queue_cnt));
WARN_ON_ONCE(atomic_read(&node->send_queue_cnt));
kfree(node);
}
@ -1372,11 +1447,13 @@ static void midcomms_shutdown(struct midcomms_node *node)
pr_debug("active shutdown timed out for node %d with state %s\n",
node->nodeid, dlm_state_str(node->state));
midcomms_node_reset(node);
dlm_lowcomms_shutdown_node(node->nodeid, true);
return;
}
pr_debug("active shutdown done for node %d with state %s\n",
node->nodeid, dlm_state_str(node->state));
dlm_lowcomms_shutdown_node(node->nodeid, false);
}
void dlm_midcomms_shutdown(void)
@ -1384,6 +1461,8 @@ void dlm_midcomms_shutdown(void)
struct midcomms_node *node;
int i, idx;
dlm_lowcomms_shutdown();
mutex_lock(&close_lock);
idx = srcu_read_lock(&nodes_srcu);
for (i = 0; i < CONN_HASH_SIZE; i++) {
@ -1401,8 +1480,6 @@ void dlm_midcomms_shutdown(void)
}
srcu_read_unlock(&nodes_srcu, idx);
mutex_unlock(&close_lock);
dlm_lowcomms_shutdown();
}
int dlm_midcomms_close(int nodeid)

7
fs/dlm/midcomms.h
View File

@ -14,12 +14,17 @@
struct midcomms_node;
int dlm_validate_incoming_buffer(int nodeid, unsigned char *buf, int len);
int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int buflen);
struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len,
gfp_t allocation, char **ppc);
void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh);
void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh, const void *name,
int namelen);
int dlm_midcomms_close(int nodeid);
int dlm_midcomms_start(void);
void dlm_midcomms_stop(void);
void dlm_midcomms_init(void);
void dlm_midcomms_exit(void);
void dlm_midcomms_shutdown(void);
void dlm_midcomms_add_member(int nodeid);
void dlm_midcomms_remove_member(int nodeid);

4
fs/dlm/rcom.c
View File

@ -91,7 +91,7 @@ static int create_rcom_stateless(struct dlm_ls *ls, int to_nodeid, int type,
static void send_rcom(struct dlm_mhandle *mh, struct dlm_rcom *rc)
{
dlm_midcomms_commit_mhandle(mh);
dlm_midcomms_commit_mhandle(mh, NULL, 0);
}
static void send_rcom_stateless(struct dlm_msg *msg, struct dlm_rcom *rc)
@ -516,7 +516,7 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
rf = (struct rcom_config *) rc->rc_buf;
rf->rf_lvblen = cpu_to_le32(~0U);
dlm_midcomms_commit_mhandle(mh);
dlm_midcomms_commit_mhandle(mh, NULL, 0);
return 0;
}

3
fs/dlm/requestqueue.c
View File

@ -44,7 +44,8 @@ void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_message *ms)
e->recover_seq = ls->ls_recover_seq & 0xFFFFFFFF;
e->nodeid = nodeid;
memcpy(&e->request, ms, le16_to_cpu(ms->m_header.h_length));
memcpy(&e->request, ms, sizeof(*ms));
memcpy(&e->request.m_extra, ms->m_extra, length);
atomic_inc(&ls->ls_requestqueue_cnt);
mutex_lock(&ls->ls_requestqueue_mutex);

74
fs/dlm/user.c
View File

@ -25,6 +25,7 @@
#include "user.h"
#include "ast.h"
#include "config.h"
#include "memory.h"
static const char name_prefix[] = "dlm";
static const struct file_operations device_fops;
@ -175,7 +176,7 @@ static int lkb_is_endoflife(int mode, int status)
being removed and then remove that lkb from the orphans list and free it */
void dlm_user_add_ast(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags, uint64_t seq)
int status, uint32_t sbflags)
{
struct dlm_ls *ls;
struct dlm_user_args *ua;
@ -209,16 +210,22 @@ void dlm_user_add_ast(struct dlm_lkb *lkb, uint32_t flags, int mode,
spin_lock(&proc->asts_spin);
rv = dlm_add_lkb_callback(lkb, flags, mode, status, sbflags, seq);
if (rv < 0) {
rv = dlm_enqueue_lkb_callback(lkb, flags, mode, status, sbflags);
switch (rv) {
case DLM_ENQUEUE_CALLBACK_FAILURE:
spin_unlock(&proc->asts_spin);
WARN_ON_ONCE(1);
goto out;
}
if (list_empty(&lkb->lkb_cb_list)) {
case DLM_ENQUEUE_CALLBACK_NEED_SCHED:
kref_get(&lkb->lkb_ref);
list_add_tail(&lkb->lkb_cb_list, &proc->asts);
wake_up_interruptible(&proc->wait);
break;
case DLM_ENQUEUE_CALLBACK_SUCCESS:
break;
default:
WARN_ON_ONCE(1);
break;
}
spin_unlock(&proc->asts_spin);
@ -800,8 +807,8 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
struct dlm_user_proc *proc = file->private_data;
struct dlm_lkb *lkb;
DECLARE_WAITQUEUE(wait, current);
struct dlm_callback cb;
int rv, resid, copy_lvb = 0;
struct dlm_callback *cb;
int rv, copy_lvb = 0;
int old_mode, new_mode;
if (count == sizeof(struct dlm_device_version)) {
@ -857,53 +864,58 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
without removing lkb_cb_list; so empty lkb_cb_list is always
consistent with empty lkb_callbacks */
lkb = list_entry(proc->asts.next, struct dlm_lkb, lkb_cb_list);
lkb = list_first_entry(&proc->asts, struct dlm_lkb, lkb_cb_list);
/* rem_lkb_callback sets a new lkb_last_cast */
old_mode = lkb->lkb_last_cast.mode;
old_mode = lkb->lkb_last_cast->mode;
rv = dlm_rem_lkb_callback(lkb->lkb_resource->res_ls, lkb, &cb, &resid);
if (rv < 0) {
rv = dlm_dequeue_lkb_callback(lkb, &cb);
switch (rv) {
case DLM_DEQUEUE_CALLBACK_EMPTY:
/* this shouldn't happen; lkb should have been removed from
list when resid was zero */
* list when last item was dequeued
*/
log_print("dlm_rem_lkb_callback empty %x", lkb->lkb_id);
list_del_init(&lkb->lkb_cb_list);
spin_unlock(&proc->asts_spin);
/* removes ref for proc->asts, may cause lkb to be freed */
dlm_put_lkb(lkb);
WARN_ON_ONCE(1);
goto try_another;
}
if (!resid)
case DLM_DEQUEUE_CALLBACK_LAST:
list_del_init(&lkb->lkb_cb_list);
lkb->lkb_flags &= ~DLM_IFL_CB_PENDING;
break;
case DLM_DEQUEUE_CALLBACK_SUCCESS:
break;
default:
WARN_ON_ONCE(1);
break;
}
spin_unlock(&proc->asts_spin);
if (cb.flags & DLM_CB_SKIP) {
/* removes ref for proc->asts, may cause lkb to be freed */
if (!resid)
dlm_put_lkb(lkb);
goto try_another;
}
if (cb->flags & DLM_CB_BAST) {
trace_dlm_bast(lkb->lkb_resource->res_ls, lkb, cb->mode);
} else if (cb->flags & DLM_CB_CAST) {
new_mode = cb->mode;
if (cb.flags & DLM_CB_BAST) {
trace_dlm_bast(lkb->lkb_resource->res_ls, lkb, cb.mode);
} else if (cb.flags & DLM_CB_CAST) {
new_mode = cb.mode;
if (!cb.sb_status && lkb->lkb_lksb->sb_lvbptr &&
if (!cb->sb_status && lkb->lkb_lksb->sb_lvbptr &&
dlm_lvb_operations[old_mode + 1][new_mode + 1])
copy_lvb = 1;
lkb->lkb_lksb->sb_status = cb.sb_status;
lkb->lkb_lksb->sb_flags = cb.sb_flags;
lkb->lkb_lksb->sb_status = cb->sb_status;
lkb->lkb_lksb->sb_flags = cb->sb_flags;
trace_dlm_ast(lkb->lkb_resource->res_ls, lkb);
}
rv = copy_result_to_user(lkb->lkb_ua,
test_bit(DLM_PROC_FLAGS_COMPAT, &proc->flags),
cb.flags, cb.mode, copy_lvb, buf, count);
cb->flags, cb->mode, copy_lvb, buf, count);
kref_put(&cb->ref, dlm_release_callback);
/* removes ref for proc->asts, may cause lkb to be freed */
if (!resid)
if (rv == DLM_DEQUEUE_CALLBACK_LAST)
dlm_put_lkb(lkb);
return rv;

2
fs/dlm/user.h
View File

@ -7,7 +7,7 @@
#define __USER_DOT_H__
void dlm_user_add_ast(struct dlm_lkb *lkb, uint32_t flags, int mode,
int status, uint32_t sbflags, uint64_t seq);
int status, uint32_t sbflags);
int dlm_user_init(void);
void dlm_user_exit(void);
int dlm_device_deregister(struct dlm_ls *ls);

303
include/trace/events/dlm.h
View File

@ -46,6 +46,56 @@
{ DLM_SBF_VALNOTVALID, "VALNOTVALID" }, \
{ DLM_SBF_ALTMODE, "ALTMODE" })
#define show_lkb_flags(flags) __print_flags(flags, "|", \
{ DLM_IFL_MSTCPY, "MSTCPY" }, \
{ DLM_IFL_RESEND, "RESEND" }, \
{ DLM_IFL_DEAD, "DEAD" }, \
{ DLM_IFL_OVERLAP_UNLOCK, "OVERLAP_UNLOCK" }, \
{ DLM_IFL_OVERLAP_CANCEL, "OVERLAP_CANCEL" }, \
{ DLM_IFL_ENDOFLIFE, "ENDOFLIFE" }, \
{ DLM_IFL_DEADLOCK_CANCEL, "DEADLOCK_CANCEL" }, \
{ DLM_IFL_STUB_MS, "STUB_MS" }, \
{ DLM_IFL_USER, "USER" }, \
{ DLM_IFL_ORPHAN, "ORPHAN" })
#define show_header_cmd(cmd) __print_symbolic(cmd, \
{ DLM_MSG, "MSG"}, \
{ DLM_RCOM, "RCOM"}, \
{ DLM_OPTS, "OPTS"}, \
{ DLM_ACK, "ACK"}, \
{ DLM_FIN, "FIN"})
#define show_message_version(version) __print_symbolic(version, \
{ DLM_VERSION_3_1, "3.1"}, \
{ DLM_VERSION_3_2, "3.2"})
#define show_message_type(type) __print_symbolic(type, \
{ DLM_MSG_REQUEST, "REQUEST"}, \
{ DLM_MSG_CONVERT, "CONVERT"}, \
{ DLM_MSG_UNLOCK, "UNLOCK"}, \
{ DLM_MSG_CANCEL, "CANCEL"}, \
{ DLM_MSG_REQUEST_REPLY, "REQUEST_REPLY"}, \
{ DLM_MSG_CONVERT_REPLY, "CONVERT_REPLY"}, \
{ DLM_MSG_UNLOCK_REPLY, "UNLOCK_REPLY"}, \
{ DLM_MSG_CANCEL_REPLY, "CANCEL_REPLY"}, \
{ DLM_MSG_GRANT, "GRANT"}, \
{ DLM_MSG_BAST, "BAST"}, \
{ DLM_MSG_LOOKUP, "LOOKUP"}, \
{ DLM_MSG_REMOVE, "REMOVE"}, \
{ DLM_MSG_LOOKUP_REPLY, "LOOKUP_REPLY"}, \
{ DLM_MSG_PURGE, "PURGE"})
#define show_rcom_type(type) __print_symbolic(type, \
{ DLM_RCOM_STATUS, "STATUS"}, \
{ DLM_RCOM_NAMES, "NAMES"}, \
{ DLM_RCOM_LOOKUP, "LOOKUP"}, \
{ DLM_RCOM_LOCK, "LOCK"}, \
{ DLM_RCOM_STATUS_REPLY, "STATUS_REPLY"}, \
{ DLM_RCOM_NAMES_REPLY, "NAMES_REPLY"}, \
{ DLM_RCOM_LOOKUP_REPLY, "LOOKUP_REPLY"}, \
{ DLM_RCOM_LOCK_REPLY, "LOCK_REPLY"})
/* note: we begin tracing dlm_lock_start() only if ls and lkb are found */
TRACE_EVENT(dlm_lock_start,
@ -290,6 +340,259 @@ TRACE_EVENT(dlm_unlock_end,
);
DECLARE_EVENT_CLASS(dlm_rcom_template,
TP_PROTO(uint32_t dst, uint32_t h_seq, const struct dlm_rcom *rc),
TP_ARGS(dst, h_seq, rc),
TP_STRUCT__entry(
__field(uint32_t, dst)
__field(uint32_t, h_seq)
__field(uint32_t, h_version)
__field(uint32_t, h_lockspace)
__field(uint32_t, h_nodeid)
__field(uint16_t, h_length)
__field(uint8_t, h_cmd)
__field(uint32_t, rc_type)
__field(int32_t, rc_result)
__field(uint64_t, rc_id)
__field(uint64_t, rc_seq)
__field(uint64_t, rc_seq_reply)
__dynamic_array(unsigned char, rc_buf,
le16_to_cpu(rc->rc_header.h_length) - sizeof(*rc))
),
TP_fast_assign(
__entry->dst = dst;
__entry->h_seq = h_seq;
__entry->h_version = le32_to_cpu(rc->rc_header.h_version);
__entry->h_lockspace = le32_to_cpu(rc->rc_header.u.h_lockspace);
__entry->h_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
__entry->h_length = le16_to_cpu(rc->rc_header.h_length);
__entry->h_cmd = rc->rc_header.h_cmd;
__entry->rc_type = le32_to_cpu(rc->rc_type);
__entry->rc_result = le32_to_cpu(rc->rc_result);
__entry->rc_id = le64_to_cpu(rc->rc_id);
__entry->rc_seq = le64_to_cpu(rc->rc_seq);
__entry->rc_seq_reply = le64_to_cpu(rc->rc_seq_reply);
memcpy(__get_dynamic_array(rc_buf), rc->rc_buf,
__get_dynamic_array_len(rc_buf));
),
TP_printk("dst=%u h_seq=%u h_version=%s h_lockspace=%u h_nodeid=%u "
"h_length=%u h_cmd=%s rc_type=%s rc_result=%d "
"rc_id=%llu rc_seq=%llu rc_seq_reply=%llu "
"rc_buf=0x%s", __entry->dst, __entry->h_seq,
show_message_version(__entry->h_version),
__entry->h_lockspace, __entry->h_nodeid, __entry->h_length,
show_header_cmd(__entry->h_cmd),
show_rcom_type(__entry->rc_type),
__entry->rc_result, __entry->rc_id, __entry->rc_seq,
__entry->rc_seq_reply,
__print_hex_str(__get_dynamic_array(rc_buf),
__get_dynamic_array_len(rc_buf)))
);
DEFINE_EVENT(dlm_rcom_template, dlm_send_rcom,
TP_PROTO(uint32_t dst, uint32_t h_seq, const struct dlm_rcom *rc),
TP_ARGS(dst, h_seq, rc));
DEFINE_EVENT(dlm_rcom_template, dlm_recv_rcom,
TP_PROTO(uint32_t dst, uint32_t h_seq, const struct dlm_rcom *rc),
TP_ARGS(dst, h_seq, rc));
TRACE_EVENT(dlm_send_message,
TP_PROTO(uint32_t dst, uint32_t h_seq, const struct dlm_message *ms,
const void *name, int namelen),
TP_ARGS(dst, h_seq, ms, name, namelen),
TP_STRUCT__entry(
__field(uint32_t, dst)
__field(uint32_t, h_seq)
__field(uint32_t, h_version)
__field(uint32_t, h_lockspace)
__field(uint32_t, h_nodeid)
__field(uint16_t, h_length)
__field(uint8_t, h_cmd)
__field(uint32_t, m_type)
__field(uint32_t, m_nodeid)
__field(uint32_t, m_pid)
__field(uint32_t, m_lkid)
__field(uint32_t, m_remid)
__field(uint32_t, m_parent_lkid)
__field(uint32_t, m_parent_remid)
__field(uint32_t, m_exflags)
__field(uint32_t, m_sbflags)
__field(uint32_t, m_flags)
__field(uint32_t, m_lvbseq)
__field(uint32_t, m_hash)
__field(int32_t, m_status)
__field(int32_t, m_grmode)
__field(int32_t, m_rqmode)
__field(int32_t, m_bastmode)
__field(int32_t, m_asts)
__field(int32_t, m_result)
__dynamic_array(unsigned char, m_extra,
le16_to_cpu(ms->m_header.h_length) - sizeof(*ms))
__dynamic_array(unsigned char, res_name, namelen)
),
TP_fast_assign(
__entry->dst = dst;
__entry->h_seq = h_seq;
__entry->h_version = le32_to_cpu(ms->m_header.h_version);
__entry->h_lockspace = le32_to_cpu(ms->m_header.u.h_lockspace);
__entry->h_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
__entry->h_length = le16_to_cpu(ms->m_header.h_length);
__entry->h_cmd = ms->m_header.h_cmd;
__entry->m_type = le32_to_cpu(ms->m_type);
__entry->m_nodeid = le32_to_cpu(ms->m_nodeid);
__entry->m_pid = le32_to_cpu(ms->m_pid);
__entry->m_lkid = le32_to_cpu(ms->m_lkid);
__entry->m_remid = le32_to_cpu(ms->m_remid);
__entry->m_parent_lkid = le32_to_cpu(ms->m_parent_lkid);
__entry->m_parent_remid = le32_to_cpu(ms->m_parent_remid);
__entry->m_exflags = le32_to_cpu(ms->m_exflags);
__entry->m_sbflags = le32_to_cpu(ms->m_sbflags);
__entry->m_flags = le32_to_cpu(ms->m_flags);
__entry->m_lvbseq = le32_to_cpu(ms->m_lvbseq);
__entry->m_hash = le32_to_cpu(ms->m_hash);
__entry->m_status = le32_to_cpu(ms->m_status);
__entry->m_grmode = le32_to_cpu(ms->m_grmode);
__entry->m_rqmode = le32_to_cpu(ms->m_rqmode);
__entry->m_bastmode = le32_to_cpu(ms->m_bastmode);
__entry->m_asts = le32_to_cpu(ms->m_asts);
__entry->m_result = le32_to_cpu(ms->m_result);
memcpy(__get_dynamic_array(m_extra), ms->m_extra,
__get_dynamic_array_len(m_extra));
memcpy(__get_dynamic_array(res_name), name,
__get_dynamic_array_len(res_name));
),
TP_printk("dst=%u h_seq=%u h_version=%s h_lockspace=%u h_nodeid=%u "
"h_length=%u h_cmd=%s m_type=%s m_nodeid=%u "
"m_pid=%u m_lkid=%u m_remid=%u m_parent_lkid=%u "
"m_parent_remid=%u m_exflags=%s m_sbflags=%s m_flags=%s "
"m_lvbseq=%u m_hash=%u m_status=%d m_grmode=%s "
"m_rqmode=%s m_bastmode=%s m_asts=%d m_result=%d "
"m_extra=0x%s res_name=0x%s", __entry->dst,
__entry->h_seq, show_message_version(__entry->h_version),
__entry->h_lockspace, __entry->h_nodeid, __entry->h_length,
show_header_cmd(__entry->h_cmd),
show_message_type(__entry->m_type),
__entry->m_nodeid, __entry->m_pid, __entry->m_lkid,
__entry->m_remid, __entry->m_parent_lkid,
__entry->m_parent_remid, show_lock_flags(__entry->m_exflags),
show_dlm_sb_flags(__entry->m_sbflags),
show_lkb_flags(__entry->m_flags), __entry->m_lvbseq,
__entry->m_hash, __entry->m_status,
show_lock_mode(__entry->m_grmode),
show_lock_mode(__entry->m_rqmode),
show_lock_mode(__entry->m_bastmode),
__entry->m_asts, __entry->m_result,
__print_hex_str(__get_dynamic_array(m_extra),
__get_dynamic_array_len(m_extra)),
__print_hex_str(__get_dynamic_array(res_name),
__get_dynamic_array_len(res_name)))
);
TRACE_EVENT(dlm_recv_message,
TP_PROTO(uint32_t dst, uint32_t h_seq, const struct dlm_message *ms),
TP_ARGS(dst, h_seq, ms),
TP_STRUCT__entry(
__field(uint32_t, dst)
__field(uint32_t, h_seq)
__field(uint32_t, h_version)
__field(uint32_t, h_lockspace)
__field(uint32_t, h_nodeid)
__field(uint16_t, h_length)
__field(uint8_t, h_cmd)
__field(uint32_t, m_type)
__field(uint32_t, m_nodeid)
__field(uint32_t, m_pid)
__field(uint32_t, m_lkid)
__field(uint32_t, m_remid)
__field(uint32_t, m_parent_lkid)
__field(uint32_t, m_parent_remid)
__field(uint32_t, m_exflags)
__field(uint32_t, m_sbflags)
__field(uint32_t, m_flags)
__field(uint32_t, m_lvbseq)
__field(uint32_t, m_hash)
__field(int32_t, m_status)
__field(int32_t, m_grmode)
__field(int32_t, m_rqmode)
__field(int32_t, m_bastmode)
__field(int32_t, m_asts)
__field(int32_t, m_result)
__dynamic_array(unsigned char, m_extra,
le16_to_cpu(ms->m_header.h_length) - sizeof(*ms))
),
TP_fast_assign(
__entry->dst = dst;
__entry->h_seq = h_seq;
__entry->h_version = le32_to_cpu(ms->m_header.h_version);
__entry->h_lockspace = le32_to_cpu(ms->m_header.u.h_lockspace);
__entry->h_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
__entry->h_length = le16_to_cpu(ms->m_header.h_length);
__entry->h_cmd = ms->m_header.h_cmd;
__entry->m_type = le32_to_cpu(ms->m_type);
__entry->m_nodeid = le32_to_cpu(ms->m_nodeid);
__entry->m_pid = le32_to_cpu(ms->m_pid);
__entry->m_lkid = le32_to_cpu(ms->m_lkid);
__entry->m_remid = le32_to_cpu(ms->m_remid);
__entry->m_parent_lkid = le32_to_cpu(ms->m_parent_lkid);
__entry->m_parent_remid = le32_to_cpu(ms->m_parent_remid);
__entry->m_exflags = le32_to_cpu(ms->m_exflags);
__entry->m_sbflags = le32_to_cpu(ms->m_sbflags);
__entry->m_flags = le32_to_cpu(ms->m_flags);
__entry->m_lvbseq = le32_to_cpu(ms->m_lvbseq);
__entry->m_hash = le32_to_cpu(ms->m_hash);
__entry->m_status = le32_to_cpu(ms->m_status);
__entry->m_grmode = le32_to_cpu(ms->m_grmode);
__entry->m_rqmode = le32_to_cpu(ms->m_rqmode);
__entry->m_bastmode = le32_to_cpu(ms->m_bastmode);
__entry->m_asts = le32_to_cpu(ms->m_asts);
__entry->m_result = le32_to_cpu(ms->m_result);
memcpy(__get_dynamic_array(m_extra), ms->m_extra,
__get_dynamic_array_len(m_extra));
),
TP_printk("dst=%u h_seq=%u h_version=%s h_lockspace=%u h_nodeid=%u "
"h_length=%u h_cmd=%s m_type=%s m_nodeid=%u "
"m_pid=%u m_lkid=%u m_remid=%u m_parent_lkid=%u "
"m_parent_remid=%u m_exflags=%s m_sbflags=%s m_flags=%s "
"m_lvbseq=%u m_hash=%u m_status=%d m_grmode=%s "
"m_rqmode=%s m_bastmode=%s m_asts=%d m_result=%d "
"m_extra=0x%s", __entry->dst,
__entry->h_seq, show_message_version(__entry->h_version),
__entry->h_lockspace, __entry->h_nodeid, __entry->h_length,
show_header_cmd(__entry->h_cmd),
show_message_type(__entry->m_type),
__entry->m_nodeid, __entry->m_pid, __entry->m_lkid,
__entry->m_remid, __entry->m_parent_lkid,
__entry->m_parent_remid, show_lock_flags(__entry->m_exflags),
show_dlm_sb_flags(__entry->m_sbflags),
show_lkb_flags(__entry->m_flags), __entry->m_lvbseq,
__entry->m_hash, __entry->m_status,
show_lock_mode(__entry->m_grmode),
show_lock_mode(__entry->m_rqmode),
show_lock_mode(__entry->m_bastmode),
__entry->m_asts, __entry->m_result,
__print_hex_str(__get_dynamic_array(m_extra),
__get_dynamic_array_len(m_extra)))
);
TRACE_EVENT(dlm_send,
TP_PROTO(int nodeid, int ret),