github-mirror/linux
1
0
Fork 0
mirror of https://github.com/torvalds/linux.git synced 2026-07-27 09:36:22 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity Graph

This adds support for manual client session reset in CephFS, allowing

Browse Source 
operators to get out of tricky livelock situations involving caps and
 file locks without evicting the problematic client instance on the MDS
 side or rebooting the client node both of which can be disruptive.
 -----BEGIN PGP SIGNATURE-----
 
 iQFHBAABCgAxFiEEydHwtzie9C7TfviiSn/eOAIR84sFAmo+rNUTHGlkcnlvbW92
 QGdtYWlsLmNvbQAKCRBKf944AhHzi++dB/0VE3YM+kdZER9Y1CoPgxB32CSD5Yoz
 cscBQCGtFM/0GHj2rMnvyeQ9+uHlQqzo/CMYURUBJ4ciHPLaZUyuJcp8R7bGNgk2
 FZ0+9yBqJjjeSJST16MKhW7/NgzuIhg8Y3c+sDrLDThYkbmEM90MYLDpopqwjLhc
 RL1mURFBQxuBLQiYujKXit1iXqYapJSRmthJAYN4pJ69w97zF+rO/qfvUGlRxYb3
 0Hb4GzD5IyWmsRGwldzc7/CXDLjTjp/N+tu1BLuCQK68U3MAw8GV39/aSocFjxN8
 rtq8TkgFb1dfSjwjD8pxei5pjBaoUjABnTg8wPDFoxEMKrVpabMeUTsb
 =xeGm
 -----END PGP SIGNATURE-----

Merge tag 'ceph-for-7.2-rc1' of https://github.com/ceph/ceph-client

Pull ceph updates from Ilya Dryomov:
 "This adds support for manual client session reset in CephFS, allowing
  operators to get out of tricky livelock situations involving caps and
  file locks without evicting the problematic client instance on the MDS
  side or rebooting the client node both of which can be disruptive"

* tag 'ceph-for-7.2-rc1' of https://github.com/ceph/ceph-client:
  ceph: add manual reset debugfs control and tracepoints
  ceph: add client reset state machine and session teardown
  ceph: add diagnostic timeout loop to wait_caps_flush()
  ceph: harden send_mds_reconnect and handle active-MDS peer reset
  ceph: use proper endian conversion for flock_len in reconnect
  ceph: convert inode flags to named bit positions and atomic bitops
  rbd: switch to dynamic root device
This commit is contained in:
Linus Torvalds 2026-06-26 16:15:53 -07:00
commit 5422e496b3
13 changed files with 1112 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
drivers/block/rbd.c
View File

@ -580,14 +580,7 @@ static const struct bus_type rbd_bus_type = {
.bus_groups = rbd_bus_groups,
};
static void rbd_root_dev_release(struct device *dev)
{
}
static struct device rbd_root_dev = {
.init_name = "rbd",
.release = rbd_root_dev_release,
};
static struct device *rbd_root_dev;
static __printf(2, 3)
void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
@ -5378,7 +5371,7 @@ static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
rbd_dev->dev.bus = &rbd_bus_type;
rbd_dev->dev.type = &rbd_device_type;
rbd_dev->dev.parent = &rbd_root_dev;
rbd_dev->dev.parent = rbd_root_dev;
device_initialize(&rbd_dev->dev);
return rbd_dev;
@ -7324,15 +7317,13 @@ static int __init rbd_sysfs_init(void)
{
int ret;
ret = device_register(&rbd_root_dev);
if (ret < 0) {
put_device(&rbd_root_dev);
return ret;
}
rbd_root_dev = root_device_register("rbd");
if (IS_ERR(rbd_root_dev))
return PTR_ERR(rbd_root_dev);
ret = bus_register(&rbd_bus_type);
if (ret < 0)
device_unregister(&rbd_root_dev);
root_device_unregister(rbd_root_dev);
return ret;
}
@ -7340,7 +7331,7 @@ static int __init rbd_sysfs_init(void)
static void __exit rbd_sysfs_cleanup(void)
{
bus_unregister(&rbd_bus_type);
device_unregister(&rbd_root_dev);
root_device_unregister(rbd_root_dev);
}
static int __init rbd_slab_init(void)

20
fs/ceph/addr.c
View File

@ -2561,7 +2561,8 @@ int ceph_pool_perm_check(struct inode *inode, int need)
struct ceph_inode_info *ci = ceph_inode(inode);
struct ceph_string *pool_ns;
s64 pool;
int ret, flags;
int ret;
unsigned long flags;
/* Only need to do this for regular files */
if (!S_ISREG(inode->i_mode))
@ -2603,20 +2604,19 @@ int ceph_pool_perm_check(struct inode *inode, int need)
if (ret < 0)
return ret;
flags = CEPH_I_POOL_PERM;
if (ret & POOL_READ)
flags |= CEPH_I_POOL_RD;
if (ret & POOL_WRITE)
flags |= CEPH_I_POOL_WR;
spin_lock(&ci->i_ceph_lock);
if (pool == ci->i_layout.pool_id &&
pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
ci->i_ceph_flags |= flags;
} else {
set_bit(CEPH_I_POOL_PERM_BIT, &ci->i_ceph_flags);
if (ret & POOL_READ)
set_bit(CEPH_I_POOL_RD_BIT, &ci->i_ceph_flags);
if (ret & POOL_WRITE)
set_bit(CEPH_I_POOL_WR_BIT, &ci->i_ceph_flags);
} else {
pool = ci->i_layout.pool_id;
flags = ci->i_ceph_flags;
}
/* Re-read flags under the lock so check: sees the updated bits. */
flags = ci->i_ceph_flags;
spin_unlock(&ci->i_ceph_lock);
goto check;
}

34
fs/ceph/caps.c
View File

@ -549,7 +549,7 @@ static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc,
doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
spin_lock(&mdsc->cap_delay_lock);
ci->i_ceph_flags |= CEPH_I_FLUSH;
set_bit(CEPH_I_FLUSH_BIT, &ci->i_ceph_flags);
if (!list_empty(&ci->i_cap_delay_list))
list_del_init(&ci->i_cap_delay_list);
list_add(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
@ -1409,7 +1409,7 @@ static void __prep_cap(struct cap_msg_args *arg, struct ceph_cap *cap,
ceph_cap_string(revoking));
BUG_ON((retain & CEPH_CAP_PIN) == 0);
ci->i_ceph_flags &= ~CEPH_I_FLUSH;
clear_bit(CEPH_I_FLUSH_BIT, &ci->i_ceph_flags);
cap->issued &= retain; /* drop bits we don't want */
/*
@ -1648,6 +1648,7 @@ static void __ceph_flush_snaps(struct ceph_inode_info *ci,
spin_lock(&mdsc->cap_dirty_lock);
capsnap->cap_flush.tid = ++mdsc->last_cap_flush_tid;
capsnap->cap_flush.ci = ci;
list_add_tail(&capsnap->cap_flush.g_list,
&mdsc->cap_flush_list);
if (oldest_flush_tid == 0)
@ -1666,7 +1667,7 @@ static void __ceph_flush_snaps(struct ceph_inode_info *ci,
last_tid = capsnap->cap_flush.tid;
}
ci->i_ceph_flags &= ~CEPH_I_FLUSH_SNAPS;
clear_bit(CEPH_I_FLUSH_SNAPS_BIT, &ci->i_ceph_flags);
while (first_tid <= last_tid) {
struct ceph_cap *cap = ci->i_auth_cap;
@ -1846,6 +1847,7 @@ struct ceph_cap_flush *ceph_alloc_cap_flush(void)
return NULL;
cf->is_capsnap = false;
cf->ci = NULL;
return cf;
}
@ -1931,6 +1933,7 @@ static u64 __mark_caps_flushing(struct inode *inode,
doutc(cl, "%p %llx.%llx now !dirty\n", inode, ceph_vinop(inode));
swap(cf, ci->i_prealloc_cap_flush);
cf->ci = ci;
cf->caps = flushing;
cf->wake = wake;
@ -2026,7 +2029,7 @@ void ceph_check_caps(struct ceph_inode_info *ci, int flags)
spin_lock(&ci->i_ceph_lock);
if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
ci->i_ceph_flags |= CEPH_I_ASYNC_CHECK_CAPS;
set_bit(CEPH_I_ASYNC_CHECK_CAPS_BIT, &ci->i_ceph_flags);
/* Don't send messages until we get async create reply */
spin_unlock(&ci->i_ceph_lock);
@ -2577,7 +2580,7 @@ static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE)
return;
ci->i_ceph_flags &= ~CEPH_I_KICK_FLUSH;
clear_bit(CEPH_I_KICK_FLUSH_BIT, &ci->i_ceph_flags);
list_for_each_entry_reverse(cf, &ci->i_cap_flush_list, i_list) {
if (cf->is_capsnap) {
@ -2686,7 +2689,7 @@ void ceph_early_kick_flushing_caps(struct ceph_mds_client *mdsc,
__kick_flushing_caps(mdsc, session, ci,
oldest_flush_tid);
} else {
ci->i_ceph_flags |= CEPH_I_KICK_FLUSH;
set_bit(CEPH_I_KICK_FLUSH_BIT, &ci->i_ceph_flags);
}
spin_unlock(&ci->i_ceph_lock);
@ -2829,7 +2832,7 @@ static int try_get_cap_refs(struct inode *inode, int need, int want,
spin_lock(&ci->i_ceph_lock);
if ((flags & CHECK_FILELOCK) &&
(ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK)) {
test_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags)) {
doutc(cl, "%p %llx.%llx error filelock\n", inode,
ceph_vinop(inode));
ret = -EIO;
@ -3207,7 +3210,7 @@ static int ceph_try_drop_cap_snap(struct ceph_inode_info *ci,
BUG_ON(capsnap->cap_flush.tid > 0);
ceph_put_snap_context(capsnap->context);
if (!list_is_last(&capsnap->ci_item, &ci->i_cap_snaps))
ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
set_bit(CEPH_I_FLUSH_SNAPS_BIT, &ci->i_ceph_flags);
list_del(&capsnap->ci_item);
ceph_put_cap_snap(capsnap);
@ -3396,7 +3399,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
if (ceph_try_drop_cap_snap(ci, capsnap)) {
put++;
} else {
ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
set_bit(CEPH_I_FLUSH_SNAPS_BIT, &ci->i_ceph_flags);
flush_snaps = true;
}
}
@ -3648,7 +3651,7 @@ static void handle_cap_grant(struct inode *inode,
if (ci->i_layout.pool_id != old_pool ||
extra_info->pool_ns != old_ns)
ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
clear_bit(CEPH_I_POOL_PERM_BIT, &ci->i_ceph_flags);
extra_info->pool_ns = old_ns;
@ -3826,6 +3829,13 @@ static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
bool wake_ci = false;
bool wake_mdsc = false;
/*
* Flush tids are monotonically increasing and acks arrive in
* order under i_ceph_lock, so this is always the latest tid.
* Diagnostic readers use READ_ONCE() without holding the lock.
*/
WRITE_ONCE(ci->i_last_cap_flush_ack, flush_tid);
list_for_each_entry_safe(cf, tmp_cf, &ci->i_cap_flush_list, i_list) {
/* Is this the one that was flushed? */
if (cf->tid == flush_tid)
@ -4815,7 +4825,7 @@ int ceph_drop_caps_for_unlink(struct inode *inode)
doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode,
ceph_vinop(inode));
spin_lock(&mdsc->cap_delay_lock);
ci->i_ceph_flags |= CEPH_I_FLUSH;
set_bit(CEPH_I_FLUSH_BIT, &ci->i_ceph_flags);
if (!list_empty(&ci->i_cap_delay_list))
list_del_init(&ci->i_cap_delay_list);
list_add_tail(&ci->i_cap_delay_list,
@ -5080,7 +5090,7 @@ int ceph_purge_inode_cap(struct inode *inode, struct ceph_cap *cap, bool *invali
if (atomic_read(&ci->i_filelock_ref) > 0) {
/* make further file lock syscall return -EIO */
ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
set_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags);
pr_warn_ratelimited_client(cl,
" dropping file locks for %p %llx.%llx\n",
inode, ceph_vinop(inode));

103
fs/ceph/debugfs.c
View File

@ -9,6 +9,7 @@
#include <linux/seq_file.h>
#include <linux/math64.h>
#include <linux/ktime.h>
#include <linux/uaccess.h>
#include <linux/atomic.h>
#include <linux/ceph/libceph.h>
@ -392,6 +393,90 @@ static int status_show(struct seq_file *s, void *p)
return 0;
}
static int reset_status_show(struct seq_file *s, void *p)
{
struct ceph_fs_client *fsc = s->private;
struct ceph_mds_client *mdsc = fsc->mdsc;
struct ceph_client_reset_state *st;
u64 trigger = 0, success = 0, failure = 0;
unsigned long last_start = 0, last_finish = 0;
int last_errno = 0;
enum ceph_client_reset_phase phase = CEPH_CLIENT_RESET_IDLE;
bool drain_timed_out = false;
int sessions_reset = 0;
int blocked_requests = 0;
char reason[CEPH_CLIENT_RESET_REASON_LEN];
if (!mdsc)
return 0;
st = &mdsc->reset_state;
spin_lock(&st->lock);
trigger = st->trigger_count;
success = st->success_count;
failure = st->failure_count;
last_start = st->last_start;
last_finish = st->last_finish;
last_errno = st->last_errno;
phase = st->phase;
drain_timed_out = st->drain_timed_out;
sessions_reset = st->sessions_reset;
strscpy(reason, st->last_reason, sizeof(reason));
spin_unlock(&st->lock);
blocked_requests = atomic_read(&st->blocked_requests);
seq_printf(s, "phase: %s\n", ceph_reset_phase_name(phase));
seq_printf(s, "trigger_count: %llu\n", trigger);
seq_printf(s, "success_count: %llu\n", success);
seq_printf(s, "failure_count: %llu\n", failure);
if (last_start)
seq_printf(s, "last_start_ms_ago: %u\n",
jiffies_to_msecs(jiffies - last_start));
else
seq_puts(s, "last_start_ms_ago: (never)\n");
if (last_finish)
seq_printf(s, "last_finish_ms_ago: %u\n",
jiffies_to_msecs(jiffies - last_finish));
else
seq_puts(s, "last_finish_ms_ago: (never)\n");
seq_printf(s, "last_errno: %d\n", last_errno);
seq_printf(s, "last_reason: %s\n",
reason[0] ? reason : "(none)");
seq_printf(s, "drain_timed_out: %s\n",
drain_timed_out ? "yes" : "no");
seq_printf(s, "sessions_reset: %d\n", sessions_reset);
seq_printf(s, "blocked_requests: %d\n", blocked_requests);
return 0;
}
static ssize_t reset_trigger_write(struct file *file, const char __user *buf,
size_t len, loff_t *ppos)
{
struct ceph_fs_client *fsc = file->private_data;
struct ceph_mds_client *mdsc = fsc->mdsc;
char reason[CEPH_CLIENT_RESET_REASON_LEN];
size_t copy;
int ret;
if (!mdsc)
return -ENODEV;
copy = min_t(size_t, len, sizeof(reason) - 1);
if (copy && copy_from_user(reason, buf, copy))
return -EFAULT;
reason[copy] = '\0';
strim(reason);
ret = ceph_mdsc_schedule_reset(mdsc, reason);
if (ret)
return ret;
return len;
}
static int subvolume_metrics_show(struct seq_file *s, void *p)
{
struct ceph_fs_client *fsc = s->private;
@ -450,6 +535,7 @@ DEFINE_SHOW_ATTRIBUTE(mdsc);
DEFINE_SHOW_ATTRIBUTE(caps);
DEFINE_SHOW_ATTRIBUTE(mds_sessions);
DEFINE_SHOW_ATTRIBUTE(status);
DEFINE_SHOW_ATTRIBUTE(reset_status);
DEFINE_SHOW_ATTRIBUTE(metrics_file);
DEFINE_SHOW_ATTRIBUTE(metrics_latency);
DEFINE_SHOW_ATTRIBUTE(metrics_size);
@ -521,6 +607,13 @@ static int metric_features_show(struct seq_file *s, void *p)
DEFINE_SHOW_ATTRIBUTE(metric_features);
static const struct file_operations ceph_reset_trigger_fops = {
.owner = THIS_MODULE,
.open = simple_open,
.write = reset_trigger_write,
.llseek = noop_llseek,
};
/*
* debugfs
*/
@ -554,6 +647,7 @@ void ceph_fs_debugfs_cleanup(struct ceph_fs_client *fsc)
debugfs_remove(fsc->debugfs_caps);
debugfs_remove(fsc->debugfs_status);
debugfs_remove(fsc->debugfs_mdsc);
debugfs_remove_recursive(fsc->debugfs_reset_dir);
debugfs_remove(fsc->debugfs_subvolume_metrics);
debugfs_remove_recursive(fsc->debugfs_metrics_dir);
doutc(fsc->client, "done\n");
@ -602,6 +696,15 @@ void ceph_fs_debugfs_init(struct ceph_fs_client *fsc)
fsc,
&caps_fops);
fsc->debugfs_reset_dir = debugfs_create_dir("reset",
fsc->client->debugfs_dir);
debugfs_create_file("trigger", 0200,
fsc->debugfs_reset_dir, fsc,
&ceph_reset_trigger_fops);
debugfs_create_file("status", 0400,
fsc->debugfs_reset_dir, fsc,
&reset_status_fops);
fsc->debugfs_status = debugfs_create_file("status",
0400,
fsc->client->debugfs_dir,

13
fs/ceph/file.c
View File

@ -598,12 +598,12 @@ static void wake_async_create_waiters(struct inode *inode,
spin_lock(&ci->i_ceph_lock);
if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
clear_and_wake_up_bit(CEPH_ASYNC_CREATE_BIT, &ci->i_ceph_flags);
/* Serialized by i_ceph_lock; the two ops touch different bits. */
clear_and_wake_up_bit(CEPH_I_ASYNC_CREATE_BIT, &ci->i_ceph_flags);
if (ci->i_ceph_flags & CEPH_I_ASYNC_CHECK_CAPS) {
ci->i_ceph_flags &= ~CEPH_I_ASYNC_CHECK_CAPS;
if (test_and_clear_bit(CEPH_I_ASYNC_CHECK_CAPS_BIT,
&ci->i_ceph_flags))
check_cap = true;
}
}
ceph_kick_flushing_inode_caps(session, ci);
spin_unlock(&ci->i_ceph_lock);
@ -766,7 +766,8 @@ static int ceph_finish_async_create(struct inode *dir, struct inode *inode,
* that point and don't worry about setting
* CEPH_I_ASYNC_CREATE.
*/
ceph_inode(inode)->i_ceph_flags = CEPH_I_ASYNC_CREATE;
set_bit(CEPH_I_ASYNC_CREATE_BIT,
&ceph_inode(inode)->i_ceph_flags);
unlock_new_inode(inode);
}
if (d_in_lookup(dentry) || d_really_is_negative(dentry)) {
@ -2486,7 +2487,7 @@ static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from)
if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
(iocb->ki_flags & IOCB_DIRECT) || (fi->flags & CEPH_F_SYNC) ||
(ci->i_ceph_flags & CEPH_I_ERROR_WRITE)) {
test_bit(CEPH_I_ERROR_WRITE_BIT, &ci->i_ceph_flags)) {
struct ceph_snap_context *snapc;
struct iov_iter data;

5
fs/ceph/inode.c
View File

@ -671,6 +671,7 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
INIT_LIST_HEAD(&ci->i_cap_snaps);
ci->i_head_snapc = NULL;
ci->i_snap_caps = 0;
ci->i_last_cap_flush_ack = 0;
ci->i_last_rd = ci->i_last_wr = jiffies - 3600 * HZ;
for (i = 0; i < CEPH_FILE_MODE_BITS; i++)
@ -1180,7 +1181,7 @@ int ceph_fill_inode(struct inode *inode, struct page *locked_page,
rcu_assign_pointer(ci->i_layout.pool_ns, pool_ns);
if (ci->i_layout.pool_id != old_pool || pool_ns != old_ns)
ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
clear_bit(CEPH_I_POOL_PERM_BIT, &ci->i_ceph_flags);
pool_ns = old_ns;
@ -3240,7 +3241,7 @@ void ceph_inode_shutdown(struct inode *inode)
bool invalidate = false;
spin_lock(&ci->i_ceph_lock);
ci->i_ceph_flags |= CEPH_I_SHUTDOWN;
set_bit(CEPH_I_SHUTDOWN_BIT, &ci->i_ceph_flags);
p = rb_first(&ci->i_caps);
while (p) {
struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);

38
fs/ceph/locks.c
View File

@ -57,9 +57,7 @@ static void ceph_fl_release_lock(struct file_lock *fl)
ci = ceph_inode(inode);
if (atomic_dec_and_test(&ci->i_filelock_ref)) {
/* clear error when all locks are released */
spin_lock(&ci->i_ceph_lock);
ci->i_ceph_flags &= ~CEPH_I_ERROR_FILELOCK;
spin_unlock(&ci->i_ceph_lock);
clear_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags);
}
fl->fl_u.ceph.inode = NULL;
iput(inode);
@ -251,6 +249,7 @@ int ceph_lock(struct file *file, int cmd, struct file_lock *fl)
{
struct inode *inode = file_inode(file);
struct ceph_inode_info *ci = ceph_inode(inode);
struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
struct ceph_client *cl = ceph_inode_to_client(inode);
int err = 0;
u16 op = CEPH_MDS_OP_SETFILELOCK;
@ -271,15 +270,17 @@ int ceph_lock(struct file *file, int cmd, struct file_lock *fl)
else if (IS_SETLKW(cmd))
wait = 1;
spin_lock(&ci->i_ceph_lock);
if (ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) {
err = -EIO;
}
spin_unlock(&ci->i_ceph_lock);
if (err < 0) {
if (test_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags)) {
if (op == CEPH_MDS_OP_SETFILELOCK && lock_is_unlock(fl))
posix_lock_file(file, fl, NULL);
return err;
return -EIO;
}
/* Wait for reset to complete before acquiring new locks */
if (op == CEPH_MDS_OP_SETFILELOCK && !lock_is_unlock(fl)) {
err = ceph_mdsc_wait_for_reset(mdsc);
if (err)
return err;
}
if (lock_is_read(fl))
@ -318,6 +319,7 @@ int ceph_flock(struct file *file, int cmd, struct file_lock *fl)
{
struct inode *inode = file_inode(file);
struct ceph_inode_info *ci = ceph_inode(inode);
struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
struct ceph_client *cl = ceph_inode_to_client(inode);
int err = 0;
u8 wait = 0;
@ -331,15 +333,17 @@ int ceph_flock(struct file *file, int cmd, struct file_lock *fl)
doutc(cl, "fl_file: %p\n", fl->c.flc_file);
spin_lock(&ci->i_ceph_lock);
if (ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) {
err = -EIO;
}
spin_unlock(&ci->i_ceph_lock);
if (err < 0) {
if (test_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags)) {
if (lock_is_unlock(fl))
locks_lock_file_wait(file, fl);
return err;
return -EIO;
}
/* Wait for reset to complete before acquiring new locks */
if (!lock_is_unlock(fl)) {
err = ceph_mdsc_wait_for_reset(mdsc);
if (err)
return err;
}
if (IS_SETLKW(cmd))

802
fs/ceph/mds_client.c
View File

@ -6,6 +6,7 @@
#include <linux/slab.h>
#include <linux/gfp.h>
#include <linux/sched.h>
#include <linux/delay.h>
#include <linux/debugfs.h>
#include <linux/seq_file.h>
#include <linux/ratelimit.h>
@ -65,6 +66,7 @@ static void __wake_requests(struct ceph_mds_client *mdsc,
struct list_head *head);
static void ceph_cap_release_work(struct work_struct *work);
static void ceph_cap_reclaim_work(struct work_struct *work);
static void ceph_mdsc_reset_workfn(struct work_struct *work);
static const struct ceph_connection_operations mds_con_ops;
@ -2330,19 +2332,112 @@ static int check_caps_flush(struct ceph_mds_client *mdsc,
}
/*
* flush all dirty inode data to disk.
* Snapshot of a single cap_flush entry for diagnostic dump.
* Collected under cap_dirty_lock, printed after releasing it.
*/
struct flush_dump_entry {
u64 ino; /* inode number */
u64 snap; /* snap id */
int caps; /* dirty cap bits */
u64 tid; /* flush transaction id */
u64 last_ack; /* most recent ack tid for this inode */
bool wake; /* whether completion was requested */
bool is_capsnap; /* true if this is a cap snap flush */
bool ci_null; /* true if cf->ci was unexpectedly NULL */
};
/*
* Dump pending cap flushes for diagnostic purposes.
*
* returns true if we've flushed through want_flush_tid
* cf->ci is safe to dereference here: cap_flush entries hold a
* reference on the inode (via the cap), and entries are removed from
* cap_flush_list under cap_dirty_lock before the cap (and thus the
* inode reference) is released. Holding cap_dirty_lock therefore
* guarantees the inode remains valid for the lifetime of the scan.
*/
static void dump_cap_flushes(struct ceph_mds_client *mdsc, u64 want_tid)
{
struct ceph_client *cl = mdsc->fsc->client;
struct flush_dump_entry entries[CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES];
struct ceph_cap_flush *cf;
int n = 0, remaining = 0;
int i;
spin_lock(&mdsc->cap_dirty_lock);
list_for_each_entry(cf, &mdsc->cap_flush_list, g_list) {
if (cf->tid > want_tid)
break;
if (n < CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES) {
struct flush_dump_entry *e = &entries[n++];
e->ci_null = WARN_ON_ONCE(!cf->ci);
if (!e->ci_null) {
e->ino = ceph_ino(&cf->ci->netfs.inode);
e->snap = ceph_snap(&cf->ci->netfs.inode);
e->last_ack = READ_ONCE(cf->ci->i_last_cap_flush_ack);
}
e->caps = cf->caps;
e->tid = cf->tid;
e->wake = cf->wake;
e->is_capsnap = cf->is_capsnap;
} else {
remaining++;
}
}
spin_unlock(&mdsc->cap_dirty_lock);
pr_info_client(cl, "still waiting for cap flushes through %llu:\n",
want_tid);
for (i = 0; i < n; i++) {
struct flush_dump_entry *e = &entries[i];
if (e->ci_null)
pr_info_client(cl,
" (null ci) %s tid=%llu wake=%d%s\n",
ceph_cap_string(e->caps), e->tid,
e->wake,
e->is_capsnap ? " is_capsnap" : "");
else
pr_info_client(cl,
" %llx.%llx %s tid=%llu last_ack=%llu wake=%d%s\n",
e->ino, e->snap,
ceph_cap_string(e->caps), e->tid,
e->last_ack, e->wake,
e->is_capsnap ? " is_capsnap" : "");
}
if (remaining)
pr_info_client(cl, " ... and %d more pending flushes\n",
remaining);
}
/*
* Wait for all cap flushes through @want_flush_tid to complete.
* Periodically dumps pending cap flush state for diagnostics.
*/
static void wait_caps_flush(struct ceph_mds_client *mdsc,
u64 want_flush_tid)
{
struct ceph_client *cl = mdsc->fsc->client;
int i = 0;
long ret;
doutc(cl, "want %llu\n", want_flush_tid);
wait_event(mdsc->cap_flushing_wq,
check_caps_flush(mdsc, want_flush_tid));
do {
/* 60 * HZ fits in a long on all supported architectures. */
ret = wait_event_timeout(mdsc->cap_flushing_wq,
check_caps_flush(mdsc, want_flush_tid),
CEPH_CAP_FLUSH_WAIT_TIMEOUT_SEC * HZ);
if (ret == 0) {
if (i < CEPH_CAP_FLUSH_MAX_DUMP_ITERS)
dump_cap_flushes(mdsc, want_flush_tid);
else if (i == CEPH_CAP_FLUSH_MAX_DUMP_ITERS)
pr_info_client(cl,
"still waiting for cap flushes; suppressing further dumps\n");
i++;
}
} while (ret == 0);
doutc(cl, "ok, flushed thru %llu\n", want_flush_tid);
}
@ -3657,7 +3752,8 @@ static void __do_request(struct ceph_mds_client *mdsc,
spin_lock(&ci->i_ceph_lock);
cap = ci->i_auth_cap;
if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE && mds != cap->mds) {
if (test_bit(CEPH_I_ASYNC_CREATE_BIT, &ci->i_ceph_flags) &&
mds != cap->mds) {
doutc(cl, "session changed for auth cap %d -> %d\n",
cap->session->s_mds, session->s_mds);
@ -3751,6 +3847,22 @@ int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,
struct ceph_client *cl = mdsc->fsc->client;
int err = 0;
/*
* If a reset is in progress, wait for it to complete.
*
* This is best-effort: a request can pass this check just
* before the phase leaves IDLE and proceed concurrently with
* reset. That is acceptable because (a) such requests will
* either complete normally or fail and be retried by the
* caller, and (b) adding lock serialization here would
* penalize every request for a rare manual operation.
*/
err = ceph_mdsc_wait_for_reset(mdsc);
if (err) {
doutc(cl, "wait_for_reset failed: %d\n", err);
return err;
}
/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
if (req->r_inode)
ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
@ -4469,9 +4581,14 @@ static void handle_session(struct ceph_mds_session *session,
break;
case CEPH_SESSION_REJECT:
WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING);
pr_info_client(cl, "mds%d rejected session\n",
session->s_mds);
WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING &&
session->s_state != CEPH_MDS_SESSION_RECONNECTING);
if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
pr_info_client(cl, "mds%d reconnect rejected\n",
session->s_mds);
else
pr_info_client(cl, "mds%d rejected session\n",
session->s_mds);
session->s_state = CEPH_MDS_SESSION_REJECTED;
cleanup_session_requests(mdsc, session);
remove_session_caps(session);
@ -4731,6 +4848,14 @@ static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
cap->mseq = 0; /* and migrate_seq */
cap->cap_gen = atomic_read(&cap->session->s_cap_gen);
/*
* Note: CEPH_I_ERROR_FILELOCK is not set during reconnect.
* Instead, locks are submitted for best-effort MDS reclaim
* via the flock_len field below. If reclaim fails (e.g.,
* another client grabbed a conflicting lock), future lock
* operations will fail and set the error flag at that point.
*/
/* These are lost when the session goes away */
if (S_ISDIR(inode->i_mode)) {
if (cap->issued & CEPH_CAP_DIR_CREATE) {
@ -4746,8 +4871,9 @@ static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
rec.v2.issued = cpu_to_le32(cap->issued);
rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
rec.v2.pathbase = cpu_to_le64(path_info.vino.ino);
rec.v2.flock_len = (__force __le32)
((ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) ? 0 : 1);
rec.v2.flock_len = cpu_to_le32(
test_bit(CEPH_I_ERROR_FILELOCK_BIT,
&ci->i_ceph_flags) ? 0 : 1);
} else {
struct timespec64 ts;
@ -4944,20 +5070,19 @@ static int encode_snap_realms(struct ceph_mds_client *mdsc,
*
* This is a relatively heavyweight operation, but it's rare.
*/
static void send_mds_reconnect(struct ceph_mds_client *mdsc,
struct ceph_mds_session *session)
static int send_mds_reconnect(struct ceph_mds_client *mdsc,
struct ceph_mds_session *session)
{
struct ceph_client *cl = mdsc->fsc->client;
struct ceph_msg *reply;
int mds = session->s_mds;
int err = -ENOMEM;
int old_state;
struct ceph_reconnect_state recon_state = {
.session = session,
};
LIST_HEAD(dispose);
pr_info_client(cl, "mds%d reconnect start\n", mds);
recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
if (!recon_state.pagelist)
goto fail_nopagelist;
@ -4966,9 +5091,37 @@ static void send_mds_reconnect(struct ceph_mds_client *mdsc,
if (!reply)
goto fail_nomsg;
xa_destroy(&session->s_delegated_inos);
mutex_lock(&session->s_mutex);
/* Serialized by s_mutex against concurrent ceph_get_deleg_ino(). */
xa_destroy(&session->s_delegated_inos);
if (session->s_state == CEPH_MDS_SESSION_CLOSED ||
session->s_state == CEPH_MDS_SESSION_REJECTED) {
pr_info_client(cl, "mds%d skipping reconnect, session %s\n",
mds,
ceph_session_state_name(session->s_state));
mutex_unlock(&session->s_mutex);
ceph_msg_put(reply);
err = -ESTALE;
goto fail_return;
}
/* s_mutex -> mdsc->mutex matches cleanup_session_requests() order. */
mutex_lock(&mdsc->mutex);
if (mds >= mdsc->max_sessions || mdsc->sessions[mds] != session) {
mutex_unlock(&mdsc->mutex);
pr_info_client(cl,
"mds%d skipping reconnect, session unregistered\n",
mds);
mutex_unlock(&session->s_mutex);
ceph_msg_put(reply);
err = -ENOENT;
goto fail_return;
}
mutex_unlock(&mdsc->mutex);
pr_info_client(cl, "mds%d reconnect start\n", mds);
old_state = session->s_state;
session->s_state = CEPH_MDS_SESSION_RECONNECTING;
session->s_seq = 0;
@ -5098,7 +5251,7 @@ static void send_mds_reconnect(struct ceph_mds_client *mdsc,
up_read(&mdsc->snap_rwsem);
ceph_pagelist_release(recon_state.pagelist);
return;
return 0;
fail_clear_cap_reconnect:
spin_lock(&session->s_cap_lock);
@ -5107,13 +5260,504 @@ static void send_mds_reconnect(struct ceph_mds_client *mdsc,
fail:
ceph_msg_put(reply);
up_read(&mdsc->snap_rwsem);
/*
* Restore prior session state so map-driven reconnect logic
* (check_new_map) can retry. Without this, a transient build
* failure strands the session in RECONNECTING indefinitely.
*/
session->s_state = old_state;
mutex_unlock(&session->s_mutex);
fail_nomsg:
ceph_pagelist_release(recon_state.pagelist);
fail_nopagelist:
pr_err_client(cl, "error %d preparing reconnect for mds%d\n",
err, mds);
return err;
fail_return:
/*
* Early-exit path for expected concurrent-teardown races
* (-ESTALE for closed/rejected sessions, -ENOENT for
* unregistered sessions). Skip the pr_err_client diagnostic
* since these are not genuine reconnect build failures.
*/
ceph_pagelist_release(recon_state.pagelist);
return err;
}
const char *ceph_reset_phase_name(enum ceph_client_reset_phase phase)
{
switch (phase) {
case CEPH_CLIENT_RESET_IDLE: return "idle";
case CEPH_CLIENT_RESET_QUIESCING: return "quiescing";
case CEPH_CLIENT_RESET_DRAINING: return "draining";
case CEPH_CLIENT_RESET_TEARDOWN: return "teardown";
default: return "unknown";
}
}
/**
* ceph_mdsc_wait_for_reset - wait for an active reset to complete
* @mdsc: MDS client
*
* Returns 0 if reset completed successfully or no reset was active.
* Returns -EAGAIN if reset completed with an error, signalling the
* caller to retry. The internal error (e.g. -ENOMEM) is not propagated
* because callers like open() or flock() have no way to act on
* work-function internals. The detailed error is available via debugfs
* reset/status and tracepoints.
* Returns -ETIMEDOUT if we timed out waiting.
* Returns -ERESTARTSYS if interrupted by signal.
*/
int ceph_mdsc_wait_for_reset(struct ceph_mds_client *mdsc)
{
struct ceph_client_reset_state *st = &mdsc->reset_state;
struct ceph_client *cl = mdsc->fsc->client;
unsigned long deadline = jiffies + CEPH_CLIENT_RESET_WAIT_TIMEOUT_SEC * HZ;
int blocked_count;
long remaining;
long wait_ret;
int ret;
if (ceph_reset_is_idle(st))
return 0;
blocked_count = atomic_inc_return(&st->blocked_requests);
doutc(cl, "request blocked during reset, %d total blocked\n",
blocked_count);
trace_ceph_client_reset_blocked(mdsc, blocked_count);
retry:
remaining = max_t(long, deadline - jiffies, 1);
wait_ret = wait_event_interruptible_timeout(st->blocked_wq,
ceph_reset_is_idle(st),
remaining);
if (wait_ret == 0) {
atomic_dec(&st->blocked_requests);
pr_warn_client(cl, "timed out waiting for reset to complete\n");
trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT);
return -ETIMEDOUT;
}
if (wait_ret < 0) {
atomic_dec(&st->blocked_requests);
trace_ceph_client_reset_unblocked(mdsc, (int)wait_ret);
return (int)wait_ret; /* -ERESTARTSYS */
}
/*
* Verify phase is still IDLE under the lock. If another reset
* was scheduled between the wake-up and this check, loop back
* and wait for it to finish rather than returning a stale result.
*/
spin_lock(&st->lock);
if (st->phase != CEPH_CLIENT_RESET_IDLE) {
spin_unlock(&st->lock);
if (time_before(jiffies, deadline))
goto retry;
atomic_dec(&st->blocked_requests);
trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT);
return -ETIMEDOUT;
}
ret = st->last_errno;
spin_unlock(&st->lock);
atomic_dec(&st->blocked_requests);
trace_ceph_client_reset_unblocked(mdsc, ret);
return ret ? -EAGAIN : 0;
}
static void ceph_mdsc_reset_complete(struct ceph_mds_client *mdsc, int ret)
{
struct ceph_client_reset_state *st = &mdsc->reset_state;
spin_lock(&st->lock);
/*
* If destroy already marked us as shut down, it owns the
* final bookkeeping and waiter wakeup. Just bail so we
* don't overwrite its state.
*/
if (st->shutdown) {
spin_unlock(&st->lock);
return;
}
st->last_finish = jiffies;
st->last_errno = ret;
st->phase = CEPH_CLIENT_RESET_IDLE;
if (ret)
st->failure_count++;
else
st->success_count++;
spin_unlock(&st->lock);
/* Wake up all requests that were blocked waiting for reset */
wake_up_all(&st->blocked_wq);
trace_ceph_client_reset_complete(mdsc, ret);
}
static void ceph_mdsc_reset_workfn(struct work_struct *work)
{
struct ceph_mds_client *mdsc =
container_of(work, struct ceph_mds_client, reset_work);
struct ceph_client_reset_state *st = &mdsc->reset_state;
struct ceph_client *cl = mdsc->fsc->client;
struct ceph_mds_session **sessions = NULL;
char reason[CEPH_CLIENT_RESET_REASON_LEN];
unsigned long drain_deadline;
int max_sessions, i, n = 0, torn_down = 0;
int ret = 0;
spin_lock(&st->lock);
strscpy(reason, st->last_reason, sizeof(reason));
spin_unlock(&st->lock);
mutex_lock(&mdsc->mutex);
max_sessions = mdsc->max_sessions;
if (max_sessions <= 0) {
mutex_unlock(&mdsc->mutex);
goto out_complete;
}
sessions = kcalloc(max_sessions, sizeof(*sessions), GFP_KERNEL);
if (!sessions) {
mutex_unlock(&mdsc->mutex);
ret = -ENOMEM;
pr_err_client(cl,
"manual session reset failed to allocate session array\n");
ceph_mdsc_reset_complete(mdsc, ret);
return;
}
for (i = 0; i < max_sessions; i++) {
struct ceph_mds_session *session = mdsc->sessions[i];
if (!session)
continue;
/*
* Read session state without s_mutex to avoid nesting
* mdsc->mutex -> s_mutex, which would invert the
* s_mutex -> mdsc->mutex order used by
* cleanup_session_requests(). s_state is an int
* so loads are atomic; the teardown loop below
* handles races with concurrent state transitions.
*/
switch (READ_ONCE(session->s_state)) {
case CEPH_MDS_SESSION_OPEN:
case CEPH_MDS_SESSION_HUNG:
case CEPH_MDS_SESSION_OPENING:
case CEPH_MDS_SESSION_RESTARTING:
case CEPH_MDS_SESSION_RECONNECTING:
case CEPH_MDS_SESSION_CLOSING:
sessions[n++] = ceph_get_mds_session(session);
break;
default:
pr_info_client(cl,
"mds%d in state %s, skipping reset\n",
session->s_mds,
ceph_session_state_name(session->s_state));
break;
}
}
mutex_unlock(&mdsc->mutex);
pr_info_client(cl,
"manual session reset executing (sessions=%d, reason=\"%s\")\n",
n, reason);
if (n == 0) {
kfree(sessions);
goto out_complete;
}
spin_lock(&st->lock);
if (st->shutdown) {
spin_unlock(&st->lock);
goto out_sessions;
}
st->phase = CEPH_CLIENT_RESET_DRAINING;
spin_unlock(&st->lock);
/*
* Best-effort drain: flush dirty state while sessions are still
* alive. New requests are blocked while phase != IDLE.
* The sessions are functional, so non-stuck state drains normally.
* Stuck state (the cause of the stalemate the operator is trying
* to break) will not drain -- that is expected, and we proceed to
* forced teardown after the timeout.
*
* Four things are drained:
* 1. MDS journal -- send_flush_mdlog asks each MDS to journal
* pending unsafe operations (creates, renames, setattrs).
* 2. Unsafe requests -- bounded wait for each unsafe write
* request to reach safe status via r_safe_completion.
* 3. Dirty caps -- ceph_flush_dirty_caps triggers cap flush on
* all sessions. Non-stuck caps flush in milliseconds.
* 4. Cap releases -- push pending cap release messages.
*
* The unsafe-request wait and cap-flush wait below provide
* the bounded drain window during which all categories can
* make progress.
*/
for (i = 0; i < n; i++)
send_flush_mdlog(sessions[i]);
/*
* Both drain legs (unsafe requests and cap flushes) share a
* single deadline so the total drain time is bounded at
* CEPH_CLIENT_RESET_DRAIN_SEC.
*/
drain_deadline = jiffies + CEPH_CLIENT_RESET_DRAIN_SEC * HZ;
/*
* Wait for unsafe write requests (creates, renames, setattrs)
* to reach safe status. Uses the same pattern as
* flush_mdlog_and_wait_mdsc_unsafe_requests() but bounded by
* the shared drain deadline. Requests that do not complete within
* the window are force-dropped during teardown.
*/
{
struct ceph_mds_request *req;
struct rb_node *rn;
u64 last_tid;
mutex_lock(&mdsc->mutex);
last_tid = mdsc->last_tid;
mutex_unlock(&mdsc->mutex);
mutex_lock(&mdsc->mutex);
rn = rb_first(&mdsc->request_tree);
while (rn) {
req = rb_entry(rn, struct ceph_mds_request, r_node);
if (req->r_tid > last_tid)
break;
if (req->r_op == CEPH_MDS_OP_SETFILELOCK ||
!(req->r_op & CEPH_MDS_OP_WRITE)) {
rn = rb_next(rn);
continue;
}
ceph_mdsc_get_request(req);
mutex_unlock(&mdsc->mutex);
wait_for_completion_timeout(&req->r_safe_completion,
max_t(long, drain_deadline - jiffies, 1));
mutex_lock(&mdsc->mutex);
ceph_mdsc_put_request(req);
if (time_after(jiffies, drain_deadline))
break;
rn = rb_first(&mdsc->request_tree);
}
mutex_unlock(&mdsc->mutex);
if (time_after_eq(jiffies, drain_deadline))
WRITE_ONCE(st->drain_timed_out, true);
}
ceph_flush_dirty_caps(mdsc);
ceph_flush_cap_releases(mdsc);
spin_lock(&mdsc->cap_dirty_lock);
if (!list_empty(&mdsc->cap_flush_list)) {
struct ceph_cap_flush *cf =
list_last_entry(&mdsc->cap_flush_list,
struct ceph_cap_flush, g_list);
u64 want_flush = mdsc->last_cap_flush_tid;
long drain_ret;
/*
* Setting wake on the last entry is sufficient: flush
* entries complete in order, so when this entry finishes
* all earlier ones are already done.
*/
cf->wake = true;
spin_unlock(&mdsc->cap_dirty_lock);
pr_info_client(cl,
"draining (want_flush=%llu, %d sessions)\n",
want_flush, n);
drain_ret = wait_event_timeout(mdsc->cap_flushing_wq,
check_caps_flush(mdsc,
want_flush),
max_t(long,
drain_deadline - jiffies,
1));
if (drain_ret == 0) {
pr_info_client(cl,
"drain timed out, proceeding with forced teardown\n");
WRITE_ONCE(st->drain_timed_out, true);
} else {
pr_info_client(cl, "drain completed successfully\n");
}
} else {
spin_unlock(&mdsc->cap_dirty_lock);
}
spin_lock(&st->lock);
if (st->shutdown) {
spin_unlock(&st->lock);
goto out_sessions;
}
st->phase = CEPH_CLIENT_RESET_TEARDOWN;
spin_unlock(&st->lock);
/*
* Ask each MDS to close the session before we tear it down
* locally. Without this the MDS sees only a connection drop and
* waits for the client to reconnect (up to session_autoclose
* seconds) before evicting the session and releasing locks.
*
* Reuse the normal close machinery so the session state/sequence
* snapshot is serialized under s_mutex and a racing s_seq bump
* retransmits REQUEST_CLOSE while the session remains CLOSING.
* We send all close requests first, then yield briefly to let the
* network stack transmit them before __unregister_session()
* closes the connections.
*/
for (i = 0; i < n; i++) {
int err;
mutex_lock(&sessions[i]->s_mutex);
err = __close_session(mdsc, sessions[i]);
mutex_unlock(&sessions[i]->s_mutex);
if (err < 0)
pr_warn_client(cl,
"mds%d failed to queue close request before reset: %d\n",
sessions[i]->s_mds, err);
}
/*
* Best-effort grace period: yield briefly so the network stack
* can transmit the queued REQUEST_CLOSE messages before we tear
* down connections. Not a correctness requirement -- the MDS
* will still evict via session_autoclose if it never receives
* the close request.
*
* Event-based waiting is not viable here: there is no completion
* event for "message left the NIC," and waiting for the MDS
* SESSION_CLOSE response would re-create the stalemate that the
* reset is meant to break.
*/
if (n > 0)
msleep(CEPH_CLIENT_RESET_CLOSE_GRACE_MS);
/*
* Tear down each session: close the connection, remove all
* caps, clean up requests, then kick pending requests so they
* re-open a fresh session on the next attempt.
*
* This is modeled on the check_new_map() forced-close path
* for stopped MDS ranks - a proven pattern for hard session
* teardown. We do NOT attempt send_mds_reconnect() because
* the MDS only accepts reconnects during its own RECONNECT
* phase (after MDS restart), not from an active client.
*
* Any state that did not drain (caps that didn't flush, unsafe
* requests that the MDS didn't journal) is force-dropped here.
* This is intentional: that state is stuck and is the reason
* the operator triggered the reset.
*/
for (i = 0; i < n; i++) {
int mds = sessions[i]->s_mds;
pr_info_client(cl, "mds%d resetting session\n", mds);
mutex_lock(&mdsc->mutex);
if (mds >= mdsc->max_sessions ||
mdsc->sessions[mds] != sessions[i]) {
pr_info_client(cl,
"mds%d session already torn down, skipping\n",
mds);
mutex_unlock(&mdsc->mutex);
ceph_put_mds_session(sessions[i]);
sessions[i] = NULL;
continue;
}
sessions[i]->s_state = CEPH_MDS_SESSION_CLOSED;
__unregister_session(mdsc, sessions[i]);
__wake_requests(mdsc, &sessions[i]->s_waiting);
mutex_unlock(&mdsc->mutex);
mutex_lock(&sessions[i]->s_mutex);
cleanup_session_requests(mdsc, sessions[i]);
remove_session_caps(sessions[i]);
mutex_unlock(&sessions[i]->s_mutex);
wake_up_all(&mdsc->session_close_wq);
ceph_put_mds_session(sessions[i]);
mutex_lock(&mdsc->mutex);
kick_requests(mdsc, mds);
mutex_unlock(&mdsc->mutex);
torn_down++;
pr_info_client(cl, "mds%d session reset complete\n", mds);
}
kfree(sessions);
spin_lock(&st->lock);
st->sessions_reset = torn_down;
spin_unlock(&st->lock);
out_complete:
ceph_mdsc_reset_complete(mdsc, ret);
return;
out_sessions:
/* shutdown == true: ceph_mdsc_destroy() owns the final transition. */
for (i = 0; i < n; i++)
ceph_put_mds_session(sessions[i]);
kfree(sessions);
}
int ceph_mdsc_schedule_reset(struct ceph_mds_client *mdsc,
const char *reason)
{
struct ceph_client_reset_state *st = &mdsc->reset_state;
struct ceph_fs_client *fsc = mdsc->fsc;
const char *msg = (reason && reason[0]) ? reason : "manual";
int mount_state;
mount_state = READ_ONCE(fsc->mount_state);
if (mount_state != CEPH_MOUNT_MOUNTED) {
pr_warn_client(fsc->client,
"reset rejected: mount_state=%d (not mounted)\n",
mount_state);
return -EINVAL;
}
spin_lock(&st->lock);
if (st->phase != CEPH_CLIENT_RESET_IDLE) {
spin_unlock(&st->lock);
return -EBUSY;
}
st->phase = CEPH_CLIENT_RESET_QUIESCING;
st->last_start = jiffies;
st->last_errno = 0;
st->drain_timed_out = false;
st->sessions_reset = 0;
st->trigger_count++;
strscpy(st->last_reason, msg, sizeof(st->last_reason));
spin_unlock(&st->lock);
if (WARN_ON_ONCE(!queue_work(system_unbound_wq, &mdsc->reset_work))) {
spin_lock(&st->lock);
st->phase = CEPH_CLIENT_RESET_IDLE;
st->last_errno = -EALREADY;
st->last_finish = jiffies;
st->failure_count++;
spin_unlock(&st->lock);
wake_up_all(&st->blocked_wq);
return -EALREADY;
}
pr_info_client(mdsc->fsc->client,
"manual session reset scheduled (reason=\"%s\")\n",
msg);
trace_ceph_client_reset_schedule(mdsc, msg);
return 0;
}
@ -5194,9 +5838,15 @@ static void check_new_map(struct ceph_mds_client *mdsc,
*/
if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
newstate >= CEPH_MDS_STATE_RECONNECT) {
int rc;
mutex_unlock(&mdsc->mutex);
clear_bit(i, targets);
send_mds_reconnect(mdsc, s);
rc = send_mds_reconnect(mdsc, s);
if (rc)
pr_warn_client(cl,
"mds%d reconnect failed: %d\n",
i, rc);
mutex_lock(&mdsc->mutex);
}
@ -5260,7 +5910,11 @@ static void check_new_map(struct ceph_mds_client *mdsc,
}
doutc(cl, "send reconnect to export target mds.%d\n", i);
mutex_unlock(&mdsc->mutex);
send_mds_reconnect(mdsc, s);
err = send_mds_reconnect(mdsc, s);
if (err)
pr_warn_client(cl,
"mds%d export target reconnect failed: %d\n",
i, err);
ceph_put_mds_session(s);
mutex_lock(&mdsc->mutex);
}
@ -5651,6 +6305,11 @@ int ceph_mdsc_init(struct ceph_fs_client *fsc)
INIT_LIST_HEAD(&mdsc->dentry_leases);
INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
spin_lock_init(&mdsc->reset_state.lock);
init_waitqueue_head(&mdsc->reset_state.blocked_wq);
atomic_set(&mdsc->reset_state.blocked_requests, 0);
INIT_WORK(&mdsc->reset_work, ceph_mdsc_reset_workfn);
ceph_caps_init(mdsc);
ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
@ -6176,6 +6835,23 @@ void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
/* flush out any connection work with references to us */
ceph_msgr_flush();
/*
* Mark reset as failed and wake any blocked waiters before
* cancelling, so unmount doesn't stall on blocked_wq timeout
* if cancel_work_sync() prevents the work from running.
*/
spin_lock(&mdsc->reset_state.lock);
mdsc->reset_state.shutdown = true;
if (mdsc->reset_state.phase != CEPH_CLIENT_RESET_IDLE) {
mdsc->reset_state.phase = CEPH_CLIENT_RESET_IDLE;
mdsc->reset_state.last_errno = -ESHUTDOWN;
mdsc->reset_state.last_finish = jiffies;
mdsc->reset_state.failure_count++;
}
spin_unlock(&mdsc->reset_state.lock);
wake_up_all(&mdsc->reset_state.blocked_wq);
cancel_work_sync(&mdsc->reset_work);
ceph_mdsc_stop(mdsc);
ceph_metric_destroy(&mdsc->metric);
@ -6348,12 +7024,92 @@ static void mds_peer_reset(struct ceph_connection *con)
{
struct ceph_mds_session *s = con->private;
struct ceph_mds_client *mdsc = s->s_mdsc;
int session_state;
pr_warn_client(mdsc->fsc->client, "mds%d closed our session\n",
s->s_mds);
if (READ_ONCE(mdsc->fsc->mount_state) != CEPH_MOUNT_FENCE_IO &&
ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) >= CEPH_MDS_STATE_RECONNECT)
send_mds_reconnect(mdsc, s);
if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO ||
ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) < CEPH_MDS_STATE_RECONNECT)
return;
/*
* Only reconnect if MDS is in its RECONNECT phase. An MDS past
* RECONNECT (REJOIN, CLIENTREPLAY, ACTIVE) will reject reconnect
* attempts, so those states fall through to session teardown below.
*/
if (ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) == CEPH_MDS_STATE_RECONNECT) {
int rc = send_mds_reconnect(mdsc, s);
if (rc)
pr_warn_client(mdsc->fsc->client,
"mds%d reconnect failed: %d\n",
s->s_mds, rc);
return;
}
/*
* MDS is active (past RECONNECT). It will not accept a
* CLIENT_RECONNECT from us, so tear the session down locally
* and let new requests re-open a fresh session.
*
* Snapshot session state with READ_ONCE, then revalidate under
* mdsc->mutex before acting. The subsequent mdsc->mutex
* section rechecks s_state to catch concurrent transitions, so
* the lockless snapshot here is safe. s->s_mutex is taken
* separately for cleanup after unregistration, which avoids
* introducing a new s->s_mutex + mdsc->mutex nesting.
*/
session_state = READ_ONCE(s->s_state);
switch (session_state) {
case CEPH_MDS_SESSION_RESTARTING:
case CEPH_MDS_SESSION_RECONNECTING:
case CEPH_MDS_SESSION_CLOSING:
case CEPH_MDS_SESSION_OPEN:
case CEPH_MDS_SESSION_HUNG:
case CEPH_MDS_SESSION_OPENING:
mutex_lock(&mdsc->mutex);
if (s->s_mds >= mdsc->max_sessions ||
mdsc->sessions[s->s_mds] != s ||
s->s_state != session_state) {
pr_info_client(mdsc->fsc->client,
"mds%d state changed to %s during peer reset\n",
s->s_mds,
ceph_session_state_name(s->s_state));
mutex_unlock(&mdsc->mutex);
return;
}
ceph_get_mds_session(s);
s->s_state = CEPH_MDS_SESSION_CLOSED;
__unregister_session(mdsc, s);
__wake_requests(mdsc, &s->s_waiting);
mutex_unlock(&mdsc->mutex);
mutex_lock(&s->s_mutex);
cleanup_session_requests(mdsc, s);
remove_session_caps(s);
mutex_unlock(&s->s_mutex);
wake_up_all(&mdsc->session_close_wq);
mutex_lock(&mdsc->mutex);
kick_requests(mdsc, s->s_mds);
mutex_unlock(&mdsc->mutex);
ceph_put_mds_session(s);
break;
case CEPH_MDS_SESSION_CLOSED:
case CEPH_MDS_SESSION_REJECTED:
break;
default:
pr_warn_client(mdsc->fsc->client,
"mds%d peer reset in unexpected state %s\n",
s->s_mds,
ceph_session_state_name(session_state));
break;
}
}
static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
@ -6365,6 +7121,8 @@ static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
mutex_lock(&mdsc->mutex);
if (__verify_registered_session(mdsc, s) < 0) {
doutc(cl, "dropping tid %llu from unregistered session %d\n",
le64_to_cpu(msg->hdr.tid), s->s_mds);
mutex_unlock(&mdsc->mutex);
goto out;
}

52
fs/ceph/mds_client.h
View File

@ -77,6 +77,50 @@ struct ceph_fs_client;
struct ceph_cap;
#define MDS_AUTH_UID_ANY -1
#define CEPH_CAP_FLUSH_WAIT_TIMEOUT_SEC 60
#define CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES 5
#define CEPH_CAP_FLUSH_MAX_DUMP_ITERS 5
#define CEPH_CLIENT_RESET_REASON_LEN 64
#define CEPH_CLIENT_RESET_DRAIN_SEC 30
#define CEPH_CLIENT_RESET_CLOSE_GRACE_MS 100
#define CEPH_CLIENT_RESET_WAIT_TIMEOUT_SEC 120
enum ceph_client_reset_phase {
CEPH_CLIENT_RESET_IDLE = 0,
/*
* QUIESCING is set synchronously by schedule_reset() before the
* workqueue item is dispatched. It gates new requests (any
* phase != IDLE blocks callers) during the window between
* scheduling and the work function's transition to DRAINING.
*/
CEPH_CLIENT_RESET_QUIESCING,
CEPH_CLIENT_RESET_DRAINING,
CEPH_CLIENT_RESET_TEARDOWN,
};
struct ceph_client_reset_state {
spinlock_t lock; /* protects all fields below */
u64 trigger_count; /* number of resets triggered */
u64 success_count; /* number of successful resets */
u64 failure_count; /* number of failed resets */
unsigned long last_start; /* jiffies when last reset started */
unsigned long last_finish; /* jiffies when last reset finished */
int last_errno; /* result of most recent reset */
enum ceph_client_reset_phase phase; /* current reset phase */
bool drain_timed_out; /* drain exceeded timeout */
bool shutdown; /* destroy in progress */
int sessions_reset; /* sessions torn down in last reset */
char last_reason[CEPH_CLIENT_RESET_REASON_LEN]; /* operator-supplied reason */
/* Request blocking during reset */
wait_queue_head_t blocked_wq; /* waitqueue for blocked callers */
atomic_t blocked_requests; /* count of blocked callers */
};
static inline bool ceph_reset_is_idle(struct ceph_client_reset_state *st)
{
return READ_ONCE(st->phase) == CEPH_CLIENT_RESET_IDLE;
}
struct ceph_mds_cap_match {
s64 uid; /* default to MDS_AUTH_UID_ANY */
@ -540,6 +584,8 @@ struct ceph_mds_client {
struct list_head dentry_dir_leases; /* lru list */
struct ceph_client_metric metric;
struct work_struct reset_work;
struct ceph_client_reset_state reset_state;
struct ceph_subvolume_metrics_tracker subvol_metrics;
/* Subvolume metrics send tracking */
@ -571,10 +617,14 @@ extern struct ceph_mds_session *
__ceph_lookup_mds_session(struct ceph_mds_client *, int mds);
extern const char *ceph_session_state_name(int s);
extern const char *ceph_reset_phase_name(enum ceph_client_reset_phase phase);
extern struct ceph_mds_session *
ceph_get_mds_session(struct ceph_mds_session *s);
extern void ceph_put_mds_session(struct ceph_mds_session *s);
int ceph_mdsc_schedule_reset(struct ceph_mds_client *mdsc,
const char *reason);
int ceph_mdsc_wait_for_reset(struct ceph_mds_client *mdsc);
extern int ceph_mdsc_init(struct ceph_fs_client *fsc);
extern void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc);
@ -670,7 +720,7 @@ static inline int ceph_wait_on_async_create(struct inode *inode)
{
struct ceph_inode_info *ci = ceph_inode(inode);
return wait_on_bit(&ci->i_ceph_flags, CEPH_ASYNC_CREATE_BIT,
return wait_on_bit(&ci->i_ceph_flags, CEPH_I_ASYNC_CREATE_BIT,
TASK_KILLABLE);
}

2
fs/ceph/snap.c
View File

@ -700,7 +700,7 @@ int __ceph_finish_cap_snap(struct ceph_inode_info *ci,
return 0;
}
ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
set_bit(CEPH_I_FLUSH_SNAPS_BIT, &ci->i_ceph_flags);
doutc(cl, "%p %llx.%llx cap_snap %p snapc %p %llu %s s=%llu\n",
inode, ceph_vinop(inode), capsnap, capsnap->context,
capsnap->context->seq, ceph_cap_string(capsnap->dirty),

70
fs/ceph/super.h
View File

@ -179,6 +179,7 @@ struct ceph_fs_client {
struct dentry *debugfs_status;
struct dentry *debugfs_mds_sessions;
struct dentry *debugfs_metrics_dir;
struct dentry *debugfs_reset_dir;
struct dentry *debugfs_subvolume_metrics;
#endif
@ -239,6 +240,7 @@ struct ceph_cap_flush {
bool is_capsnap; /* true means capsnap */
struct list_head g_list; // global
struct list_head i_list; // per inode
struct ceph_inode_info *ci;
};
/*
@ -453,6 +455,11 @@ struct ceph_inode_info {
struct ceph_snap_context *i_head_snapc; /* set if wr_buffer_head > 0 or
dirty|flushing caps */
unsigned i_snap_caps; /* cap bits for snapped files */
/*
* Written under i_ceph_lock, read via READ_ONCE()
* from diagnostic paths.
*/
u64 i_last_cap_flush_ack;
unsigned long i_last_rd;
unsigned long i_last_wr;
@ -665,23 +672,33 @@ static inline struct inode *ceph_find_inode(struct super_block *sb,
/*
* Ceph inode.
*/
#define CEPH_I_DIR_ORDERED (1 << 0) /* dentries in dir are ordered */
#define CEPH_I_FLUSH (1 << 2) /* do not delay flush of dirty metadata */
#define CEPH_I_POOL_PERM (1 << 3) /* pool rd/wr bits are valid */
#define CEPH_I_POOL_RD (1 << 4) /* can read from pool */
#define CEPH_I_POOL_WR (1 << 5) /* can write to pool */
#define CEPH_I_SEC_INITED (1 << 6) /* security initialized */
#define CEPH_I_KICK_FLUSH (1 << 7) /* kick flushing caps */
#define CEPH_I_FLUSH_SNAPS (1 << 8) /* need flush snapss */
#define CEPH_I_ERROR_WRITE (1 << 9) /* have seen write errors */
#define CEPH_I_ERROR_FILELOCK (1 << 10) /* have seen file lock errors */
#define CEPH_I_ODIRECT_BIT (11) /* inode in direct I/O mode */
#define CEPH_I_ODIRECT (1 << CEPH_I_ODIRECT_BIT)
#define CEPH_ASYNC_CREATE_BIT (12) /* async create in flight for this */
#define CEPH_I_ASYNC_CREATE (1 << CEPH_ASYNC_CREATE_BIT)
#define CEPH_I_SHUTDOWN (1 << 13) /* inode is no longer usable */
#define CEPH_I_ASYNC_CHECK_CAPS (1 << 14) /* check caps immediately after async
creating finishes */
#define CEPH_I_DIR_ORDERED_BIT (0) /* dentries in dir are ordered */
/* bit 1 historically unused */
#define CEPH_I_FLUSH_BIT (2) /* do not delay flush of dirty metadata */
#define CEPH_I_POOL_PERM_BIT (3) /* pool rd/wr bits are valid */
#define CEPH_I_POOL_RD_BIT (4) /* can read from pool */
#define CEPH_I_POOL_WR_BIT (5) /* can write to pool */
#define CEPH_I_SEC_INITED_BIT (6) /* security initialized */
#define CEPH_I_KICK_FLUSH_BIT (7) /* kick flushing caps */
#define CEPH_I_FLUSH_SNAPS_BIT (8) /* need flush snaps */
#define CEPH_I_ERROR_WRITE_BIT (9) /* have seen write errors */
#define CEPH_I_ERROR_FILELOCK_BIT (10) /* have seen file lock errors */
#define CEPH_I_ODIRECT_BIT (11) /* inode in direct I/O mode */
#define CEPH_I_ASYNC_CREATE_BIT (12) /* async create in flight for this */
#define CEPH_I_SHUTDOWN_BIT (13) /* inode is no longer usable */
#define CEPH_I_ASYNC_CHECK_CAPS_BIT (14) /* check caps after async creating finishes */
#define CEPH_I_DIR_ORDERED (1 << CEPH_I_DIR_ORDERED_BIT)
#define CEPH_I_FLUSH (1 << CEPH_I_FLUSH_BIT)
#define CEPH_I_POOL_PERM (1 << CEPH_I_POOL_PERM_BIT)
#define CEPH_I_POOL_RD (1 << CEPH_I_POOL_RD_BIT)
#define CEPH_I_POOL_WR (1 << CEPH_I_POOL_WR_BIT)
#define CEPH_I_SEC_INITED (1 << CEPH_I_SEC_INITED_BIT)
#define CEPH_I_KICK_FLUSH (1 << CEPH_I_KICK_FLUSH_BIT)
#define CEPH_I_FLUSH_SNAPS (1 << CEPH_I_FLUSH_SNAPS_BIT)
#define CEPH_I_ODIRECT (1 << CEPH_I_ODIRECT_BIT)
#define CEPH_I_ASYNC_CREATE (1 << CEPH_I_ASYNC_CREATE_BIT)
#define CEPH_I_SHUTDOWN (1 << CEPH_I_SHUTDOWN_BIT)
/*
* Masks of ceph inode work.
@ -694,27 +711,18 @@ static inline struct inode *ceph_find_inode(struct super_block *sb,
/*
* We set the ERROR_WRITE bit when we start seeing write errors on an inode
* and then clear it when they start succeeding. Note that we do a lockless
* check first, and only take the lock if it looks like it needs to be changed.
* The write submission code just takes this as a hint, so we're not too
* worried if a few slip through in either direction.
* and then clear it when they start succeeding. The write submission code
* just takes this as a hint, so we're not too worried if a few slip through
* in either direction.
*/
static inline void ceph_set_error_write(struct ceph_inode_info *ci)
{
if (!(READ_ONCE(ci->i_ceph_flags) & CEPH_I_ERROR_WRITE)) {
spin_lock(&ci->i_ceph_lock);
ci->i_ceph_flags |= CEPH_I_ERROR_WRITE;
spin_unlock(&ci->i_ceph_lock);
}
set_bit(CEPH_I_ERROR_WRITE_BIT, &ci->i_ceph_flags);
}
static inline void ceph_clear_error_write(struct ceph_inode_info *ci)
{
if (READ_ONCE(ci->i_ceph_flags) & CEPH_I_ERROR_WRITE) {
spin_lock(&ci->i_ceph_lock);
ci->i_ceph_flags &= ~CEPH_I_ERROR_WRITE;
spin_unlock(&ci->i_ceph_lock);
}
clear_bit(CEPH_I_ERROR_WRITE_BIT, &ci->i_ceph_flags);
}
static inline void __ceph_dir_set_complete(struct ceph_inode_info *ci,

2
fs/ceph/xattr.c
View File

@ -1054,7 +1054,7 @@ ssize_t __ceph_getxattr(struct inode *inode, const char *name, void *value,
if (current->journal_info &&
!strncmp(name, XATTR_SECURITY_PREFIX, XATTR_SECURITY_PREFIX_LEN) &&
security_ismaclabel(name + XATTR_SECURITY_PREFIX_LEN))
ci->i_ceph_flags |= CEPH_I_SEC_INITED;
set_bit(CEPH_I_SEC_INITED_BIT, &ci->i_ceph_flags);
out:
spin_unlock(&ci->i_ceph_lock);
return err;

67
include/trace/events/ceph.h
View File

@ -226,6 +226,73 @@ TRACE_EVENT(ceph_handle_caps,
__entry->mseq)
);
/*
* Client reset tracepoints - identify the client by its monitor-
* assigned global_id so traces remain meaningful when kernel pointer
* hashing is enabled.
*/
TRACE_EVENT(ceph_client_reset_schedule,
TP_PROTO(const struct ceph_mds_client *mdsc, const char *reason),
TP_ARGS(mdsc, reason),
TP_STRUCT__entry(
__field(u64, client_id)
__string(reason, reason ? reason : "")
),
TP_fast_assign(
__entry->client_id = mdsc->fsc->client->monc.auth ?
mdsc->fsc->client->monc.auth->global_id : 0;
__assign_str(reason);
),
TP_printk("client_id=%llu reason=%s",
__entry->client_id, __get_str(reason))
);
TRACE_EVENT(ceph_client_reset_complete,
TP_PROTO(const struct ceph_mds_client *mdsc, int ret),
TP_ARGS(mdsc, ret),
TP_STRUCT__entry(
__field(u64, client_id)
__field(int, ret)
),
TP_fast_assign(
__entry->client_id = mdsc->fsc->client->monc.auth ?
mdsc->fsc->client->monc.auth->global_id : 0;
__entry->ret = ret;
),
TP_printk("client_id=%llu ret=%d", __entry->client_id, __entry->ret)
);
TRACE_EVENT(ceph_client_reset_blocked,
TP_PROTO(const struct ceph_mds_client *mdsc, int blocked_count),
TP_ARGS(mdsc, blocked_count),
TP_STRUCT__entry(
__field(u64, client_id)
__field(int, blocked_count)
),
TP_fast_assign(
__entry->client_id = mdsc->fsc->client->monc.auth ?
mdsc->fsc->client->monc.auth->global_id : 0;
__entry->blocked_count = blocked_count;
),
TP_printk("client_id=%llu blocked_count=%d", __entry->client_id,
__entry->blocked_count)
);
TRACE_EVENT(ceph_client_reset_unblocked,
TP_PROTO(const struct ceph_mds_client *mdsc, int ret),
TP_ARGS(mdsc, ret),
TP_STRUCT__entry(
__field(u64, client_id)
__field(int, ret)
),
TP_fast_assign(
__entry->client_id = mdsc->fsc->client->monc.auth ?
mdsc->fsc->client->monc.auth->global_id : 0;
__entry->ret = ret;
),
TP_printk("client_id=%llu ret=%d", __entry->client_id, __entry->ret)
);
#undef EM
#undef E_
#endif /* _TRACE_CEPH_H */