1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
5 #include <linux/wait.h>
6 #include <linux/slab.h>
8 #include <linux/sched.h>
9 #include <linux/debugfs.h>
10 #include <linux/seq_file.h>
11 #include <linux/ratelimit.h>
14 #include "mds_client.h"
16 #include <linux/ceph/ceph_features.h>
17 #include <linux/ceph/messenger.h>
18 #include <linux/ceph/decode.h>
19 #include <linux/ceph/pagelist.h>
20 #include <linux/ceph/auth.h>
21 #include <linux/ceph/debugfs.h>
24 * A cluster of MDS (metadata server) daemons is responsible for
25 * managing the file system namespace (the directory hierarchy and
26 * inodes) and for coordinating shared access to storage. Metadata is
27 * partitioning hierarchically across a number of servers, and that
28 * partition varies over time as the cluster adjusts the distribution
29 * in order to balance load.
31 * The MDS client is primarily responsible to managing synchronous
32 * metadata requests for operations like open, unlink, and so forth.
33 * If there is a MDS failure, we find out about it when we (possibly
34 * request and) receive a new MDS map, and can resubmit affected
37 * For the most part, though, we take advantage of a lossless
38 * communications channel to the MDS, and do not need to worry about
39 * timing out or resubmitting requests.
41 * We maintain a stateful "session" with each MDS we interact with.
42 * Within each session, we sent periodic heartbeat messages to ensure
43 * any capabilities or leases we have been issues remain valid. If
44 * the session times out and goes stale, our leases and capabilities
45 * are no longer valid.
48 struct ceph_reconnect_state
{
50 struct ceph_pagelist
*pagelist
;
54 static void __wake_requests(struct ceph_mds_client
*mdsc
,
55 struct list_head
*head
);
57 static const struct ceph_connection_operations mds_con_ops
;
65 * parse individual inode info
67 static int parse_reply_info_in(void **p
, void *end
,
68 struct ceph_mds_reply_info_in
*info
,
74 *p
+= sizeof(struct ceph_mds_reply_inode
) +
75 sizeof(*info
->in
->fragtree
.splits
) *
76 le32_to_cpu(info
->in
->fragtree
.nsplits
);
78 ceph_decode_32_safe(p
, end
, info
->symlink_len
, bad
);
79 ceph_decode_need(p
, end
, info
->symlink_len
, bad
);
81 *p
+= info
->symlink_len
;
83 if (features
& CEPH_FEATURE_DIRLAYOUTHASH
)
84 ceph_decode_copy_safe(p
, end
, &info
->dir_layout
,
85 sizeof(info
->dir_layout
), bad
);
87 memset(&info
->dir_layout
, 0, sizeof(info
->dir_layout
));
89 ceph_decode_32_safe(p
, end
, info
->xattr_len
, bad
);
90 ceph_decode_need(p
, end
, info
->xattr_len
, bad
);
91 info
->xattr_data
= *p
;
92 *p
+= info
->xattr_len
;
94 if (features
& CEPH_FEATURE_MDS_INLINE_DATA
) {
95 ceph_decode_64_safe(p
, end
, info
->inline_version
, bad
);
96 ceph_decode_32_safe(p
, end
, info
->inline_len
, bad
);
97 ceph_decode_need(p
, end
, info
->inline_len
, bad
);
98 info
->inline_data
= *p
;
99 *p
+= info
->inline_len
;
101 info
->inline_version
= CEPH_INLINE_NONE
;
103 if (features
& CEPH_FEATURE_MDS_QUOTA
) {
104 u8 struct_v
, struct_compat
;
108 * both struct_v and struct_compat are expected to be >= 1
110 ceph_decode_8_safe(p
, end
, struct_v
, bad
);
111 ceph_decode_8_safe(p
, end
, struct_compat
, bad
);
112 if (!struct_v
|| !struct_compat
)
114 ceph_decode_32_safe(p
, end
, struct_len
, bad
);
115 ceph_decode_need(p
, end
, struct_len
, bad
);
116 ceph_decode_64_safe(p
, end
, info
->max_bytes
, bad
);
117 ceph_decode_64_safe(p
, end
, info
->max_files
, bad
);
123 info
->pool_ns_len
= 0;
124 info
->pool_ns_data
= NULL
;
125 if (features
& CEPH_FEATURE_FS_FILE_LAYOUT_V2
) {
126 ceph_decode_32_safe(p
, end
, info
->pool_ns_len
, bad
);
127 if (info
->pool_ns_len
> 0) {
128 ceph_decode_need(p
, end
, info
->pool_ns_len
, bad
);
129 info
->pool_ns_data
= *p
;
130 *p
+= info
->pool_ns_len
;
140 * parse a normal reply, which may contain a (dir+)dentry and/or a
143 static int parse_reply_info_trace(void **p
, void *end
,
144 struct ceph_mds_reply_info_parsed
*info
,
149 if (info
->head
->is_dentry
) {
150 err
= parse_reply_info_in(p
, end
, &info
->diri
, features
);
154 if (unlikely(*p
+ sizeof(*info
->dirfrag
) > end
))
157 *p
+= sizeof(*info
->dirfrag
) +
158 sizeof(u32
)*le32_to_cpu(info
->dirfrag
->ndist
);
159 if (unlikely(*p
> end
))
162 ceph_decode_32_safe(p
, end
, info
->dname_len
, bad
);
163 ceph_decode_need(p
, end
, info
->dname_len
, bad
);
165 *p
+= info
->dname_len
;
167 *p
+= sizeof(*info
->dlease
);
170 if (info
->head
->is_target
) {
171 err
= parse_reply_info_in(p
, end
, &info
->targeti
, features
);
176 if (unlikely(*p
!= end
))
183 pr_err("problem parsing mds trace %d\n", err
);
188 * parse readdir results
190 static int parse_reply_info_dir(void **p
, void *end
,
191 struct ceph_mds_reply_info_parsed
*info
,
198 if (*p
+ sizeof(*info
->dir_dir
) > end
)
200 *p
+= sizeof(*info
->dir_dir
) +
201 sizeof(u32
)*le32_to_cpu(info
->dir_dir
->ndist
);
205 ceph_decode_need(p
, end
, sizeof(num
) + 2, bad
);
206 num
= ceph_decode_32(p
);
208 u16 flags
= ceph_decode_16(p
);
209 info
->dir_end
= !!(flags
& CEPH_READDIR_FRAG_END
);
210 info
->dir_complete
= !!(flags
& CEPH_READDIR_FRAG_COMPLETE
);
211 info
->hash_order
= !!(flags
& CEPH_READDIR_HASH_ORDER
);
212 info
->offset_hash
= !!(flags
& CEPH_READDIR_OFFSET_HASH
);
217 BUG_ON(!info
->dir_entries
);
218 if ((unsigned long)(info
->dir_entries
+ num
) >
219 (unsigned long)info
->dir_entries
+ info
->dir_buf_size
) {
220 pr_err("dir contents are larger than expected\n");
227 struct ceph_mds_reply_dir_entry
*rde
= info
->dir_entries
+ i
;
229 ceph_decode_need(p
, end
, sizeof(u32
)*2, bad
);
230 rde
->name_len
= ceph_decode_32(p
);
231 ceph_decode_need(p
, end
, rde
->name_len
, bad
);
234 dout("parsed dir dname '%.*s'\n", rde
->name_len
, rde
->name
);
236 *p
+= sizeof(struct ceph_mds_reply_lease
);
239 err
= parse_reply_info_in(p
, end
, &rde
->inode
, features
);
242 /* ceph_readdir_prepopulate() will update it */
256 pr_err("problem parsing dir contents %d\n", err
);
261 * parse fcntl F_GETLK results
263 static int parse_reply_info_filelock(void **p
, void *end
,
264 struct ceph_mds_reply_info_parsed
*info
,
267 if (*p
+ sizeof(*info
->filelock_reply
) > end
)
270 info
->filelock_reply
= *p
;
271 *p
+= sizeof(*info
->filelock_reply
);
273 if (unlikely(*p
!= end
))
282 * parse create results
284 static int parse_reply_info_create(void **p
, void *end
,
285 struct ceph_mds_reply_info_parsed
*info
,
288 if (features
& CEPH_FEATURE_REPLY_CREATE_INODE
) {
290 info
->has_create_ino
= false;
292 info
->has_create_ino
= true;
293 info
->ino
= ceph_decode_64(p
);
297 if (unlikely(*p
!= end
))
306 * parse extra results
308 static int parse_reply_info_extra(void **p
, void *end
,
309 struct ceph_mds_reply_info_parsed
*info
,
312 u32 op
= le32_to_cpu(info
->head
->op
);
314 if (op
== CEPH_MDS_OP_GETFILELOCK
)
315 return parse_reply_info_filelock(p
, end
, info
, features
);
316 else if (op
== CEPH_MDS_OP_READDIR
|| op
== CEPH_MDS_OP_LSSNAP
)
317 return parse_reply_info_dir(p
, end
, info
, features
);
318 else if (op
== CEPH_MDS_OP_CREATE
)
319 return parse_reply_info_create(p
, end
, info
, features
);
325 * parse entire mds reply
327 static int parse_reply_info(struct ceph_msg
*msg
,
328 struct ceph_mds_reply_info_parsed
*info
,
335 info
->head
= msg
->front
.iov_base
;
336 p
= msg
->front
.iov_base
+ sizeof(struct ceph_mds_reply_head
);
337 end
= p
+ msg
->front
.iov_len
- sizeof(struct ceph_mds_reply_head
);
340 ceph_decode_32_safe(&p
, end
, len
, bad
);
342 ceph_decode_need(&p
, end
, len
, bad
);
343 err
= parse_reply_info_trace(&p
, p
+len
, info
, features
);
349 ceph_decode_32_safe(&p
, end
, len
, bad
);
351 ceph_decode_need(&p
, end
, len
, bad
);
352 err
= parse_reply_info_extra(&p
, p
+len
, info
, features
);
358 ceph_decode_32_safe(&p
, end
, len
, bad
);
359 info
->snapblob_len
= len
;
370 pr_err("mds parse_reply err %d\n", err
);
374 static void destroy_reply_info(struct ceph_mds_reply_info_parsed
*info
)
376 if (!info
->dir_entries
)
378 free_pages((unsigned long)info
->dir_entries
, get_order(info
->dir_buf_size
));
385 const char *ceph_session_state_name(int s
)
388 case CEPH_MDS_SESSION_NEW
: return "new";
389 case CEPH_MDS_SESSION_OPENING
: return "opening";
390 case CEPH_MDS_SESSION_OPEN
: return "open";
391 case CEPH_MDS_SESSION_HUNG
: return "hung";
392 case CEPH_MDS_SESSION_CLOSING
: return "closing";
393 case CEPH_MDS_SESSION_RESTARTING
: return "restarting";
394 case CEPH_MDS_SESSION_RECONNECTING
: return "reconnecting";
395 case CEPH_MDS_SESSION_REJECTED
: return "rejected";
396 default: return "???";
400 static struct ceph_mds_session
*get_session(struct ceph_mds_session
*s
)
402 if (refcount_inc_not_zero(&s
->s_ref
)) {
403 dout("mdsc get_session %p %d -> %d\n", s
,
404 refcount_read(&s
->s_ref
)-1, refcount_read(&s
->s_ref
));
407 dout("mdsc get_session %p 0 -- FAIL\n", s
);
412 void ceph_put_mds_session(struct ceph_mds_session
*s
)
414 dout("mdsc put_session %p %d -> %d\n", s
,
415 refcount_read(&s
->s_ref
), refcount_read(&s
->s_ref
)-1);
416 if (refcount_dec_and_test(&s
->s_ref
)) {
417 if (s
->s_auth
.authorizer
)
418 ceph_auth_destroy_authorizer(s
->s_auth
.authorizer
);
424 * called under mdsc->mutex
426 struct ceph_mds_session
*__ceph_lookup_mds_session(struct ceph_mds_client
*mdsc
,
429 struct ceph_mds_session
*session
;
431 if (mds
>= mdsc
->max_sessions
|| !mdsc
->sessions
[mds
])
433 session
= mdsc
->sessions
[mds
];
434 dout("lookup_mds_session %p %d\n", session
,
435 refcount_read(&session
->s_ref
));
436 get_session(session
);
440 static bool __have_session(struct ceph_mds_client
*mdsc
, int mds
)
442 if (mds
>= mdsc
->max_sessions
|| !mdsc
->sessions
[mds
])
448 static int __verify_registered_session(struct ceph_mds_client
*mdsc
,
449 struct ceph_mds_session
*s
)
451 if (s
->s_mds
>= mdsc
->max_sessions
||
452 mdsc
->sessions
[s
->s_mds
] != s
)
458 * create+register a new session for given mds.
459 * called under mdsc->mutex.
461 static struct ceph_mds_session
*register_session(struct ceph_mds_client
*mdsc
,
464 struct ceph_mds_session
*s
;
466 if (mds
>= mdsc
->mdsmap
->m_num_mds
)
467 return ERR_PTR(-EINVAL
);
469 s
= kzalloc(sizeof(*s
), GFP_NOFS
);
471 return ERR_PTR(-ENOMEM
);
473 if (mds
>= mdsc
->max_sessions
) {
474 int newmax
= 1 << get_count_order(mds
+ 1);
475 struct ceph_mds_session
**sa
;
477 dout("%s: realloc to %d\n", __func__
, newmax
);
478 sa
= kcalloc(newmax
, sizeof(void *), GFP_NOFS
);
481 if (mdsc
->sessions
) {
482 memcpy(sa
, mdsc
->sessions
,
483 mdsc
->max_sessions
* sizeof(void *));
484 kfree(mdsc
->sessions
);
487 mdsc
->max_sessions
= newmax
;
490 dout("%s: mds%d\n", __func__
, mds
);
493 s
->s_state
= CEPH_MDS_SESSION_NEW
;
496 mutex_init(&s
->s_mutex
);
498 ceph_con_init(&s
->s_con
, s
, &mds_con_ops
, &mdsc
->fsc
->client
->msgr
);
500 spin_lock_init(&s
->s_gen_ttl_lock
);
502 s
->s_cap_ttl
= jiffies
- 1;
504 spin_lock_init(&s
->s_cap_lock
);
505 s
->s_renew_requested
= 0;
507 INIT_LIST_HEAD(&s
->s_caps
);
510 refcount_set(&s
->s_ref
, 1);
511 INIT_LIST_HEAD(&s
->s_waiting
);
512 INIT_LIST_HEAD(&s
->s_unsafe
);
513 s
->s_num_cap_releases
= 0;
514 s
->s_cap_reconnect
= 0;
515 s
->s_cap_iterator
= NULL
;
516 INIT_LIST_HEAD(&s
->s_cap_releases
);
517 INIT_LIST_HEAD(&s
->s_cap_flushing
);
519 mdsc
->sessions
[mds
] = s
;
520 atomic_inc(&mdsc
->num_sessions
);
521 refcount_inc(&s
->s_ref
); /* one ref to sessions[], one to caller */
523 ceph_con_open(&s
->s_con
, CEPH_ENTITY_TYPE_MDS
, mds
,
524 ceph_mdsmap_get_addr(mdsc
->mdsmap
, mds
));
530 return ERR_PTR(-ENOMEM
);
534 * called under mdsc->mutex
536 static void __unregister_session(struct ceph_mds_client
*mdsc
,
537 struct ceph_mds_session
*s
)
539 dout("__unregister_session mds%d %p\n", s
->s_mds
, s
);
540 BUG_ON(mdsc
->sessions
[s
->s_mds
] != s
);
541 mdsc
->sessions
[s
->s_mds
] = NULL
;
542 ceph_con_close(&s
->s_con
);
543 ceph_put_mds_session(s
);
544 atomic_dec(&mdsc
->num_sessions
);
548 * drop session refs in request.
550 * should be last request ref, or hold mdsc->mutex
552 static void put_request_session(struct ceph_mds_request
*req
)
554 if (req
->r_session
) {
555 ceph_put_mds_session(req
->r_session
);
556 req
->r_session
= NULL
;
560 void ceph_mdsc_release_request(struct kref
*kref
)
562 struct ceph_mds_request
*req
= container_of(kref
,
563 struct ceph_mds_request
,
565 destroy_reply_info(&req
->r_reply_info
);
567 ceph_msg_put(req
->r_request
);
569 ceph_msg_put(req
->r_reply
);
571 ceph_put_cap_refs(ceph_inode(req
->r_inode
), CEPH_CAP_PIN
);
575 ceph_put_cap_refs(ceph_inode(req
->r_parent
), CEPH_CAP_PIN
);
576 iput(req
->r_target_inode
);
579 if (req
->r_old_dentry
)
580 dput(req
->r_old_dentry
);
581 if (req
->r_old_dentry_dir
) {
583 * track (and drop pins for) r_old_dentry_dir
584 * separately, since r_old_dentry's d_parent may have
585 * changed between the dir mutex being dropped and
586 * this request being freed.
588 ceph_put_cap_refs(ceph_inode(req
->r_old_dentry_dir
),
590 iput(req
->r_old_dentry_dir
);
595 ceph_pagelist_release(req
->r_pagelist
);
596 put_request_session(req
);
597 ceph_unreserve_caps(req
->r_mdsc
, &req
->r_caps_reservation
);
601 DEFINE_RB_FUNCS(request
, struct ceph_mds_request
, r_tid
, r_node
)
604 * lookup session, bump ref if found.
606 * called under mdsc->mutex.
608 static struct ceph_mds_request
*
609 lookup_get_request(struct ceph_mds_client
*mdsc
, u64 tid
)
611 struct ceph_mds_request
*req
;
613 req
= lookup_request(&mdsc
->request_tree
, tid
);
615 ceph_mdsc_get_request(req
);
621 * Register an in-flight request, and assign a tid. Link to directory
622 * are modifying (if any).
624 * Called under mdsc->mutex.
626 static void __register_request(struct ceph_mds_client
*mdsc
,
627 struct ceph_mds_request
*req
,
632 req
->r_tid
= ++mdsc
->last_tid
;
633 if (req
->r_num_caps
) {
634 ret
= ceph_reserve_caps(mdsc
, &req
->r_caps_reservation
,
637 pr_err("__register_request %p "
638 "failed to reserve caps: %d\n", req
, ret
);
639 /* set req->r_err to fail early from __do_request */
644 dout("__register_request %p tid %lld\n", req
, req
->r_tid
);
645 ceph_mdsc_get_request(req
);
646 insert_request(&mdsc
->request_tree
, req
);
648 req
->r_uid
= current_fsuid();
649 req
->r_gid
= current_fsgid();
651 if (mdsc
->oldest_tid
== 0 && req
->r_op
!= CEPH_MDS_OP_SETFILELOCK
)
652 mdsc
->oldest_tid
= req
->r_tid
;
656 req
->r_unsafe_dir
= dir
;
660 static void __unregister_request(struct ceph_mds_client
*mdsc
,
661 struct ceph_mds_request
*req
)
663 dout("__unregister_request %p tid %lld\n", req
, req
->r_tid
);
665 /* Never leave an unregistered request on an unsafe list! */
666 list_del_init(&req
->r_unsafe_item
);
668 if (req
->r_tid
== mdsc
->oldest_tid
) {
669 struct rb_node
*p
= rb_next(&req
->r_node
);
670 mdsc
->oldest_tid
= 0;
672 struct ceph_mds_request
*next_req
=
673 rb_entry(p
, struct ceph_mds_request
, r_node
);
674 if (next_req
->r_op
!= CEPH_MDS_OP_SETFILELOCK
) {
675 mdsc
->oldest_tid
= next_req
->r_tid
;
682 erase_request(&mdsc
->request_tree
, req
);
684 if (req
->r_unsafe_dir
&&
685 test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
)) {
686 struct ceph_inode_info
*ci
= ceph_inode(req
->r_unsafe_dir
);
687 spin_lock(&ci
->i_unsafe_lock
);
688 list_del_init(&req
->r_unsafe_dir_item
);
689 spin_unlock(&ci
->i_unsafe_lock
);
691 if (req
->r_target_inode
&&
692 test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
)) {
693 struct ceph_inode_info
*ci
= ceph_inode(req
->r_target_inode
);
694 spin_lock(&ci
->i_unsafe_lock
);
695 list_del_init(&req
->r_unsafe_target_item
);
696 spin_unlock(&ci
->i_unsafe_lock
);
699 if (req
->r_unsafe_dir
) {
700 iput(req
->r_unsafe_dir
);
701 req
->r_unsafe_dir
= NULL
;
704 complete_all(&req
->r_safe_completion
);
706 ceph_mdsc_put_request(req
);
710 * Walk back up the dentry tree until we hit a dentry representing a
711 * non-snapshot inode. We do this using the rcu_read_lock (which must be held
712 * when calling this) to ensure that the objects won't disappear while we're
713 * working with them. Once we hit a candidate dentry, we attempt to take a
714 * reference to it, and return that as the result.
716 static struct inode
*get_nonsnap_parent(struct dentry
*dentry
)
718 struct inode
*inode
= NULL
;
720 while (dentry
&& !IS_ROOT(dentry
)) {
721 inode
= d_inode_rcu(dentry
);
722 if (!inode
|| ceph_snap(inode
) == CEPH_NOSNAP
)
724 dentry
= dentry
->d_parent
;
727 inode
= igrab(inode
);
732 * Choose mds to send request to next. If there is a hint set in the
733 * request (e.g., due to a prior forward hint from the mds), use that.
734 * Otherwise, consult frag tree and/or caps to identify the
735 * appropriate mds. If all else fails, choose randomly.
737 * Called under mdsc->mutex.
739 static int __choose_mds(struct ceph_mds_client
*mdsc
,
740 struct ceph_mds_request
*req
)
743 struct ceph_inode_info
*ci
;
744 struct ceph_cap
*cap
;
745 int mode
= req
->r_direct_mode
;
747 u32 hash
= req
->r_direct_hash
;
748 bool is_hash
= test_bit(CEPH_MDS_R_DIRECT_IS_HASH
, &req
->r_req_flags
);
751 * is there a specific mds we should try? ignore hint if we have
752 * no session and the mds is not up (active or recovering).
754 if (req
->r_resend_mds
>= 0 &&
755 (__have_session(mdsc
, req
->r_resend_mds
) ||
756 ceph_mdsmap_get_state(mdsc
->mdsmap
, req
->r_resend_mds
) > 0)) {
757 dout("choose_mds using resend_mds mds%d\n",
759 return req
->r_resend_mds
;
762 if (mode
== USE_RANDOM_MDS
)
767 if (ceph_snap(req
->r_inode
) != CEPH_SNAPDIR
) {
768 inode
= req
->r_inode
;
771 /* req->r_dentry is non-null for LSSNAP request */
773 inode
= get_nonsnap_parent(req
->r_dentry
);
775 dout("__choose_mds using snapdir's parent %p\n", inode
);
777 } else if (req
->r_dentry
) {
778 /* ignore race with rename; old or new d_parent is okay */
779 struct dentry
*parent
;
783 parent
= req
->r_dentry
->d_parent
;
784 dir
= req
->r_parent
? : d_inode_rcu(parent
);
786 if (!dir
|| dir
->i_sb
!= mdsc
->fsc
->sb
) {
787 /* not this fs or parent went negative */
788 inode
= d_inode(req
->r_dentry
);
791 } else if (ceph_snap(dir
) != CEPH_NOSNAP
) {
792 /* direct snapped/virtual snapdir requests
793 * based on parent dir inode */
794 inode
= get_nonsnap_parent(parent
);
795 dout("__choose_mds using nonsnap parent %p\n", inode
);
798 inode
= d_inode(req
->r_dentry
);
799 if (!inode
|| mode
== USE_AUTH_MDS
) {
802 hash
= ceph_dentry_hash(dir
, req
->r_dentry
);
811 dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode
, (int)is_hash
,
815 ci
= ceph_inode(inode
);
817 if (is_hash
&& S_ISDIR(inode
->i_mode
)) {
818 struct ceph_inode_frag frag
;
821 ceph_choose_frag(ci
, hash
, &frag
, &found
);
823 if (mode
== USE_ANY_MDS
&& frag
.ndist
> 0) {
826 /* choose a random replica */
827 get_random_bytes(&r
, 1);
830 dout("choose_mds %p %llx.%llx "
831 "frag %u mds%d (%d/%d)\n",
832 inode
, ceph_vinop(inode
),
835 if (ceph_mdsmap_get_state(mdsc
->mdsmap
, mds
) >=
836 CEPH_MDS_STATE_ACTIVE
)
840 /* since this file/dir wasn't known to be
841 * replicated, then we want to look for the
842 * authoritative mds. */
845 /* choose auth mds */
847 dout("choose_mds %p %llx.%llx "
848 "frag %u mds%d (auth)\n",
849 inode
, ceph_vinop(inode
), frag
.frag
, mds
);
850 if (ceph_mdsmap_get_state(mdsc
->mdsmap
, mds
) >=
851 CEPH_MDS_STATE_ACTIVE
)
857 spin_lock(&ci
->i_ceph_lock
);
859 if (mode
== USE_AUTH_MDS
)
860 cap
= ci
->i_auth_cap
;
861 if (!cap
&& !RB_EMPTY_ROOT(&ci
->i_caps
))
862 cap
= rb_entry(rb_first(&ci
->i_caps
), struct ceph_cap
, ci_node
);
864 spin_unlock(&ci
->i_ceph_lock
);
868 mds
= cap
->session
->s_mds
;
869 dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
870 inode
, ceph_vinop(inode
), mds
,
871 cap
== ci
->i_auth_cap
? "auth " : "", cap
);
872 spin_unlock(&ci
->i_ceph_lock
);
878 mds
= ceph_mdsmap_get_random_mds(mdsc
->mdsmap
);
879 dout("choose_mds chose random mds%d\n", mds
);
887 static struct ceph_msg
*create_session_msg(u32 op
, u64 seq
)
889 struct ceph_msg
*msg
;
890 struct ceph_mds_session_head
*h
;
892 msg
= ceph_msg_new(CEPH_MSG_CLIENT_SESSION
, sizeof(*h
), GFP_NOFS
,
895 pr_err("create_session_msg ENOMEM creating msg\n");
898 h
= msg
->front
.iov_base
;
899 h
->op
= cpu_to_le32(op
);
900 h
->seq
= cpu_to_le64(seq
);
906 * session message, specialization for CEPH_SESSION_REQUEST_OPEN
907 * to include additional client metadata fields.
909 static struct ceph_msg
*create_session_open_msg(struct ceph_mds_client
*mdsc
, u64 seq
)
911 struct ceph_msg
*msg
;
912 struct ceph_mds_session_head
*h
;
914 int metadata_bytes
= 0;
915 int metadata_key_count
= 0;
916 struct ceph_options
*opt
= mdsc
->fsc
->client
->options
;
917 struct ceph_mount_options
*fsopt
= mdsc
->fsc
->mount_options
;
920 const char* metadata
[][2] = {
921 {"hostname", mdsc
->nodename
},
922 {"kernel_version", init_utsname()->release
},
923 {"entity_id", opt
->name
? : ""},
924 {"root", fsopt
->server_path
? : "/"},
928 /* Calculate serialized length of metadata */
929 metadata_bytes
= 4; /* map length */
930 for (i
= 0; metadata
[i
][0]; ++i
) {
931 metadata_bytes
+= 8 + strlen(metadata
[i
][0]) +
932 strlen(metadata
[i
][1]);
933 metadata_key_count
++;
936 /* Allocate the message */
937 msg
= ceph_msg_new(CEPH_MSG_CLIENT_SESSION
, sizeof(*h
) + metadata_bytes
,
940 pr_err("create_session_msg ENOMEM creating msg\n");
943 h
= msg
->front
.iov_base
;
944 h
->op
= cpu_to_le32(CEPH_SESSION_REQUEST_OPEN
);
945 h
->seq
= cpu_to_le64(seq
);
948 * Serialize client metadata into waiting buffer space, using
949 * the format that userspace expects for map<string, string>
951 * ClientSession messages with metadata are v2
953 msg
->hdr
.version
= cpu_to_le16(2);
954 msg
->hdr
.compat_version
= cpu_to_le16(1);
956 /* The write pointer, following the session_head structure */
957 p
= msg
->front
.iov_base
+ sizeof(*h
);
959 /* Number of entries in the map */
960 ceph_encode_32(&p
, metadata_key_count
);
962 /* Two length-prefixed strings for each entry in the map */
963 for (i
= 0; metadata
[i
][0]; ++i
) {
964 size_t const key_len
= strlen(metadata
[i
][0]);
965 size_t const val_len
= strlen(metadata
[i
][1]);
967 ceph_encode_32(&p
, key_len
);
968 memcpy(p
, metadata
[i
][0], key_len
);
970 ceph_encode_32(&p
, val_len
);
971 memcpy(p
, metadata
[i
][1], val_len
);
979 * send session open request.
981 * called under mdsc->mutex
983 static int __open_session(struct ceph_mds_client
*mdsc
,
984 struct ceph_mds_session
*session
)
986 struct ceph_msg
*msg
;
988 int mds
= session
->s_mds
;
990 /* wait for mds to go active? */
991 mstate
= ceph_mdsmap_get_state(mdsc
->mdsmap
, mds
);
992 dout("open_session to mds%d (%s)\n", mds
,
993 ceph_mds_state_name(mstate
));
994 session
->s_state
= CEPH_MDS_SESSION_OPENING
;
995 session
->s_renew_requested
= jiffies
;
997 /* send connect message */
998 msg
= create_session_open_msg(mdsc
, session
->s_seq
);
1001 ceph_con_send(&session
->s_con
, msg
);
1006 * open sessions for any export targets for the given mds
1008 * called under mdsc->mutex
1010 static struct ceph_mds_session
*
1011 __open_export_target_session(struct ceph_mds_client
*mdsc
, int target
)
1013 struct ceph_mds_session
*session
;
1015 session
= __ceph_lookup_mds_session(mdsc
, target
);
1017 session
= register_session(mdsc
, target
);
1018 if (IS_ERR(session
))
1021 if (session
->s_state
== CEPH_MDS_SESSION_NEW
||
1022 session
->s_state
== CEPH_MDS_SESSION_CLOSING
)
1023 __open_session(mdsc
, session
);
1028 struct ceph_mds_session
*
1029 ceph_mdsc_open_export_target_session(struct ceph_mds_client
*mdsc
, int target
)
1031 struct ceph_mds_session
*session
;
1033 dout("open_export_target_session to mds%d\n", target
);
1035 mutex_lock(&mdsc
->mutex
);
1036 session
= __open_export_target_session(mdsc
, target
);
1037 mutex_unlock(&mdsc
->mutex
);
1042 static void __open_export_target_sessions(struct ceph_mds_client
*mdsc
,
1043 struct ceph_mds_session
*session
)
1045 struct ceph_mds_info
*mi
;
1046 struct ceph_mds_session
*ts
;
1047 int i
, mds
= session
->s_mds
;
1049 if (mds
>= mdsc
->mdsmap
->m_num_mds
)
1052 mi
= &mdsc
->mdsmap
->m_info
[mds
];
1053 dout("open_export_target_sessions for mds%d (%d targets)\n",
1054 session
->s_mds
, mi
->num_export_targets
);
1056 for (i
= 0; i
< mi
->num_export_targets
; i
++) {
1057 ts
= __open_export_target_session(mdsc
, mi
->export_targets
[i
]);
1059 ceph_put_mds_session(ts
);
1063 void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client
*mdsc
,
1064 struct ceph_mds_session
*session
)
1066 mutex_lock(&mdsc
->mutex
);
1067 __open_export_target_sessions(mdsc
, session
);
1068 mutex_unlock(&mdsc
->mutex
);
1075 static void detach_cap_releases(struct ceph_mds_session
*session
,
1076 struct list_head
*target
)
1078 lockdep_assert_held(&session
->s_cap_lock
);
1080 list_splice_init(&session
->s_cap_releases
, target
);
1081 session
->s_num_cap_releases
= 0;
1082 dout("dispose_cap_releases mds%d\n", session
->s_mds
);
1085 static void dispose_cap_releases(struct ceph_mds_client
*mdsc
,
1086 struct list_head
*dispose
)
1088 while (!list_empty(dispose
)) {
1089 struct ceph_cap
*cap
;
1090 /* zero out the in-progress message */
1091 cap
= list_first_entry(dispose
, struct ceph_cap
, session_caps
);
1092 list_del(&cap
->session_caps
);
1093 ceph_put_cap(mdsc
, cap
);
1097 static void cleanup_session_requests(struct ceph_mds_client
*mdsc
,
1098 struct ceph_mds_session
*session
)
1100 struct ceph_mds_request
*req
;
1103 dout("cleanup_session_requests mds%d\n", session
->s_mds
);
1104 mutex_lock(&mdsc
->mutex
);
1105 while (!list_empty(&session
->s_unsafe
)) {
1106 req
= list_first_entry(&session
->s_unsafe
,
1107 struct ceph_mds_request
, r_unsafe_item
);
1108 pr_warn_ratelimited(" dropping unsafe request %llu\n",
1110 __unregister_request(mdsc
, req
);
1112 /* zero r_attempts, so kick_requests() will re-send requests */
1113 p
= rb_first(&mdsc
->request_tree
);
1115 req
= rb_entry(p
, struct ceph_mds_request
, r_node
);
1117 if (req
->r_session
&&
1118 req
->r_session
->s_mds
== session
->s_mds
)
1119 req
->r_attempts
= 0;
1121 mutex_unlock(&mdsc
->mutex
);
1125 * Helper to safely iterate over all caps associated with a session, with
1126 * special care taken to handle a racing __ceph_remove_cap().
1128 * Caller must hold session s_mutex.
1130 static int iterate_session_caps(struct ceph_mds_session
*session
,
1131 int (*cb
)(struct inode
*, struct ceph_cap
*,
1134 struct list_head
*p
;
1135 struct ceph_cap
*cap
;
1136 struct inode
*inode
, *last_inode
= NULL
;
1137 struct ceph_cap
*old_cap
= NULL
;
1140 dout("iterate_session_caps %p mds%d\n", session
, session
->s_mds
);
1141 spin_lock(&session
->s_cap_lock
);
1142 p
= session
->s_caps
.next
;
1143 while (p
!= &session
->s_caps
) {
1144 cap
= list_entry(p
, struct ceph_cap
, session_caps
);
1145 inode
= igrab(&cap
->ci
->vfs_inode
);
1150 session
->s_cap_iterator
= cap
;
1151 spin_unlock(&session
->s_cap_lock
);
1158 ceph_put_cap(session
->s_mdsc
, old_cap
);
1162 ret
= cb(inode
, cap
, arg
);
1165 spin_lock(&session
->s_cap_lock
);
1168 dout("iterate_session_caps finishing cap %p removal\n",
1170 BUG_ON(cap
->session
!= session
);
1171 cap
->session
= NULL
;
1172 list_del_init(&cap
->session_caps
);
1173 session
->s_nr_caps
--;
1174 if (cap
->queue_release
) {
1175 list_add_tail(&cap
->session_caps
,
1176 &session
->s_cap_releases
);
1177 session
->s_num_cap_releases
++;
1179 old_cap
= cap
; /* put_cap it w/o locks held */
1187 session
->s_cap_iterator
= NULL
;
1188 spin_unlock(&session
->s_cap_lock
);
1192 ceph_put_cap(session
->s_mdsc
, old_cap
);
1197 static int remove_session_caps_cb(struct inode
*inode
, struct ceph_cap
*cap
,
1200 struct ceph_fs_client
*fsc
= (struct ceph_fs_client
*)arg
;
1201 struct ceph_inode_info
*ci
= ceph_inode(inode
);
1202 LIST_HEAD(to_remove
);
1204 bool invalidate
= false;
1206 dout("removing cap %p, ci is %p, inode is %p\n",
1207 cap
, ci
, &ci
->vfs_inode
);
1208 spin_lock(&ci
->i_ceph_lock
);
1209 __ceph_remove_cap(cap
, false);
1210 if (!ci
->i_auth_cap
) {
1211 struct ceph_cap_flush
*cf
;
1212 struct ceph_mds_client
*mdsc
= fsc
->mdsc
;
1214 ci
->i_ceph_flags
|= CEPH_I_CAP_DROPPED
;
1216 if (ci
->i_wrbuffer_ref
> 0 &&
1217 READ_ONCE(fsc
->mount_state
) == CEPH_MOUNT_SHUTDOWN
)
1220 while (!list_empty(&ci
->i_cap_flush_list
)) {
1221 cf
= list_first_entry(&ci
->i_cap_flush_list
,
1222 struct ceph_cap_flush
, i_list
);
1223 list_move(&cf
->i_list
, &to_remove
);
1226 spin_lock(&mdsc
->cap_dirty_lock
);
1228 list_for_each_entry(cf
, &to_remove
, i_list
)
1229 list_del(&cf
->g_list
);
1231 if (!list_empty(&ci
->i_dirty_item
)) {
1232 pr_warn_ratelimited(
1233 " dropping dirty %s state for %p %lld\n",
1234 ceph_cap_string(ci
->i_dirty_caps
),
1235 inode
, ceph_ino(inode
));
1236 ci
->i_dirty_caps
= 0;
1237 list_del_init(&ci
->i_dirty_item
);
1240 if (!list_empty(&ci
->i_flushing_item
)) {
1241 pr_warn_ratelimited(
1242 " dropping dirty+flushing %s state for %p %lld\n",
1243 ceph_cap_string(ci
->i_flushing_caps
),
1244 inode
, ceph_ino(inode
));
1245 ci
->i_flushing_caps
= 0;
1246 list_del_init(&ci
->i_flushing_item
);
1247 mdsc
->num_cap_flushing
--;
1250 spin_unlock(&mdsc
->cap_dirty_lock
);
1252 if (atomic_read(&ci
->i_filelock_ref
) > 0) {
1253 /* make further file lock syscall return -EIO */
1254 ci
->i_ceph_flags
|= CEPH_I_ERROR_FILELOCK
;
1255 pr_warn_ratelimited(" dropping file locks for %p %lld\n",
1256 inode
, ceph_ino(inode
));
1259 if (!ci
->i_dirty_caps
&& ci
->i_prealloc_cap_flush
) {
1260 list_add(&ci
->i_prealloc_cap_flush
->i_list
, &to_remove
);
1261 ci
->i_prealloc_cap_flush
= NULL
;
1264 spin_unlock(&ci
->i_ceph_lock
);
1265 while (!list_empty(&to_remove
)) {
1266 struct ceph_cap_flush
*cf
;
1267 cf
= list_first_entry(&to_remove
,
1268 struct ceph_cap_flush
, i_list
);
1269 list_del(&cf
->i_list
);
1270 ceph_free_cap_flush(cf
);
1273 wake_up_all(&ci
->i_cap_wq
);
1275 ceph_queue_invalidate(inode
);
1282 * caller must hold session s_mutex
1284 static void remove_session_caps(struct ceph_mds_session
*session
)
1286 struct ceph_fs_client
*fsc
= session
->s_mdsc
->fsc
;
1287 struct super_block
*sb
= fsc
->sb
;
1290 dout("remove_session_caps on %p\n", session
);
1291 iterate_session_caps(session
, remove_session_caps_cb
, fsc
);
1293 wake_up_all(&fsc
->mdsc
->cap_flushing_wq
);
1295 spin_lock(&session
->s_cap_lock
);
1296 if (session
->s_nr_caps
> 0) {
1297 struct inode
*inode
;
1298 struct ceph_cap
*cap
, *prev
= NULL
;
1299 struct ceph_vino vino
;
1301 * iterate_session_caps() skips inodes that are being
1302 * deleted, we need to wait until deletions are complete.
1303 * __wait_on_freeing_inode() is designed for the job,
1304 * but it is not exported, so use lookup inode function
1307 while (!list_empty(&session
->s_caps
)) {
1308 cap
= list_entry(session
->s_caps
.next
,
1309 struct ceph_cap
, session_caps
);
1313 vino
= cap
->ci
->i_vino
;
1314 spin_unlock(&session
->s_cap_lock
);
1316 inode
= ceph_find_inode(sb
, vino
);
1319 spin_lock(&session
->s_cap_lock
);
1323 // drop cap expires and unlock s_cap_lock
1324 detach_cap_releases(session
, &dispose
);
1326 BUG_ON(session
->s_nr_caps
> 0);
1327 BUG_ON(!list_empty(&session
->s_cap_flushing
));
1328 spin_unlock(&session
->s_cap_lock
);
1329 dispose_cap_releases(session
->s_mdsc
, &dispose
);
1333 * wake up any threads waiting on this session's caps. if the cap is
1334 * old (didn't get renewed on the client reconnect), remove it now.
1336 * caller must hold s_mutex.
1338 static int wake_up_session_cb(struct inode
*inode
, struct ceph_cap
*cap
,
1341 struct ceph_inode_info
*ci
= ceph_inode(inode
);
1344 spin_lock(&ci
->i_ceph_lock
);
1345 ci
->i_wanted_max_size
= 0;
1346 ci
->i_requested_max_size
= 0;
1347 spin_unlock(&ci
->i_ceph_lock
);
1349 wake_up_all(&ci
->i_cap_wq
);
1353 static void wake_up_session_caps(struct ceph_mds_session
*session
,
1356 dout("wake_up_session_caps %p mds%d\n", session
, session
->s_mds
);
1357 iterate_session_caps(session
, wake_up_session_cb
,
1358 (void *)(unsigned long)reconnect
);
1362 * Send periodic message to MDS renewing all currently held caps. The
1363 * ack will reset the expiration for all caps from this session.
1365 * caller holds s_mutex
1367 static int send_renew_caps(struct ceph_mds_client
*mdsc
,
1368 struct ceph_mds_session
*session
)
1370 struct ceph_msg
*msg
;
1373 if (time_after_eq(jiffies
, session
->s_cap_ttl
) &&
1374 time_after_eq(session
->s_cap_ttl
, session
->s_renew_requested
))
1375 pr_info("mds%d caps stale\n", session
->s_mds
);
1376 session
->s_renew_requested
= jiffies
;
1378 /* do not try to renew caps until a recovering mds has reconnected
1379 * with its clients. */
1380 state
= ceph_mdsmap_get_state(mdsc
->mdsmap
, session
->s_mds
);
1381 if (state
< CEPH_MDS_STATE_RECONNECT
) {
1382 dout("send_renew_caps ignoring mds%d (%s)\n",
1383 session
->s_mds
, ceph_mds_state_name(state
));
1387 dout("send_renew_caps to mds%d (%s)\n", session
->s_mds
,
1388 ceph_mds_state_name(state
));
1389 msg
= create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS
,
1390 ++session
->s_renew_seq
);
1393 ceph_con_send(&session
->s_con
, msg
);
1397 static int send_flushmsg_ack(struct ceph_mds_client
*mdsc
,
1398 struct ceph_mds_session
*session
, u64 seq
)
1400 struct ceph_msg
*msg
;
1402 dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
1403 session
->s_mds
, ceph_session_state_name(session
->s_state
), seq
);
1404 msg
= create_session_msg(CEPH_SESSION_FLUSHMSG_ACK
, seq
);
1407 ceph_con_send(&session
->s_con
, msg
);
1413 * Note new cap ttl, and any transition from stale -> not stale (fresh?).
1415 * Called under session->s_mutex
1417 static void renewed_caps(struct ceph_mds_client
*mdsc
,
1418 struct ceph_mds_session
*session
, int is_renew
)
1423 spin_lock(&session
->s_cap_lock
);
1424 was_stale
= is_renew
&& time_after_eq(jiffies
, session
->s_cap_ttl
);
1426 session
->s_cap_ttl
= session
->s_renew_requested
+
1427 mdsc
->mdsmap
->m_session_timeout
*HZ
;
1430 if (time_before(jiffies
, session
->s_cap_ttl
)) {
1431 pr_info("mds%d caps renewed\n", session
->s_mds
);
1434 pr_info("mds%d caps still stale\n", session
->s_mds
);
1437 dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
1438 session
->s_mds
, session
->s_cap_ttl
, was_stale
? "stale" : "fresh",
1439 time_before(jiffies
, session
->s_cap_ttl
) ? "stale" : "fresh");
1440 spin_unlock(&session
->s_cap_lock
);
1443 wake_up_session_caps(session
, 0);
1447 * send a session close request
1449 static int request_close_session(struct ceph_mds_client
*mdsc
,
1450 struct ceph_mds_session
*session
)
1452 struct ceph_msg
*msg
;
1454 dout("request_close_session mds%d state %s seq %lld\n",
1455 session
->s_mds
, ceph_session_state_name(session
->s_state
),
1457 msg
= create_session_msg(CEPH_SESSION_REQUEST_CLOSE
, session
->s_seq
);
1460 ceph_con_send(&session
->s_con
, msg
);
1465 * Called with s_mutex held.
1467 static int __close_session(struct ceph_mds_client
*mdsc
,
1468 struct ceph_mds_session
*session
)
1470 if (session
->s_state
>= CEPH_MDS_SESSION_CLOSING
)
1472 session
->s_state
= CEPH_MDS_SESSION_CLOSING
;
1473 return request_close_session(mdsc
, session
);
1476 static bool drop_negative_children(struct dentry
*dentry
)
1478 struct dentry
*child
;
1479 bool all_negative
= true;
1481 if (!d_is_dir(dentry
))
1484 spin_lock(&dentry
->d_lock
);
1485 list_for_each_entry(child
, &dentry
->d_subdirs
, d_child
) {
1486 if (d_really_is_positive(child
)) {
1487 all_negative
= false;
1491 spin_unlock(&dentry
->d_lock
);
1494 shrink_dcache_parent(dentry
);
1496 return all_negative
;
1500 * Trim old(er) caps.
1502 * Because we can't cache an inode without one or more caps, we do
1503 * this indirectly: if a cap is unused, we prune its aliases, at which
1504 * point the inode will hopefully get dropped to.
1506 * Yes, this is a bit sloppy. Our only real goal here is to respond to
1507 * memory pressure from the MDS, though, so it needn't be perfect.
1509 static int trim_caps_cb(struct inode
*inode
, struct ceph_cap
*cap
, void *arg
)
1511 struct ceph_mds_session
*session
= arg
;
1512 struct ceph_inode_info
*ci
= ceph_inode(inode
);
1513 int used
, wanted
, oissued
, mine
;
1515 if (session
->s_trim_caps
<= 0)
1518 spin_lock(&ci
->i_ceph_lock
);
1519 mine
= cap
->issued
| cap
->implemented
;
1520 used
= __ceph_caps_used(ci
);
1521 wanted
= __ceph_caps_file_wanted(ci
);
1522 oissued
= __ceph_caps_issued_other(ci
, cap
);
1524 dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
1525 inode
, cap
, ceph_cap_string(mine
), ceph_cap_string(oissued
),
1526 ceph_cap_string(used
), ceph_cap_string(wanted
));
1527 if (cap
== ci
->i_auth_cap
) {
1528 if (ci
->i_dirty_caps
|| ci
->i_flushing_caps
||
1529 !list_empty(&ci
->i_cap_snaps
))
1531 if ((used
| wanted
) & CEPH_CAP_ANY_WR
)
1533 /* Note: it's possible that i_filelock_ref becomes non-zero
1534 * after dropping auth caps. It doesn't hurt because reply
1535 * of lock mds request will re-add auth caps. */
1536 if (atomic_read(&ci
->i_filelock_ref
) > 0)
1539 /* The inode has cached pages, but it's no longer used.
1540 * we can safely drop it */
1541 if (wanted
== 0 && used
== CEPH_CAP_FILE_CACHE
&&
1542 !(oissued
& CEPH_CAP_FILE_CACHE
)) {
1546 if ((used
| wanted
) & ~oissued
& mine
)
1547 goto out
; /* we need these caps */
1550 /* we aren't the only cap.. just remove us */
1551 __ceph_remove_cap(cap
, true);
1552 session
->s_trim_caps
--;
1554 struct dentry
*dentry
;
1555 /* try dropping referring dentries */
1556 spin_unlock(&ci
->i_ceph_lock
);
1557 dentry
= d_find_any_alias(inode
);
1558 if (dentry
&& drop_negative_children(dentry
)) {
1561 d_prune_aliases(inode
);
1562 count
= atomic_read(&inode
->i_count
);
1564 session
->s_trim_caps
--;
1565 dout("trim_caps_cb %p cap %p pruned, count now %d\n",
1574 spin_unlock(&ci
->i_ceph_lock
);
1579 * Trim session cap count down to some max number.
1581 int ceph_trim_caps(struct ceph_mds_client
*mdsc
,
1582 struct ceph_mds_session
*session
,
1585 int trim_caps
= session
->s_nr_caps
- max_caps
;
1587 dout("trim_caps mds%d start: %d / %d, trim %d\n",
1588 session
->s_mds
, session
->s_nr_caps
, max_caps
, trim_caps
);
1589 if (trim_caps
> 0) {
1590 session
->s_trim_caps
= trim_caps
;
1591 iterate_session_caps(session
, trim_caps_cb
, session
);
1592 dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
1593 session
->s_mds
, session
->s_nr_caps
, max_caps
,
1594 trim_caps
- session
->s_trim_caps
);
1595 session
->s_trim_caps
= 0;
1598 ceph_send_cap_releases(mdsc
, session
);
1602 static int check_caps_flush(struct ceph_mds_client
*mdsc
,
1607 spin_lock(&mdsc
->cap_dirty_lock
);
1608 if (!list_empty(&mdsc
->cap_flush_list
)) {
1609 struct ceph_cap_flush
*cf
=
1610 list_first_entry(&mdsc
->cap_flush_list
,
1611 struct ceph_cap_flush
, g_list
);
1612 if (cf
->tid
<= want_flush_tid
) {
1613 dout("check_caps_flush still flushing tid "
1614 "%llu <= %llu\n", cf
->tid
, want_flush_tid
);
1618 spin_unlock(&mdsc
->cap_dirty_lock
);
1623 * flush all dirty inode data to disk.
1625 * returns true if we've flushed through want_flush_tid
1627 static void wait_caps_flush(struct ceph_mds_client
*mdsc
,
1630 dout("check_caps_flush want %llu\n", want_flush_tid
);
1632 wait_event(mdsc
->cap_flushing_wq
,
1633 check_caps_flush(mdsc
, want_flush_tid
));
1635 dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid
);
1639 * called under s_mutex
1641 void ceph_send_cap_releases(struct ceph_mds_client
*mdsc
,
1642 struct ceph_mds_session
*session
)
1644 struct ceph_msg
*msg
= NULL
;
1645 struct ceph_mds_cap_release
*head
;
1646 struct ceph_mds_cap_item
*item
;
1647 struct ceph_osd_client
*osdc
= &mdsc
->fsc
->client
->osdc
;
1648 struct ceph_cap
*cap
;
1649 LIST_HEAD(tmp_list
);
1650 int num_cap_releases
;
1651 __le32 barrier
, *cap_barrier
;
1653 down_read(&osdc
->lock
);
1654 barrier
= cpu_to_le32(osdc
->epoch_barrier
);
1655 up_read(&osdc
->lock
);
1657 spin_lock(&session
->s_cap_lock
);
1659 list_splice_init(&session
->s_cap_releases
, &tmp_list
);
1660 num_cap_releases
= session
->s_num_cap_releases
;
1661 session
->s_num_cap_releases
= 0;
1662 spin_unlock(&session
->s_cap_lock
);
1664 while (!list_empty(&tmp_list
)) {
1666 msg
= ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE
,
1667 PAGE_SIZE
, GFP_NOFS
, false);
1670 head
= msg
->front
.iov_base
;
1671 head
->num
= cpu_to_le32(0);
1672 msg
->front
.iov_len
= sizeof(*head
);
1674 msg
->hdr
.version
= cpu_to_le16(2);
1675 msg
->hdr
.compat_version
= cpu_to_le16(1);
1678 cap
= list_first_entry(&tmp_list
, struct ceph_cap
,
1680 list_del(&cap
->session_caps
);
1683 head
= msg
->front
.iov_base
;
1684 le32_add_cpu(&head
->num
, 1);
1685 item
= msg
->front
.iov_base
+ msg
->front
.iov_len
;
1686 item
->ino
= cpu_to_le64(cap
->cap_ino
);
1687 item
->cap_id
= cpu_to_le64(cap
->cap_id
);
1688 item
->migrate_seq
= cpu_to_le32(cap
->mseq
);
1689 item
->seq
= cpu_to_le32(cap
->issue_seq
);
1690 msg
->front
.iov_len
+= sizeof(*item
);
1692 ceph_put_cap(mdsc
, cap
);
1694 if (le32_to_cpu(head
->num
) == CEPH_CAPS_PER_RELEASE
) {
1695 // Append cap_barrier field
1696 cap_barrier
= msg
->front
.iov_base
+ msg
->front
.iov_len
;
1697 *cap_barrier
= barrier
;
1698 msg
->front
.iov_len
+= sizeof(*cap_barrier
);
1700 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
1701 dout("send_cap_releases mds%d %p\n", session
->s_mds
, msg
);
1702 ceph_con_send(&session
->s_con
, msg
);
1707 BUG_ON(num_cap_releases
!= 0);
1709 spin_lock(&session
->s_cap_lock
);
1710 if (!list_empty(&session
->s_cap_releases
))
1712 spin_unlock(&session
->s_cap_lock
);
1715 // Append cap_barrier field
1716 cap_barrier
= msg
->front
.iov_base
+ msg
->front
.iov_len
;
1717 *cap_barrier
= barrier
;
1718 msg
->front
.iov_len
+= sizeof(*cap_barrier
);
1720 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
1721 dout("send_cap_releases mds%d %p\n", session
->s_mds
, msg
);
1722 ceph_con_send(&session
->s_con
, msg
);
1726 pr_err("send_cap_releases mds%d, failed to allocate message\n",
1728 spin_lock(&session
->s_cap_lock
);
1729 list_splice(&tmp_list
, &session
->s_cap_releases
);
1730 session
->s_num_cap_releases
+= num_cap_releases
;
1731 spin_unlock(&session
->s_cap_lock
);
1738 int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request
*req
,
1741 struct ceph_inode_info
*ci
= ceph_inode(dir
);
1742 struct ceph_mds_reply_info_parsed
*rinfo
= &req
->r_reply_info
;
1743 struct ceph_mount_options
*opt
= req
->r_mdsc
->fsc
->mount_options
;
1744 size_t size
= sizeof(struct ceph_mds_reply_dir_entry
);
1745 int order
, num_entries
;
1747 spin_lock(&ci
->i_ceph_lock
);
1748 num_entries
= ci
->i_files
+ ci
->i_subdirs
;
1749 spin_unlock(&ci
->i_ceph_lock
);
1750 num_entries
= max(num_entries
, 1);
1751 num_entries
= min(num_entries
, opt
->max_readdir
);
1753 order
= get_order(size
* num_entries
);
1754 while (order
>= 0) {
1755 rinfo
->dir_entries
= (void*)__get_free_pages(GFP_KERNEL
|
1758 if (rinfo
->dir_entries
)
1762 if (!rinfo
->dir_entries
)
1765 num_entries
= (PAGE_SIZE
<< order
) / size
;
1766 num_entries
= min(num_entries
, opt
->max_readdir
);
1768 rinfo
->dir_buf_size
= PAGE_SIZE
<< order
;
1769 req
->r_num_caps
= num_entries
+ 1;
1770 req
->r_args
.readdir
.max_entries
= cpu_to_le32(num_entries
);
1771 req
->r_args
.readdir
.max_bytes
= cpu_to_le32(opt
->max_readdir_bytes
);
1776 * Create an mds request.
1778 struct ceph_mds_request
*
1779 ceph_mdsc_create_request(struct ceph_mds_client
*mdsc
, int op
, int mode
)
1781 struct ceph_mds_request
*req
= kzalloc(sizeof(*req
), GFP_NOFS
);
1784 return ERR_PTR(-ENOMEM
);
1786 mutex_init(&req
->r_fill_mutex
);
1788 req
->r_started
= jiffies
;
1789 req
->r_resend_mds
= -1;
1790 INIT_LIST_HEAD(&req
->r_unsafe_dir_item
);
1791 INIT_LIST_HEAD(&req
->r_unsafe_target_item
);
1793 kref_init(&req
->r_kref
);
1794 RB_CLEAR_NODE(&req
->r_node
);
1795 INIT_LIST_HEAD(&req
->r_wait
);
1796 init_completion(&req
->r_completion
);
1797 init_completion(&req
->r_safe_completion
);
1798 INIT_LIST_HEAD(&req
->r_unsafe_item
);
1800 req
->r_stamp
= timespec_trunc(current_kernel_time(), mdsc
->fsc
->sb
->s_time_gran
);
1803 req
->r_direct_mode
= mode
;
1808 * return oldest (lowest) request, tid in request tree, 0 if none.
1810 * called under mdsc->mutex.
1812 static struct ceph_mds_request
*__get_oldest_req(struct ceph_mds_client
*mdsc
)
1814 if (RB_EMPTY_ROOT(&mdsc
->request_tree
))
1816 return rb_entry(rb_first(&mdsc
->request_tree
),
1817 struct ceph_mds_request
, r_node
);
1820 static inline u64
__get_oldest_tid(struct ceph_mds_client
*mdsc
)
1822 return mdsc
->oldest_tid
;
1826 * Build a dentry's path. Allocate on heap; caller must kfree. Based
1827 * on build_path_from_dentry in fs/cifs/dir.c.
1829 * If @stop_on_nosnap, generate path relative to the first non-snapped
1832 * Encode hidden .snap dirs as a double /, i.e.
1833 * foo/.snap/bar -> foo//bar
1835 char *ceph_mdsc_build_path(struct dentry
*dentry
, int *plen
, u64
*base
,
1838 struct dentry
*temp
;
1844 return ERR_PTR(-EINVAL
);
1848 seq
= read_seqbegin(&rename_lock
);
1850 for (temp
= dentry
; !IS_ROOT(temp
);) {
1851 struct inode
*inode
= d_inode(temp
);
1852 if (inode
&& ceph_snap(inode
) == CEPH_SNAPDIR
)
1853 len
++; /* slash only */
1854 else if (stop_on_nosnap
&& inode
&&
1855 ceph_snap(inode
) == CEPH_NOSNAP
)
1858 len
+= 1 + temp
->d_name
.len
;
1859 temp
= temp
->d_parent
;
1863 len
--; /* no leading '/' */
1865 path
= kmalloc(len
+1, GFP_NOFS
);
1867 return ERR_PTR(-ENOMEM
);
1869 path
[pos
] = 0; /* trailing null */
1871 for (temp
= dentry
; !IS_ROOT(temp
) && pos
!= 0; ) {
1872 struct inode
*inode
;
1874 spin_lock(&temp
->d_lock
);
1875 inode
= d_inode(temp
);
1876 if (inode
&& ceph_snap(inode
) == CEPH_SNAPDIR
) {
1877 dout("build_path path+%d: %p SNAPDIR\n",
1879 } else if (stop_on_nosnap
&& inode
&&
1880 ceph_snap(inode
) == CEPH_NOSNAP
) {
1881 spin_unlock(&temp
->d_lock
);
1884 pos
-= temp
->d_name
.len
;
1886 spin_unlock(&temp
->d_lock
);
1889 strncpy(path
+ pos
, temp
->d_name
.name
,
1892 spin_unlock(&temp
->d_lock
);
1895 temp
= temp
->d_parent
;
1898 if (pos
!= 0 || read_seqretry(&rename_lock
, seq
)) {
1899 pr_err("build_path did not end path lookup where "
1900 "expected, namelen is %d, pos is %d\n", len
, pos
);
1901 /* presumably this is only possible if racing with a
1902 rename of one of the parent directories (we can not
1903 lock the dentries above us to prevent this, but
1904 retrying should be harmless) */
1909 *base
= ceph_ino(d_inode(temp
));
1911 dout("build_path on %p %d built %llx '%.*s'\n",
1912 dentry
, d_count(dentry
), *base
, len
, path
);
1916 static int build_dentry_path(struct dentry
*dentry
, struct inode
*dir
,
1917 const char **ppath
, int *ppathlen
, u64
*pino
,
1924 dir
= d_inode_rcu(dentry
->d_parent
);
1925 if (dir
&& ceph_snap(dir
) == CEPH_NOSNAP
) {
1926 *pino
= ceph_ino(dir
);
1928 *ppath
= dentry
->d_name
.name
;
1929 *ppathlen
= dentry
->d_name
.len
;
1933 path
= ceph_mdsc_build_path(dentry
, ppathlen
, pino
, 1);
1935 return PTR_ERR(path
);
1941 static int build_inode_path(struct inode
*inode
,
1942 const char **ppath
, int *ppathlen
, u64
*pino
,
1945 struct dentry
*dentry
;
1948 if (ceph_snap(inode
) == CEPH_NOSNAP
) {
1949 *pino
= ceph_ino(inode
);
1953 dentry
= d_find_alias(inode
);
1954 path
= ceph_mdsc_build_path(dentry
, ppathlen
, pino
, 1);
1957 return PTR_ERR(path
);
1964 * request arguments may be specified via an inode *, a dentry *, or
1965 * an explicit ino+path.
1967 static int set_request_path_attr(struct inode
*rinode
, struct dentry
*rdentry
,
1968 struct inode
*rdiri
, const char *rpath
,
1969 u64 rino
, const char **ppath
, int *pathlen
,
1970 u64
*ino
, int *freepath
)
1975 r
= build_inode_path(rinode
, ppath
, pathlen
, ino
, freepath
);
1976 dout(" inode %p %llx.%llx\n", rinode
, ceph_ino(rinode
),
1978 } else if (rdentry
) {
1979 r
= build_dentry_path(rdentry
, rdiri
, ppath
, pathlen
, ino
,
1981 dout(" dentry %p %llx/%.*s\n", rdentry
, *ino
, *pathlen
,
1983 } else if (rpath
|| rino
) {
1986 *pathlen
= rpath
? strlen(rpath
) : 0;
1987 dout(" path %.*s\n", *pathlen
, rpath
);
1994 * called under mdsc->mutex
1996 static struct ceph_msg
*create_request_message(struct ceph_mds_client
*mdsc
,
1997 struct ceph_mds_request
*req
,
1998 int mds
, bool drop_cap_releases
)
2000 struct ceph_msg
*msg
;
2001 struct ceph_mds_request_head
*head
;
2002 const char *path1
= NULL
;
2003 const char *path2
= NULL
;
2004 u64 ino1
= 0, ino2
= 0;
2005 int pathlen1
= 0, pathlen2
= 0;
2006 int freepath1
= 0, freepath2
= 0;
2012 ret
= set_request_path_attr(req
->r_inode
, req
->r_dentry
,
2013 req
->r_parent
, req
->r_path1
, req
->r_ino1
.ino
,
2014 &path1
, &pathlen1
, &ino1
, &freepath1
);
2020 ret
= set_request_path_attr(NULL
, req
->r_old_dentry
,
2021 req
->r_old_dentry_dir
,
2022 req
->r_path2
, req
->r_ino2
.ino
,
2023 &path2
, &pathlen2
, &ino2
, &freepath2
);
2029 len
= sizeof(*head
) +
2030 pathlen1
+ pathlen2
+ 2*(1 + sizeof(u32
) + sizeof(u64
)) +
2031 sizeof(struct ceph_timespec
);
2033 /* calculate (max) length for cap releases */
2034 len
+= sizeof(struct ceph_mds_request_release
) *
2035 (!!req
->r_inode_drop
+ !!req
->r_dentry_drop
+
2036 !!req
->r_old_inode_drop
+ !!req
->r_old_dentry_drop
);
2037 if (req
->r_dentry_drop
)
2038 len
+= req
->r_dentry
->d_name
.len
;
2039 if (req
->r_old_dentry_drop
)
2040 len
+= req
->r_old_dentry
->d_name
.len
;
2042 msg
= ceph_msg_new(CEPH_MSG_CLIENT_REQUEST
, len
, GFP_NOFS
, false);
2044 msg
= ERR_PTR(-ENOMEM
);
2048 msg
->hdr
.version
= cpu_to_le16(2);
2049 msg
->hdr
.tid
= cpu_to_le64(req
->r_tid
);
2051 head
= msg
->front
.iov_base
;
2052 p
= msg
->front
.iov_base
+ sizeof(*head
);
2053 end
= msg
->front
.iov_base
+ msg
->front
.iov_len
;
2055 head
->mdsmap_epoch
= cpu_to_le32(mdsc
->mdsmap
->m_epoch
);
2056 head
->op
= cpu_to_le32(req
->r_op
);
2057 head
->caller_uid
= cpu_to_le32(from_kuid(&init_user_ns
, req
->r_uid
));
2058 head
->caller_gid
= cpu_to_le32(from_kgid(&init_user_ns
, req
->r_gid
));
2059 head
->args
= req
->r_args
;
2061 ceph_encode_filepath(&p
, end
, ino1
, path1
);
2062 ceph_encode_filepath(&p
, end
, ino2
, path2
);
2064 /* make note of release offset, in case we need to replay */
2065 req
->r_request_release_offset
= p
- msg
->front
.iov_base
;
2069 if (req
->r_inode_drop
)
2070 releases
+= ceph_encode_inode_release(&p
,
2071 req
->r_inode
? req
->r_inode
: d_inode(req
->r_dentry
),
2072 mds
, req
->r_inode_drop
, req
->r_inode_unless
, 0);
2073 if (req
->r_dentry_drop
)
2074 releases
+= ceph_encode_dentry_release(&p
, req
->r_dentry
,
2075 req
->r_parent
, mds
, req
->r_dentry_drop
,
2076 req
->r_dentry_unless
);
2077 if (req
->r_old_dentry_drop
)
2078 releases
+= ceph_encode_dentry_release(&p
, req
->r_old_dentry
,
2079 req
->r_old_dentry_dir
, mds
,
2080 req
->r_old_dentry_drop
,
2081 req
->r_old_dentry_unless
);
2082 if (req
->r_old_inode_drop
)
2083 releases
+= ceph_encode_inode_release(&p
,
2084 d_inode(req
->r_old_dentry
),
2085 mds
, req
->r_old_inode_drop
, req
->r_old_inode_unless
, 0);
2087 if (drop_cap_releases
) {
2089 p
= msg
->front
.iov_base
+ req
->r_request_release_offset
;
2092 head
->num_releases
= cpu_to_le16(releases
);
2096 struct ceph_timespec ts
;
2097 ceph_encode_timespec(&ts
, &req
->r_stamp
);
2098 ceph_encode_copy(&p
, &ts
, sizeof(ts
));
2102 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
2103 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
2105 if (req
->r_pagelist
) {
2106 struct ceph_pagelist
*pagelist
= req
->r_pagelist
;
2107 refcount_inc(&pagelist
->refcnt
);
2108 ceph_msg_data_add_pagelist(msg
, pagelist
);
2109 msg
->hdr
.data_len
= cpu_to_le32(pagelist
->length
);
2111 msg
->hdr
.data_len
= 0;
2114 msg
->hdr
.data_off
= cpu_to_le16(0);
2118 kfree((char *)path2
);
2121 kfree((char *)path1
);
2127 * called under mdsc->mutex if error, under no mutex if
2130 static void complete_request(struct ceph_mds_client
*mdsc
,
2131 struct ceph_mds_request
*req
)
2133 if (req
->r_callback
)
2134 req
->r_callback(mdsc
, req
);
2136 complete_all(&req
->r_completion
);
2140 * called under mdsc->mutex
2142 static int __prepare_send_request(struct ceph_mds_client
*mdsc
,
2143 struct ceph_mds_request
*req
,
2144 int mds
, bool drop_cap_releases
)
2146 struct ceph_mds_request_head
*rhead
;
2147 struct ceph_msg
*msg
;
2152 struct ceph_cap
*cap
=
2153 ceph_get_cap_for_mds(ceph_inode(req
->r_inode
), mds
);
2156 req
->r_sent_on_mseq
= cap
->mseq
;
2158 req
->r_sent_on_mseq
= -1;
2160 dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req
,
2161 req
->r_tid
, ceph_mds_op_name(req
->r_op
), req
->r_attempts
);
2163 if (test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
)) {
2166 * Replay. Do not regenerate message (and rebuild
2167 * paths, etc.); just use the original message.
2168 * Rebuilding paths will break for renames because
2169 * d_move mangles the src name.
2171 msg
= req
->r_request
;
2172 rhead
= msg
->front
.iov_base
;
2174 flags
= le32_to_cpu(rhead
->flags
);
2175 flags
|= CEPH_MDS_FLAG_REPLAY
;
2176 rhead
->flags
= cpu_to_le32(flags
);
2178 if (req
->r_target_inode
)
2179 rhead
->ino
= cpu_to_le64(ceph_ino(req
->r_target_inode
));
2181 rhead
->num_retry
= req
->r_attempts
- 1;
2183 /* remove cap/dentry releases from message */
2184 rhead
->num_releases
= 0;
2187 p
= msg
->front
.iov_base
+ req
->r_request_release_offset
;
2189 struct ceph_timespec ts
;
2190 ceph_encode_timespec(&ts
, &req
->r_stamp
);
2191 ceph_encode_copy(&p
, &ts
, sizeof(ts
));
2194 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
2195 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
2199 if (req
->r_request
) {
2200 ceph_msg_put(req
->r_request
);
2201 req
->r_request
= NULL
;
2203 msg
= create_request_message(mdsc
, req
, mds
, drop_cap_releases
);
2205 req
->r_err
= PTR_ERR(msg
);
2206 return PTR_ERR(msg
);
2208 req
->r_request
= msg
;
2210 rhead
= msg
->front
.iov_base
;
2211 rhead
->oldest_client_tid
= cpu_to_le64(__get_oldest_tid(mdsc
));
2212 if (test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
))
2213 flags
|= CEPH_MDS_FLAG_REPLAY
;
2215 flags
|= CEPH_MDS_FLAG_WANT_DENTRY
;
2216 rhead
->flags
= cpu_to_le32(flags
);
2217 rhead
->num_fwd
= req
->r_num_fwd
;
2218 rhead
->num_retry
= req
->r_attempts
- 1;
2221 dout(" r_parent = %p\n", req
->r_parent
);
2226 * send request, or put it on the appropriate wait list.
2228 static int __do_request(struct ceph_mds_client
*mdsc
,
2229 struct ceph_mds_request
*req
)
2231 struct ceph_mds_session
*session
= NULL
;
2235 if (req
->r_err
|| test_bit(CEPH_MDS_R_GOT_RESULT
, &req
->r_req_flags
)) {
2236 if (test_bit(CEPH_MDS_R_ABORTED
, &req
->r_req_flags
))
2237 __unregister_request(mdsc
, req
);
2241 if (req
->r_timeout
&&
2242 time_after_eq(jiffies
, req
->r_started
+ req
->r_timeout
)) {
2243 dout("do_request timed out\n");
2247 if (READ_ONCE(mdsc
->fsc
->mount_state
) == CEPH_MOUNT_SHUTDOWN
) {
2248 dout("do_request forced umount\n");
2252 if (READ_ONCE(mdsc
->fsc
->mount_state
) == CEPH_MOUNT_MOUNTING
) {
2253 if (mdsc
->mdsmap_err
) {
2254 err
= mdsc
->mdsmap_err
;
2255 dout("do_request mdsmap err %d\n", err
);
2258 if (mdsc
->mdsmap
->m_epoch
== 0) {
2259 dout("do_request no mdsmap, waiting for map\n");
2260 list_add(&req
->r_wait
, &mdsc
->waiting_for_map
);
2263 if (!(mdsc
->fsc
->mount_options
->flags
&
2264 CEPH_MOUNT_OPT_MOUNTWAIT
) &&
2265 !ceph_mdsmap_is_cluster_available(mdsc
->mdsmap
)) {
2267 pr_info("probably no mds server is up\n");
2272 put_request_session(req
);
2274 mds
= __choose_mds(mdsc
, req
);
2276 ceph_mdsmap_get_state(mdsc
->mdsmap
, mds
) < CEPH_MDS_STATE_ACTIVE
) {
2277 dout("do_request no mds or not active, waiting for map\n");
2278 list_add(&req
->r_wait
, &mdsc
->waiting_for_map
);
2282 /* get, open session */
2283 session
= __ceph_lookup_mds_session(mdsc
, mds
);
2285 session
= register_session(mdsc
, mds
);
2286 if (IS_ERR(session
)) {
2287 err
= PTR_ERR(session
);
2291 req
->r_session
= get_session(session
);
2293 dout("do_request mds%d session %p state %s\n", mds
, session
,
2294 ceph_session_state_name(session
->s_state
));
2295 if (session
->s_state
!= CEPH_MDS_SESSION_OPEN
&&
2296 session
->s_state
!= CEPH_MDS_SESSION_HUNG
) {
2297 if (session
->s_state
== CEPH_MDS_SESSION_REJECTED
) {
2301 if (session
->s_state
== CEPH_MDS_SESSION_NEW
||
2302 session
->s_state
== CEPH_MDS_SESSION_CLOSING
)
2303 __open_session(mdsc
, session
);
2304 list_add(&req
->r_wait
, &session
->s_waiting
);
2309 req
->r_resend_mds
= -1; /* forget any previous mds hint */
2311 if (req
->r_request_started
== 0) /* note request start time */
2312 req
->r_request_started
= jiffies
;
2314 err
= __prepare_send_request(mdsc
, req
, mds
, false);
2316 ceph_msg_get(req
->r_request
);
2317 ceph_con_send(&session
->s_con
, req
->r_request
);
2321 ceph_put_mds_session(session
);
2324 dout("__do_request early error %d\n", err
);
2326 complete_request(mdsc
, req
);
2327 __unregister_request(mdsc
, req
);
2334 * called under mdsc->mutex
2336 static void __wake_requests(struct ceph_mds_client
*mdsc
,
2337 struct list_head
*head
)
2339 struct ceph_mds_request
*req
;
2340 LIST_HEAD(tmp_list
);
2342 list_splice_init(head
, &tmp_list
);
2344 while (!list_empty(&tmp_list
)) {
2345 req
= list_entry(tmp_list
.next
,
2346 struct ceph_mds_request
, r_wait
);
2347 list_del_init(&req
->r_wait
);
2348 dout(" wake request %p tid %llu\n", req
, req
->r_tid
);
2349 __do_request(mdsc
, req
);
2354 * Wake up threads with requests pending for @mds, so that they can
2355 * resubmit their requests to a possibly different mds.
2357 static void kick_requests(struct ceph_mds_client
*mdsc
, int mds
)
2359 struct ceph_mds_request
*req
;
2360 struct rb_node
*p
= rb_first(&mdsc
->request_tree
);
2362 dout("kick_requests mds%d\n", mds
);
2364 req
= rb_entry(p
, struct ceph_mds_request
, r_node
);
2366 if (test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
))
2368 if (req
->r_attempts
> 0)
2369 continue; /* only new requests */
2370 if (req
->r_session
&&
2371 req
->r_session
->s_mds
== mds
) {
2372 dout(" kicking tid %llu\n", req
->r_tid
);
2373 list_del_init(&req
->r_wait
);
2374 __do_request(mdsc
, req
);
2379 void ceph_mdsc_submit_request(struct ceph_mds_client
*mdsc
,
2380 struct ceph_mds_request
*req
)
2382 dout("submit_request on %p\n", req
);
2383 mutex_lock(&mdsc
->mutex
);
2384 __register_request(mdsc
, req
, NULL
);
2385 __do_request(mdsc
, req
);
2386 mutex_unlock(&mdsc
->mutex
);
2390 * Synchrously perform an mds request. Take care of all of the
2391 * session setup, forwarding, retry details.
2393 int ceph_mdsc_do_request(struct ceph_mds_client
*mdsc
,
2395 struct ceph_mds_request
*req
)
2399 dout("do_request on %p\n", req
);
2401 /* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
2403 ceph_get_cap_refs(ceph_inode(req
->r_inode
), CEPH_CAP_PIN
);
2405 ceph_get_cap_refs(ceph_inode(req
->r_parent
), CEPH_CAP_PIN
);
2406 if (req
->r_old_dentry_dir
)
2407 ceph_get_cap_refs(ceph_inode(req
->r_old_dentry_dir
),
2411 mutex_lock(&mdsc
->mutex
);
2412 __register_request(mdsc
, req
, dir
);
2413 __do_request(mdsc
, req
);
2421 mutex_unlock(&mdsc
->mutex
);
2422 dout("do_request waiting\n");
2423 if (!req
->r_timeout
&& req
->r_wait_for_completion
) {
2424 err
= req
->r_wait_for_completion(mdsc
, req
);
2426 long timeleft
= wait_for_completion_killable_timeout(
2428 ceph_timeout_jiffies(req
->r_timeout
));
2432 err
= -EIO
; /* timed out */
2434 err
= timeleft
; /* killed */
2436 dout("do_request waited, got %d\n", err
);
2437 mutex_lock(&mdsc
->mutex
);
2439 /* only abort if we didn't race with a real reply */
2440 if (test_bit(CEPH_MDS_R_GOT_RESULT
, &req
->r_req_flags
)) {
2441 err
= le32_to_cpu(req
->r_reply_info
.head
->result
);
2442 } else if (err
< 0) {
2443 dout("aborted request %lld with %d\n", req
->r_tid
, err
);
2446 * ensure we aren't running concurrently with
2447 * ceph_fill_trace or ceph_readdir_prepopulate, which
2448 * rely on locks (dir mutex) held by our caller.
2450 mutex_lock(&req
->r_fill_mutex
);
2452 set_bit(CEPH_MDS_R_ABORTED
, &req
->r_req_flags
);
2453 mutex_unlock(&req
->r_fill_mutex
);
2455 if (req
->r_parent
&&
2456 (req
->r_op
& CEPH_MDS_OP_WRITE
))
2457 ceph_invalidate_dir_request(req
);
2463 mutex_unlock(&mdsc
->mutex
);
2464 dout("do_request %p done, result %d\n", req
, err
);
2469 * Invalidate dir's completeness, dentry lease state on an aborted MDS
2470 * namespace request.
2472 void ceph_invalidate_dir_request(struct ceph_mds_request
*req
)
2474 struct inode
*dir
= req
->r_parent
;
2475 struct inode
*old_dir
= req
->r_old_dentry_dir
;
2477 dout("invalidate_dir_request %p %p (complete, lease(s))\n", dir
, old_dir
);
2479 ceph_dir_clear_complete(dir
);
2481 ceph_dir_clear_complete(old_dir
);
2483 ceph_invalidate_dentry_lease(req
->r_dentry
);
2484 if (req
->r_old_dentry
)
2485 ceph_invalidate_dentry_lease(req
->r_old_dentry
);
2491 * We take the session mutex and parse and process the reply immediately.
2492 * This preserves the logical ordering of replies, capabilities, etc., sent
2493 * by the MDS as they are applied to our local cache.
2495 static void handle_reply(struct ceph_mds_session
*session
, struct ceph_msg
*msg
)
2497 struct ceph_mds_client
*mdsc
= session
->s_mdsc
;
2498 struct ceph_mds_request
*req
;
2499 struct ceph_mds_reply_head
*head
= msg
->front
.iov_base
;
2500 struct ceph_mds_reply_info_parsed
*rinfo
; /* parsed reply info */
2501 struct ceph_snap_realm
*realm
;
2504 int mds
= session
->s_mds
;
2506 if (msg
->front
.iov_len
< sizeof(*head
)) {
2507 pr_err("mdsc_handle_reply got corrupt (short) reply\n");
2512 /* get request, session */
2513 tid
= le64_to_cpu(msg
->hdr
.tid
);
2514 mutex_lock(&mdsc
->mutex
);
2515 req
= lookup_get_request(mdsc
, tid
);
2517 dout("handle_reply on unknown tid %llu\n", tid
);
2518 mutex_unlock(&mdsc
->mutex
);
2521 dout("handle_reply %p\n", req
);
2523 /* correct session? */
2524 if (req
->r_session
!= session
) {
2525 pr_err("mdsc_handle_reply got %llu on session mds%d"
2526 " not mds%d\n", tid
, session
->s_mds
,
2527 req
->r_session
? req
->r_session
->s_mds
: -1);
2528 mutex_unlock(&mdsc
->mutex
);
2533 if ((test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
) && !head
->safe
) ||
2534 (test_bit(CEPH_MDS_R_GOT_SAFE
, &req
->r_req_flags
) && head
->safe
)) {
2535 pr_warn("got a dup %s reply on %llu from mds%d\n",
2536 head
->safe
? "safe" : "unsafe", tid
, mds
);
2537 mutex_unlock(&mdsc
->mutex
);
2540 if (test_bit(CEPH_MDS_R_GOT_SAFE
, &req
->r_req_flags
)) {
2541 pr_warn("got unsafe after safe on %llu from mds%d\n",
2543 mutex_unlock(&mdsc
->mutex
);
2547 result
= le32_to_cpu(head
->result
);
2551 * if we're not talking to the authority, send to them
2552 * if the authority has changed while we weren't looking,
2553 * send to new authority
2554 * Otherwise we just have to return an ESTALE
2556 if (result
== -ESTALE
) {
2557 dout("got ESTALE on request %llu\n", req
->r_tid
);
2558 req
->r_resend_mds
= -1;
2559 if (req
->r_direct_mode
!= USE_AUTH_MDS
) {
2560 dout("not using auth, setting for that now\n");
2561 req
->r_direct_mode
= USE_AUTH_MDS
;
2562 __do_request(mdsc
, req
);
2563 mutex_unlock(&mdsc
->mutex
);
2566 int mds
= __choose_mds(mdsc
, req
);
2567 if (mds
>= 0 && mds
!= req
->r_session
->s_mds
) {
2568 dout("but auth changed, so resending\n");
2569 __do_request(mdsc
, req
);
2570 mutex_unlock(&mdsc
->mutex
);
2574 dout("have to return ESTALE on request %llu\n", req
->r_tid
);
2579 set_bit(CEPH_MDS_R_GOT_SAFE
, &req
->r_req_flags
);
2580 __unregister_request(mdsc
, req
);
2582 if (test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
)) {
2584 * We already handled the unsafe response, now do the
2585 * cleanup. No need to examine the response; the MDS
2586 * doesn't include any result info in the safe
2587 * response. And even if it did, there is nothing
2588 * useful we could do with a revised return value.
2590 dout("got safe reply %llu, mds%d\n", tid
, mds
);
2592 /* last unsafe request during umount? */
2593 if (mdsc
->stopping
&& !__get_oldest_req(mdsc
))
2594 complete_all(&mdsc
->safe_umount_waiters
);
2595 mutex_unlock(&mdsc
->mutex
);
2599 set_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
);
2600 list_add_tail(&req
->r_unsafe_item
, &req
->r_session
->s_unsafe
);
2601 if (req
->r_unsafe_dir
) {
2602 struct ceph_inode_info
*ci
=
2603 ceph_inode(req
->r_unsafe_dir
);
2604 spin_lock(&ci
->i_unsafe_lock
);
2605 list_add_tail(&req
->r_unsafe_dir_item
,
2606 &ci
->i_unsafe_dirops
);
2607 spin_unlock(&ci
->i_unsafe_lock
);
2611 dout("handle_reply tid %lld result %d\n", tid
, result
);
2612 rinfo
= &req
->r_reply_info
;
2613 err
= parse_reply_info(msg
, rinfo
, session
->s_con
.peer_features
);
2614 mutex_unlock(&mdsc
->mutex
);
2616 mutex_lock(&session
->s_mutex
);
2618 pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds
, tid
);
2625 if (rinfo
->snapblob_len
) {
2626 down_write(&mdsc
->snap_rwsem
);
2627 ceph_update_snap_trace(mdsc
, rinfo
->snapblob
,
2628 rinfo
->snapblob
+ rinfo
->snapblob_len
,
2629 le32_to_cpu(head
->op
) == CEPH_MDS_OP_RMSNAP
,
2631 downgrade_write(&mdsc
->snap_rwsem
);
2633 down_read(&mdsc
->snap_rwsem
);
2636 /* insert trace into our cache */
2637 mutex_lock(&req
->r_fill_mutex
);
2638 current
->journal_info
= req
;
2639 err
= ceph_fill_trace(mdsc
->fsc
->sb
, req
);
2641 if (result
== 0 && (req
->r_op
== CEPH_MDS_OP_READDIR
||
2642 req
->r_op
== CEPH_MDS_OP_LSSNAP
))
2643 ceph_readdir_prepopulate(req
, req
->r_session
);
2644 ceph_unreserve_caps(mdsc
, &req
->r_caps_reservation
);
2646 current
->journal_info
= NULL
;
2647 mutex_unlock(&req
->r_fill_mutex
);
2649 up_read(&mdsc
->snap_rwsem
);
2651 ceph_put_snap_realm(mdsc
, realm
);
2653 if (err
== 0 && req
->r_target_inode
&&
2654 test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
)) {
2655 struct ceph_inode_info
*ci
= ceph_inode(req
->r_target_inode
);
2656 spin_lock(&ci
->i_unsafe_lock
);
2657 list_add_tail(&req
->r_unsafe_target_item
, &ci
->i_unsafe_iops
);
2658 spin_unlock(&ci
->i_unsafe_lock
);
2661 mutex_lock(&mdsc
->mutex
);
2662 if (!test_bit(CEPH_MDS_R_ABORTED
, &req
->r_req_flags
)) {
2666 req
->r_reply
= ceph_msg_get(msg
);
2667 set_bit(CEPH_MDS_R_GOT_RESULT
, &req
->r_req_flags
);
2670 dout("reply arrived after request %lld was aborted\n", tid
);
2672 mutex_unlock(&mdsc
->mutex
);
2674 mutex_unlock(&session
->s_mutex
);
2676 /* kick calling process */
2677 complete_request(mdsc
, req
);
2679 ceph_mdsc_put_request(req
);
2686 * handle mds notification that our request has been forwarded.
2688 static void handle_forward(struct ceph_mds_client
*mdsc
,
2689 struct ceph_mds_session
*session
,
2690 struct ceph_msg
*msg
)
2692 struct ceph_mds_request
*req
;
2693 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
2697 void *p
= msg
->front
.iov_base
;
2698 void *end
= p
+ msg
->front
.iov_len
;
2700 ceph_decode_need(&p
, end
, 2*sizeof(u32
), bad
);
2701 next_mds
= ceph_decode_32(&p
);
2702 fwd_seq
= ceph_decode_32(&p
);
2704 mutex_lock(&mdsc
->mutex
);
2705 req
= lookup_get_request(mdsc
, tid
);
2707 dout("forward tid %llu to mds%d - req dne\n", tid
, next_mds
);
2708 goto out
; /* dup reply? */
2711 if (test_bit(CEPH_MDS_R_ABORTED
, &req
->r_req_flags
)) {
2712 dout("forward tid %llu aborted, unregistering\n", tid
);
2713 __unregister_request(mdsc
, req
);
2714 } else if (fwd_seq
<= req
->r_num_fwd
) {
2715 dout("forward tid %llu to mds%d - old seq %d <= %d\n",
2716 tid
, next_mds
, req
->r_num_fwd
, fwd_seq
);
2718 /* resend. forward race not possible; mds would drop */
2719 dout("forward tid %llu to mds%d (we resend)\n", tid
, next_mds
);
2721 BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT
, &req
->r_req_flags
));
2722 req
->r_attempts
= 0;
2723 req
->r_num_fwd
= fwd_seq
;
2724 req
->r_resend_mds
= next_mds
;
2725 put_request_session(req
);
2726 __do_request(mdsc
, req
);
2728 ceph_mdsc_put_request(req
);
2730 mutex_unlock(&mdsc
->mutex
);
2734 pr_err("mdsc_handle_forward decode error err=%d\n", err
);
2738 * handle a mds session control message
2740 static void handle_session(struct ceph_mds_session
*session
,
2741 struct ceph_msg
*msg
)
2743 struct ceph_mds_client
*mdsc
= session
->s_mdsc
;
2746 int mds
= session
->s_mds
;
2747 struct ceph_mds_session_head
*h
= msg
->front
.iov_base
;
2751 if (msg
->front
.iov_len
!= sizeof(*h
))
2753 op
= le32_to_cpu(h
->op
);
2754 seq
= le64_to_cpu(h
->seq
);
2756 mutex_lock(&mdsc
->mutex
);
2757 if (op
== CEPH_SESSION_CLOSE
) {
2758 get_session(session
);
2759 __unregister_session(mdsc
, session
);
2761 /* FIXME: this ttl calculation is generous */
2762 session
->s_ttl
= jiffies
+ HZ
*mdsc
->mdsmap
->m_session_autoclose
;
2763 mutex_unlock(&mdsc
->mutex
);
2765 mutex_lock(&session
->s_mutex
);
2767 dout("handle_session mds%d %s %p state %s seq %llu\n",
2768 mds
, ceph_session_op_name(op
), session
,
2769 ceph_session_state_name(session
->s_state
), seq
);
2771 if (session
->s_state
== CEPH_MDS_SESSION_HUNG
) {
2772 session
->s_state
= CEPH_MDS_SESSION_OPEN
;
2773 pr_info("mds%d came back\n", session
->s_mds
);
2777 case CEPH_SESSION_OPEN
:
2778 if (session
->s_state
== CEPH_MDS_SESSION_RECONNECTING
)
2779 pr_info("mds%d reconnect success\n", session
->s_mds
);
2780 session
->s_state
= CEPH_MDS_SESSION_OPEN
;
2781 renewed_caps(mdsc
, session
, 0);
2784 __close_session(mdsc
, session
);
2787 case CEPH_SESSION_RENEWCAPS
:
2788 if (session
->s_renew_seq
== seq
)
2789 renewed_caps(mdsc
, session
, 1);
2792 case CEPH_SESSION_CLOSE
:
2793 if (session
->s_state
== CEPH_MDS_SESSION_RECONNECTING
)
2794 pr_info("mds%d reconnect denied\n", session
->s_mds
);
2795 cleanup_session_requests(mdsc
, session
);
2796 remove_session_caps(session
);
2797 wake
= 2; /* for good measure */
2798 wake_up_all(&mdsc
->session_close_wq
);
2801 case CEPH_SESSION_STALE
:
2802 pr_info("mds%d caps went stale, renewing\n",
2804 spin_lock(&session
->s_gen_ttl_lock
);
2805 session
->s_cap_gen
++;
2806 session
->s_cap_ttl
= jiffies
- 1;
2807 spin_unlock(&session
->s_gen_ttl_lock
);
2808 send_renew_caps(mdsc
, session
);
2811 case CEPH_SESSION_RECALL_STATE
:
2812 ceph_trim_caps(mdsc
, session
, le32_to_cpu(h
->max_caps
));
2815 case CEPH_SESSION_FLUSHMSG
:
2816 send_flushmsg_ack(mdsc
, session
, seq
);
2819 case CEPH_SESSION_FORCE_RO
:
2820 dout("force_session_readonly %p\n", session
);
2821 spin_lock(&session
->s_cap_lock
);
2822 session
->s_readonly
= true;
2823 spin_unlock(&session
->s_cap_lock
);
2824 wake_up_session_caps(session
, 0);
2827 case CEPH_SESSION_REJECT
:
2828 WARN_ON(session
->s_state
!= CEPH_MDS_SESSION_OPENING
);
2829 pr_info("mds%d rejected session\n", session
->s_mds
);
2830 session
->s_state
= CEPH_MDS_SESSION_REJECTED
;
2831 cleanup_session_requests(mdsc
, session
);
2832 remove_session_caps(session
);
2833 wake
= 2; /* for good measure */
2837 pr_err("mdsc_handle_session bad op %d mds%d\n", op
, mds
);
2841 mutex_unlock(&session
->s_mutex
);
2843 mutex_lock(&mdsc
->mutex
);
2844 __wake_requests(mdsc
, &session
->s_waiting
);
2846 kick_requests(mdsc
, mds
);
2847 mutex_unlock(&mdsc
->mutex
);
2849 if (op
== CEPH_SESSION_CLOSE
)
2850 ceph_put_mds_session(session
);
2854 pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds
,
2855 (int)msg
->front
.iov_len
);
2862 * called under session->mutex.
2864 static void replay_unsafe_requests(struct ceph_mds_client
*mdsc
,
2865 struct ceph_mds_session
*session
)
2867 struct ceph_mds_request
*req
, *nreq
;
2871 dout("replay_unsafe_requests mds%d\n", session
->s_mds
);
2873 mutex_lock(&mdsc
->mutex
);
2874 list_for_each_entry_safe(req
, nreq
, &session
->s_unsafe
, r_unsafe_item
) {
2875 err
= __prepare_send_request(mdsc
, req
, session
->s_mds
, true);
2877 ceph_msg_get(req
->r_request
);
2878 ceph_con_send(&session
->s_con
, req
->r_request
);
2883 * also re-send old requests when MDS enters reconnect stage. So that MDS
2884 * can process completed request in clientreplay stage.
2886 p
= rb_first(&mdsc
->request_tree
);
2888 req
= rb_entry(p
, struct ceph_mds_request
, r_node
);
2890 if (test_bit(CEPH_MDS_R_GOT_UNSAFE
, &req
->r_req_flags
))
2892 if (req
->r_attempts
== 0)
2893 continue; /* only old requests */
2894 if (req
->r_session
&&
2895 req
->r_session
->s_mds
== session
->s_mds
) {
2896 err
= __prepare_send_request(mdsc
, req
,
2897 session
->s_mds
, true);
2899 ceph_msg_get(req
->r_request
);
2900 ceph_con_send(&session
->s_con
, req
->r_request
);
2904 mutex_unlock(&mdsc
->mutex
);
2908 * Encode information about a cap for a reconnect with the MDS.
2910 static int encode_caps_cb(struct inode
*inode
, struct ceph_cap
*cap
,
2914 struct ceph_mds_cap_reconnect v2
;
2915 struct ceph_mds_cap_reconnect_v1 v1
;
2917 struct ceph_inode_info
*ci
= cap
->ci
;
2918 struct ceph_reconnect_state
*recon_state
= arg
;
2919 struct ceph_pagelist
*pagelist
= recon_state
->pagelist
;
2924 struct dentry
*dentry
;
2926 dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
2927 inode
, ceph_vinop(inode
), cap
, cap
->cap_id
,
2928 ceph_cap_string(cap
->issued
));
2929 err
= ceph_pagelist_encode_64(pagelist
, ceph_ino(inode
));
2933 dentry
= d_find_alias(inode
);
2935 path
= ceph_mdsc_build_path(dentry
, &pathlen
, &pathbase
, 0);
2937 err
= PTR_ERR(path
);
2946 spin_lock(&ci
->i_ceph_lock
);
2947 cap
->seq
= 0; /* reset cap seq */
2948 cap
->issue_seq
= 0; /* and issue_seq */
2949 cap
->mseq
= 0; /* and migrate_seq */
2950 cap
->cap_gen
= cap
->session
->s_cap_gen
;
2952 if (recon_state
->msg_version
>= 2) {
2953 rec
.v2
.cap_id
= cpu_to_le64(cap
->cap_id
);
2954 rec
.v2
.wanted
= cpu_to_le32(__ceph_caps_wanted(ci
));
2955 rec
.v2
.issued
= cpu_to_le32(cap
->issued
);
2956 rec
.v2
.snaprealm
= cpu_to_le64(ci
->i_snap_realm
->ino
);
2957 rec
.v2
.pathbase
= cpu_to_le64(pathbase
);
2958 rec
.v2
.flock_len
= (__force __le32
)
2959 ((ci
->i_ceph_flags
& CEPH_I_ERROR_FILELOCK
) ? 0 : 1);
2962 rec
.v1
.cap_id
= cpu_to_le64(cap
->cap_id
);
2963 rec
.v1
.wanted
= cpu_to_le32(__ceph_caps_wanted(ci
));
2964 rec
.v1
.issued
= cpu_to_le32(cap
->issued
);
2965 rec
.v1
.size
= cpu_to_le64(inode
->i_size
);
2966 ts
= timespec64_to_timespec(inode
->i_mtime
);
2967 ceph_encode_timespec(&rec
.v1
.mtime
, &ts
);
2968 ts
= timespec64_to_timespec(inode
->i_atime
);
2969 ceph_encode_timespec(&rec
.v1
.atime
, &ts
);
2970 rec
.v1
.snaprealm
= cpu_to_le64(ci
->i_snap_realm
->ino
);
2971 rec
.v1
.pathbase
= cpu_to_le64(pathbase
);
2974 if (list_empty(&ci
->i_cap_snaps
)) {
2975 snap_follows
= ci
->i_head_snapc
? ci
->i_head_snapc
->seq
: 0;
2977 struct ceph_cap_snap
*capsnap
=
2978 list_first_entry(&ci
->i_cap_snaps
,
2979 struct ceph_cap_snap
, ci_item
);
2980 snap_follows
= capsnap
->follows
;
2982 spin_unlock(&ci
->i_ceph_lock
);
2984 if (recon_state
->msg_version
>= 2) {
2985 int num_fcntl_locks
, num_flock_locks
;
2986 struct ceph_filelock
*flocks
= NULL
;
2987 size_t struct_len
, total_len
= 0;
2991 if (rec
.v2
.flock_len
) {
2992 ceph_count_locks(inode
, &num_fcntl_locks
, &num_flock_locks
);
2994 num_fcntl_locks
= 0;
2995 num_flock_locks
= 0;
2997 if (num_fcntl_locks
+ num_flock_locks
> 0) {
2998 flocks
= kmalloc_array(num_fcntl_locks
+ num_flock_locks
,
2999 sizeof(struct ceph_filelock
),
3005 err
= ceph_encode_locks_to_buffer(inode
, flocks
,
3020 if (recon_state
->msg_version
>= 3) {
3021 /* version, compat_version and struct_len */
3022 total_len
= 2 * sizeof(u8
) + sizeof(u32
);
3026 * number of encoded locks is stable, so copy to pagelist
3028 struct_len
= 2 * sizeof(u32
) +
3029 (num_fcntl_locks
+ num_flock_locks
) *
3030 sizeof(struct ceph_filelock
);
3031 rec
.v2
.flock_len
= cpu_to_le32(struct_len
);
3033 struct_len
+= sizeof(rec
.v2
);
3034 struct_len
+= sizeof(u32
) + pathlen
;
3037 struct_len
+= sizeof(u64
); /* snap_follows */
3039 total_len
+= struct_len
;
3040 err
= ceph_pagelist_reserve(pagelist
, total_len
);
3043 if (recon_state
->msg_version
>= 3) {
3044 ceph_pagelist_encode_8(pagelist
, struct_v
);
3045 ceph_pagelist_encode_8(pagelist
, 1);
3046 ceph_pagelist_encode_32(pagelist
, struct_len
);
3048 ceph_pagelist_encode_string(pagelist
, path
, pathlen
);
3049 ceph_pagelist_append(pagelist
, &rec
, sizeof(rec
.v2
));
3050 ceph_locks_to_pagelist(flocks
, pagelist
,
3054 ceph_pagelist_encode_64(pagelist
, snap_follows
);
3058 size_t size
= sizeof(u32
) + pathlen
+ sizeof(rec
.v1
);
3059 err
= ceph_pagelist_reserve(pagelist
, size
);
3061 ceph_pagelist_encode_string(pagelist
, path
, pathlen
);
3062 ceph_pagelist_append(pagelist
, &rec
, sizeof(rec
.v1
));
3066 recon_state
->nr_caps
++;
3076 * If an MDS fails and recovers, clients need to reconnect in order to
3077 * reestablish shared state. This includes all caps issued through
3078 * this session _and_ the snap_realm hierarchy. Because it's not
3079 * clear which snap realms the mds cares about, we send everything we
3080 * know about.. that ensures we'll then get any new info the
3081 * recovering MDS might have.
3083 * This is a relatively heavyweight operation, but it's rare.
3085 * called with mdsc->mutex held.
3087 static void send_mds_reconnect(struct ceph_mds_client
*mdsc
,
3088 struct ceph_mds_session
*session
)
3090 struct ceph_msg
*reply
;
3092 int mds
= session
->s_mds
;
3095 struct ceph_pagelist
*pagelist
;
3096 struct ceph_reconnect_state recon_state
;
3099 pr_info("mds%d reconnect start\n", mds
);
3101 pagelist
= kmalloc(sizeof(*pagelist
), GFP_NOFS
);
3103 goto fail_nopagelist
;
3104 ceph_pagelist_init(pagelist
);
3106 reply
= ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT
, 0, GFP_NOFS
, false);
3110 mutex_lock(&session
->s_mutex
);
3111 session
->s_state
= CEPH_MDS_SESSION_RECONNECTING
;
3114 dout("session %p state %s\n", session
,
3115 ceph_session_state_name(session
->s_state
));
3117 spin_lock(&session
->s_gen_ttl_lock
);
3118 session
->s_cap_gen
++;
3119 spin_unlock(&session
->s_gen_ttl_lock
);
3121 spin_lock(&session
->s_cap_lock
);
3122 /* don't know if session is readonly */
3123 session
->s_readonly
= 0;
3125 * notify __ceph_remove_cap() that we are composing cap reconnect.
3126 * If a cap get released before being added to the cap reconnect,
3127 * __ceph_remove_cap() should skip queuing cap release.
3129 session
->s_cap_reconnect
= 1;
3130 /* drop old cap expires; we're about to reestablish that state */
3131 detach_cap_releases(session
, &dispose
);
3132 spin_unlock(&session
->s_cap_lock
);
3133 dispose_cap_releases(mdsc
, &dispose
);
3135 /* trim unused caps to reduce MDS's cache rejoin time */
3136 if (mdsc
->fsc
->sb
->s_root
)
3137 shrink_dcache_parent(mdsc
->fsc
->sb
->s_root
);
3139 ceph_con_close(&session
->s_con
);
3140 ceph_con_open(&session
->s_con
,
3141 CEPH_ENTITY_TYPE_MDS
, mds
,
3142 ceph_mdsmap_get_addr(mdsc
->mdsmap
, mds
));
3144 /* replay unsafe requests */
3145 replay_unsafe_requests(mdsc
, session
);
3147 down_read(&mdsc
->snap_rwsem
);
3149 /* traverse this session's caps */
3150 s_nr_caps
= session
->s_nr_caps
;
3151 err
= ceph_pagelist_encode_32(pagelist
, s_nr_caps
);
3155 recon_state
.nr_caps
= 0;
3156 recon_state
.pagelist
= pagelist
;
3157 if (session
->s_con
.peer_features
& CEPH_FEATURE_MDSENC
)
3158 recon_state
.msg_version
= 3;
3159 else if (session
->s_con
.peer_features
& CEPH_FEATURE_FLOCK
)
3160 recon_state
.msg_version
= 2;
3162 recon_state
.msg_version
= 1;
3163 err
= iterate_session_caps(session
, encode_caps_cb
, &recon_state
);
3167 spin_lock(&session
->s_cap_lock
);
3168 session
->s_cap_reconnect
= 0;
3169 spin_unlock(&session
->s_cap_lock
);
3172 * snaprealms. we provide mds with the ino, seq (version), and
3173 * parent for all of our realms. If the mds has any newer info,
3176 for (p
= rb_first(&mdsc
->snap_realms
); p
; p
= rb_next(p
)) {
3177 struct ceph_snap_realm
*realm
=
3178 rb_entry(p
, struct ceph_snap_realm
, node
);
3179 struct ceph_mds_snaprealm_reconnect sr_rec
;
3181 dout(" adding snap realm %llx seq %lld parent %llx\n",
3182 realm
->ino
, realm
->seq
, realm
->parent_ino
);
3183 sr_rec
.ino
= cpu_to_le64(realm
->ino
);
3184 sr_rec
.seq
= cpu_to_le64(realm
->seq
);
3185 sr_rec
.parent
= cpu_to_le64(realm
->parent_ino
);
3186 err
= ceph_pagelist_append(pagelist
, &sr_rec
, sizeof(sr_rec
));
3191 reply
->hdr
.version
= cpu_to_le16(recon_state
.msg_version
);
3193 /* raced with cap release? */
3194 if (s_nr_caps
!= recon_state
.nr_caps
) {
3195 struct page
*page
= list_first_entry(&pagelist
->head
,
3197 __le32
*addr
= kmap_atomic(page
);
3198 *addr
= cpu_to_le32(recon_state
.nr_caps
);
3199 kunmap_atomic(addr
);
3202 reply
->hdr
.data_len
= cpu_to_le32(pagelist
->length
);
3203 ceph_msg_data_add_pagelist(reply
, pagelist
);
3205 ceph_early_kick_flushing_caps(mdsc
, session
);
3207 ceph_con_send(&session
->s_con
, reply
);
3209 mutex_unlock(&session
->s_mutex
);
3211 mutex_lock(&mdsc
->mutex
);
3212 __wake_requests(mdsc
, &session
->s_waiting
);
3213 mutex_unlock(&mdsc
->mutex
);
3215 up_read(&mdsc
->snap_rwsem
);
3219 ceph_msg_put(reply
);
3220 up_read(&mdsc
->snap_rwsem
);
3221 mutex_unlock(&session
->s_mutex
);
3223 ceph_pagelist_release(pagelist
);
3225 pr_err("error %d preparing reconnect for mds%d\n", err
, mds
);
3231 * compare old and new mdsmaps, kicking requests
3232 * and closing out old connections as necessary
3234 * called under mdsc->mutex.
3236 static void check_new_map(struct ceph_mds_client
*mdsc
,
3237 struct ceph_mdsmap
*newmap
,
3238 struct ceph_mdsmap
*oldmap
)
3241 int oldstate
, newstate
;
3242 struct ceph_mds_session
*s
;
3244 dout("check_new_map new %u old %u\n",
3245 newmap
->m_epoch
, oldmap
->m_epoch
);
3247 for (i
= 0; i
< oldmap
->m_num_mds
&& i
< mdsc
->max_sessions
; i
++) {
3248 if (!mdsc
->sessions
[i
])
3250 s
= mdsc
->sessions
[i
];
3251 oldstate
= ceph_mdsmap_get_state(oldmap
, i
);
3252 newstate
= ceph_mdsmap_get_state(newmap
, i
);
3254 dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
3255 i
, ceph_mds_state_name(oldstate
),
3256 ceph_mdsmap_is_laggy(oldmap
, i
) ? " (laggy)" : "",
3257 ceph_mds_state_name(newstate
),
3258 ceph_mdsmap_is_laggy(newmap
, i
) ? " (laggy)" : "",
3259 ceph_session_state_name(s
->s_state
));
3261 if (i
>= newmap
->m_num_mds
||
3262 memcmp(ceph_mdsmap_get_addr(oldmap
, i
),
3263 ceph_mdsmap_get_addr(newmap
, i
),
3264 sizeof(struct ceph_entity_addr
))) {
3265 if (s
->s_state
== CEPH_MDS_SESSION_OPENING
) {
3266 /* the session never opened, just close it
3269 __unregister_session(mdsc
, s
);
3270 __wake_requests(mdsc
, &s
->s_waiting
);
3271 ceph_put_mds_session(s
);
3272 } else if (i
>= newmap
->m_num_mds
) {
3273 /* force close session for stopped mds */
3275 __unregister_session(mdsc
, s
);
3276 __wake_requests(mdsc
, &s
->s_waiting
);
3277 kick_requests(mdsc
, i
);
3278 mutex_unlock(&mdsc
->mutex
);
3280 mutex_lock(&s
->s_mutex
);
3281 cleanup_session_requests(mdsc
, s
);
3282 remove_session_caps(s
);
3283 mutex_unlock(&s
->s_mutex
);
3285 ceph_put_mds_session(s
);
3287 mutex_lock(&mdsc
->mutex
);
3290 mutex_unlock(&mdsc
->mutex
);
3291 mutex_lock(&s
->s_mutex
);
3292 mutex_lock(&mdsc
->mutex
);
3293 ceph_con_close(&s
->s_con
);
3294 mutex_unlock(&s
->s_mutex
);
3295 s
->s_state
= CEPH_MDS_SESSION_RESTARTING
;
3297 } else if (oldstate
== newstate
) {
3298 continue; /* nothing new with this mds */
3304 if (s
->s_state
== CEPH_MDS_SESSION_RESTARTING
&&
3305 newstate
>= CEPH_MDS_STATE_RECONNECT
) {
3306 mutex_unlock(&mdsc
->mutex
);
3307 send_mds_reconnect(mdsc
, s
);
3308 mutex_lock(&mdsc
->mutex
);
3312 * kick request on any mds that has gone active.
3314 if (oldstate
< CEPH_MDS_STATE_ACTIVE
&&
3315 newstate
>= CEPH_MDS_STATE_ACTIVE
) {
3316 if (oldstate
!= CEPH_MDS_STATE_CREATING
&&
3317 oldstate
!= CEPH_MDS_STATE_STARTING
)
3318 pr_info("mds%d recovery completed\n", s
->s_mds
);
3319 kick_requests(mdsc
, i
);
3320 ceph_kick_flushing_caps(mdsc
, s
);
3321 wake_up_session_caps(s
, 1);
3325 for (i
= 0; i
< newmap
->m_num_mds
&& i
< mdsc
->max_sessions
; i
++) {
3326 s
= mdsc
->sessions
[i
];
3329 if (!ceph_mdsmap_is_laggy(newmap
, i
))
3331 if (s
->s_state
== CEPH_MDS_SESSION_OPEN
||
3332 s
->s_state
== CEPH_MDS_SESSION_HUNG
||
3333 s
->s_state
== CEPH_MDS_SESSION_CLOSING
) {
3334 dout(" connecting to export targets of laggy mds%d\n",
3336 __open_export_target_sessions(mdsc
, s
);
3348 * caller must hold session s_mutex, dentry->d_lock
3350 void __ceph_mdsc_drop_dentry_lease(struct dentry
*dentry
)
3352 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
3354 ceph_put_mds_session(di
->lease_session
);
3355 di
->lease_session
= NULL
;
3358 static void handle_lease(struct ceph_mds_client
*mdsc
,
3359 struct ceph_mds_session
*session
,
3360 struct ceph_msg
*msg
)
3362 struct super_block
*sb
= mdsc
->fsc
->sb
;
3363 struct inode
*inode
;
3364 struct dentry
*parent
, *dentry
;
3365 struct ceph_dentry_info
*di
;
3366 int mds
= session
->s_mds
;
3367 struct ceph_mds_lease
*h
= msg
->front
.iov_base
;
3369 struct ceph_vino vino
;
3373 dout("handle_lease from mds%d\n", mds
);
3376 if (msg
->front
.iov_len
< sizeof(*h
) + sizeof(u32
))
3378 vino
.ino
= le64_to_cpu(h
->ino
);
3379 vino
.snap
= CEPH_NOSNAP
;
3380 seq
= le32_to_cpu(h
->seq
);
3381 dname
.name
= (void *)h
+ sizeof(*h
) + sizeof(u32
);
3382 dname
.len
= msg
->front
.iov_len
- sizeof(*h
) - sizeof(u32
);
3383 if (dname
.len
!= get_unaligned_le32(h
+1))
3387 inode
= ceph_find_inode(sb
, vino
);
3388 dout("handle_lease %s, ino %llx %p %.*s\n",
3389 ceph_lease_op_name(h
->action
), vino
.ino
, inode
,
3390 dname
.len
, dname
.name
);
3392 mutex_lock(&session
->s_mutex
);
3396 dout("handle_lease no inode %llx\n", vino
.ino
);
3401 parent
= d_find_alias(inode
);
3403 dout("no parent dentry on inode %p\n", inode
);
3405 goto release
; /* hrm... */
3407 dname
.hash
= full_name_hash(parent
, dname
.name
, dname
.len
);
3408 dentry
= d_lookup(parent
, &dname
);
3413 spin_lock(&dentry
->d_lock
);
3414 di
= ceph_dentry(dentry
);
3415 switch (h
->action
) {
3416 case CEPH_MDS_LEASE_REVOKE
:
3417 if (di
->lease_session
== session
) {
3418 if (ceph_seq_cmp(di
->lease_seq
, seq
) > 0)
3419 h
->seq
= cpu_to_le32(di
->lease_seq
);
3420 __ceph_mdsc_drop_dentry_lease(dentry
);
3425 case CEPH_MDS_LEASE_RENEW
:
3426 if (di
->lease_session
== session
&&
3427 di
->lease_gen
== session
->s_cap_gen
&&
3428 di
->lease_renew_from
&&
3429 di
->lease_renew_after
== 0) {
3430 unsigned long duration
=
3431 msecs_to_jiffies(le32_to_cpu(h
->duration_ms
));
3433 di
->lease_seq
= seq
;
3434 di
->time
= di
->lease_renew_from
+ duration
;
3435 di
->lease_renew_after
= di
->lease_renew_from
+
3437 di
->lease_renew_from
= 0;
3441 spin_unlock(&dentry
->d_lock
);
3448 /* let's just reuse the same message */
3449 h
->action
= CEPH_MDS_LEASE_REVOKE_ACK
;
3451 ceph_con_send(&session
->s_con
, msg
);
3455 mutex_unlock(&session
->s_mutex
);
3459 pr_err("corrupt lease message\n");
3463 void ceph_mdsc_lease_send_msg(struct ceph_mds_session
*session
,
3464 struct inode
*inode
,
3465 struct dentry
*dentry
, char action
,
3468 struct ceph_msg
*msg
;
3469 struct ceph_mds_lease
*lease
;
3470 int len
= sizeof(*lease
) + sizeof(u32
);
3473 dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
3474 inode
, dentry
, ceph_lease_op_name(action
), session
->s_mds
);
3475 dnamelen
= dentry
->d_name
.len
;
3478 msg
= ceph_msg_new(CEPH_MSG_CLIENT_LEASE
, len
, GFP_NOFS
, false);
3481 lease
= msg
->front
.iov_base
;
3482 lease
->action
= action
;
3483 lease
->ino
= cpu_to_le64(ceph_vino(inode
).ino
);
3484 lease
->first
= lease
->last
= cpu_to_le64(ceph_vino(inode
).snap
);
3485 lease
->seq
= cpu_to_le32(seq
);
3486 put_unaligned_le32(dnamelen
, lease
+ 1);
3487 memcpy((void *)(lease
+ 1) + 4, dentry
->d_name
.name
, dnamelen
);
3490 * if this is a preemptive lease RELEASE, no need to
3491 * flush request stream, since the actual request will
3494 msg
->more_to_follow
= (action
== CEPH_MDS_LEASE_RELEASE
);
3496 ceph_con_send(&session
->s_con
, msg
);
3500 * lock unlock sessions, to wait ongoing session activities
3502 static void lock_unlock_sessions(struct ceph_mds_client
*mdsc
)
3506 mutex_lock(&mdsc
->mutex
);
3507 for (i
= 0; i
< mdsc
->max_sessions
; i
++) {
3508 struct ceph_mds_session
*s
= __ceph_lookup_mds_session(mdsc
, i
);
3511 mutex_unlock(&mdsc
->mutex
);
3512 mutex_lock(&s
->s_mutex
);
3513 mutex_unlock(&s
->s_mutex
);
3514 ceph_put_mds_session(s
);
3515 mutex_lock(&mdsc
->mutex
);
3517 mutex_unlock(&mdsc
->mutex
);
3523 * delayed work -- periodically trim expired leases, renew caps with mds
3525 static void schedule_delayed(struct ceph_mds_client
*mdsc
)
3528 unsigned hz
= round_jiffies_relative(HZ
* delay
);
3529 schedule_delayed_work(&mdsc
->delayed_work
, hz
);
3532 static void delayed_work(struct work_struct
*work
)
3535 struct ceph_mds_client
*mdsc
=
3536 container_of(work
, struct ceph_mds_client
, delayed_work
.work
);
3540 dout("mdsc delayed_work\n");
3541 ceph_check_delayed_caps(mdsc
);
3543 mutex_lock(&mdsc
->mutex
);
3544 renew_interval
= mdsc
->mdsmap
->m_session_timeout
>> 2;
3545 renew_caps
= time_after_eq(jiffies
, HZ
*renew_interval
+
3546 mdsc
->last_renew_caps
);
3548 mdsc
->last_renew_caps
= jiffies
;
3550 for (i
= 0; i
< mdsc
->max_sessions
; i
++) {
3551 struct ceph_mds_session
*s
= __ceph_lookup_mds_session(mdsc
, i
);
3554 if (s
->s_state
== CEPH_MDS_SESSION_CLOSING
) {
3555 dout("resending session close request for mds%d\n",
3557 request_close_session(mdsc
, s
);
3558 ceph_put_mds_session(s
);
3561 if (s
->s_ttl
&& time_after(jiffies
, s
->s_ttl
)) {
3562 if (s
->s_state
== CEPH_MDS_SESSION_OPEN
) {
3563 s
->s_state
= CEPH_MDS_SESSION_HUNG
;
3564 pr_info("mds%d hung\n", s
->s_mds
);
3567 if (s
->s_state
< CEPH_MDS_SESSION_OPEN
) {
3568 /* this mds is failed or recovering, just wait */
3569 ceph_put_mds_session(s
);
3572 mutex_unlock(&mdsc
->mutex
);
3574 mutex_lock(&s
->s_mutex
);
3576 send_renew_caps(mdsc
, s
);
3578 ceph_con_keepalive(&s
->s_con
);
3579 if (s
->s_state
== CEPH_MDS_SESSION_OPEN
||
3580 s
->s_state
== CEPH_MDS_SESSION_HUNG
)
3581 ceph_send_cap_releases(mdsc
, s
);
3582 mutex_unlock(&s
->s_mutex
);
3583 ceph_put_mds_session(s
);
3585 mutex_lock(&mdsc
->mutex
);
3587 mutex_unlock(&mdsc
->mutex
);
3589 schedule_delayed(mdsc
);
3592 int ceph_mdsc_init(struct ceph_fs_client
*fsc
)
3595 struct ceph_mds_client
*mdsc
;
3597 mdsc
= kzalloc(sizeof(struct ceph_mds_client
), GFP_NOFS
);
3601 mutex_init(&mdsc
->mutex
);
3602 mdsc
->mdsmap
= kzalloc(sizeof(*mdsc
->mdsmap
), GFP_NOFS
);
3603 if (!mdsc
->mdsmap
) {
3609 init_completion(&mdsc
->safe_umount_waiters
);
3610 init_waitqueue_head(&mdsc
->session_close_wq
);
3611 INIT_LIST_HEAD(&mdsc
->waiting_for_map
);
3612 mdsc
->sessions
= NULL
;
3613 atomic_set(&mdsc
->num_sessions
, 0);
3614 mdsc
->max_sessions
= 0;
3616 atomic64_set(&mdsc
->quotarealms_count
, 0);
3617 mdsc
->last_snap_seq
= 0;
3618 init_rwsem(&mdsc
->snap_rwsem
);
3619 mdsc
->snap_realms
= RB_ROOT
;
3620 INIT_LIST_HEAD(&mdsc
->snap_empty
);
3621 spin_lock_init(&mdsc
->snap_empty_lock
);
3623 mdsc
->oldest_tid
= 0;
3624 mdsc
->request_tree
= RB_ROOT
;
3625 INIT_DELAYED_WORK(&mdsc
->delayed_work
, delayed_work
);
3626 mdsc
->last_renew_caps
= jiffies
;
3627 INIT_LIST_HEAD(&mdsc
->cap_delay_list
);
3628 spin_lock_init(&mdsc
->cap_delay_lock
);
3629 INIT_LIST_HEAD(&mdsc
->snap_flush_list
);
3630 spin_lock_init(&mdsc
->snap_flush_lock
);
3631 mdsc
->last_cap_flush_tid
= 1;
3632 INIT_LIST_HEAD(&mdsc
->cap_flush_list
);
3633 INIT_LIST_HEAD(&mdsc
->cap_dirty
);
3634 INIT_LIST_HEAD(&mdsc
->cap_dirty_migrating
);
3635 mdsc
->num_cap_flushing
= 0;
3636 spin_lock_init(&mdsc
->cap_dirty_lock
);
3637 init_waitqueue_head(&mdsc
->cap_flushing_wq
);
3638 spin_lock_init(&mdsc
->dentry_lru_lock
);
3639 INIT_LIST_HEAD(&mdsc
->dentry_lru
);
3641 ceph_caps_init(mdsc
);
3642 ceph_adjust_min_caps(mdsc
, fsc
->min_caps
);
3644 init_rwsem(&mdsc
->pool_perm_rwsem
);
3645 mdsc
->pool_perm_tree
= RB_ROOT
;
3647 strscpy(mdsc
->nodename
, utsname()->nodename
,
3648 sizeof(mdsc
->nodename
));
3653 * Wait for safe replies on open mds requests. If we time out, drop
3654 * all requests from the tree to avoid dangling dentry refs.
3656 static void wait_requests(struct ceph_mds_client
*mdsc
)
3658 struct ceph_options
*opts
= mdsc
->fsc
->client
->options
;
3659 struct ceph_mds_request
*req
;
3661 mutex_lock(&mdsc
->mutex
);
3662 if (__get_oldest_req(mdsc
)) {
3663 mutex_unlock(&mdsc
->mutex
);
3665 dout("wait_requests waiting for requests\n");
3666 wait_for_completion_timeout(&mdsc
->safe_umount_waiters
,
3667 ceph_timeout_jiffies(opts
->mount_timeout
));
3669 /* tear down remaining requests */
3670 mutex_lock(&mdsc
->mutex
);
3671 while ((req
= __get_oldest_req(mdsc
))) {
3672 dout("wait_requests timed out on tid %llu\n",
3674 __unregister_request(mdsc
, req
);
3677 mutex_unlock(&mdsc
->mutex
);
3678 dout("wait_requests done\n");
3682 * called before mount is ro, and before dentries are torn down.
3683 * (hmm, does this still race with new lookups?)
3685 void ceph_mdsc_pre_umount(struct ceph_mds_client
*mdsc
)
3687 dout("pre_umount\n");
3690 lock_unlock_sessions(mdsc
);
3691 ceph_flush_dirty_caps(mdsc
);
3692 wait_requests(mdsc
);
3695 * wait for reply handlers to drop their request refs and
3696 * their inode/dcache refs
3702 * wait for all write mds requests to flush.
3704 static void wait_unsafe_requests(struct ceph_mds_client
*mdsc
, u64 want_tid
)
3706 struct ceph_mds_request
*req
= NULL
, *nextreq
;
3709 mutex_lock(&mdsc
->mutex
);
3710 dout("wait_unsafe_requests want %lld\n", want_tid
);
3712 req
= __get_oldest_req(mdsc
);
3713 while (req
&& req
->r_tid
<= want_tid
) {
3714 /* find next request */
3715 n
= rb_next(&req
->r_node
);
3717 nextreq
= rb_entry(n
, struct ceph_mds_request
, r_node
);
3720 if (req
->r_op
!= CEPH_MDS_OP_SETFILELOCK
&&
3721 (req
->r_op
& CEPH_MDS_OP_WRITE
)) {
3723 ceph_mdsc_get_request(req
);
3725 ceph_mdsc_get_request(nextreq
);
3726 mutex_unlock(&mdsc
->mutex
);
3727 dout("wait_unsafe_requests wait on %llu (want %llu)\n",
3728 req
->r_tid
, want_tid
);
3729 wait_for_completion(&req
->r_safe_completion
);
3730 mutex_lock(&mdsc
->mutex
);
3731 ceph_mdsc_put_request(req
);
3733 break; /* next dne before, so we're done! */
3734 if (RB_EMPTY_NODE(&nextreq
->r_node
)) {
3735 /* next request was removed from tree */
3736 ceph_mdsc_put_request(nextreq
);
3739 ceph_mdsc_put_request(nextreq
); /* won't go away */
3743 mutex_unlock(&mdsc
->mutex
);
3744 dout("wait_unsafe_requests done\n");
3747 void ceph_mdsc_sync(struct ceph_mds_client
*mdsc
)
3749 u64 want_tid
, want_flush
;
3751 if (READ_ONCE(mdsc
->fsc
->mount_state
) == CEPH_MOUNT_SHUTDOWN
)
3755 mutex_lock(&mdsc
->mutex
);
3756 want_tid
= mdsc
->last_tid
;
3757 mutex_unlock(&mdsc
->mutex
);
3759 ceph_flush_dirty_caps(mdsc
);
3760 spin_lock(&mdsc
->cap_dirty_lock
);
3761 want_flush
= mdsc
->last_cap_flush_tid
;
3762 if (!list_empty(&mdsc
->cap_flush_list
)) {
3763 struct ceph_cap_flush
*cf
=
3764 list_last_entry(&mdsc
->cap_flush_list
,
3765 struct ceph_cap_flush
, g_list
);
3768 spin_unlock(&mdsc
->cap_dirty_lock
);
3770 dout("sync want tid %lld flush_seq %lld\n",
3771 want_tid
, want_flush
);
3773 wait_unsafe_requests(mdsc
, want_tid
);
3774 wait_caps_flush(mdsc
, want_flush
);
3778 * true if all sessions are closed, or we force unmount
3780 static bool done_closing_sessions(struct ceph_mds_client
*mdsc
, int skipped
)
3782 if (READ_ONCE(mdsc
->fsc
->mount_state
) == CEPH_MOUNT_SHUTDOWN
)
3784 return atomic_read(&mdsc
->num_sessions
) <= skipped
;
3788 * called after sb is ro.
3790 void ceph_mdsc_close_sessions(struct ceph_mds_client
*mdsc
)
3792 struct ceph_options
*opts
= mdsc
->fsc
->client
->options
;
3793 struct ceph_mds_session
*session
;
3797 dout("close_sessions\n");
3799 /* close sessions */
3800 mutex_lock(&mdsc
->mutex
);
3801 for (i
= 0; i
< mdsc
->max_sessions
; i
++) {
3802 session
= __ceph_lookup_mds_session(mdsc
, i
);
3805 mutex_unlock(&mdsc
->mutex
);
3806 mutex_lock(&session
->s_mutex
);
3807 if (__close_session(mdsc
, session
) <= 0)
3809 mutex_unlock(&session
->s_mutex
);
3810 ceph_put_mds_session(session
);
3811 mutex_lock(&mdsc
->mutex
);
3813 mutex_unlock(&mdsc
->mutex
);
3815 dout("waiting for sessions to close\n");
3816 wait_event_timeout(mdsc
->session_close_wq
,
3817 done_closing_sessions(mdsc
, skipped
),
3818 ceph_timeout_jiffies(opts
->mount_timeout
));
3820 /* tear down remaining sessions */
3821 mutex_lock(&mdsc
->mutex
);
3822 for (i
= 0; i
< mdsc
->max_sessions
; i
++) {
3823 if (mdsc
->sessions
[i
]) {
3824 session
= get_session(mdsc
->sessions
[i
]);
3825 __unregister_session(mdsc
, session
);
3826 mutex_unlock(&mdsc
->mutex
);
3827 mutex_lock(&session
->s_mutex
);
3828 remove_session_caps(session
);
3829 mutex_unlock(&session
->s_mutex
);
3830 ceph_put_mds_session(session
);
3831 mutex_lock(&mdsc
->mutex
);
3834 WARN_ON(!list_empty(&mdsc
->cap_delay_list
));
3835 mutex_unlock(&mdsc
->mutex
);
3837 ceph_cleanup_empty_realms(mdsc
);
3839 cancel_delayed_work_sync(&mdsc
->delayed_work
); /* cancel timer */
3844 void ceph_mdsc_force_umount(struct ceph_mds_client
*mdsc
)
3846 struct ceph_mds_session
*session
;
3849 dout("force umount\n");
3851 mutex_lock(&mdsc
->mutex
);
3852 for (mds
= 0; mds
< mdsc
->max_sessions
; mds
++) {
3853 session
= __ceph_lookup_mds_session(mdsc
, mds
);
3856 mutex_unlock(&mdsc
->mutex
);
3857 mutex_lock(&session
->s_mutex
);
3858 __close_session(mdsc
, session
);
3859 if (session
->s_state
== CEPH_MDS_SESSION_CLOSING
) {
3860 cleanup_session_requests(mdsc
, session
);
3861 remove_session_caps(session
);
3863 mutex_unlock(&session
->s_mutex
);
3864 ceph_put_mds_session(session
);
3865 mutex_lock(&mdsc
->mutex
);
3866 kick_requests(mdsc
, mds
);
3868 __wake_requests(mdsc
, &mdsc
->waiting_for_map
);
3869 mutex_unlock(&mdsc
->mutex
);
3872 static void ceph_mdsc_stop(struct ceph_mds_client
*mdsc
)
3875 cancel_delayed_work_sync(&mdsc
->delayed_work
); /* cancel timer */
3877 ceph_mdsmap_destroy(mdsc
->mdsmap
);
3878 kfree(mdsc
->sessions
);
3879 ceph_caps_finalize(mdsc
);
3880 ceph_pool_perm_destroy(mdsc
);
3883 void ceph_mdsc_destroy(struct ceph_fs_client
*fsc
)
3885 struct ceph_mds_client
*mdsc
= fsc
->mdsc
;
3886 dout("mdsc_destroy %p\n", mdsc
);
3891 /* flush out any connection work with references to us */
3894 ceph_mdsc_stop(mdsc
);
3898 dout("mdsc_destroy %p done\n", mdsc
);
3901 void ceph_mdsc_handle_fsmap(struct ceph_mds_client
*mdsc
, struct ceph_msg
*msg
)
3903 struct ceph_fs_client
*fsc
= mdsc
->fsc
;
3904 const char *mds_namespace
= fsc
->mount_options
->mds_namespace
;
3905 void *p
= msg
->front
.iov_base
;
3906 void *end
= p
+ msg
->front
.iov_len
;
3910 u32 mount_fscid
= (u32
)-1;
3911 u8 struct_v
, struct_cv
;
3914 ceph_decode_need(&p
, end
, sizeof(u32
), bad
);
3915 epoch
= ceph_decode_32(&p
);
3917 dout("handle_fsmap epoch %u\n", epoch
);
3919 ceph_decode_need(&p
, end
, 2 + sizeof(u32
), bad
);
3920 struct_v
= ceph_decode_8(&p
);
3921 struct_cv
= ceph_decode_8(&p
);
3922 map_len
= ceph_decode_32(&p
);
3924 ceph_decode_need(&p
, end
, sizeof(u32
) * 3, bad
);
3925 p
+= sizeof(u32
) * 2; /* skip epoch and legacy_client_fscid */
3927 num_fs
= ceph_decode_32(&p
);
3928 while (num_fs
-- > 0) {
3929 void *info_p
, *info_end
;
3934 ceph_decode_need(&p
, end
, 2 + sizeof(u32
), bad
);
3935 info_v
= ceph_decode_8(&p
);
3936 info_cv
= ceph_decode_8(&p
);
3937 info_len
= ceph_decode_32(&p
);
3938 ceph_decode_need(&p
, end
, info_len
, bad
);
3940 info_end
= p
+ info_len
;
3943 ceph_decode_need(&info_p
, info_end
, sizeof(u32
) * 2, bad
);
3944 fscid
= ceph_decode_32(&info_p
);
3945 namelen
= ceph_decode_32(&info_p
);
3946 ceph_decode_need(&info_p
, info_end
, namelen
, bad
);
3948 if (mds_namespace
&&
3949 strlen(mds_namespace
) == namelen
&&
3950 !strncmp(mds_namespace
, (char *)info_p
, namelen
)) {
3951 mount_fscid
= fscid
;
3956 ceph_monc_got_map(&fsc
->client
->monc
, CEPH_SUB_FSMAP
, epoch
);
3957 if (mount_fscid
!= (u32
)-1) {
3958 fsc
->client
->monc
.fs_cluster_id
= mount_fscid
;
3959 ceph_monc_want_map(&fsc
->client
->monc
, CEPH_SUB_MDSMAP
,
3961 ceph_monc_renew_subs(&fsc
->client
->monc
);
3969 pr_err("error decoding fsmap\n");
3971 mutex_lock(&mdsc
->mutex
);
3972 mdsc
->mdsmap_err
= err
;
3973 __wake_requests(mdsc
, &mdsc
->waiting_for_map
);
3974 mutex_unlock(&mdsc
->mutex
);
3978 * handle mds map update.
3980 void ceph_mdsc_handle_mdsmap(struct ceph_mds_client
*mdsc
, struct ceph_msg
*msg
)
3984 void *p
= msg
->front
.iov_base
;
3985 void *end
= p
+ msg
->front
.iov_len
;
3986 struct ceph_mdsmap
*newmap
, *oldmap
;
3987 struct ceph_fsid fsid
;
3990 ceph_decode_need(&p
, end
, sizeof(fsid
)+2*sizeof(u32
), bad
);
3991 ceph_decode_copy(&p
, &fsid
, sizeof(fsid
));
3992 if (ceph_check_fsid(mdsc
->fsc
->client
, &fsid
) < 0)
3994 epoch
= ceph_decode_32(&p
);
3995 maplen
= ceph_decode_32(&p
);
3996 dout("handle_map epoch %u len %d\n", epoch
, (int)maplen
);
3998 /* do we need it? */
3999 mutex_lock(&mdsc
->mutex
);
4000 if (mdsc
->mdsmap
&& epoch
<= mdsc
->mdsmap
->m_epoch
) {
4001 dout("handle_map epoch %u <= our %u\n",
4002 epoch
, mdsc
->mdsmap
->m_epoch
);
4003 mutex_unlock(&mdsc
->mutex
);
4007 newmap
= ceph_mdsmap_decode(&p
, end
);
4008 if (IS_ERR(newmap
)) {
4009 err
= PTR_ERR(newmap
);
4013 /* swap into place */
4015 oldmap
= mdsc
->mdsmap
;
4016 mdsc
->mdsmap
= newmap
;
4017 check_new_map(mdsc
, newmap
, oldmap
);
4018 ceph_mdsmap_destroy(oldmap
);
4020 mdsc
->mdsmap
= newmap
; /* first mds map */
4022 mdsc
->fsc
->sb
->s_maxbytes
= mdsc
->mdsmap
->m_max_file_size
;
4024 __wake_requests(mdsc
, &mdsc
->waiting_for_map
);
4025 ceph_monc_got_map(&mdsc
->fsc
->client
->monc
, CEPH_SUB_MDSMAP
,
4026 mdsc
->mdsmap
->m_epoch
);
4028 mutex_unlock(&mdsc
->mutex
);
4029 schedule_delayed(mdsc
);
4033 mutex_unlock(&mdsc
->mutex
);
4035 pr_err("error decoding mdsmap %d\n", err
);
4039 static struct ceph_connection
*con_get(struct ceph_connection
*con
)
4041 struct ceph_mds_session
*s
= con
->private;
4043 if (get_session(s
)) {
4044 dout("mdsc con_get %p ok (%d)\n", s
, refcount_read(&s
->s_ref
));
4047 dout("mdsc con_get %p FAIL\n", s
);
4051 static void con_put(struct ceph_connection
*con
)
4053 struct ceph_mds_session
*s
= con
->private;
4055 dout("mdsc con_put %p (%d)\n", s
, refcount_read(&s
->s_ref
) - 1);
4056 ceph_put_mds_session(s
);
4060 * if the client is unresponsive for long enough, the mds will kill
4061 * the session entirely.
4063 static void peer_reset(struct ceph_connection
*con
)
4065 struct ceph_mds_session
*s
= con
->private;
4066 struct ceph_mds_client
*mdsc
= s
->s_mdsc
;
4068 pr_warn("mds%d closed our session\n", s
->s_mds
);
4069 send_mds_reconnect(mdsc
, s
);
4072 static void dispatch(struct ceph_connection
*con
, struct ceph_msg
*msg
)
4074 struct ceph_mds_session
*s
= con
->private;
4075 struct ceph_mds_client
*mdsc
= s
->s_mdsc
;
4076 int type
= le16_to_cpu(msg
->hdr
.type
);
4078 mutex_lock(&mdsc
->mutex
);
4079 if (__verify_registered_session(mdsc
, s
) < 0) {
4080 mutex_unlock(&mdsc
->mutex
);
4083 mutex_unlock(&mdsc
->mutex
);
4086 case CEPH_MSG_MDS_MAP
:
4087 ceph_mdsc_handle_mdsmap(mdsc
, msg
);
4089 case CEPH_MSG_FS_MAP_USER
:
4090 ceph_mdsc_handle_fsmap(mdsc
, msg
);
4092 case CEPH_MSG_CLIENT_SESSION
:
4093 handle_session(s
, msg
);
4095 case CEPH_MSG_CLIENT_REPLY
:
4096 handle_reply(s
, msg
);
4098 case CEPH_MSG_CLIENT_REQUEST_FORWARD
:
4099 handle_forward(mdsc
, s
, msg
);
4101 case CEPH_MSG_CLIENT_CAPS
:
4102 ceph_handle_caps(s
, msg
);
4104 case CEPH_MSG_CLIENT_SNAP
:
4105 ceph_handle_snap(mdsc
, s
, msg
);
4107 case CEPH_MSG_CLIENT_LEASE
:
4108 handle_lease(mdsc
, s
, msg
);
4110 case CEPH_MSG_CLIENT_QUOTA
:
4111 ceph_handle_quota(mdsc
, s
, msg
);
4115 pr_err("received unknown message type %d %s\n", type
,
4116 ceph_msg_type_name(type
));
4127 * Note: returned pointer is the address of a structure that's
4128 * managed separately. Caller must *not* attempt to free it.
4130 static struct ceph_auth_handshake
*get_authorizer(struct ceph_connection
*con
,
4131 int *proto
, int force_new
)
4133 struct ceph_mds_session
*s
= con
->private;
4134 struct ceph_mds_client
*mdsc
= s
->s_mdsc
;
4135 struct ceph_auth_client
*ac
= mdsc
->fsc
->client
->monc
.auth
;
4136 struct ceph_auth_handshake
*auth
= &s
->s_auth
;
4138 if (force_new
&& auth
->authorizer
) {
4139 ceph_auth_destroy_authorizer(auth
->authorizer
);
4140 auth
->authorizer
= NULL
;
4142 if (!auth
->authorizer
) {
4143 int ret
= ceph_auth_create_authorizer(ac
, CEPH_ENTITY_TYPE_MDS
,
4146 return ERR_PTR(ret
);
4148 int ret
= ceph_auth_update_authorizer(ac
, CEPH_ENTITY_TYPE_MDS
,
4151 return ERR_PTR(ret
);
4153 *proto
= ac
->protocol
;
4159 static int verify_authorizer_reply(struct ceph_connection
*con
)
4161 struct ceph_mds_session
*s
= con
->private;
4162 struct ceph_mds_client
*mdsc
= s
->s_mdsc
;
4163 struct ceph_auth_client
*ac
= mdsc
->fsc
->client
->monc
.auth
;
4165 return ceph_auth_verify_authorizer_reply(ac
, s
->s_auth
.authorizer
);
4168 static int invalidate_authorizer(struct ceph_connection
*con
)
4170 struct ceph_mds_session
*s
= con
->private;
4171 struct ceph_mds_client
*mdsc
= s
->s_mdsc
;
4172 struct ceph_auth_client
*ac
= mdsc
->fsc
->client
->monc
.auth
;
4174 ceph_auth_invalidate_authorizer(ac
, CEPH_ENTITY_TYPE_MDS
);
4176 return ceph_monc_validate_auth(&mdsc
->fsc
->client
->monc
);
4179 static struct ceph_msg
*mds_alloc_msg(struct ceph_connection
*con
,
4180 struct ceph_msg_header
*hdr
, int *skip
)
4182 struct ceph_msg
*msg
;
4183 int type
= (int) le16_to_cpu(hdr
->type
);
4184 int front_len
= (int) le32_to_cpu(hdr
->front_len
);
4190 msg
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
4192 pr_err("unable to allocate msg type %d len %d\n",
4200 static int mds_sign_message(struct ceph_msg
*msg
)
4202 struct ceph_mds_session
*s
= msg
->con
->private;
4203 struct ceph_auth_handshake
*auth
= &s
->s_auth
;
4205 return ceph_auth_sign_message(auth
, msg
);
4208 static int mds_check_message_signature(struct ceph_msg
*msg
)
4210 struct ceph_mds_session
*s
= msg
->con
->private;
4211 struct ceph_auth_handshake
*auth
= &s
->s_auth
;
4213 return ceph_auth_check_message_signature(auth
, msg
);
4216 static const struct ceph_connection_operations mds_con_ops
= {
4219 .dispatch
= dispatch
,
4220 .get_authorizer
= get_authorizer
,
4221 .verify_authorizer_reply
= verify_authorizer_reply
,
4222 .invalidate_authorizer
= invalidate_authorizer
,
4223 .peer_reset
= peer_reset
,
4224 .alloc_msg
= mds_alloc_msg
,
4225 .sign_message
= mds_sign_message
,
4226 .check_message_signature
= mds_check_message_signature
,