projects/linux·blob master · refs

7439 lines · 199.0 KiB

1// SPDX-License-Identifier: GPL-2.0
2#include <linux/ceph/ceph_debug.h>
3
4#include <linux/fs.h>
5#include <linux/wait.h>
6#include <linux/slab.h>
7#include <linux/gfp.h>
8#include <linux/sched.h>
9#include <linux/sched/mm.h>
10#include <linux/delay.h>
11#include <linux/debugfs.h>
12#include <linux/seq_file.h>
13#include <linux/ratelimit.h>
14#include <linux/bits.h>
15#include <linux/ktime.h>
16#include <linux/bitmap.h>
17#include <linux/mnt_idmapping.h>
18
19#include "super.h"
20#include "mds_client.h"
21#include "crypto.h"
22
23#include <linux/ceph/ceph_features.h>
24#include <linux/ceph/messenger.h>
25#include <linux/ceph/decode.h>
26#include <linux/ceph/pagelist.h>
27#include <linux/ceph/auth.h>
28#include <linux/ceph/debugfs.h>
29#include <trace/events/ceph.h>
30
31#define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)
32
33/*
34 * A cluster of MDS (metadata server) daemons is responsible for
35 * managing the file system namespace (the directory hierarchy and
36 * inodes) and for coordinating shared access to storage. Metadata is
37 * partitioning hierarchically across a number of servers, and that
38 * partition varies over time as the cluster adjusts the distribution
39 * in order to balance load.
40 *
41 * The MDS client is primarily responsible to managing synchronous
42 * metadata requests for operations like open, unlink, and so forth.
43 * If there is a MDS failure, we find out about it when we (possibly
44 * request and) receive a new MDS map, and can resubmit affected
45 * requests.
46 *
47 * For the most part, though, we take advantage of a lossless
48 * communications channel to the MDS, and do not need to worry about
49 * timing out or resubmitting requests.
50 *
51 * We maintain a stateful "session" with each MDS we interact with.
52 * Within each session, we sent periodic heartbeat messages to ensure
53 * any capabilities or leases we have been issues remain valid. If
54 * the session times out and goes stale, our leases and capabilities
55 * are no longer valid.
56 */
57
58struct ceph_reconnect_state {
59 struct ceph_mds_session *session;
60 int nr_caps, nr_realms;
61 struct ceph_pagelist *pagelist;
62 unsigned msg_version;
63 bool allow_multi;
64};
65
66static void __wake_requests(struct ceph_mds_client *mdsc,
67 struct list_head *head);
68static void ceph_cap_release_work(struct work_struct *work);
69static void ceph_cap_reclaim_work(struct work_struct *work);
70static void ceph_mdsc_reset_workfn(struct work_struct *work);
71
72static const struct ceph_connection_operations mds_con_ops;
73
74static void ceph_metric_bind_session(struct ceph_mds_client *mdsc,
75 struct ceph_mds_session *session)
76{
77 struct ceph_mds_session *old;
78
79 if (!mdsc || !session || disable_send_metrics)
80 return;
81
82 old = mdsc->metric.session;
83 mdsc->metric.session = ceph_get_mds_session(session);
84 if (old)
85 ceph_put_mds_session(old);
86
87 metric_schedule_delayed(&mdsc->metric);
88}
89
90/*
91 * mds reply parsing
92 */
93
94static int parse_reply_info_quota(void **p, void *end,
95 struct ceph_mds_reply_info_in *info)
96{
97 u8 struct_v, struct_compat;
98 u32 struct_len;
99
100 ceph_decode_8_safe(p, end, struct_v, bad);
101 ceph_decode_8_safe(p, end, struct_compat, bad);
102 /* struct_v is expected to be >= 1. we only
103 * understand encoding with struct_compat == 1. */
104 if (!struct_v || struct_compat != 1)
105 goto bad;
106 ceph_decode_32_safe(p, end, struct_len, bad);
107 ceph_decode_need(p, end, struct_len, bad);
108 end = *p + struct_len;
109 ceph_decode_64_safe(p, end, info->max_bytes, bad);
110 ceph_decode_64_safe(p, end, info->max_files, bad);
111 *p = end;
112 return 0;
113bad:
114 return -EIO;
115}
116
117static int parse_reply_info_in(void **p, void *end,
118 struct ceph_mds_reply_info_in *info,
119 u64 features,
120 struct ceph_mds_client *mdsc)
121{
122 int err = 0;
123 u8 struct_v = 0;
124 u8 struct_compat = 0;
125 u32 struct_len = 0;
126
127 info->subvolume_id = CEPH_SUBVOLUME_ID_NONE;
128
129 if (features == (u64)-1) {
130 ceph_decode_8_safe(p, end, struct_v, bad);
131 ceph_decode_8_safe(p, end, struct_compat, bad);
132 /* struct_v is expected to be >= 1. we only understand
133 * encoding with struct_compat == 1. */
134 if (!struct_v || struct_compat != 1)
135 goto bad;
136 ceph_decode_32_safe(p, end, struct_len, bad);
137 ceph_decode_need(p, end, struct_len, bad);
138 end = *p + struct_len;
139 }
140
141 ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);
142 info->in = *p;
143 *p += sizeof(struct ceph_mds_reply_inode) +
144 sizeof(*info->in->fragtree.splits) *
145 le32_to_cpu(info->in->fragtree.nsplits);
146
147 ceph_decode_32_safe(p, end, info->symlink_len, bad);
148 ceph_decode_need(p, end, info->symlink_len, bad);
149 info->symlink = *p;
150 *p += info->symlink_len;
151
152 ceph_decode_copy_safe(p, end, &info->dir_layout,
153 sizeof(info->dir_layout), bad);
154 ceph_decode_32_safe(p, end, info->xattr_len, bad);
155 ceph_decode_need(p, end, info->xattr_len, bad);
156 info->xattr_data = *p;
157 *p += info->xattr_len;
158
159 if (features == (u64)-1) {
160 /* inline data */
161 ceph_decode_64_safe(p, end, info->inline_version, bad);
162 ceph_decode_32_safe(p, end, info->inline_len, bad);
163 ceph_decode_need(p, end, info->inline_len, bad);
164 info->inline_data = *p;
165 *p += info->inline_len;
166 /* quota */
167 err = parse_reply_info_quota(p, end, info);
168 if (err < 0)
169 goto out_bad;
170 /* pool namespace */
171 ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
172 if (info->pool_ns_len > 0) {
173 ceph_decode_need(p, end, info->pool_ns_len, bad);
174 info->pool_ns_data = *p;
175 *p += info->pool_ns_len;
176 }
177
178 /* btime */
179 ceph_decode_need(p, end, sizeof(info->btime), bad);
180 ceph_decode_copy(p, &info->btime, sizeof(info->btime));
181
182 /* change attribute */
183 ceph_decode_64_safe(p, end, info->change_attr, bad);
184
185 /* dir pin */
186 if (struct_v >= 2) {
187 ceph_decode_32_safe(p, end, info->dir_pin, bad);
188 } else {
189 info->dir_pin = -ENODATA;
190 }
191
192 /* snapshot birth time, remains zero for v<=2 */
193 if (struct_v >= 3) {
194 ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
195 ceph_decode_copy(p, &info->snap_btime,
196 sizeof(info->snap_btime));
197 } else {
198 memset(&info->snap_btime, 0, sizeof(info->snap_btime));
199 }
200
201 /* snapshot count, remains zero for v<=3 */
202 if (struct_v >= 4) {
203 ceph_decode_64_safe(p, end, info->rsnaps, bad);
204 } else {
205 info->rsnaps = 0;
206 }
207
208 if (struct_v >= 5) {
209 u32 alen;
210
211 ceph_decode_32_safe(p, end, alen, bad);
212
213 while (alen--) {
214 u32 len;
215
216 /* key */
217 ceph_decode_32_safe(p, end, len, bad);
218 ceph_decode_skip_n(p, end, len, bad);
219 /* value */
220 ceph_decode_32_safe(p, end, len, bad);
221 ceph_decode_skip_n(p, end, len, bad);
222 }
223 }
224
225 /* fscrypt flag -- ignore */
226 if (struct_v >= 6)
227 ceph_decode_skip_8(p, end, bad);
228
229 info->fscrypt_auth = NULL;
230 info->fscrypt_auth_len = 0;
231 info->fscrypt_file = NULL;
232 info->fscrypt_file_len = 0;
233 if (struct_v >= 7) {
234 ceph_decode_32_safe(p, end, info->fscrypt_auth_len, bad);
235 if (info->fscrypt_auth_len) {
236 info->fscrypt_auth = kmalloc(info->fscrypt_auth_len,
237 GFP_KERNEL);
238 if (!info->fscrypt_auth)
239 return -ENOMEM;
240 ceph_decode_copy_safe(p, end, info->fscrypt_auth,
241 info->fscrypt_auth_len, bad);
242 }
243 ceph_decode_32_safe(p, end, info->fscrypt_file_len, bad);
244 if (info->fscrypt_file_len) {
245 info->fscrypt_file = kmalloc(info->fscrypt_file_len,
246 GFP_KERNEL);
247 if (!info->fscrypt_file)
248 return -ENOMEM;
249 ceph_decode_copy_safe(p, end, info->fscrypt_file,
250 info->fscrypt_file_len, bad);
251 }
252 }
253
254 /*
255 * InodeStat encoding versions:
256 * v1-v7: various fields added over time
257 * v8: added optmetadata (versioned sub-structure containing
258 * optional inode metadata like charmap for case-insensitive
259 * filesystems). The kernel client doesn't support
260 * case-insensitive lookups, so we skip this field.
261 * v9: added subvolume_id (parsed below)
262 */
263 if (struct_v >= 8) {
264 u32 v8_struct_len;
265
266 /* skip optmetadata versioned sub-structure */
267 ceph_decode_skip_8(p, end, bad); /* struct_v */
268 ceph_decode_skip_8(p, end, bad); /* struct_compat */
269 ceph_decode_32_safe(p, end, v8_struct_len, bad);
270 ceph_decode_skip_n(p, end, v8_struct_len, bad);
271 }
272
273 /* struct_v 9 added subvolume_id */
274 if (struct_v >= 9)
275 ceph_decode_64_safe(p, end, info->subvolume_id, bad);
276
277 *p = end;
278 } else {
279 /* legacy (unversioned) struct */
280 if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
281 ceph_decode_64_safe(p, end, info->inline_version, bad);
282 ceph_decode_32_safe(p, end, info->inline_len, bad);
283 ceph_decode_need(p, end, info->inline_len, bad);
284 info->inline_data = *p;
285 *p += info->inline_len;
286 } else
287 info->inline_version = CEPH_INLINE_NONE;
288
289 if (features & CEPH_FEATURE_MDS_QUOTA) {
290 err = parse_reply_info_quota(p, end, info);
291 if (err < 0)
292 goto out_bad;
293 } else {
294 info->max_bytes = 0;
295 info->max_files = 0;
296 }
297
298 info->pool_ns_len = 0;
299 info->pool_ns_data = NULL;
300 if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
301 ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
302 if (info->pool_ns_len > 0) {
303 ceph_decode_need(p, end, info->pool_ns_len, bad);
304 info->pool_ns_data = *p;
305 *p += info->pool_ns_len;
306 }
307 }
308
309 if (features & CEPH_FEATURE_FS_BTIME) {
310 ceph_decode_need(p, end, sizeof(info->btime), bad);
311 ceph_decode_copy(p, &info->btime, sizeof(info->btime));
312 ceph_decode_64_safe(p, end, info->change_attr, bad);
313 }
314
315 info->dir_pin = -ENODATA;
316 /* info->snap_btime and info->rsnaps remain zero */
317 }
318 return 0;
319bad:
320 err = -EIO;
321out_bad:
322 return err;
323}
324
325static int parse_reply_info_dir(void **p, void *end,
326 struct ceph_mds_reply_dirfrag **dirfrag,
327 u64 features)
328{
329 if (features == (u64)-1) {
330 u8 struct_v, struct_compat;
331 u32 struct_len;
332 ceph_decode_8_safe(p, end, struct_v, bad);
333 ceph_decode_8_safe(p, end, struct_compat, bad);
334 /* struct_v is expected to be >= 1. we only understand
335 * encoding whose struct_compat == 1. */
336 if (!struct_v || struct_compat != 1)
337 goto bad;
338 ceph_decode_32_safe(p, end, struct_len, bad);
339 ceph_decode_need(p, end, struct_len, bad);
340 end = *p + struct_len;
341 }
342
343 ceph_decode_need(p, end, sizeof(**dirfrag), bad);
344 *dirfrag = *p;
345 *p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist);
346 if (unlikely(*p > end))
347 goto bad;
348 if (features == (u64)-1)
349 *p = end;
350 return 0;
351bad:
352 return -EIO;
353}
354
355static int parse_reply_info_lease(void **p, void *end,
356 struct ceph_mds_reply_lease **lease,
357 u64 features, u32 *altname_len, u8 **altname)
358{
359 u8 struct_v;
360 u32 struct_len;
361 void *lend;
362
363 if (features == (u64)-1) {
364 u8 struct_compat;
365
366 ceph_decode_8_safe(p, end, struct_v, bad);
367 ceph_decode_8_safe(p, end, struct_compat, bad);
368
369 /* struct_v is expected to be >= 1. we only understand
370 * encoding whose struct_compat == 1. */
371 if (!struct_v || struct_compat != 1)
372 goto bad;
373
374 ceph_decode_32_safe(p, end, struct_len, bad);
375 } else {
376 struct_len = sizeof(**lease);
377 *altname_len = 0;
378 *altname = NULL;
379 }
380
381 lend = *p + struct_len;
382 ceph_decode_need(p, end, struct_len, bad);
383 *lease = *p;
384 *p += sizeof(**lease);
385
386 if (features == (u64)-1) {
387 if (struct_v >= 2) {
388 ceph_decode_32_safe(p, end, *altname_len, bad);
389 ceph_decode_need(p, end, *altname_len, bad);
390 *altname = *p;
391 *p += *altname_len;
392 } else {
393 *altname = NULL;
394 *altname_len = 0;
395 }
396 }
397 *p = lend;
398 return 0;
399bad:
400 return -EIO;
401}
402
403/*
404 * parse a normal reply, which may contain a (dir+)dentry and/or a
405 * target inode.
406 */
407static int parse_reply_info_trace(void **p, void *end,
408 struct ceph_mds_reply_info_parsed *info,
409 u64 features,
410 struct ceph_mds_client *mdsc)
411{
412 int err;
413
414 if (info->head->is_dentry) {
415 err = parse_reply_info_in(p, end, &info->diri, features, mdsc);
416 if (err < 0)
417 goto out_bad;
418
419 err = parse_reply_info_dir(p, end, &info->dirfrag, features);
420 if (err < 0)
421 goto out_bad;
422
423 ceph_decode_32_safe(p, end, info->dname_len, bad);
424 ceph_decode_need(p, end, info->dname_len, bad);
425 info->dname = *p;
426 *p += info->dname_len;
427
428 err = parse_reply_info_lease(p, end, &info->dlease, features,
429 &info->altname_len, &info->altname);
430 if (err < 0)
431 goto out_bad;
432 }
433
434 if (info->head->is_target) {
435 err = parse_reply_info_in(p, end, &info->targeti, features,
436 mdsc);
437 if (err < 0)
438 goto out_bad;
439 }
440
441 if (unlikely(*p != end))
442 goto bad;
443 return 0;
444
445bad:
446 err = -EIO;
447out_bad:
448 pr_err("problem parsing mds trace %d\n", err);
449 return err;
450}
451
452/*
453 * parse readdir results
454 */
455static int parse_reply_info_readdir(void **p, void *end,
456 struct ceph_mds_request *req,
457 u64 features,
458 struct ceph_mds_client *mdsc)
459{
460 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info;
461 struct ceph_client *cl = req->r_mdsc->fsc->client;
462 u32 num, i = 0;
463 int err;
464
465 err = parse_reply_info_dir(p, end, &info->dir_dir, features);
466 if (err < 0)
467 goto out_bad;
468
469 ceph_decode_need(p, end, sizeof(num) + 2, bad);
470 num = ceph_decode_32(p);
471 {
472 u16 flags = ceph_decode_16(p);
473 info->dir_end = !!(flags & CEPH_READDIR_FRAG_END);
474 info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE);
475 info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER);
476 info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH);
477 }
478 if (num == 0)
479 goto done;
480
481 BUG_ON(!info->dir_entries);
482 if ((unsigned long)(info->dir_entries + num) >
483 (unsigned long)info->dir_entries + info->dir_buf_size) {
484 pr_err_client(cl, "dir contents are larger than expected\n");
485 WARN_ON(1);
486 goto bad;
487 }
488
489 info->dir_nr = num;
490 while (num) {
491 struct inode *inode = d_inode(req->r_dentry);
492 struct ceph_inode_info *ci = ceph_inode(inode);
493 struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
494 struct fscrypt_str oname = FSTR_INIT(NULL, 0);
495 struct ceph_fname fname;
496 u32 altname_len, _name_len;
497 u8 *altname, *_name;
498 u8 *tname = NULL;
499
500 /* dentry */
501 ceph_decode_32_safe(p, end, _name_len, bad);
502 ceph_decode_need(p, end, _name_len, bad);
503 _name = *p;
504 *p += _name_len;
505 doutc(cl, "parsed dir dname '%.*s'\n", _name_len, _name);
506
507 if (info->hash_order)
508 rde->raw_hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
509 _name, _name_len);
510
511 /* dentry lease */
512 err = parse_reply_info_lease(p, end, &rde->lease, features,
513 &altname_len, &altname);
514 if (err)
515 goto out_bad;
516
517 /*
518 * Try to dencrypt the dentry names and update them
519 * in the ceph_mds_reply_dir_entry struct.
520 */
521 fname.dir = inode;
522 fname.name = _name;
523 fname.name_len = _name_len;
524 fname.ctext = altname;
525 fname.ctext_len = altname_len;
526 /*
527 * The _name_len maybe larger than altname_len, such as
528 * when the human readable name length is in range of
529 * (CEPH_NOHASH_NAME_MAX, CEPH_NOHASH_NAME_MAX + SHA256_DIGEST_SIZE),
530 * then the copy in ceph_fname_to_usr will corrupt the
531 * data if there has no encryption key.
532 *
533 * Just set the no_copy flag and then if there has no
534 * encryption key the oname.name will be assigned to
535 * _name always.
536 */
537 fname.no_copy = true;
538 if (altname_len == 0) {
539 /*
540 * Set tname to _name, and this will be used
541 * to do the base64_decode in-place. It's
542 * safe because the decoded string should
543 * always be shorter, which is 3/4 of origin
544 * string. If this message was allocated with
545 * vmalloc() (happens, but rarely), leave it
546 * NULL and let ceph_fname_to_usr() allocate
547 * suitable temporary working space instead.
548 */
549 if (likely(!is_vmalloc_addr(_name)))
550 tname = _name;
551
552 /*
553 * Set oname to _name too, and this will be
554 * used to do the dencryption in-place.
555 */
556 oname.name = _name;
557 oname.len = _name_len;
558 } else {
559 /*
560 * This will do the decryption only in-place
561 * from altname cryptext directly.
562 */
563 oname.name = altname;
564 oname.len = altname_len;
565 }
566 rde->is_nokey = false;
567 err = ceph_fname_to_usr(&fname, tname, &oname, &rde->is_nokey);
568 if (err) {
569 pr_err_client(cl, "unable to decode %.*s, got %d\n",
570 _name_len, _name, err);
571 goto out_bad;
572 }
573 rde->name = oname.name;
574 rde->name_len = oname.len;
575
576 /* inode */
577 err = parse_reply_info_in(p, end, &rde->inode, features, mdsc);
578 if (err < 0)
579 goto out_bad;
580 /* ceph_readdir_prepopulate() will update it */
581 rde->offset = 0;
582 i++;
583 num--;
584 }
585
586done:
587 /* Skip over any unrecognized fields */
588 *p = end;
589 return 0;
590
591bad:
592 err = -EIO;
593out_bad:
594 pr_err_client(cl, "problem parsing dir contents %d\n", err);
595 return err;
596}
597
598/*
599 * parse fcntl F_GETLK results
600 */
601static int parse_reply_info_filelock(void **p, void *end,
602 struct ceph_mds_reply_info_parsed *info,
603 u64 features)
604{
605 if (*p + sizeof(*info->filelock_reply) > end)
606 goto bad;
607
608 info->filelock_reply = *p;
609
610 /* Skip over any unrecognized fields */
611 *p = end;
612 return 0;
613bad:
614 return -EIO;
615}
616
617
618#if BITS_PER_LONG == 64
619
620#define DELEGATED_INO_AVAILABLE xa_mk_value(1)
621
622static int ceph_insert_deleg_ino(struct ceph_mds_session *s, u64 ino)
623{
624 struct ceph_client *cl = s->s_mdsc->fsc->client;
625 int err;
626
627 /*
628 * Cap how many delegated inodes a single session may hold. This is
629 * the only place that grows the count, so atomic_add_unless() bounds
630 * it at exactly CEPH_MAX_DELEG_INOS; s_num_deleg_inos can never exceed
631 * that.
632 */
633 if (!atomic_add_unless(&s->s_num_deleg_inos, 1, CEPH_MAX_DELEG_INOS)) {
634 pr_warn_ratelimited_client(cl,
635 "MDS session already holds %d delegated inodes\n",
636 CEPH_MAX_DELEG_INOS);
637 return -EOVERFLOW;
638 }
639
640 err = xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
641 GFP_KERNEL);
642 if (err)
643 atomic_dec(&s->s_num_deleg_inos);
644 return err;
645}
646
647static int ceph_parse_deleg_inos(void **p, void *end,
648 struct ceph_mds_session *s)
649{
650 struct ceph_client *cl = s->s_mdsc->fsc->client;
651 u64 msg_deleg_inos = 0;
652 u32 sets;
653
654 ceph_decode_32_safe(p, end, sets, bad);
655 doutc(cl, "got %u sets of delegated inodes\n", sets);
656 while (sets--) {
657 u64 start, len;
658
659 ceph_decode_64_safe(p, end, start, bad);
660 ceph_decode_64_safe(p, end, len, bad);
661
662 /* Don't accept a delegation of system inodes */
663 if (start < CEPH_INO_SYSTEM_BASE) {
664 pr_warn_ratelimited_client(cl,
665 "ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n",
666 start, len);
667 continue;
668 }
669
670 /*
671 * Bound the number of inodes one reply may delegate.
672 * ceph_insert_deleg_ino() separately caps the per-session
673 * population, so this only has to stop one reply from spinning
674 * the insert loop under an attacker-controlled len.
675 */
676 if (len > (u64)CEPH_MAX_DELEG_INOS ||
677 msg_deleg_inos > (u64)CEPH_MAX_DELEG_INOS - len) {
678 pr_warn_ratelimited_client(cl,
679 "MDS reply delegates too many inodes (have %llu, +%llu, max %d)\n",
680 msg_deleg_inos, len, CEPH_MAX_DELEG_INOS);
681 return -EIO;
682 }
683 msg_deleg_inos += len;
684
685 while (len--) {
686 int err = ceph_insert_deleg_ino(s, start++);
687
688 if (!err) {
689 doutc(cl, "added delegated inode 0x%llx\n", start - 1);
690 } else if (err == -EBUSY) {
691 pr_warn_client(cl,
692 "MDS delegated inode 0x%llx more than once.\n",
693 start - 1);
694 } else if (err == -EOVERFLOW) {
695 /* ceph_insert_deleg_ino() already warned. */
696 return -EIO;
697 } else {
698 return err;
699 }
700 }
701 }
702 return 0;
703bad:
704 return -EIO;
705}
706
707u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
708{
709 unsigned long ino;
710 void *val;
711
712 xa_for_each(&s->s_delegated_inos, ino, val) {
713 val = xa_erase(&s->s_delegated_inos, ino);
714 if (val == DELEGATED_INO_AVAILABLE) {
715 atomic_dec(&s->s_num_deleg_inos);
716 return ino;
717 }
718 }
719 return 0;
720}
721
722int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
723{
724 return ceph_insert_deleg_ino(s, ino);
725}
726#else /* BITS_PER_LONG == 64 */
727/*
728 * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
729 * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
730 * and bottom words?
731 */
732static int ceph_parse_deleg_inos(void **p, void *end,
733 struct ceph_mds_session *s)
734{
735 u32 sets;
736
737 ceph_decode_32_safe(p, end, sets, bad);
738 if (sets)
739 ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
740 return 0;
741bad:
742 return -EIO;
743}
744
745u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
746{
747 return 0;
748}
749
750int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
751{
752 return 0;
753}
754#endif /* BITS_PER_LONG == 64 */
755
756/*
757 * parse create results
758 */
759static int parse_reply_info_create(void **p, void *end,
760 struct ceph_mds_reply_info_parsed *info,
761 u64 features, struct ceph_mds_session *s)
762{
763 int ret;
764
765 if (features == (u64)-1 ||
766 (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {
767 if (*p == end) {
768 /* Malformed reply? */
769 info->has_create_ino = false;
770 } else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
771 info->has_create_ino = true;
772 /* struct_v, struct_compat, and len */
773 ceph_decode_skip_n(p, end, 2 + sizeof(u32), bad);
774 ceph_decode_64_safe(p, end, info->ino, bad);
775 ret = ceph_parse_deleg_inos(p, end, s);
776 if (ret)
777 return ret;
778 } else {
779 /* legacy */
780 ceph_decode_64_safe(p, end, info->ino, bad);
781 info->has_create_ino = true;
782 }
783 } else {
784 if (*p != end)
785 goto bad;
786 }
787
788 /* Skip over any unrecognized fields */
789 *p = end;
790 return 0;
791bad:
792 return -EIO;
793}
794
795static int parse_reply_info_getvxattr(void **p, void *end,
796 struct ceph_mds_reply_info_parsed *info,
797 u64 features)
798{
799 u32 value_len;
800
801 ceph_decode_skip_8(p, end, bad); /* skip current version: 1 */
802 ceph_decode_skip_8(p, end, bad); /* skip first version: 1 */
803 ceph_decode_skip_32(p, end, bad); /* skip payload length */
804
805 ceph_decode_32_safe(p, end, value_len, bad);
806
807 if (value_len == end - *p) {
808 info->xattr_info.xattr_value = *p;
809 info->xattr_info.xattr_value_len = value_len;
810 *p = end;
811 return value_len;
812 }
813bad:
814 return -EIO;
815}
816
817/*
818 * parse extra results
819 */
820static int parse_reply_info_extra(void **p, void *end,
821 struct ceph_mds_request *req,
822 u64 features, struct ceph_mds_session *s)
823{
824 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info;
825 u32 op = le32_to_cpu(info->head->op);
826
827 if (op == CEPH_MDS_OP_GETFILELOCK)
828 return parse_reply_info_filelock(p, end, info, features);
829 else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
830 return parse_reply_info_readdir(p, end, req, features,
831 req->r_mdsc);
832 else if (op == CEPH_MDS_OP_CREATE)
833 return parse_reply_info_create(p, end, info, features, s);
834 else if (op == CEPH_MDS_OP_GETVXATTR)
835 return parse_reply_info_getvxattr(p, end, info, features);
836 else
837 return -EIO;
838}
839
840/*
841 * parse entire mds reply
842 */
843static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg,
844 struct ceph_mds_request *req, u64 features)
845{
846 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info;
847 struct ceph_client *cl = s->s_mdsc->fsc->client;
848 void *p, *end;
849 u32 len;
850 int err;
851
852 info->head = msg->front.iov_base;
853 p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
854 end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
855
856 /* trace */
857 ceph_decode_32_safe(&p, end, len, bad);
858 if (len > 0) {
859 ceph_decode_need(&p, end, len, bad);
860 err = parse_reply_info_trace(&p, p + len, info, features,
861 s->s_mdsc);
862 if (err < 0)
863 goto out_bad;
864 }
865
866 /* extra */
867 ceph_decode_32_safe(&p, end, len, bad);
868 if (len > 0) {
869 ceph_decode_need(&p, end, len, bad);
870 err = parse_reply_info_extra(&p, p + len, req, features, s);
871 if (err < 0)
872 goto out_bad;
873 }
874
875 /* snap blob */
876 ceph_decode_32_safe(&p, end, len, bad);
877 info->snapblob_len = len;
878 info->snapblob = p;
879 p += len;
880
881 if (p != end)
882 goto bad;
883 return 0;
884
885bad:
886 err = -EIO;
887out_bad:
888 pr_err_client(cl, "mds parse_reply err %d\n", err);
889 ceph_msg_dump(msg);
890 return err;
891}
892
893static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
894{
895 int i;
896
897 kfree(info->diri.fscrypt_auth);
898 kfree(info->diri.fscrypt_file);
899 kfree(info->targeti.fscrypt_auth);
900 kfree(info->targeti.fscrypt_file);
901 if (!info->dir_entries)
902 return;
903
904 for (i = 0; i < info->dir_nr; i++) {
905 struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
906
907 kfree(rde->inode.fscrypt_auth);
908 kfree(rde->inode.fscrypt_file);
909 }
910 free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size));
911}
912
913/*
914 * In async unlink case the kclient won't wait for the first reply
915 * from MDS and just drop all the links and unhash the dentry and then
916 * succeeds immediately.
917 *
918 * For any new create/link/rename,etc requests followed by using the
919 * same file names we must wait for the first reply of the inflight
920 * unlink request, or the MDS possibly will fail these following
921 * requests with -EEXIST if the inflight async unlink request was
922 * delayed for some reasons.
923 *
924 * And the worst case is that for the none async openc request it will
925 * successfully open the file if the CDentry hasn't been unlinked yet,
926 * but later the previous delayed async unlink request will remove the
927 * CDentry. That means the just created file is possibly deleted later
928 * by accident.
929 *
930 * We need to wait for the inflight async unlink requests to finish
931 * when creating new files/directories by using the same file names.
932 */
933int ceph_wait_on_conflict_unlink(struct dentry *dentry)
934{
935 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb);
936 struct ceph_client *cl = fsc->client;
937 struct dentry *pdentry = dentry->d_parent;
938 struct dentry *udentry, *found = NULL;
939 struct ceph_dentry_info *di;
940 struct qstr dname;
941 u32 hash = dentry->d_name.hash;
942 int err;
943
944 dname.name = dentry->d_name.name;
945 dname.len = dentry->d_name.len;
946
947 rcu_read_lock();
948 hash_for_each_possible_rcu(fsc->async_unlink_conflict, di,
949 hnode, hash) {
950 udentry = di->dentry;
951
952 spin_lock(&udentry->d_lock);
953 if (udentry->d_name.hash != hash)
954 goto next;
955 if (unlikely(udentry->d_parent != pdentry))
956 goto next;
957 if (!hash_hashed(&di->hnode))
958 goto next;
959
960 if (!test_bit(CEPH_DENTRY_ASYNC_UNLINK_BIT, &di->flags))
961 pr_warn_client(cl, "dentry %p:%pd async unlink bit is not set\n",
962 dentry, dentry);
963
964 if (!d_same_name(udentry, pdentry, &dname))
965 goto next;
966
967 found = dget_dlock(udentry);
968 spin_unlock(&udentry->d_lock);
969 break;
970next:
971 spin_unlock(&udentry->d_lock);
972 }
973 rcu_read_unlock();
974
975 if (likely(!found))
976 return 0;
977
978 doutc(cl, "dentry %p:%pd conflict with old %p:%pd\n", dentry, dentry,
979 found, found);
980
981 err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_UNLINK_BIT,
982 TASK_KILLABLE);
983 dput(found);
984 return err;
985}
986
987
988/*
989 * sessions
990 */
991const char *ceph_session_state_name(int s)
992{
993 switch (s) {
994 case CEPH_MDS_SESSION_NEW: return "new";
995 case CEPH_MDS_SESSION_OPENING: return "opening";
996 case CEPH_MDS_SESSION_OPEN: return "open";
997 case CEPH_MDS_SESSION_HUNG: return "hung";
998 case CEPH_MDS_SESSION_CLOSING: return "closing";
999 case CEPH_MDS_SESSION_CLOSED: return "closed";
1000 case CEPH_MDS_SESSION_RESTARTING: return "restarting";
1001 case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
1002 case CEPH_MDS_SESSION_REJECTED: return "rejected";
1003 default: return "???";
1004 }
1005}
1006
1007struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s)
1008{
1009 if (refcount_inc_not_zero(&s->s_ref))
1010 return s;
1011 return NULL;
1012}
1013
1014void ceph_put_mds_session(struct ceph_mds_session *s)
1015{
1016 if (IS_ERR_OR_NULL(s))
1017 return;
1018
1019 if (refcount_dec_and_test(&s->s_ref)) {
1020 if (s->s_auth.authorizer)
1021 ceph_auth_destroy_authorizer(s->s_auth.authorizer);
1022 WARN_ON(mutex_is_locked(&s->s_mutex));
1023 xa_destroy(&s->s_delegated_inos);
1024 kfree(s);
1025 }
1026}
1027
1028/*
1029 * called under mdsc->mutex
1030 */
1031struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
1032 int mds)
1033{
1034 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
1035 return NULL;
1036 return ceph_get_mds_session(mdsc->sessions[mds]);
1037}
1038
1039static bool __have_session(struct ceph_mds_client *mdsc, int mds)
1040{
1041 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
1042 return false;
1043 else
1044 return true;
1045}
1046
1047static int __verify_registered_session(struct ceph_mds_client *mdsc,
1048 struct ceph_mds_session *s)
1049{
1050 if (s->s_mds >= mdsc->max_sessions ||
1051 mdsc->sessions[s->s_mds] != s)
1052 return -ENOENT;
1053 return 0;
1054}
1055
1056/*
1057 * create+register a new session for given mds.
1058 * called under mdsc->mutex.
1059 */
1060static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
1061 int mds)
1062{
1063 struct ceph_client *cl = mdsc->fsc->client;
1064 struct ceph_mds_session *s;
1065
1066 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO)
1067 return ERR_PTR(-EIO);
1068
1069 if (mds >= mdsc->mdsmap->possible_max_rank)
1070 return ERR_PTR(-EINVAL);
1071
1072 s = kzalloc_obj(*s, GFP_NOFS);
1073 if (!s)
1074 return ERR_PTR(-ENOMEM);
1075
1076 if (mds >= mdsc->max_sessions) {
1077 int newmax = 1 << get_count_order(mds + 1);
1078 struct ceph_mds_session **sa;
1079 size_t ptr_size = sizeof(struct ceph_mds_session *);
1080
1081 doutc(cl, "realloc to %d\n", newmax);
1082 sa = kcalloc(newmax, ptr_size, GFP_NOFS);
1083 if (!sa)
1084 goto fail_realloc;
1085 if (mdsc->sessions) {
1086 memcpy(sa, mdsc->sessions,
1087 mdsc->max_sessions * ptr_size);
1088 kfree(mdsc->sessions);
1089 }
1090 mdsc->sessions = sa;
1091 mdsc->max_sessions = newmax;
1092 }
1093
1094 doutc(cl, "mds%d\n", mds);
1095 s->s_mdsc = mdsc;
1096 s->s_mds = mds;
1097 s->s_state = CEPH_MDS_SESSION_NEW;
1098 mutex_init(&s->s_mutex);
1099
1100 ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
1101
1102 atomic_set(&s->s_cap_gen, 1);
1103 s->s_cap_ttl = jiffies - 1;
1104
1105 spin_lock_init(&s->s_cap_lock);
1106 INIT_LIST_HEAD(&s->s_caps);
1107 refcount_set(&s->s_ref, 1);
1108 INIT_LIST_HEAD(&s->s_waiting);
1109 INIT_LIST_HEAD(&s->s_unsafe);
1110 xa_init(&s->s_delegated_inos);
1111 atomic_set(&s->s_num_deleg_inos, 0);
1112 INIT_LIST_HEAD(&s->s_cap_releases);
1113 INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);
1114
1115 INIT_LIST_HEAD(&s->s_cap_dirty);
1116 INIT_LIST_HEAD(&s->s_cap_flushing);
1117
1118 mdsc->sessions[mds] = s;
1119 atomic_inc(&mdsc->num_sessions);
1120 refcount_inc(&s->s_ref); /* one ref to sessions[], one to caller */
1121
1122 ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
1123 ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
1124
1125 return s;
1126
1127fail_realloc:
1128 kfree(s);
1129 return ERR_PTR(-ENOMEM);
1130}
1131
1132/*
1133 * called under mdsc->mutex
1134 */
1135static void __unregister_session(struct ceph_mds_client *mdsc,
1136 struct ceph_mds_session *s)
1137{
1138 doutc(mdsc->fsc->client, "mds%d %p\n", s->s_mds, s);
1139 BUG_ON(mdsc->sessions[s->s_mds] != s);
1140 mdsc->sessions[s->s_mds] = NULL;
1141 ceph_con_close(&s->s_con);
1142 ceph_put_mds_session(s);
1143 atomic_dec(&mdsc->num_sessions);
1144}
1145
1146/*
1147 * drop session refs in request.
1148 *
1149 * should be last request ref, or hold mdsc->mutex
1150 */
1151static void put_request_session(struct ceph_mds_request *req)
1152{
1153 if (req->r_session) {
1154 ceph_put_mds_session(req->r_session);
1155 req->r_session = NULL;
1156 }
1157}
1158
1159void ceph_mdsc_iterate_sessions(struct ceph_mds_client *mdsc,
1160 void (*cb)(struct ceph_mds_session *),
1161 bool check_state)
1162{
1163 int mds;
1164
1165 mutex_lock(&mdsc->mutex);
1166 for (mds = 0; mds < mdsc->max_sessions; ++mds) {
1167 struct ceph_mds_session *s;
1168
1169 s = __ceph_lookup_mds_session(mdsc, mds);
1170 if (!s)
1171 continue;
1172
1173 if (check_state && !check_session_state(s)) {
1174 ceph_put_mds_session(s);
1175 continue;
1176 }
1177
1178 mutex_unlock(&mdsc->mutex);
1179 cb(s);
1180 ceph_put_mds_session(s);
1181 mutex_lock(&mdsc->mutex);
1182 }
1183 mutex_unlock(&mdsc->mutex);
1184}
1185
1186void ceph_mdsc_release_request(struct kref *kref)
1187{
1188 struct ceph_mds_request *req = container_of(kref,
1189 struct ceph_mds_request,
1190 r_kref);
1191 ceph_mdsc_release_dir_caps_async(req);
1192 destroy_reply_info(&req->r_reply_info);
1193 if (req->r_request)
1194 ceph_msg_put(req->r_request);
1195 if (req->r_reply)
1196 ceph_msg_put(req->r_reply);
1197 if (req->r_inode) {
1198 ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
1199 iput(req->r_inode);
1200 }
1201 if (req->r_parent) {
1202 ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
1203 iput(req->r_parent);
1204 }
1205 iput(req->r_target_inode);
1206 iput(req->r_new_inode);
1207 if (req->r_dentry)
1208 dput(req->r_dentry);
1209 if (req->r_old_dentry)
1210 dput(req->r_old_dentry);
1211 if (req->r_old_dentry_dir) {
1212 /*
1213 * track (and drop pins for) r_old_dentry_dir
1214 * separately, since r_old_dentry's d_parent may have
1215 * changed between the dir mutex being dropped and
1216 * this request being freed.
1217 */
1218 ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
1219 CEPH_CAP_PIN);
1220 iput(req->r_old_dentry_dir);
1221 }
1222 kfree(req->r_path1);
1223 kfree(req->r_path2);
1224 put_cred(req->r_cred);
1225 if (req->r_mnt_idmap)
1226 mnt_idmap_put(req->r_mnt_idmap);
1227 if (req->r_pagelist)
1228 ceph_pagelist_release(req->r_pagelist);
1229 kfree(req->r_fscrypt_auth);
1230 kfree(req->r_altname);
1231 put_request_session(req);
1232 ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
1233 WARN_ON_ONCE(!list_empty(&req->r_wait));
1234 kmem_cache_free(ceph_mds_request_cachep, req);
1235}
1236
1237DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)
1238
1239/*
1240 * lookup session, bump ref if found.
1241 *
1242 * called under mdsc->mutex.
1243 */
1244static struct ceph_mds_request *
1245lookup_get_request(struct ceph_mds_client *mdsc, u64 tid)
1246{
1247 struct ceph_mds_request *req;
1248
1249 req = lookup_request(&mdsc->request_tree, tid);
1250 if (req)
1251 ceph_mdsc_get_request(req);
1252
1253 return req;
1254}
1255
1256/*
1257 * Register an in-flight request, and assign a tid. Link to directory
1258 * are modifying (if any).
1259 *
1260 * Called under mdsc->mutex.
1261 */
1262static void __register_request(struct ceph_mds_client *mdsc,
1263 struct ceph_mds_request *req,
1264 struct inode *dir)
1265{
1266 struct ceph_client *cl = mdsc->fsc->client;
1267 int ret = 0;
1268
1269 req->r_tid = ++mdsc->last_tid;
1270 if (req->r_num_caps) {
1271 ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
1272 req->r_num_caps);
1273 if (ret < 0) {
1274 pr_err_client(cl, "%p failed to reserve caps: %d\n",
1275 req, ret);
1276 /* set req->r_err to fail early from __do_request */
1277 req->r_err = ret;
1278 return;
1279 }
1280 }
1281 doutc(cl, "%p tid %lld\n", req, req->r_tid);
1282 ceph_mdsc_get_request(req);
1283 insert_request(&mdsc->request_tree, req);
1284
1285 req->r_cred = get_current_cred();
1286 if (!req->r_mnt_idmap)
1287 req->r_mnt_idmap = &nop_mnt_idmap;
1288
1289 if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
1290 mdsc->oldest_tid = req->r_tid;
1291
1292 if (dir) {
1293 struct ceph_inode_info *ci = ceph_inode(dir);
1294
1295 ihold(dir);
1296 req->r_unsafe_dir = dir;
1297 spin_lock(&ci->i_unsafe_lock);
1298 list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
1299 spin_unlock(&ci->i_unsafe_lock);
1300 }
1301}
1302
1303static void __unregister_request(struct ceph_mds_client *mdsc,
1304 struct ceph_mds_request *req)
1305{
1306 doutc(mdsc->fsc->client, "%p tid %lld\n", req, req->r_tid);
1307
1308 /* Never leave an unregistered request on an unsafe list! */
1309 list_del_init(&req->r_unsafe_item);
1310
1311 if (req->r_tid == mdsc->oldest_tid) {
1312 struct rb_node *p = rb_next(&req->r_node);
1313 mdsc->oldest_tid = 0;
1314 while (p) {
1315 struct ceph_mds_request *next_req =
1316 rb_entry(p, struct ceph_mds_request, r_node);
1317 if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
1318 mdsc->oldest_tid = next_req->r_tid;
1319 break;
1320 }
1321 p = rb_next(p);
1322 }
1323 }
1324
1325 erase_request(&mdsc->request_tree, req);
1326
1327 if (req->r_unsafe_dir) {
1328 struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
1329 spin_lock(&ci->i_unsafe_lock);
1330 list_del_init(&req->r_unsafe_dir_item);
1331 spin_unlock(&ci->i_unsafe_lock);
1332 }
1333 if (req->r_target_inode &&
1334 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
1335 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
1336 spin_lock(&ci->i_unsafe_lock);
1337 list_del_init(&req->r_unsafe_target_item);
1338 spin_unlock(&ci->i_unsafe_lock);
1339 }
1340
1341 if (req->r_unsafe_dir) {
1342 iput(req->r_unsafe_dir);
1343 req->r_unsafe_dir = NULL;
1344 }
1345
1346 complete_all(&req->r_safe_completion);
1347
1348 ceph_mdsc_put_request(req);
1349}
1350
1351/*
1352 * Walk back up the dentry tree until we hit a dentry representing a
1353 * non-snapshot inode. We do this using the rcu_read_lock (which must be held
1354 * when calling this) to ensure that the objects won't disappear while we're
1355 * working with them. Once we hit a candidate dentry, we attempt to take a
1356 * reference to it, and return that as the result.
1357 */
1358static struct inode *get_nonsnap_parent(struct dentry *dentry)
1359{
1360 struct inode *inode = NULL;
1361
1362 while (dentry && !IS_ROOT(dentry)) {
1363 inode = d_inode_rcu(dentry);
1364 if (!inode || ceph_snap(inode) == CEPH_NOSNAP)
1365 break;
1366 dentry = dentry->d_parent;
1367 }
1368 if (inode)
1369 inode = igrab(inode);
1370 return inode;
1371}
1372
1373/*
1374 * Choose mds to send request to next. If there is a hint set in the
1375 * request (e.g., due to a prior forward hint from the mds), use that.
1376 * Otherwise, consult frag tree and/or caps to identify the
1377 * appropriate mds. If all else fails, choose randomly.
1378 *
1379 * Called under mdsc->mutex.
1380 */
1381static int __choose_mds(struct ceph_mds_client *mdsc,
1382 struct ceph_mds_request *req,
1383 bool *random)
1384{
1385 struct inode *inode;
1386 struct ceph_inode_info *ci;
1387 struct ceph_cap *cap;
1388 int mode = req->r_direct_mode;
1389 int mds = -1;
1390 u32 hash = req->r_direct_hash;
1391 bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
1392 struct ceph_client *cl = mdsc->fsc->client;
1393
1394 if (random)
1395 *random = false;
1396
1397 /*
1398 * is there a specific mds we should try? ignore hint if we have
1399 * no session and the mds is not up (active or recovering).
1400 */
1401 if (req->r_resend_mds >= 0 &&
1402 (__have_session(mdsc, req->r_resend_mds) ||
1403 ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
1404 doutc(cl, "using resend_mds mds%d\n", req->r_resend_mds);
1405 return req->r_resend_mds;
1406 }
1407
1408 if (mode == USE_RANDOM_MDS)
1409 goto random;
1410
1411 inode = NULL;
1412 if (req->r_inode) {
1413 if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) {
1414 inode = req->r_inode;
1415 ihold(inode);
1416 } else {
1417 /* req->r_dentry is non-null for LSSNAP request */
1418 rcu_read_lock();
1419 inode = get_nonsnap_parent(req->r_dentry);
1420 rcu_read_unlock();
1421 doutc(cl, "using snapdir's parent %p %llx.%llx\n",
1422 inode, ceph_vinop(inode));
1423 }
1424 } else if (req->r_dentry) {
1425 /* ignore race with rename; old or new d_parent is okay */
1426 struct dentry *parent;
1427 struct inode *dir;
1428
1429 rcu_read_lock();
1430 parent = READ_ONCE(req->r_dentry->d_parent);
1431 dir = req->r_parent ? : d_inode_rcu(parent);
1432
1433 if (!dir || dir->i_sb != mdsc->fsc->sb) {
1434 /* not this fs or parent went negative */
1435 inode = d_inode(req->r_dentry);
1436 if (inode)
1437 ihold(inode);
1438 } else if (ceph_snap(dir) != CEPH_NOSNAP) {
1439 /* direct snapped/virtual snapdir requests
1440 * based on parent dir inode */
1441 inode = get_nonsnap_parent(parent);
1442 doutc(cl, "using nonsnap parent %p %llx.%llx\n",
1443 inode, ceph_vinop(inode));
1444 } else {
1445 /* dentry target */
1446 inode = d_inode(req->r_dentry);
1447 if (!inode || mode == USE_AUTH_MDS) {
1448 /* dir + name */
1449 inode = igrab(dir);
1450 hash = ceph_dentry_hash(dir, req->r_dentry);
1451 is_hash = true;
1452 } else {
1453 ihold(inode);
1454 }
1455 }
1456 rcu_read_unlock();
1457 }
1458
1459 if (!inode)
1460 goto random;
1461
1462 doutc(cl, "%p %llx.%llx is_hash=%d (0x%x) mode %d\n", inode,
1463 ceph_vinop(inode), (int)is_hash, hash, mode);
1464 ci = ceph_inode(inode);
1465
1466 if (is_hash && S_ISDIR(inode->i_mode)) {
1467 struct ceph_inode_frag frag;
1468 int found;
1469
1470 ceph_choose_frag(ci, hash, &frag, &found);
1471 if (found) {
1472 if (mode == USE_ANY_MDS && frag.ndist > 0) {
1473 u8 r;
1474
1475 /* choose a random replica */
1476 get_random_bytes(&r, 1);
1477 r %= frag.ndist;
1478 mds = frag.dist[r];
1479 doutc(cl, "%p %llx.%llx frag %u mds%d (%d/%d)\n",
1480 inode, ceph_vinop(inode), frag.frag,
1481 mds, (int)r, frag.ndist);
1482 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
1483 CEPH_MDS_STATE_ACTIVE &&
1484 !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))
1485 goto out;
1486 }
1487
1488 /* since this file/dir wasn't known to be
1489 * replicated, then we want to look for the
1490 * authoritative mds. */
1491 if (frag.mds >= 0) {
1492 /* choose auth mds */
1493 mds = frag.mds;
1494 doutc(cl, "%p %llx.%llx frag %u mds%d (auth)\n",
1495 inode, ceph_vinop(inode), frag.frag, mds);
1496 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
1497 CEPH_MDS_STATE_ACTIVE) {
1498 if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,
1499 mds))
1500 goto out;
1501 }
1502 }
1503 mode = USE_AUTH_MDS;
1504 }
1505 }
1506
1507 spin_lock(&ci->i_ceph_lock);
1508 cap = NULL;
1509 if (mode == USE_AUTH_MDS)
1510 cap = ci->i_auth_cap;
1511 if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
1512 cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
1513 if (!cap) {
1514 spin_unlock(&ci->i_ceph_lock);
1515 iput(inode);
1516 goto random;
1517 }
1518 mds = cap->session->s_mds;
1519 doutc(cl, "%p %llx.%llx mds%d (%scap %p)\n", inode,
1520 ceph_vinop(inode), mds,
1521 cap == ci->i_auth_cap ? "auth " : "", cap);
1522 spin_unlock(&ci->i_ceph_lock);
1523out:
1524 iput(inode);
1525 return mds;
1526
1527random:
1528 if (random)
1529 *random = true;
1530
1531 mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
1532 doutc(cl, "chose random mds%d\n", mds);
1533 return mds;
1534}
1535
1536
1537/*
1538 * session messages
1539 */
1540struct ceph_msg *ceph_create_session_msg(u32 op, u64 seq)
1541{
1542 struct ceph_msg *msg;
1543 struct ceph_mds_session_head *h;
1544
1545 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
1546 false);
1547 if (!msg) {
1548 pr_err("ENOMEM creating session %s msg\n",
1549 ceph_session_op_name(op));
1550 return NULL;
1551 }
1552 h = msg->front.iov_base;
1553 h->op = cpu_to_le32(op);
1554 h->seq = cpu_to_le64(seq);
1555
1556 return msg;
1557}
1558
1559static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
1560#define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)
1561static int encode_supported_features(void **p, void *end)
1562{
1563 static const size_t count = ARRAY_SIZE(feature_bits);
1564
1565 if (count > 0) {
1566 size_t i;
1567 size_t size = FEATURE_BYTES(count);
1568 unsigned long bit;
1569
1570 if (WARN_ON_ONCE(*p + 4 + size > end))
1571 return -ERANGE;
1572
1573 ceph_encode_32(p, size);
1574 memset(*p, 0, size);
1575 for (i = 0; i < count; i++) {
1576 bit = feature_bits[i];
1577 ((unsigned char *)(*p))[bit / 8] |= BIT(bit % 8);
1578 }
1579 *p += size;
1580 } else {
1581 if (WARN_ON_ONCE(*p + 4 > end))
1582 return -ERANGE;
1583
1584 ceph_encode_32(p, 0);
1585 }
1586
1587 return 0;
1588}
1589
1590static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
1591#define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
1592static int encode_metric_spec(void **p, void *end)
1593{
1594 static const size_t count = ARRAY_SIZE(metric_bits);
1595
1596 /* header */
1597 if (WARN_ON_ONCE(*p + 2 > end))
1598 return -ERANGE;
1599
1600 ceph_encode_8(p, 1); /* version */
1601 ceph_encode_8(p, 1); /* compat */
1602
1603 if (count > 0) {
1604 size_t i;
1605 size_t size = METRIC_BYTES(count);
1606
1607 if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
1608 return -ERANGE;
1609
1610 /* metric spec info length */
1611 ceph_encode_32(p, 4 + size);
1612
1613 /* metric spec */
1614 ceph_encode_32(p, size);
1615 memset(*p, 0, size);
1616 for (i = 0; i < count; i++)
1617 ((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8);
1618 *p += size;
1619 } else {
1620 if (WARN_ON_ONCE(*p + 4 + 4 > end))
1621 return -ERANGE;
1622
1623 /* metric spec info length */
1624 ceph_encode_32(p, 4);
1625 /* metric spec */
1626 ceph_encode_32(p, 0);
1627 }
1628
1629 return 0;
1630}
1631
1632/*
1633 * session message, specialization for CEPH_SESSION_REQUEST_OPEN
1634 * to include additional client metadata fields.
1635 */
1636static struct ceph_msg *
1637create_session_full_msg(struct ceph_mds_client *mdsc, int op, u64 seq)
1638{
1639 struct ceph_msg *msg;
1640 struct ceph_mds_session_head *h;
1641 int i;
1642 int extra_bytes = 0;
1643 int metadata_key_count = 0;
1644 struct ceph_options *opt = mdsc->fsc->client->options;
1645 struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;
1646 struct ceph_client *cl = mdsc->fsc->client;
1647 size_t size, count;
1648 void *p, *end;
1649 int ret;
1650
1651 const char* metadata[][2] = {
1652 {"hostname", mdsc->nodename},
1653 {"kernel_version", init_utsname()->release},
1654 {"entity_id", opt->name ? : ""},
1655 {"root", fsopt->server_path ? : "/"},
1656 {NULL, NULL}
1657 };
1658
1659 /* Calculate serialized length of metadata */
1660 extra_bytes = 4; /* map length */
1661 for (i = 0; metadata[i][0]; ++i) {
1662 extra_bytes += 8 + strlen(metadata[i][0]) +
1663 strlen(metadata[i][1]);
1664 metadata_key_count++;
1665 }
1666
1667 /* supported feature */
1668 size = 0;
1669 count = ARRAY_SIZE(feature_bits);
1670 if (count > 0)
1671 size = FEATURE_BYTES(count);
1672 extra_bytes += 4 + size;
1673
1674 /* metric spec */
1675 size = 0;
1676 count = ARRAY_SIZE(metric_bits);
1677 if (count > 0)
1678 size = METRIC_BYTES(count);
1679 extra_bytes += 2 + 4 + 4 + size;
1680
1681 /* flags, mds auth caps and oldest_client_tid */
1682 extra_bytes += 4 + 4 + 8;
1683
1684 /* Allocate the message */
1685 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,
1686 GFP_NOFS, false);
1687 if (!msg) {
1688 pr_err_client(cl, "ENOMEM creating session open msg\n");
1689 return ERR_PTR(-ENOMEM);
1690 }
1691 p = msg->front.iov_base;
1692 end = p + msg->front.iov_len;
1693
1694 h = p;
1695 h->op = cpu_to_le32(op);
1696 h->seq = cpu_to_le64(seq);
1697
1698 /*
1699 * Serialize client metadata into waiting buffer space, using
1700 * the format that userspace expects for map<string, string>
1701 *
1702 * ClientSession messages with metadata are v7
1703 */
1704 msg->hdr.version = cpu_to_le16(7);
1705 msg->hdr.compat_version = cpu_to_le16(1);
1706
1707 /* The write pointer, following the session_head structure */
1708 p += sizeof(*h);
1709
1710 /* Number of entries in the map */
1711 ceph_encode_32(&p, metadata_key_count);
1712
1713 /* Two length-prefixed strings for each entry in the map */
1714 for (i = 0; metadata[i][0]; ++i) {
1715 size_t const key_len = strlen(metadata[i][0]);
1716 size_t const val_len = strlen(metadata[i][1]);
1717
1718 ceph_encode_32(&p, key_len);
1719 memcpy(p, metadata[i][0], key_len);
1720 p += key_len;
1721 ceph_encode_32(&p, val_len);
1722 memcpy(p, metadata[i][1], val_len);
1723 p += val_len;
1724 }
1725
1726 ret = encode_supported_features(&p, end);
1727 if (ret) {
1728 pr_err_client(cl, "encode_supported_features failed!\n");
1729 ceph_msg_put(msg);
1730 return ERR_PTR(ret);
1731 }
1732
1733 ret = encode_metric_spec(&p, end);
1734 if (ret) {
1735 pr_err_client(cl, "encode_metric_spec failed!\n");
1736 ceph_msg_put(msg);
1737 return ERR_PTR(ret);
1738 }
1739
1740 /* version == 5, flags */
1741 ceph_encode_32(&p, 0);
1742
1743 /* version == 6, mds auth caps */
1744 ceph_encode_32(&p, 0);
1745
1746 /* version == 7, oldest_client_tid */
1747 ceph_encode_64(&p, mdsc->oldest_tid);
1748
1749 msg->front.iov_len = p - msg->front.iov_base;
1750 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1751
1752 return msg;
1753}
1754
1755/*
1756 * send session open request.
1757 *
1758 * called under mdsc->mutex
1759 */
1760static int __open_session(struct ceph_mds_client *mdsc,
1761 struct ceph_mds_session *session)
1762{
1763 struct ceph_msg *msg;
1764 int mstate;
1765 int mds = session->s_mds;
1766
1767 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO)
1768 return -EIO;
1769
1770 /* wait for mds to go active? */
1771 mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
1772 doutc(mdsc->fsc->client, "open_session to mds%d (%s)\n", mds,
1773 ceph_mds_state_name(mstate));
1774 session->s_state = CEPH_MDS_SESSION_OPENING;
1775 session->s_renew_requested = jiffies;
1776
1777 /* send connect message */
1778 msg = create_session_full_msg(mdsc, CEPH_SESSION_REQUEST_OPEN,
1779 session->s_seq);
1780 if (IS_ERR(msg))
1781 return PTR_ERR(msg);
1782 ceph_con_send(&session->s_con, msg);
1783 return 0;
1784}
1785
1786/*
1787 * open sessions for any export targets for the given mds
1788 *
1789 * called under mdsc->mutex
1790 */
1791static struct ceph_mds_session *
1792__open_export_target_session(struct ceph_mds_client *mdsc, int target)
1793{
1794 struct ceph_mds_session *session;
1795 int ret;
1796
1797 session = __ceph_lookup_mds_session(mdsc, target);
1798 if (!session) {
1799 session = register_session(mdsc, target);
1800 if (IS_ERR(session))
1801 return session;
1802 }
1803 if (session->s_state == CEPH_MDS_SESSION_NEW ||
1804 session->s_state == CEPH_MDS_SESSION_CLOSING) {
1805 ret = __open_session(mdsc, session);
1806 if (ret)
1807 return ERR_PTR(ret);
1808 }
1809
1810 return session;
1811}
1812
1813struct ceph_mds_session *
1814ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
1815{
1816 struct ceph_mds_session *session;
1817 struct ceph_client *cl = mdsc->fsc->client;
1818
1819 doutc(cl, "to mds%d\n", target);
1820
1821 mutex_lock(&mdsc->mutex);
1822 session = __open_export_target_session(mdsc, target);
1823 mutex_unlock(&mdsc->mutex);
1824
1825 return session;
1826}
1827
1828static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
1829 struct ceph_mds_session *session)
1830{
1831 struct ceph_mds_info *mi;
1832 struct ceph_mds_session *ts;
1833 int i, mds = session->s_mds;
1834 struct ceph_client *cl = mdsc->fsc->client;
1835
1836 if (mds >= mdsc->mdsmap->possible_max_rank)
1837 return;
1838
1839 mi = &mdsc->mdsmap->m_info[mds];
1840 doutc(cl, "for mds%d (%d targets)\n", session->s_mds,
1841 mi->num_export_targets);
1842
1843 for (i = 0; i < mi->num_export_targets; i++) {
1844 ts = __open_export_target_session(mdsc, mi->export_targets[i]);
1845 ceph_put_mds_session(ts);
1846 }
1847}
1848
1849/*
1850 * session caps
1851 */
1852
1853static int detach_cap_releases(struct ceph_mds_session *session,
1854 struct list_head *target)
1855{
1856 struct ceph_client *cl = session->s_mdsc->fsc->client;
1857 const int num_cap_releases = session->s_num_cap_releases;
1858
1859 lockdep_assert_held(&session->s_cap_lock);
1860
1861 list_splice_init(&session->s_cap_releases, target);
1862 session->s_num_cap_releases = 0;
1863 doutc(cl, "mds%d\n", session->s_mds);
1864
1865 return num_cap_releases;
1866}
1867
1868static void dispose_cap_releases(struct ceph_mds_client *mdsc,
1869 struct list_head *dispose)
1870{
1871 while (!list_empty(dispose)) {
1872 struct ceph_cap *cap;
1873 /* zero out the in-progress message */
1874 cap = list_first_entry(dispose, struct ceph_cap, session_caps);
1875 list_del(&cap->session_caps);
1876 ceph_put_cap(mdsc, cap);
1877 }
1878}
1879
1880static void cleanup_session_requests(struct ceph_mds_client *mdsc,
1881 struct ceph_mds_session *session)
1882{
1883 struct ceph_client *cl = mdsc->fsc->client;
1884 struct ceph_mds_request *req;
1885 struct rb_node *p;
1886
1887 doutc(cl, "mds%d\n", session->s_mds);
1888 mutex_lock(&mdsc->mutex);
1889 while (!list_empty(&session->s_unsafe)) {
1890 req = list_first_entry(&session->s_unsafe,
1891 struct ceph_mds_request, r_unsafe_item);
1892 pr_warn_ratelimited_client(cl, " dropping unsafe request %llu\n",
1893 req->r_tid);
1894 if (req->r_target_inode)
1895 mapping_set_error(req->r_target_inode->i_mapping, -EIO);
1896 if (req->r_unsafe_dir)
1897 mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);
1898 __unregister_request(mdsc, req);
1899 }
1900 /* zero r_attempts, so kick_requests() will re-send requests */
1901 p = rb_first(&mdsc->request_tree);
1902 while (p) {
1903 req = rb_entry(p, struct ceph_mds_request, r_node);
1904 p = rb_next(p);
1905 if (req->r_session &&
1906 req->r_session->s_mds == session->s_mds)
1907 req->r_attempts = 0;
1908 }
1909 mutex_unlock(&mdsc->mutex);
1910}
1911
1912/*
1913 * Helper to safely iterate over all caps associated with a session, with
1914 * special care taken to handle a racing __ceph_remove_cap().
1915 *
1916 * Caller must hold session s_mutex.
1917 */
1918int ceph_iterate_session_caps(struct ceph_mds_session *session,
1919 int (*cb)(struct inode *, int mds, void *),
1920 void *arg)
1921{
1922 struct ceph_client *cl = session->s_mdsc->fsc->client;
1923 struct list_head *p;
1924 struct ceph_cap *cap;
1925 struct inode *inode, *last_inode = NULL;
1926 struct ceph_cap *old_cap = NULL;
1927 int ret;
1928
1929 doutc(cl, "%p mds%d\n", session, session->s_mds);
1930 spin_lock(&session->s_cap_lock);
1931 p = session->s_caps.next;
1932 while (p != &session->s_caps) {
1933 int mds;
1934
1935 cap = list_entry(p, struct ceph_cap, session_caps);
1936 inode = igrab(&cap->ci->netfs.inode);
1937 if (!inode) {
1938 p = p->next;
1939 continue;
1940 }
1941 session->s_cap_iterator = cap;
1942 mds = cap->mds;
1943 spin_unlock(&session->s_cap_lock);
1944
1945 if (last_inode) {
1946 iput(last_inode);
1947 last_inode = NULL;
1948 }
1949 if (old_cap) {
1950 ceph_put_cap(session->s_mdsc, old_cap);
1951 old_cap = NULL;
1952 }
1953
1954 ret = cb(inode, mds, arg);
1955 last_inode = inode;
1956
1957 spin_lock(&session->s_cap_lock);
1958 p = p->next;
1959 if (ceph_cap_is_removed(cap)) {
1960 doutc(cl, "finishing cap %p removal\n", cap);
1961 BUG_ON(cap->session != session);
1962 cap->session = NULL;
1963 list_del_init(&cap->session_caps);
1964 session->s_nr_caps--;
1965 atomic64_dec(&session->s_mdsc->metric.total_caps);
1966 if (cap->queue_release)
1967 __ceph_queue_cap_release(session, cap);
1968 else
1969 old_cap = cap; /* put_cap it w/o locks held */
1970 }
1971 if (ret < 0)
1972 goto out;
1973 }
1974 ret = 0;
1975out:
1976 session->s_cap_iterator = NULL;
1977 spin_unlock(&session->s_cap_lock);
1978
1979 iput(last_inode);
1980 if (old_cap)
1981 ceph_put_cap(session->s_mdsc, old_cap);
1982
1983 return ret;
1984}
1985
1986static int remove_session_caps_cb(struct inode *inode, int mds, void *arg)
1987{
1988 struct ceph_inode_info *ci = ceph_inode(inode);
1989 struct ceph_client *cl = ceph_inode_to_client(inode);
1990 bool invalidate = false;
1991 struct ceph_cap *cap;
1992 int iputs = 0;
1993
1994 spin_lock(&ci->i_ceph_lock);
1995 cap = __get_cap_for_mds(ci, mds);
1996 if (cap) {
1997 doutc(cl, " removing cap %p, ci is %p, inode is %p\n",
1998 cap, ci, &ci->netfs.inode);
1999
2000 iputs = ceph_purge_inode_cap(inode, cap, &invalidate);
2001 }
2002 spin_unlock(&ci->i_ceph_lock);
2003
2004 if (cap)
2005 wake_up_all(&ci->i_cap_wq);
2006 if (invalidate)
2007 ceph_queue_invalidate(inode);
2008 while (iputs--)
2009 iput(inode);
2010 return 0;
2011}
2012
2013/*
2014 * caller must hold session s_mutex
2015 */
2016static void remove_session_caps(struct ceph_mds_session *session)
2017{
2018 struct ceph_fs_client *fsc = session->s_mdsc->fsc;
2019 struct super_block *sb = fsc->sb;
2020 LIST_HEAD(dispose);
2021
2022 doutc(fsc->client, "on %p\n", session);
2023 ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);
2024
2025 wake_up_all(&fsc->mdsc->cap_flushing_wq);
2026
2027 spin_lock(&session->s_cap_lock);
2028 if (session->s_nr_caps > 0) {
2029 struct inode *inode;
2030 struct ceph_cap *cap, *prev = NULL;
2031 struct ceph_vino vino;
2032 /*
2033 * iterate_session_caps() skips inodes that are being
2034 * deleted, we need to wait until deletions are complete.
2035 * __wait_on_freeing_inode() is designed for the job,
2036 * but it is not exported, so use lookup inode function
2037 * to access it.
2038 */
2039 while (!list_empty(&session->s_caps)) {
2040 cap = list_entry(session->s_caps.next,
2041 struct ceph_cap, session_caps);
2042 if (cap == prev)
2043 break;
2044 prev = cap;
2045 vino = cap->ci->i_vino;
2046 spin_unlock(&session->s_cap_lock);
2047
2048 inode = ceph_find_inode(sb, vino);
2049 iput(inode);
2050
2051 spin_lock(&session->s_cap_lock);
2052 }
2053 }
2054
2055 // drop cap expires and unlock s_cap_lock
2056 detach_cap_releases(session, &dispose);
2057
2058 BUG_ON(session->s_nr_caps > 0);
2059 BUG_ON(!list_empty(&session->s_cap_flushing));
2060 spin_unlock(&session->s_cap_lock);
2061 dispose_cap_releases(session->s_mdsc, &dispose);
2062}
2063
2064enum {
2065 RECONNECT,
2066 RENEWCAPS,
2067 FORCE_RO,
2068};
2069
2070/*
2071 * wake up any threads waiting on this session's caps. if the cap is
2072 * old (didn't get renewed on the client reconnect), remove it now.
2073 *
2074 * caller must hold s_mutex.
2075 */
2076static int wake_up_session_cb(struct inode *inode, int mds, void *arg)
2077{
2078 struct ceph_inode_info *ci = ceph_inode(inode);
2079 unsigned long ev = (unsigned long)arg;
2080
2081 if (ev == RECONNECT) {
2082 spin_lock(&ci->i_ceph_lock);
2083 ci->i_wanted_max_size = 0;
2084 ci->i_requested_max_size = 0;
2085 spin_unlock(&ci->i_ceph_lock);
2086 } else if (ev == RENEWCAPS) {
2087 struct ceph_cap *cap;
2088
2089 spin_lock(&ci->i_ceph_lock);
2090 cap = __get_cap_for_mds(ci, mds);
2091 /* mds did not re-issue stale cap */
2092 if (cap && cap->cap_gen < atomic_read(&cap->session->s_cap_gen))
2093 cap->issued = cap->implemented = CEPH_CAP_PIN;
2094 spin_unlock(&ci->i_ceph_lock);
2095 } else if (ev == FORCE_RO) {
2096 }
2097 wake_up_all(&ci->i_cap_wq);
2098 return 0;
2099}
2100
2101static void wake_up_session_caps(struct ceph_mds_session *session, int ev)
2102{
2103 struct ceph_client *cl = session->s_mdsc->fsc->client;
2104
2105 doutc(cl, "session %p mds%d\n", session, session->s_mds);
2106 ceph_iterate_session_caps(session, wake_up_session_cb,
2107 (void *)(unsigned long)ev);
2108}
2109
2110/*
2111 * Send periodic message to MDS renewing all currently held caps. The
2112 * ack will reset the expiration for all caps from this session.
2113 *
2114 * caller holds s_mutex
2115 */
2116static int send_renew_caps(struct ceph_mds_client *mdsc,
2117 struct ceph_mds_session *session)
2118{
2119 struct ceph_client *cl = mdsc->fsc->client;
2120 struct ceph_msg *msg;
2121 int state;
2122
2123 if (time_after_eq(jiffies, session->s_cap_ttl) &&
2124 time_after_eq(session->s_cap_ttl, session->s_renew_requested))
2125 pr_info_client(cl, "mds%d caps stale\n", session->s_mds);
2126 session->s_renew_requested = jiffies;
2127
2128 /* do not try to renew caps until a recovering mds has reconnected
2129 * with its clients. */
2130 state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
2131 if (state < CEPH_MDS_STATE_RECONNECT) {
2132 doutc(cl, "ignoring mds%d (%s)\n", session->s_mds,
2133 ceph_mds_state_name(state));
2134 return 0;
2135 }
2136
2137 doutc(cl, "to mds%d (%s)\n", session->s_mds,
2138 ceph_mds_state_name(state));
2139 msg = create_session_full_msg(mdsc, CEPH_SESSION_REQUEST_RENEWCAPS,
2140 ++session->s_renew_seq);
2141 if (IS_ERR(msg))
2142 return PTR_ERR(msg);
2143 ceph_con_send(&session->s_con, msg);
2144 return 0;
2145}
2146
2147static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
2148 struct ceph_mds_session *session, u64 seq)
2149{
2150 struct ceph_client *cl = mdsc->fsc->client;
2151 struct ceph_msg *msg;
2152
2153 doutc(cl, "to mds%d (%s)s seq %lld\n", session->s_mds,
2154 ceph_session_state_name(session->s_state), seq);
2155 msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
2156 if (!msg)
2157 return -ENOMEM;
2158 ceph_con_send(&session->s_con, msg);
2159 return 0;
2160}
2161
2162
2163/*
2164 * Note new cap ttl, and any transition from stale -> not stale (fresh?).
2165 *
2166 * Called under session->s_mutex
2167 */
2168static void renewed_caps(struct ceph_mds_client *mdsc,
2169 struct ceph_mds_session *session, int is_renew)
2170{
2171 struct ceph_client *cl = mdsc->fsc->client;
2172 int was_stale;
2173 int wake = 0;
2174
2175 spin_lock(&session->s_cap_lock);
2176 was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
2177
2178 session->s_cap_ttl = session->s_renew_requested +
2179 mdsc->mdsmap->m_session_timeout*HZ;
2180
2181 if (was_stale) {
2182 if (time_before(jiffies, session->s_cap_ttl)) {
2183 pr_info_client(cl, "mds%d caps renewed\n",
2184 session->s_mds);
2185 wake = 1;
2186 } else {
2187 pr_info_client(cl, "mds%d caps still stale\n",
2188 session->s_mds);
2189 }
2190 }
2191 doutc(cl, "mds%d ttl now %lu, was %s, now %s\n", session->s_mds,
2192 session->s_cap_ttl, was_stale ? "stale" : "fresh",
2193 time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
2194 spin_unlock(&session->s_cap_lock);
2195
2196 if (wake)
2197 wake_up_session_caps(session, RENEWCAPS);
2198}
2199
2200/*
2201 * send a session close request
2202 */
2203static int request_close_session(struct ceph_mds_session *session)
2204{
2205 struct ceph_client *cl = session->s_mdsc->fsc->client;
2206 struct ceph_msg *msg;
2207
2208 doutc(cl, "mds%d state %s seq %lld\n", session->s_mds,
2209 ceph_session_state_name(session->s_state), session->s_seq);
2210 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE,
2211 session->s_seq);
2212 if (!msg)
2213 return -ENOMEM;
2214 ceph_con_send(&session->s_con, msg);
2215 return 1;
2216}
2217
2218/*
2219 * Called with s_mutex held.
2220 */
2221static int __close_session(struct ceph_mds_client *mdsc,
2222 struct ceph_mds_session *session)
2223{
2224 if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
2225 return 0;
2226 session->s_state = CEPH_MDS_SESSION_CLOSING;
2227 return request_close_session(session);
2228}
2229
2230static bool drop_negative_children(struct dentry *dentry)
2231{
2232 struct dentry *child;
2233 bool all_negative = true;
2234
2235 if (!d_is_dir(dentry))
2236 goto out;
2237
2238 spin_lock(&dentry->d_lock);
2239 hlist_for_each_entry(child, &dentry->d_children, d_sib) {
2240 if (d_really_is_positive(child)) {
2241 all_negative = false;
2242 break;
2243 }
2244 }
2245 spin_unlock(&dentry->d_lock);
2246
2247 if (all_negative)
2248 shrink_dcache_parent(dentry);
2249out:
2250 return all_negative;
2251}
2252
2253/*
2254 * Trim old(er) caps.
2255 *
2256 * Because we can't cache an inode without one or more caps, we do
2257 * this indirectly: if a cap is unused, we prune its aliases, at which
2258 * point the inode will hopefully get dropped to.
2259 *
2260 * Yes, this is a bit sloppy. Our only real goal here is to respond to
2261 * memory pressure from the MDS, though, so it needn't be perfect.
2262 */
2263static int trim_caps_cb(struct inode *inode, int mds, void *arg)
2264{
2265 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
2266 struct ceph_client *cl = mdsc->fsc->client;
2267 int *remaining = arg;
2268 struct ceph_inode_info *ci = ceph_inode(inode);
2269 int used, wanted, oissued, mine;
2270 struct ceph_cap *cap;
2271
2272 if (*remaining <= 0)
2273 return -1;
2274
2275 spin_lock(&ci->i_ceph_lock);
2276 cap = __get_cap_for_mds(ci, mds);
2277 if (!cap) {
2278 spin_unlock(&ci->i_ceph_lock);
2279 return 0;
2280 }
2281 mine = cap->issued | cap->implemented;
2282 used = __ceph_caps_used(ci);
2283 wanted = __ceph_caps_file_wanted(ci);
2284 oissued = __ceph_caps_issued_other(ci, cap);
2285
2286 doutc(cl, "%p %llx.%llx cap %p mine %s oissued %s used %s wanted %s\n",
2287 inode, ceph_vinop(inode), cap, ceph_cap_string(mine),
2288 ceph_cap_string(oissued), ceph_cap_string(used),
2289 ceph_cap_string(wanted));
2290 if (cap == ci->i_auth_cap) {
2291 if (ci->i_dirty_caps || ci->i_flushing_caps ||
2292 !list_empty(&ci->i_cap_snaps))
2293 goto out;
2294 if ((used | wanted) & CEPH_CAP_ANY_WR)
2295 goto out;
2296 /* Note: it's possible that i_filelock_ref becomes non-zero
2297 * after dropping auth caps. It doesn't hurt because reply
2298 * of lock mds request will re-add auth caps. */
2299 if (atomic_read(&ci->i_filelock_ref) > 0)
2300 goto out;
2301 }
2302 /* The inode has cached pages, but it's no longer used.
2303 * we can safely drop it */
2304 if (S_ISREG(inode->i_mode) &&
2305 wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
2306 !(oissued & CEPH_CAP_FILE_CACHE)) {
2307 used = 0;
2308 oissued = 0;
2309 }
2310 if ((used | wanted) & ~oissued & mine)
2311 goto out; /* we need these caps */
2312
2313 if (oissued) {
2314 /* we aren't the only cap.. just remove us */
2315 ceph_remove_cap(mdsc, cap, ci, true);
2316 (*remaining)--;
2317 } else {
2318 struct dentry *dentry;
2319 /* try dropping referring dentries */
2320 spin_unlock(&ci->i_ceph_lock);
2321 dentry = d_find_any_alias(inode);
2322 if (dentry && drop_negative_children(dentry)) {
2323 int count;
2324 dput(dentry);
2325 d_prune_aliases(inode);
2326 count = icount_read_once(inode);
2327 if (count == 1)
2328 (*remaining)--;
2329 doutc(cl, "%p %llx.%llx cap %p pruned, count now %d\n",
2330 inode, ceph_vinop(inode), cap, count);
2331 } else {
2332 dput(dentry);
2333 }
2334 return 0;
2335 }
2336
2337out:
2338 spin_unlock(&ci->i_ceph_lock);
2339 return 0;
2340}
2341
2342/*
2343 * Trim session cap count down to some max number.
2344 */
2345int ceph_trim_caps(struct ceph_mds_client *mdsc,
2346 struct ceph_mds_session *session,
2347 int max_caps)
2348{
2349 struct ceph_client *cl = mdsc->fsc->client;
2350 int trim_caps = session->s_nr_caps - max_caps;
2351
2352 doutc(cl, "mds%d start: %d / %d, trim %d\n", session->s_mds,
2353 session->s_nr_caps, max_caps, trim_caps);
2354 if (trim_caps > 0) {
2355 int remaining = trim_caps;
2356
2357 ceph_iterate_session_caps(session, trim_caps_cb, &remaining);
2358 doutc(cl, "mds%d done: %d / %d, trimmed %d\n",
2359 session->s_mds, session->s_nr_caps, max_caps,
2360 trim_caps - remaining);
2361 }
2362
2363 ceph_flush_session_cap_releases(mdsc, session);
2364 return 0;
2365}
2366
2367static int check_caps_flush(struct ceph_mds_client *mdsc,
2368 u64 want_flush_tid)
2369{
2370 struct ceph_client *cl = mdsc->fsc->client;
2371 int ret = 1;
2372
2373 spin_lock(&mdsc->cap_dirty_lock);
2374 if (!list_empty(&mdsc->cap_flush_list)) {
2375 struct ceph_cap_flush *cf =
2376 list_first_entry(&mdsc->cap_flush_list,
2377 struct ceph_cap_flush, g_list);
2378 if (cf->tid <= want_flush_tid) {
2379 doutc(cl, "still flushing tid %llu <= %llu\n",
2380 cf->tid, want_flush_tid);
2381 ret = 0;
2382 }
2383 }
2384 spin_unlock(&mdsc->cap_dirty_lock);
2385 return ret;
2386}
2387
2388/*
2389 * Snapshot of a single cap_flush entry for diagnostic dump.
2390 * Collected under cap_dirty_lock, printed after releasing it.
2391 */
2392struct flush_dump_entry {
2393 u64 ino; /* inode number */
2394 u64 snap; /* snap id */
2395 int caps; /* dirty cap bits */
2396 u64 tid; /* flush transaction id */
2397 u64 last_ack; /* most recent ack tid for this inode */
2398 bool wake; /* whether completion was requested */
2399 bool is_capsnap; /* true if this is a cap snap flush */
2400 bool ci_null; /* true if cf->ci was unexpectedly NULL */
2401};
2402
2403/*
2404 * Dump pending cap flushes for diagnostic purposes.
2405 *
2406 * cf->ci is safe to dereference here: cap_flush entries hold a
2407 * reference on the inode (via the cap), and entries are removed from
2408 * cap_flush_list under cap_dirty_lock before the cap (and thus the
2409 * inode reference) is released. Holding cap_dirty_lock therefore
2410 * guarantees the inode remains valid for the lifetime of the scan.
2411 */
2412
2413static void dump_cap_flushes(struct ceph_mds_client *mdsc, u64 want_tid)
2414{
2415 struct ceph_client *cl = mdsc->fsc->client;
2416 struct flush_dump_entry entries[CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES];
2417 struct ceph_cap_flush *cf;
2418 int n = 0, remaining = 0;
2419 int i;
2420
2421 spin_lock(&mdsc->cap_dirty_lock);
2422 list_for_each_entry(cf, &mdsc->cap_flush_list, g_list) {
2423 if (cf->tid > want_tid)
2424 break;
2425 if (n < CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES) {
2426 struct flush_dump_entry *e = &entries[n++];
2427
2428 e->ci_null = WARN_ON_ONCE(!cf->ci);
2429 if (!e->ci_null) {
2430 e->ino = ceph_ino(&cf->ci->netfs.inode);
2431 e->snap = ceph_snap(&cf->ci->netfs.inode);
2432 e->last_ack = READ_ONCE(cf->ci->i_last_cap_flush_ack);
2433 }
2434 e->caps = cf->caps;
2435 e->tid = cf->tid;
2436 e->wake = cf->wake;
2437 e->is_capsnap = cf->is_capsnap;
2438 } else {
2439 remaining++;
2440 }
2441 }
2442 spin_unlock(&mdsc->cap_dirty_lock);
2443
2444 pr_info_client(cl, "still waiting for cap flushes through %llu:\n",
2445 want_tid);
2446 for (i = 0; i < n; i++) {
2447 struct flush_dump_entry *e = &entries[i];
2448
2449 if (e->ci_null)
2450 pr_info_client(cl,
2451 " (null ci) %s tid=%llu wake=%d%s\n",
2452 ceph_cap_string(e->caps), e->tid,
2453 e->wake,
2454 e->is_capsnap ? " is_capsnap" : "");
2455 else
2456 pr_info_client(cl,
2457 " %llx.%llx %s tid=%llu last_ack=%llu wake=%d%s\n",
2458 e->ino, e->snap,
2459 ceph_cap_string(e->caps), e->tid,
2460 e->last_ack, e->wake,
2461 e->is_capsnap ? " is_capsnap" : "");
2462 }
2463 if (remaining)
2464 pr_info_client(cl, " ... and %d more pending flushes\n",
2465 remaining);
2466}
2467
2468/*
2469 * Wait for all cap flushes through @want_flush_tid to complete.
2470 * Periodically dumps pending cap flush state for diagnostics.
2471 */
2472static void wait_caps_flush(struct ceph_mds_client *mdsc,
2473 u64 want_flush_tid)
2474{
2475 struct ceph_client *cl = mdsc->fsc->client;
2476 int i = 0;
2477 long ret;
2478
2479 doutc(cl, "want %llu\n", want_flush_tid);
2480
2481 do {
2482 /* 60 * HZ fits in a long on all supported architectures. */
2483 ret = wait_event_timeout(mdsc->cap_flushing_wq,
2484 check_caps_flush(mdsc, want_flush_tid),
2485 CEPH_CAP_FLUSH_WAIT_TIMEOUT_SEC * HZ);
2486 if (ret == 0) {
2487 if (i < CEPH_CAP_FLUSH_MAX_DUMP_ITERS)
2488 dump_cap_flushes(mdsc, want_flush_tid);
2489 else if (i == CEPH_CAP_FLUSH_MAX_DUMP_ITERS)
2490 pr_info_client(cl,
2491 "still waiting for cap flushes; suppressing further dumps\n");
2492 i++;
2493 }
2494 } while (ret == 0);
2495
2496 doutc(cl, "ok, flushed thru %llu\n", want_flush_tid);
2497}
2498
2499/*
2500 * called under s_mutex
2501 */
2502static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
2503 struct ceph_mds_session *session)
2504{
2505 struct ceph_client *cl = mdsc->fsc->client;
2506 struct ceph_msg *msg = NULL;
2507 struct ceph_mds_cap_release *head;
2508 struct ceph_mds_cap_item *item;
2509 struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
2510 struct ceph_cap *cap;
2511 LIST_HEAD(tmp_list);
2512 int num_cap_releases;
2513 __le32 barrier, *cap_barrier;
2514
2515 down_read(&osdc->lock);
2516 barrier = cpu_to_le32(osdc->epoch_barrier);
2517 up_read(&osdc->lock);
2518
2519 spin_lock(&session->s_cap_lock);
2520again:
2521 num_cap_releases = detach_cap_releases(session, &tmp_list);
2522 spin_unlock(&session->s_cap_lock);
2523
2524 while (!list_empty(&tmp_list)) {
2525 if (!msg) {
2526 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
2527 PAGE_SIZE, GFP_NOFS, false);
2528 if (!msg)
2529 goto out_err;
2530 head = msg->front.iov_base;
2531 head->num = cpu_to_le32(0);
2532 msg->front.iov_len = sizeof(*head);
2533
2534 msg->hdr.version = cpu_to_le16(2);
2535 msg->hdr.compat_version = cpu_to_le16(1);
2536 }
2537
2538 cap = list_first_entry(&tmp_list, struct ceph_cap,
2539 session_caps);
2540 list_del(&cap->session_caps);
2541 num_cap_releases--;
2542
2543 head = msg->front.iov_base;
2544 put_unaligned_le32(get_unaligned_le32(&head->num) + 1,
2545 &head->num);
2546 item = msg->front.iov_base + msg->front.iov_len;
2547 item->ino = cpu_to_le64(cap->cap_ino);
2548 item->cap_id = cpu_to_le64(cap->cap_id);
2549 item->migrate_seq = cpu_to_le32(cap->mseq);
2550 item->issue_seq = cpu_to_le32(cap->issue_seq);
2551 msg->front.iov_len += sizeof(*item);
2552
2553 ceph_put_cap(mdsc, cap);
2554
2555 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
2556 // Append cap_barrier field
2557 cap_barrier = msg->front.iov_base + msg->front.iov_len;
2558 *cap_barrier = barrier;
2559 msg->front.iov_len += sizeof(*cap_barrier);
2560
2561 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2562 doutc(cl, "mds%d %p\n", session->s_mds, msg);
2563 ceph_con_send(&session->s_con, msg);
2564 msg = NULL;
2565 }
2566 }
2567
2568 BUG_ON(num_cap_releases != 0);
2569
2570 spin_lock(&session->s_cap_lock);
2571 if (!list_empty(&session->s_cap_releases))
2572 goto again;
2573 spin_unlock(&session->s_cap_lock);
2574
2575 if (msg) {
2576 // Append cap_barrier field
2577 cap_barrier = msg->front.iov_base + msg->front.iov_len;
2578 *cap_barrier = barrier;
2579 msg->front.iov_len += sizeof(*cap_barrier);
2580
2581 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2582 doutc(cl, "mds%d %p\n", session->s_mds, msg);
2583 ceph_con_send(&session->s_con, msg);
2584 }
2585 return;
2586out_err:
2587 pr_err_client(cl, "mds%d, failed to allocate message\n",
2588 session->s_mds);
2589 spin_lock(&session->s_cap_lock);
2590 list_splice(&tmp_list, &session->s_cap_releases);
2591 session->s_num_cap_releases += num_cap_releases;
2592 spin_unlock(&session->s_cap_lock);
2593}
2594
2595static void ceph_cap_release_work(struct work_struct *work)
2596{
2597 struct ceph_mds_session *session =
2598 container_of(work, struct ceph_mds_session, s_cap_release_work);
2599
2600 mutex_lock(&session->s_mutex);
2601 if (session->s_state == CEPH_MDS_SESSION_OPEN ||
2602 session->s_state == CEPH_MDS_SESSION_HUNG)
2603 ceph_send_cap_releases(session->s_mdsc, session);
2604 mutex_unlock(&session->s_mutex);
2605 ceph_put_mds_session(session);
2606}
2607
2608void ceph_flush_session_cap_releases(struct ceph_mds_client *mdsc,
2609 struct ceph_mds_session *session)
2610{
2611 struct ceph_client *cl = mdsc->fsc->client;
2612 if (mdsc->stopping)
2613 return;
2614
2615 ceph_get_mds_session(session);
2616 if (queue_work(mdsc->fsc->cap_wq,
2617 &session->s_cap_release_work)) {
2618 doutc(cl, "cap release work queued\n");
2619 } else {
2620 ceph_put_mds_session(session);
2621 doutc(cl, "failed to queue cap release work\n");
2622 }
2623}
2624
2625/*
2626 * caller holds session->s_cap_lock
2627 */
2628void __ceph_queue_cap_release(struct ceph_mds_session *session,
2629 struct ceph_cap *cap)
2630{
2631 list_add_tail(&cap->session_caps, &session->s_cap_releases);
2632 session->s_num_cap_releases++;
2633
2634 if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))
2635 ceph_flush_session_cap_releases(session->s_mdsc, session);
2636}
2637
2638static void ceph_cap_reclaim_work(struct work_struct *work)
2639{
2640 struct ceph_mds_client *mdsc =
2641 container_of(work, struct ceph_mds_client, cap_reclaim_work);
2642 int ret = ceph_trim_dentries(mdsc);
2643 if (ret == -EAGAIN)
2644 ceph_queue_cap_reclaim_work(mdsc);
2645}
2646
2647void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)
2648{
2649 struct ceph_client *cl = mdsc->fsc->client;
2650 if (mdsc->stopping)
2651 return;
2652
2653 if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {
2654 doutc(cl, "caps reclaim work queued\n");
2655 } else {
2656 doutc(cl, "failed to queue caps release work\n");
2657 }
2658}
2659
2660void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)
2661{
2662 int val;
2663 if (!nr)
2664 return;
2665 val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);
2666 if ((val % CEPH_CAPS_PER_RELEASE) < nr) {
2667 atomic_set(&mdsc->cap_reclaim_pending, 0);
2668 ceph_queue_cap_reclaim_work(mdsc);
2669 }
2670}
2671
2672void ceph_queue_cap_unlink_work(struct ceph_mds_client *mdsc)
2673{
2674 struct ceph_client *cl = mdsc->fsc->client;
2675 if (mdsc->stopping)
2676 return;
2677
2678 if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_unlink_work)) {
2679 doutc(cl, "caps unlink work queued\n");
2680 } else {
2681 doutc(cl, "failed to queue caps unlink work\n");
2682 }
2683}
2684
2685static void ceph_cap_unlink_work(struct work_struct *work)
2686{
2687 struct ceph_mds_client *mdsc =
2688 container_of(work, struct ceph_mds_client, cap_unlink_work);
2689 struct ceph_client *cl = mdsc->fsc->client;
2690
2691 doutc(cl, "begin\n");
2692 spin_lock(&mdsc->cap_delay_lock);
2693 while (!list_empty(&mdsc->cap_unlink_delay_list)) {
2694 struct ceph_inode_info *ci;
2695 struct inode *inode;
2696
2697 ci = list_first_entry(&mdsc->cap_unlink_delay_list,
2698 struct ceph_inode_info,
2699 i_cap_delay_list);
2700 list_del_init(&ci->i_cap_delay_list);
2701
2702 inode = igrab(&ci->netfs.inode);
2703 if (inode) {
2704 spin_unlock(&mdsc->cap_delay_lock);
2705 doutc(cl, "on %p %llx.%llx\n", inode,
2706 ceph_vinop(inode));
2707 ceph_check_caps(ci, CHECK_CAPS_FLUSH);
2708 iput(inode);
2709 spin_lock(&mdsc->cap_delay_lock);
2710 }
2711 }
2712 spin_unlock(&mdsc->cap_delay_lock);
2713 doutc(cl, "done\n");
2714}
2715
2716/*
2717 * requests
2718 */
2719
2720int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
2721 struct inode *dir)
2722{
2723 struct ceph_inode_info *ci = ceph_inode(dir);
2724 struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
2725 struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
2726 size_t size = sizeof(struct ceph_mds_reply_dir_entry);
2727 unsigned int num_entries;
2728 u64 bytes_count;
2729 int order;
2730
2731 spin_lock(&ci->i_ceph_lock);
2732 num_entries = ci->i_files + ci->i_subdirs;
2733 spin_unlock(&ci->i_ceph_lock);
2734 num_entries = max(num_entries, 1U);
2735 num_entries = min(num_entries, opt->max_readdir);
2736
2737 bytes_count = (u64)size * num_entries;
2738 if (unlikely(bytes_count > ULONG_MAX))
2739 bytes_count = ULONG_MAX;
2740
2741 order = get_order((unsigned long)bytes_count);
2742 while (order >= 0) {
2743 rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL |
2744 __GFP_NOWARN |
2745 __GFP_ZERO,
2746 order);
2747 if (rinfo->dir_entries)
2748 break;
2749 order--;
2750 }
2751 if (!rinfo->dir_entries || unlikely(order < 0))
2752 return -ENOMEM;
2753
2754 num_entries = (PAGE_SIZE << order) / size;
2755 num_entries = min(num_entries, opt->max_readdir);
2756
2757 rinfo->dir_buf_size = PAGE_SIZE << order;
2758 req->r_num_caps = num_entries + 1;
2759 req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
2760 req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
2761 return 0;
2762}
2763
2764/*
2765 * Create an mds request.
2766 */
2767struct ceph_mds_request *
2768ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
2769{
2770 struct ceph_mds_request *req;
2771
2772 req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);
2773 if (!req)
2774 return ERR_PTR(-ENOMEM);
2775
2776 mutex_init(&req->r_fill_mutex);
2777 req->r_mdsc = mdsc;
2778 req->r_started = jiffies;
2779 req->r_start_latency = ktime_get();
2780 req->r_resend_mds = -1;
2781 INIT_LIST_HEAD(&req->r_unsafe_dir_item);
2782 INIT_LIST_HEAD(&req->r_unsafe_target_item);
2783 req->r_fmode = -1;
2784 req->r_feature_needed = -1;
2785 kref_init(&req->r_kref);
2786 RB_CLEAR_NODE(&req->r_node);
2787 INIT_LIST_HEAD(&req->r_wait);
2788 init_completion(&req->r_completion);
2789 init_completion(&req->r_safe_completion);
2790 INIT_LIST_HEAD(&req->r_unsafe_item);
2791
2792 ktime_get_coarse_real_ts64(&req->r_stamp);
2793
2794 req->r_op = op;
2795 req->r_direct_mode = mode;
2796 return req;
2797}
2798
2799/*
2800 * return oldest (lowest) request, tid in request tree, 0 if none.
2801 *
2802 * called under mdsc->mutex.
2803 */
2804static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
2805{
2806 if (RB_EMPTY_ROOT(&mdsc->request_tree))
2807 return NULL;
2808 return rb_entry(rb_first(&mdsc->request_tree),
2809 struct ceph_mds_request, r_node);
2810}
2811
2812static inline u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
2813{
2814 return mdsc->oldest_tid;
2815}
2816
2817#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
2818static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen)
2819{
2820 struct inode *dir = req->r_parent;
2821 struct dentry *dentry = req->r_dentry;
2822 const struct qstr *name = req->r_dname;
2823 u8 *cryptbuf = NULL;
2824 u32 len = 0;
2825 int ret = 0;
2826
2827 /* only encode if we have parent and dentry */
2828 if (!dir || !dentry)
2829 goto success;
2830
2831 /* No-op unless this is encrypted */
2832 if (!IS_ENCRYPTED(dir))
2833 goto success;
2834
2835 ret = ceph_fscrypt_prepare_readdir(dir);
2836 if (ret < 0)
2837 return ERR_PTR(ret);
2838
2839 /* No key? Just ignore it. */
2840 if (!fscrypt_has_encryption_key(dir))
2841 goto success;
2842
2843 if (!name)
2844 name = &dentry->d_name;
2845
2846 if (!fscrypt_fname_encrypted_size(dir, name->len, NAME_MAX, &len)) {
2847 WARN_ON_ONCE(1);
2848 return ERR_PTR(-ENAMETOOLONG);
2849 }
2850
2851 /* No need to append altname if name is short enough */
2852 if (len <= CEPH_NOHASH_NAME_MAX) {
2853 len = 0;
2854 goto success;
2855 }
2856
2857 cryptbuf = kmalloc(len, GFP_KERNEL);
2858 if (!cryptbuf)
2859 return ERR_PTR(-ENOMEM);
2860
2861 ret = fscrypt_fname_encrypt(dir, name, cryptbuf, len);
2862 if (ret) {
2863 kfree(cryptbuf);
2864 return ERR_PTR(ret);
2865 }
2866success:
2867 *plen = len;
2868 return cryptbuf;
2869}
2870#else
2871static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen)
2872{
2873 *plen = 0;
2874 return NULL;
2875}
2876#endif
2877
2878/**
2879 * ceph_mdsc_build_path - build a path string to a given dentry
2880 * @mdsc: mds client
2881 * @dentry: dentry to which path should be built
2882 * @path_info: output path, length, base ino+snap, and freepath ownership flag
2883 * @for_wire: is this path going to be sent to the MDS?
2884 *
2885 * Build a string that represents the path to the dentry. This is mostly called
2886 * for two different purposes:
2887 *
2888 * 1) we need to build a path string to send to the MDS (for_wire == true)
2889 * 2) we need a path string for local presentation (e.g. debugfs)
2890 * (for_wire == false)
2891 *
2892 * The path is built in reverse, starting with the dentry. Walk back up toward
2893 * the root, building the path until the first non-snapped inode is reached
2894 * (for_wire) or the root inode is reached (!for_wire).
2895 *
2896 * Encode hidden .snap dirs as a double /, i.e.
2897 * foo/.snap/bar -> foo//bar
2898 */
2899char *ceph_mdsc_build_path(struct ceph_mds_client *mdsc, struct dentry *dentry,
2900 struct ceph_path_info *path_info, int for_wire)
2901{
2902 struct ceph_client *cl = mdsc->fsc->client;
2903 struct dentry *cur;
2904 struct inode *inode;
2905 char *path;
2906 int pos;
2907 unsigned seq;
2908 u64 base;
2909
2910 if (!dentry)
2911 return ERR_PTR(-EINVAL);
2912
2913 path = __getname();
2914 if (!path)
2915 return ERR_PTR(-ENOMEM);
2916retry:
2917 pos = PATH_MAX - 1;
2918 path[pos] = '\0';
2919
2920 seq = read_seqbegin(&rename_lock);
2921 cur = dget(dentry);
2922 for (;;) {
2923 struct dentry *parent;
2924
2925 spin_lock(&cur->d_lock);
2926 inode = d_inode(cur);
2927 if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
2928 doutc(cl, "path+%d: %p SNAPDIR\n", pos, cur);
2929 spin_unlock(&cur->d_lock);
2930 parent = dget_parent(cur);
2931 } else if (for_wire && inode && dentry != cur &&
2932 ceph_snap(inode) == CEPH_NOSNAP) {
2933 spin_unlock(&cur->d_lock);
2934 pos++; /* get rid of any prepended '/' */
2935 break;
2936 } else if (!for_wire || !IS_ENCRYPTED(d_inode(cur->d_parent))) {
2937 pos -= cur->d_name.len;
2938 if (pos < 0) {
2939 spin_unlock(&cur->d_lock);
2940 break;
2941 }
2942 memcpy(path + pos, cur->d_name.name, cur->d_name.len);
2943 spin_unlock(&cur->d_lock);
2944 parent = dget_parent(cur);
2945 } else {
2946 int len, ret;
2947 char buf[NAME_MAX];
2948
2949 /*
2950 * Proactively copy name into buf, in case we need to
2951 * present it as-is.
2952 */
2953 memcpy(buf, cur->d_name.name, cur->d_name.len);
2954 len = cur->d_name.len;
2955 spin_unlock(&cur->d_lock);
2956 parent = dget_parent(cur);
2957
2958 ret = ceph_fscrypt_prepare_readdir(d_inode(parent));
2959 if (ret < 0) {
2960 dput(parent);
2961 dput(cur);
2962 __putname(path);
2963 return ERR_PTR(ret);
2964 }
2965
2966 if (fscrypt_has_encryption_key(d_inode(parent))) {
2967 len = ceph_encode_encrypted_dname(d_inode(parent),
2968 buf, len);
2969 if (len < 0) {
2970 dput(parent);
2971 dput(cur);
2972 __putname(path);
2973 return ERR_PTR(len);
2974 }
2975 }
2976 pos -= len;
2977 if (pos < 0) {
2978 dput(parent);
2979 break;
2980 }
2981 memcpy(path + pos, buf, len);
2982 }
2983 dput(cur);
2984 cur = parent;
2985
2986 /* Are we at the root? */
2987 if (IS_ROOT(cur))
2988 break;
2989
2990 /* Are we out of buffer? */
2991 if (--pos < 0)
2992 break;
2993
2994 path[pos] = '/';
2995 }
2996 inode = d_inode(cur);
2997 base = inode ? ceph_ino(inode) : 0;
2998 dput(cur);
2999
3000 if (read_seqretry(&rename_lock, seq))
3001 goto retry;
3002
3003 if (pos < 0) {
3004 /*
3005 * The path is longer than PATH_MAX and this function
3006 * cannot ever succeed. Creating paths that long is
3007 * possible with Ceph, but Linux cannot use them.
3008 */
3009 __putname(path);
3010 return ERR_PTR(-ENAMETOOLONG);
3011 }
3012
3013 /* Initialize the output structure */
3014 memset(path_info, 0, sizeof(*path_info));
3015
3016 path_info->vino.ino = base;
3017 path_info->pathlen = PATH_MAX - 1 - pos;
3018 path_info->path = path + pos;
3019 path_info->freepath = true;
3020
3021 /* Set snap from dentry if available */
3022 if (d_inode(dentry))
3023 path_info->vino.snap = ceph_snap(d_inode(dentry));
3024 else
3025 path_info->vino.snap = CEPH_NOSNAP;
3026
3027 doutc(cl, "on %p %d built %llx '%.*s'\n", dentry, d_count(dentry),
3028 base, PATH_MAX - 1 - pos, path + pos);
3029 return path + pos;
3030}
3031
3032static int build_dentry_path(struct ceph_mds_client *mdsc, struct dentry *dentry,
3033 struct inode *dir, struct ceph_path_info *path_info,
3034 bool parent_locked)
3035{
3036 char *path;
3037
3038 rcu_read_lock();
3039 if (!dir)
3040 dir = d_inode_rcu(dentry->d_parent);
3041 if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP &&
3042 !IS_ENCRYPTED(dir)) {
3043 path_info->vino.ino = ceph_ino(dir);
3044 path_info->vino.snap = ceph_snap(dir);
3045 rcu_read_unlock();
3046 path_info->path = dentry->d_name.name;
3047 path_info->pathlen = dentry->d_name.len;
3048 path_info->freepath = false;
3049 return 0;
3050 }
3051 rcu_read_unlock();
3052 path = ceph_mdsc_build_path(mdsc, dentry, path_info, 1);
3053 if (IS_ERR(path))
3054 return PTR_ERR(path);
3055 /*
3056 * ceph_mdsc_build_path already fills path_info, including snap handling.
3057 */
3058 return 0;
3059}
3060
3061static int build_inode_path(struct inode *inode, struct ceph_path_info *path_info)
3062{
3063 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
3064 struct dentry *dentry;
3065 char *path;
3066
3067 if (ceph_snap(inode) == CEPH_NOSNAP) {
3068 path_info->vino.ino = ceph_ino(inode);
3069 path_info->vino.snap = ceph_snap(inode);
3070 path_info->pathlen = 0;
3071 path_info->freepath = false;
3072 return 0;
3073 }
3074 dentry = d_find_alias(inode);
3075 path = ceph_mdsc_build_path(mdsc, dentry, path_info, 1);
3076 dput(dentry);
3077 if (IS_ERR(path))
3078 return PTR_ERR(path);
3079 /*
3080 * ceph_mdsc_build_path already fills path_info, including snap from dentry.
3081 * Override with inode's snap since that's what this function is for.
3082 */
3083 path_info->vino.snap = ceph_snap(inode);
3084 return 0;
3085}
3086
3087/*
3088 * request arguments may be specified via an inode *, a dentry *, or
3089 * an explicit ino+path.
3090 */
3091static int set_request_path_attr(struct ceph_mds_client *mdsc, struct inode *rinode,
3092 struct dentry *rdentry, struct inode *rdiri,
3093 const char *rpath, u64 rino,
3094 struct ceph_path_info *path_info,
3095 bool parent_locked)
3096{
3097 struct ceph_client *cl = mdsc->fsc->client;
3098 int r = 0;
3099
3100 /* Initialize the output structure */
3101 memset(path_info, 0, sizeof(*path_info));
3102
3103 if (rinode) {
3104 r = build_inode_path(rinode, path_info);
3105 doutc(cl, " inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
3106 ceph_snap(rinode));
3107 } else if (rdentry) {
3108 r = build_dentry_path(mdsc, rdentry, rdiri, path_info, parent_locked);
3109 doutc(cl, " dentry %p %llx/%.*s\n", rdentry, path_info->vino.ino,
3110 path_info->pathlen, path_info->path);
3111 } else if (rpath || rino) {
3112 path_info->vino.ino = rino;
3113 path_info->vino.snap = CEPH_NOSNAP;
3114 path_info->path = rpath;
3115 path_info->pathlen = rpath ? strlen(rpath) : 0;
3116 path_info->freepath = false;
3117
3118 doutc(cl, " path %.*s\n", path_info->pathlen, rpath);
3119 }
3120
3121 return r;
3122}
3123
3124static void encode_mclientrequest_tail(void **p,
3125 const struct ceph_mds_request *req)
3126{
3127 struct ceph_timespec ts;
3128 int i;
3129
3130 ceph_encode_timespec64(&ts, &req->r_stamp);
3131 ceph_encode_copy(p, &ts, sizeof(ts));
3132
3133 /* v4: gid_list */
3134 ceph_encode_32(p, req->r_cred->group_info->ngroups);
3135 for (i = 0; i < req->r_cred->group_info->ngroups; i++)
3136 ceph_encode_64(p, from_kgid(&init_user_ns,
3137 req->r_cred->group_info->gid[i]));
3138
3139 /* v5: altname */
3140 ceph_encode_32(p, req->r_altname_len);
3141 ceph_encode_copy(p, req->r_altname, req->r_altname_len);
3142
3143 /* v6: fscrypt_auth and fscrypt_file */
3144 if (req->r_fscrypt_auth) {
3145 u32 authlen = ceph_fscrypt_auth_len(req->r_fscrypt_auth);
3146
3147 ceph_encode_32(p, authlen);
3148 ceph_encode_copy(p, req->r_fscrypt_auth, authlen);
3149 } else {
3150 ceph_encode_32(p, 0);
3151 }
3152 if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags)) {
3153 ceph_encode_32(p, sizeof(__le64));
3154 ceph_encode_64(p, req->r_fscrypt_file);
3155 } else {
3156 ceph_encode_32(p, 0);
3157 }
3158}
3159
3160static inline u16 mds_supported_head_version(struct ceph_mds_session *session)
3161{
3162 if (!test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD, &session->s_features))
3163 return 1;
3164
3165 if (!test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features))
3166 return 2;
3167
3168 return CEPH_MDS_REQUEST_HEAD_VERSION;
3169}
3170
3171static struct ceph_mds_request_head_legacy *
3172find_legacy_request_head(void *p, u64 features)
3173{
3174 bool legacy = !(features & CEPH_FEATURE_FS_BTIME);
3175 struct ceph_mds_request_head *head;
3176
3177 if (legacy)
3178 return (struct ceph_mds_request_head_legacy *)p;
3179 head = (struct ceph_mds_request_head *)p;
3180 return (struct ceph_mds_request_head_legacy *)&head->oldest_client_tid;
3181}
3182
3183/*
3184 * called under mdsc->mutex
3185 */
3186static struct ceph_msg *create_request_message(struct ceph_mds_session *session,
3187 struct ceph_mds_request *req,
3188 bool drop_cap_releases)
3189{
3190 int mds = session->s_mds;
3191 struct ceph_mds_client *mdsc = session->s_mdsc;
3192 struct ceph_client *cl = mdsc->fsc->client;
3193 struct ceph_msg *msg;
3194 struct ceph_mds_request_head_legacy *lhead;
3195 struct ceph_path_info path_info1 = {0};
3196 struct ceph_path_info path_info2 = {0};
3197 struct dentry *old_dentry = NULL;
3198 int len;
3199 u16 releases;
3200 void *p, *end;
3201 int ret;
3202 bool legacy = !(session->s_con.peer_features & CEPH_FEATURE_FS_BTIME);
3203 u16 request_head_version = mds_supported_head_version(session);
3204 kuid_t caller_fsuid = req->r_cred->fsuid;
3205 kgid_t caller_fsgid = req->r_cred->fsgid;
3206 bool parent_locked = test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
3207
3208 ret = set_request_path_attr(mdsc, req->r_inode, req->r_dentry,
3209 req->r_parent, req->r_path1, req->r_ino1.ino,
3210 &path_info1, parent_locked);
3211 if (ret < 0) {
3212 msg = ERR_PTR(ret);
3213 goto out;
3214 }
3215
3216 /*
3217 * When the parent directory's i_rwsem is *not* locked, req->r_parent may
3218 * have become stale (e.g. after a concurrent rename) between the time the
3219 * dentry was looked up and now. If we detect that the stored r_parent
3220 * does not match the inode number we just encoded for the request, switch
3221 * to the correct inode so that the MDS receives a valid parent reference.
3222 */
3223 if (!parent_locked && req->r_parent && path_info1.vino.ino &&
3224 ceph_ino(req->r_parent) != path_info1.vino.ino) {
3225 struct inode *old_parent = req->r_parent;
3226 struct inode *correct_dir = ceph_get_inode(mdsc->fsc->sb, path_info1.vino, NULL);
3227 if (!IS_ERR(correct_dir)) {
3228 WARN_ONCE(1, "ceph: r_parent mismatch (had %llx wanted %llx) - updating\n",
3229 ceph_ino(old_parent), path_info1.vino.ino);
3230 /*
3231 * Transfer CEPH_CAP_PIN from the old parent to the new one.
3232 * The pin was taken earlier in ceph_mdsc_submit_request().
3233 */
3234 ceph_put_cap_refs(ceph_inode(old_parent), CEPH_CAP_PIN);
3235 iput(old_parent);
3236 req->r_parent = correct_dir;
3237 ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
3238 }
3239 }
3240
3241 /* If r_old_dentry is set, then assume that its parent is locked */
3242 if (req->r_old_dentry &&
3243 !(req->r_old_dentry->d_flags & DCACHE_DISCONNECTED))
3244 old_dentry = req->r_old_dentry;
3245 ret = set_request_path_attr(mdsc, NULL, old_dentry,
3246 req->r_old_dentry_dir,
3247 req->r_path2, req->r_ino2.ino,
3248 &path_info2, true);
3249 if (ret < 0) {
3250 msg = ERR_PTR(ret);
3251 goto out_free1;
3252 }
3253
3254 req->r_altname = get_fscrypt_altname(req, &req->r_altname_len);
3255 if (IS_ERR(req->r_altname)) {
3256 msg = ERR_CAST(req->r_altname);
3257 req->r_altname = NULL;
3258 goto out_free2;
3259 }
3260
3261 /*
3262 * For old cephs without supporting the 32bit retry/fwd feature
3263 * it will copy the raw memories directly when decoding the
3264 * requests. While new cephs will decode the head depending the
3265 * version member, so we need to make sure it will be compatible
3266 * with them both.
3267 */
3268 if (legacy)
3269 len = sizeof(struct ceph_mds_request_head_legacy);
3270 else if (request_head_version == 1)
3271 len = offsetofend(struct ceph_mds_request_head, args);
3272 else if (request_head_version == 2)
3273 len = offsetofend(struct ceph_mds_request_head, ext_num_fwd);
3274 else
3275 len = sizeof(struct ceph_mds_request_head);
3276
3277 /* filepaths */
3278 len += 2 * (1 + sizeof(u32) + sizeof(u64));
3279 len += path_info1.pathlen + path_info2.pathlen;
3280
3281 /* cap releases */
3282 len += sizeof(struct ceph_mds_request_release) *
3283 (!!req->r_inode_drop + !!req->r_dentry_drop +
3284 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
3285
3286 if (req->r_dentry_drop)
3287 len += path_info1.pathlen;
3288 if (req->r_old_dentry_drop)
3289 len += path_info2.pathlen;
3290
3291 /* MClientRequest tail */
3292
3293 /* req->r_stamp */
3294 len += sizeof(struct ceph_timespec);
3295
3296 /* gid list */
3297 len += sizeof(u32) + (sizeof(u64) * req->r_cred->group_info->ngroups);
3298
3299 /* alternate name */
3300 len += sizeof(u32) + req->r_altname_len;
3301
3302 /* fscrypt_auth */
3303 len += sizeof(u32); // fscrypt_auth
3304 if (req->r_fscrypt_auth)
3305 len += ceph_fscrypt_auth_len(req->r_fscrypt_auth);
3306
3307 /* fscrypt_file */
3308 len += sizeof(u32);
3309 if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags))
3310 len += sizeof(__le64);
3311
3312 msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);
3313 if (!msg) {
3314 msg = ERR_PTR(-ENOMEM);
3315 goto out_free2;
3316 }
3317
3318 msg->hdr.tid = cpu_to_le64(req->r_tid);
3319
3320 lhead = find_legacy_request_head(msg->front.iov_base,
3321 session->s_con.peer_features);
3322
3323 if ((req->r_mnt_idmap != &nop_mnt_idmap) &&
3324 !test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features)) {
3325 WARN_ON_ONCE(!IS_CEPH_MDS_OP_NEWINODE(req->r_op));
3326
3327 if (enable_unsafe_idmap) {
3328 pr_warn_once_client(cl,
3329 "idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID"
3330 " is not supported by MDS. UID/GID-based restrictions may"
3331 " not work properly.\n");
3332
3333 caller_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns,
3334 VFSUIDT_INIT(req->r_cred->fsuid));
3335 caller_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns,
3336 VFSGIDT_INIT(req->r_cred->fsgid));
3337 } else {
3338 pr_err_ratelimited_client(cl,
3339 "idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID"
3340 " is not supported by MDS. Fail request with -EIO.\n");
3341
3342 ret = -EIO;
3343 goto out_err;
3344 }
3345 }
3346
3347 /*
3348 * The ceph_mds_request_head_legacy didn't contain a version field, and
3349 * one was added when we moved the message version from 3->4.
3350 */
3351 if (legacy) {
3352 msg->hdr.version = cpu_to_le16(3);
3353 p = msg->front.iov_base + sizeof(*lhead);
3354 } else if (request_head_version == 1) {
3355 struct ceph_mds_request_head *nhead = msg->front.iov_base;
3356
3357 msg->hdr.version = cpu_to_le16(4);
3358 nhead->version = cpu_to_le16(1);
3359 p = msg->front.iov_base + offsetofend(struct ceph_mds_request_head, args);
3360 } else if (request_head_version == 2) {
3361 struct ceph_mds_request_head *nhead = msg->front.iov_base;
3362
3363 msg->hdr.version = cpu_to_le16(6);
3364 nhead->version = cpu_to_le16(2);
3365
3366 p = msg->front.iov_base + offsetofend(struct ceph_mds_request_head, ext_num_fwd);
3367 } else {
3368 struct ceph_mds_request_head *nhead = msg->front.iov_base;
3369 kuid_t owner_fsuid;
3370 kgid_t owner_fsgid;
3371
3372 msg->hdr.version = cpu_to_le16(6);
3373 nhead->version = cpu_to_le16(CEPH_MDS_REQUEST_HEAD_VERSION);
3374 nhead->struct_len = cpu_to_le32(sizeof(struct ceph_mds_request_head));
3375
3376 if (IS_CEPH_MDS_OP_NEWINODE(req->r_op)) {
3377 owner_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns,
3378 VFSUIDT_INIT(req->r_cred->fsuid));
3379 owner_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns,
3380 VFSGIDT_INIT(req->r_cred->fsgid));
3381 nhead->owner_uid = cpu_to_le32(from_kuid(&init_user_ns, owner_fsuid));
3382 nhead->owner_gid = cpu_to_le32(from_kgid(&init_user_ns, owner_fsgid));
3383 } else {
3384 nhead->owner_uid = cpu_to_le32(-1);
3385 nhead->owner_gid = cpu_to_le32(-1);
3386 }
3387
3388 p = msg->front.iov_base + sizeof(*nhead);
3389 }
3390
3391 end = msg->front.iov_base + msg->front.iov_len;
3392
3393 lhead->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
3394 lhead->op = cpu_to_le32(req->r_op);
3395 lhead->caller_uid = cpu_to_le32(from_kuid(&init_user_ns,
3396 caller_fsuid));
3397 lhead->caller_gid = cpu_to_le32(from_kgid(&init_user_ns,
3398 caller_fsgid));
3399 lhead->ino = cpu_to_le64(req->r_deleg_ino);
3400 lhead->args = req->r_args;
3401
3402 ceph_encode_filepath(&p, end, path_info1.vino.ino, path_info1.path);
3403 ceph_encode_filepath(&p, end, path_info2.vino.ino, path_info2.path);
3404
3405 /* make note of release offset, in case we need to replay */
3406 req->r_request_release_offset = p - msg->front.iov_base;
3407
3408 /* cap releases */
3409 releases = 0;
3410 if (req->r_inode_drop)
3411 releases += ceph_encode_inode_release(&p,
3412 req->r_inode ? req->r_inode : d_inode(req->r_dentry),
3413 mds, req->r_inode_drop, req->r_inode_unless,
3414 req->r_op == CEPH_MDS_OP_READDIR);
3415 if (req->r_dentry_drop) {
3416 ret = ceph_encode_dentry_release(&p, req->r_dentry,
3417 req->r_parent, mds, req->r_dentry_drop,
3418 req->r_dentry_unless);
3419 if (ret < 0)
3420 goto out_err;
3421 releases += ret;
3422 }
3423 if (req->r_old_dentry_drop) {
3424 ret = ceph_encode_dentry_release(&p, req->r_old_dentry,
3425 req->r_old_dentry_dir, mds,
3426 req->r_old_dentry_drop,
3427 req->r_old_dentry_unless);
3428 if (ret < 0)
3429 goto out_err;
3430 releases += ret;
3431 }
3432 if (req->r_old_inode_drop)
3433 releases += ceph_encode_inode_release(&p,
3434 d_inode(req->r_old_dentry),
3435 mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
3436
3437 if (drop_cap_releases) {
3438 releases = 0;
3439 p = msg->front.iov_base + req->r_request_release_offset;
3440 }
3441
3442 lhead->num_releases = cpu_to_le16(releases);
3443
3444 encode_mclientrequest_tail(&p, req);
3445
3446 if (WARN_ON_ONCE(p > end)) {
3447 ceph_msg_put(msg);
3448 msg = ERR_PTR(-ERANGE);
3449 goto out_free2;
3450 }
3451
3452 msg->front.iov_len = p - msg->front.iov_base;
3453 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
3454
3455 if (req->r_pagelist) {
3456 struct ceph_pagelist *pagelist = req->r_pagelist;
3457 ceph_msg_data_add_pagelist(msg, pagelist);
3458 msg->hdr.data_len = cpu_to_le32(pagelist->length);
3459 } else {
3460 msg->hdr.data_len = 0;
3461 }
3462
3463 msg->hdr.data_off = cpu_to_le16(0);
3464
3465out_free2:
3466 ceph_mdsc_free_path_info(&path_info2);
3467out_free1:
3468 ceph_mdsc_free_path_info(&path_info1);
3469out:
3470 return msg;
3471out_err:
3472 ceph_msg_put(msg);
3473 msg = ERR_PTR(ret);
3474 goto out_free2;
3475}
3476
3477/*
3478 * called under mdsc->mutex if error, under no mutex if
3479 * success.
3480 */
3481static void complete_request(struct ceph_mds_client *mdsc,
3482 struct ceph_mds_request *req)
3483{
3484 req->r_end_latency = ktime_get();
3485
3486 trace_ceph_mdsc_complete_request(mdsc, req);
3487
3488 if (req->r_callback)
3489 req->r_callback(mdsc, req);
3490 complete_all(&req->r_completion);
3491}
3492
3493/*
3494 * called under mdsc->mutex
3495 */
3496static int __prepare_send_request(struct ceph_mds_session *session,
3497 struct ceph_mds_request *req,
3498 bool drop_cap_releases)
3499{
3500 int mds = session->s_mds;
3501 struct ceph_mds_client *mdsc = session->s_mdsc;
3502 struct ceph_client *cl = mdsc->fsc->client;
3503 struct ceph_mds_request_head_legacy *lhead;
3504 struct ceph_mds_request_head *nhead;
3505 struct ceph_msg *msg;
3506 int flags = 0, old_max_retry;
3507 bool old_version = !test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD,
3508 &session->s_features);
3509
3510 /*
3511 * Avoid infinite retrying after overflow. The client will
3512 * increase the retry count and if the MDS is old version,
3513 * so we limit to retry at most 256 times.
3514 */
3515 if (req->r_attempts) {
3516 old_max_retry = sizeof_field(struct ceph_mds_request_head,
3517 num_retry);
3518 old_max_retry = 1 << (old_max_retry * BITS_PER_BYTE);
3519 if ((old_version && req->r_attempts >= old_max_retry) ||
3520 ((uint32_t)req->r_attempts >= U32_MAX)) {
3521 pr_warn_ratelimited_client(cl, "request tid %llu seq overflow\n",
3522 req->r_tid);
3523 return -EMULTIHOP;
3524 }
3525 }
3526
3527 req->r_attempts++;
3528 if (req->r_inode) {
3529 struct ceph_cap *cap =
3530 ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
3531
3532 if (cap)
3533 req->r_sent_on_mseq = cap->mseq;
3534 else
3535 req->r_sent_on_mseq = -1;
3536 }
3537 doutc(cl, "%p tid %lld %s (attempt %d)\n", req, req->r_tid,
3538 ceph_mds_op_name(req->r_op), req->r_attempts);
3539
3540 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
3541 void *p;
3542
3543 /*
3544 * Replay. Do not regenerate message (and rebuild
3545 * paths, etc.); just use the original message.
3546 * Rebuilding paths will break for renames because
3547 * d_move mangles the src name.
3548 */
3549 msg = req->r_request;
3550 lhead = find_legacy_request_head(msg->front.iov_base,
3551 session->s_con.peer_features);
3552
3553 flags = le32_to_cpu(lhead->flags);
3554 flags |= CEPH_MDS_FLAG_REPLAY;
3555 lhead->flags = cpu_to_le32(flags);
3556
3557 if (req->r_target_inode)
3558 lhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
3559
3560 lhead->num_retry = req->r_attempts - 1;
3561 if (!old_version) {
3562 nhead = (struct ceph_mds_request_head*)msg->front.iov_base;
3563 nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1);
3564 }
3565
3566 /* remove cap/dentry releases from message */
3567 lhead->num_releases = 0;
3568
3569 p = msg->front.iov_base + req->r_request_release_offset;
3570 encode_mclientrequest_tail(&p, req);
3571
3572 msg->front.iov_len = p - msg->front.iov_base;
3573 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
3574 return 0;
3575 }
3576
3577 if (req->r_request) {
3578 ceph_msg_put(req->r_request);
3579 req->r_request = NULL;
3580 }
3581 msg = create_request_message(session, req, drop_cap_releases);
3582 if (IS_ERR(msg)) {
3583 req->r_err = PTR_ERR(msg);
3584 return PTR_ERR(msg);
3585 }
3586 req->r_request = msg;
3587
3588 lhead = find_legacy_request_head(msg->front.iov_base,
3589 session->s_con.peer_features);
3590 lhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
3591 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
3592 flags |= CEPH_MDS_FLAG_REPLAY;
3593 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))
3594 flags |= CEPH_MDS_FLAG_ASYNC;
3595 if (req->r_parent)
3596 flags |= CEPH_MDS_FLAG_WANT_DENTRY;
3597 lhead->flags = cpu_to_le32(flags);
3598 lhead->num_fwd = req->r_num_fwd;
3599 lhead->num_retry = req->r_attempts - 1;
3600 if (!old_version) {
3601 nhead = (struct ceph_mds_request_head*)msg->front.iov_base;
3602 nhead->ext_num_fwd = cpu_to_le32(req->r_num_fwd);
3603 nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1);
3604 }
3605
3606 doutc(cl, " r_parent = %p\n", req->r_parent);
3607 return 0;
3608}
3609
3610/*
3611 * called under mdsc->mutex
3612 */
3613static int __send_request(struct ceph_mds_session *session,
3614 struct ceph_mds_request *req,
3615 bool drop_cap_releases)
3616{
3617 int err;
3618
3619 trace_ceph_mdsc_send_request(session, req);
3620
3621 err = __prepare_send_request(session, req, drop_cap_releases);
3622 if (!err) {
3623 ceph_msg_get(req->r_request);
3624 ceph_con_send(&session->s_con, req->r_request);
3625 }
3626
3627 return err;
3628}
3629
3630/*
3631 * send request, or put it on the appropriate wait list.
3632 */
3633static void __do_request(struct ceph_mds_client *mdsc,
3634 struct ceph_mds_request *req)
3635{
3636 struct ceph_client *cl = mdsc->fsc->client;
3637 struct ceph_mds_session *session = NULL;
3638 int mds = -1;
3639 int err = 0;
3640 bool random;
3641
3642 if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
3643 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
3644 __unregister_request(mdsc, req);
3645 return;
3646 }
3647
3648 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) {
3649 doutc(cl, "metadata corrupted\n");
3650 err = -EIO;
3651 goto finish;
3652 }
3653 if (req->r_timeout &&
3654 time_after_eq(jiffies, req->r_started + req->r_timeout)) {
3655 doutc(cl, "timed out\n");
3656 err = -ETIMEDOUT;
3657 goto finish;
3658 }
3659 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
3660 doutc(cl, "forced umount\n");
3661 err = -EIO;
3662 goto finish;
3663 }
3664 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) {
3665 if (mdsc->mdsmap_err) {
3666 err = mdsc->mdsmap_err;
3667 doutc(cl, "mdsmap err %d\n", err);
3668 goto finish;
3669 }
3670 if (mdsc->mdsmap->m_epoch == 0) {
3671 doutc(cl, "no mdsmap, waiting for map\n");
3672 trace_ceph_mdsc_suspend_request(mdsc, session, req,
3673 ceph_mdsc_suspend_reason_no_mdsmap);
3674 list_add(&req->r_wait, &mdsc->waiting_for_map);
3675 return;
3676 }
3677 if (!(mdsc->fsc->mount_options->flags &
3678 CEPH_MOUNT_OPT_MOUNTWAIT) &&
3679 !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) {
3680 err = -EHOSTUNREACH;
3681 goto finish;
3682 }
3683 }
3684
3685 put_request_session(req);
3686
3687 mds = __choose_mds(mdsc, req, &random);
3688 if (mds < 0 ||
3689 ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
3690 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
3691 err = -EJUKEBOX;
3692 goto finish;
3693 }
3694 doutc(cl, "no mds or not active, waiting for map\n");
3695 trace_ceph_mdsc_suspend_request(mdsc, session, req,
3696 ceph_mdsc_suspend_reason_no_active_mds);
3697 list_add(&req->r_wait, &mdsc->waiting_for_map);
3698 return;
3699 }
3700
3701 /* get, open session */
3702 session = __ceph_lookup_mds_session(mdsc, mds);
3703 if (!session) {
3704 session = register_session(mdsc, mds);
3705 if (IS_ERR(session)) {
3706 err = PTR_ERR(session);
3707 goto finish;
3708 }
3709 }
3710 req->r_session = ceph_get_mds_session(session);
3711
3712 doutc(cl, "mds%d session %p state %s\n", mds, session,
3713 ceph_session_state_name(session->s_state));
3714
3715 /*
3716 * The old ceph will crash the MDSs when see unknown OPs
3717 */
3718 if (req->r_feature_needed > 0 &&
3719 !test_bit(req->r_feature_needed, &session->s_features)) {
3720 err = -EOPNOTSUPP;
3721 goto out_session;
3722 }
3723
3724 if (session->s_state != CEPH_MDS_SESSION_OPEN &&
3725 session->s_state != CEPH_MDS_SESSION_HUNG) {
3726 /*
3727 * We cannot queue async requests since the caps and delegated
3728 * inodes are bound to the session. Just return -EJUKEBOX and
3729 * let the caller retry a sync request in that case.
3730 */
3731 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
3732 err = -EJUKEBOX;
3733 goto out_session;
3734 }
3735
3736 /*
3737 * If the session has been REJECTED, then return a hard error,
3738 * unless it's a CLEANRECOVER mount, in which case we'll queue
3739 * it to the mdsc queue.
3740 */
3741 if (session->s_state == CEPH_MDS_SESSION_REJECTED) {
3742 if (ceph_test_mount_opt(mdsc->fsc, CLEANRECOVER)) {
3743 trace_ceph_mdsc_suspend_request(mdsc, session, req,
3744 ceph_mdsc_suspend_reason_rejected);
3745 list_add(&req->r_wait, &mdsc->waiting_for_map);
3746 } else
3747 err = -EACCES;
3748 goto out_session;
3749 }
3750
3751 if (session->s_state == CEPH_MDS_SESSION_NEW ||
3752 session->s_state == CEPH_MDS_SESSION_CLOSING) {
3753 err = __open_session(mdsc, session);
3754 if (err)
3755 goto out_session;
3756 /* retry the same mds later */
3757 if (random)
3758 req->r_resend_mds = mds;
3759 }
3760 trace_ceph_mdsc_suspend_request(mdsc, session, req,
3761 ceph_mdsc_suspend_reason_session);
3762 list_add(&req->r_wait, &session->s_waiting);
3763 goto out_session;
3764 }
3765
3766 /* send request */
3767 req->r_resend_mds = -1; /* forget any previous mds hint */
3768
3769 if (req->r_request_started == 0) /* note request start time */
3770 req->r_request_started = jiffies;
3771
3772 /*
3773 * For async create we will choose the auth MDS of frag in parent
3774 * directory to send the request and usually this works fine, but
3775 * if the migrated the dirtory to another MDS before it could handle
3776 * it the request will be forwarded.
3777 *
3778 * And then the auth cap will be changed.
3779 */
3780 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) && req->r_num_fwd) {
3781 struct ceph_dentry_info *di = ceph_dentry(req->r_dentry);
3782 struct ceph_inode_info *ci;
3783 struct ceph_cap *cap;
3784
3785 /*
3786 * The request maybe handled very fast and the new inode
3787 * hasn't been linked to the dentry yet. We need to wait
3788 * for the ceph_finish_async_create(), which shouldn't be
3789 * stuck too long or fail in thoery, to finish when forwarding
3790 * the request.
3791 */
3792 if (!d_inode(req->r_dentry)) {
3793 err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT,
3794 TASK_KILLABLE);
3795 if (err) {
3796 mutex_lock(&req->r_fill_mutex);
3797 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
3798 mutex_unlock(&req->r_fill_mutex);
3799 goto out_session;
3800 }
3801 }
3802
3803 ci = ceph_inode(d_inode(req->r_dentry));
3804
3805 spin_lock(&ci->i_ceph_lock);
3806 cap = ci->i_auth_cap;
3807 if (test_bit(CEPH_I_ASYNC_CREATE_BIT, &ci->i_ceph_flags) &&
3808 mds != cap->mds) {
3809 doutc(cl, "session changed for auth cap %d -> %d\n",
3810 cap->session->s_mds, session->s_mds);
3811
3812 /* Remove the auth cap from old session */
3813 spin_lock(&cap->session->s_cap_lock);
3814 cap->session->s_nr_caps--;
3815 list_del_init(&cap->session_caps);
3816 spin_unlock(&cap->session->s_cap_lock);
3817
3818 /* Add the auth cap to the new session */
3819 cap->mds = mds;
3820 cap->session = session;
3821 spin_lock(&session->s_cap_lock);
3822 session->s_nr_caps++;
3823 list_add_tail(&cap->session_caps, &session->s_caps);
3824 spin_unlock(&session->s_cap_lock);
3825
3826 change_auth_cap_ses(ci, session);
3827 }
3828 spin_unlock(&ci->i_ceph_lock);
3829 }
3830
3831 err = __send_request(session, req, false);
3832
3833out_session:
3834 ceph_put_mds_session(session);
3835finish:
3836 if (err) {
3837 doutc(cl, "early error %d\n", err);
3838 req->r_err = err;
3839 complete_request(mdsc, req);
3840 __unregister_request(mdsc, req);
3841 }
3842 return;
3843}
3844
3845/*
3846 * called under mdsc->mutex
3847 */
3848static void __wake_requests(struct ceph_mds_client *mdsc,
3849 struct list_head *head)
3850{
3851 struct ceph_client *cl = mdsc->fsc->client;
3852 struct ceph_mds_request *req;
3853 LIST_HEAD(tmp_list);
3854
3855 list_splice_init(head, &tmp_list);
3856
3857 while (!list_empty(&tmp_list)) {
3858 req = list_entry(tmp_list.next,
3859 struct ceph_mds_request, r_wait);
3860 list_del_init(&req->r_wait);
3861 doutc(cl, " wake request %p tid %llu\n", req,
3862 req->r_tid);
3863 trace_ceph_mdsc_resume_request(mdsc, req);
3864 __do_request(mdsc, req);
3865 }
3866}
3867
3868/*
3869 * Wake up threads with requests pending for @mds, so that they can
3870 * resubmit their requests to a possibly different mds.
3871 */
3872static void kick_requests(struct ceph_mds_client *mdsc, int mds)
3873{
3874 struct ceph_client *cl = mdsc->fsc->client;
3875 struct ceph_mds_request *req;
3876 struct rb_node *p = rb_first(&mdsc->request_tree);
3877
3878 doutc(cl, "kick_requests mds%d\n", mds);
3879 while (p) {
3880 req = rb_entry(p, struct ceph_mds_request, r_node);
3881 p = rb_next(p);
3882 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
3883 continue;
3884 if (req->r_attempts > 0)
3885 continue; /* only new requests */
3886 if (req->r_session &&
3887 req->r_session->s_mds == mds) {
3888 doutc(cl, " kicking tid %llu\n", req->r_tid);
3889 list_del_init(&req->r_wait);
3890 trace_ceph_mdsc_resume_request(mdsc, req);
3891 __do_request(mdsc, req);
3892 }
3893 }
3894}
3895
3896int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,
3897 struct ceph_mds_request *req)
3898{
3899 struct ceph_client *cl = mdsc->fsc->client;
3900 int err = 0;
3901
3902 /*
3903 * If a reset is in progress, wait for it to complete.
3904 *
3905 * This is best-effort: a request can pass this check just
3906 * before the phase leaves IDLE and proceed concurrently with
3907 * reset. That is acceptable because (a) such requests will
3908 * either complete normally or fail and be retried by the
3909 * caller, and (b) adding lock serialization here would
3910 * penalize every request for a rare manual operation.
3911 */
3912 err = ceph_mdsc_wait_for_reset(mdsc);
3913 if (err) {
3914 doutc(cl, "wait_for_reset failed: %d\n", err);
3915 return err;
3916 }
3917
3918 /* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
3919 if (req->r_inode)
3920 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
3921 if (req->r_parent) {
3922 struct ceph_inode_info *ci = ceph_inode(req->r_parent);
3923 int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?
3924 CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;
3925 spin_lock(&ci->i_ceph_lock);
3926 ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);
3927 __ceph_touch_fmode(ci, mdsc, fmode);
3928 spin_unlock(&ci->i_ceph_lock);
3929 }
3930 if (req->r_old_dentry_dir)
3931 ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
3932 CEPH_CAP_PIN);
3933
3934 if (req->r_inode) {
3935 err = ceph_wait_on_async_create(req->r_inode);
3936 if (err) {
3937 doutc(cl, "wait for async create returned: %d\n", err);
3938 return err;
3939 }
3940 }
3941
3942 if (!err && req->r_old_inode) {
3943 err = ceph_wait_on_async_create(req->r_old_inode);
3944 if (err) {
3945 doutc(cl, "wait for async create returned: %d\n", err);
3946 return err;
3947 }
3948 }
3949
3950 doutc(cl, "submit_request on %p for inode %p\n", req, dir);
3951 mutex_lock(&mdsc->mutex);
3952 __register_request(mdsc, req, dir);
3953 trace_ceph_mdsc_submit_request(mdsc, req);
3954 __do_request(mdsc, req);
3955 err = req->r_err;
3956 mutex_unlock(&mdsc->mutex);
3957 return err;
3958}
3959
3960int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
3961 struct ceph_mds_request *req,
3962 ceph_mds_request_wait_callback_t wait_func)
3963{
3964 struct ceph_client *cl = mdsc->fsc->client;
3965 int err;
3966
3967 /* wait */
3968 doutc(cl, "do_request waiting\n");
3969 if (wait_func) {
3970 err = wait_func(mdsc, req);
3971 } else {
3972 long timeleft = wait_for_completion_killable_timeout(
3973 &req->r_completion,
3974 ceph_timeout_jiffies(req->r_timeout));
3975 if (timeleft > 0)
3976 err = 0;
3977 else if (!timeleft)
3978 err = -ETIMEDOUT; /* timed out */
3979 else
3980 err = timeleft; /* killed */
3981 }
3982 doutc(cl, "do_request waited, got %d\n", err);
3983 mutex_lock(&mdsc->mutex);
3984
3985 /* only abort if we didn't race with a real reply */
3986 if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
3987 err = le32_to_cpu(req->r_reply_info.head->result);
3988 } else if (err < 0) {
3989 doutc(cl, "aborted request %lld with %d\n", req->r_tid, err);
3990
3991 /*
3992 * ensure we aren't running concurrently with
3993 * ceph_fill_trace or ceph_readdir_prepopulate, which
3994 * rely on locks (dir mutex) held by our caller.
3995 */
3996 mutex_lock(&req->r_fill_mutex);
3997 req->r_err = err;
3998 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
3999 mutex_unlock(&req->r_fill_mutex);
4000
4001 if (req->r_parent &&
4002 (req->r_op & CEPH_MDS_OP_WRITE))
4003 ceph_invalidate_dir_request(req);
4004 } else {
4005 err = req->r_err;
4006 }
4007
4008 mutex_unlock(&mdsc->mutex);
4009 return err;
4010}
4011
4012/*
4013 * Synchrously perform an mds request. Take care of all of the
4014 * session setup, forwarding, retry details.
4015 */
4016int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
4017 struct inode *dir,
4018 struct ceph_mds_request *req)
4019{
4020 struct ceph_client *cl = mdsc->fsc->client;
4021 int err;
4022
4023 doutc(cl, "do_request on %p\n", req);
4024
4025 /* issue */
4026 err = ceph_mdsc_submit_request(mdsc, dir, req);
4027 if (!err)
4028 err = ceph_mdsc_wait_request(mdsc, req, NULL);
4029 doutc(cl, "do_request %p done, result %d\n", req, err);
4030 return err;
4031}
4032
4033/*
4034 * Invalidate dir's completeness, dentry lease state on an aborted MDS
4035 * namespace request.
4036 */
4037void ceph_invalidate_dir_request(struct ceph_mds_request *req)
4038{
4039 struct inode *dir = req->r_parent;
4040 struct inode *old_dir = req->r_old_dentry_dir;
4041 struct ceph_client *cl = req->r_mdsc->fsc->client;
4042
4043 doutc(cl, "invalidate_dir_request %p %p (complete, lease(s))\n",
4044 dir, old_dir);
4045
4046 ceph_dir_clear_complete(dir);
4047 if (old_dir)
4048 ceph_dir_clear_complete(old_dir);
4049 if (req->r_dentry)
4050 ceph_invalidate_dentry_lease(req->r_dentry);
4051 if (req->r_old_dentry)
4052 ceph_invalidate_dentry_lease(req->r_old_dentry);
4053}
4054
4055/*
4056 * Handle mds reply.
4057 *
4058 * We take the session mutex and parse and process the reply immediately.
4059 * This preserves the logical ordering of replies, capabilities, etc., sent
4060 * by the MDS as they are applied to our local cache.
4061 */
4062static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
4063{
4064 struct ceph_mds_client *mdsc = session->s_mdsc;
4065 struct ceph_client *cl = mdsc->fsc->client;
4066 struct ceph_mds_request *req;
4067 struct ceph_mds_reply_head *head = msg->front.iov_base;
4068 struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */
4069 struct ceph_snap_realm *realm;
4070 unsigned int nofs_flags;
4071 u64 tid;
4072 int err, result;
4073 int mds = session->s_mds;
4074 bool close_sessions = false;
4075
4076 if (msg->front.iov_len < sizeof(*head)) {
4077 pr_err_client(cl, "got corrupt (short) reply\n");
4078 ceph_msg_dump(msg);
4079 return;
4080 }
4081
4082 /* get request, session */
4083 tid = le64_to_cpu(msg->hdr.tid);
4084 mutex_lock(&mdsc->mutex);
4085 req = lookup_get_request(mdsc, tid);
4086 if (!req) {
4087 doutc(cl, "on unknown tid %llu\n", tid);
4088 mutex_unlock(&mdsc->mutex);
4089 return;
4090 }
4091 doutc(cl, "handle_reply %p\n", req);
4092
4093 /* correct session? */
4094 if (req->r_session != session) {
4095 pr_err_client(cl, "got %llu on session mds%d not mds%d\n",
4096 tid, session->s_mds,
4097 req->r_session ? req->r_session->s_mds : -1);
4098 mutex_unlock(&mdsc->mutex);
4099 goto out;
4100 }
4101
4102 /* dup? */
4103 if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
4104 (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
4105 pr_warn_client(cl, "got a dup %s reply on %llu from mds%d\n",
4106 head->safe ? "safe" : "unsafe", tid, mds);
4107 mutex_unlock(&mdsc->mutex);
4108 goto out;
4109 }
4110 if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
4111 pr_warn_client(cl, "got unsafe after safe on %llu from mds%d\n",
4112 tid, mds);
4113 mutex_unlock(&mdsc->mutex);
4114 goto out;
4115 }
4116
4117 result = le32_to_cpu(head->result);
4118
4119 if (head->safe) {
4120 set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
4121 __unregister_request(mdsc, req);
4122
4123 /* last request during umount? */
4124 if (mdsc->stopping && !__get_oldest_req(mdsc))
4125 complete_all(&mdsc->safe_umount_waiters);
4126
4127 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
4128 /*
4129 * We already handled the unsafe response, now do the
4130 * cleanup. No need to examine the response; the MDS
4131 * doesn't include any result info in the safe
4132 * response. And even if it did, there is nothing
4133 * useful we could do with a revised return value.
4134 */
4135 doutc(cl, "got safe reply %llu, mds%d\n", tid, mds);
4136
4137 mutex_unlock(&mdsc->mutex);
4138 goto out;
4139 }
4140 } else {
4141 set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
4142 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
4143 }
4144
4145 /*
4146 * Now that all mutex-protected state has been updated above
4147 * (the request has been unregistered or added to the
4148 * session's unsafe list), we can unlock it.
4149 */
4150 mutex_unlock(&mdsc->mutex);
4151
4152 doutc(cl, "tid %lld result %d\n", tid, result);
4153 if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))
4154 err = parse_reply_info(session, msg, req, (u64)-1);
4155 else
4156 err = parse_reply_info(session, msg, req,
4157 session->s_con.peer_features);
4158
4159 /* Must find target inode outside of mutexes to avoid deadlocks */
4160 rinfo = &req->r_reply_info;
4161 if ((err >= 0) && rinfo->head->is_target) {
4162 struct inode *in = xchg(&req->r_new_inode, NULL);
4163 struct ceph_vino tvino = {
4164 .ino = le64_to_cpu(rinfo->targeti.in->ino),
4165 .snap = le64_to_cpu(rinfo->targeti.in->snapid)
4166 };
4167
4168 /*
4169 * If we ended up opening an existing inode, discard
4170 * r_new_inode
4171 */
4172 if (req->r_op == CEPH_MDS_OP_CREATE &&
4173 !req->r_reply_info.has_create_ino) {
4174 /* This should never happen on an async create */
4175 WARN_ON_ONCE(req->r_deleg_ino);
4176 iput(in);
4177 in = NULL;
4178 }
4179
4180 in = ceph_get_inode(mdsc->fsc->sb, tvino, in);
4181 if (IS_ERR(in)) {
4182 err = PTR_ERR(in);
4183 mutex_lock(&session->s_mutex);
4184 goto out_err;
4185 }
4186 req->r_target_inode = in;
4187 ceph_inode_set_subvolume(in, rinfo->targeti.subvolume_id);
4188 }
4189
4190 mutex_lock(&session->s_mutex);
4191 if (err < 0) {
4192 pr_err_client(cl, "got corrupt reply mds%d(tid:%lld)\n",
4193 mds, tid);
4194 ceph_msg_dump(msg);
4195 goto out_err;
4196 }
4197
4198 /* snap trace */
4199 realm = NULL;
4200 if (rinfo->snapblob_len) {
4201 down_write(&mdsc->snap_rwsem);
4202 err = ceph_update_snap_trace(mdsc, rinfo->snapblob,
4203 rinfo->snapblob + rinfo->snapblob_len,
4204 le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
4205 &realm);
4206 if (err) {
4207 up_write(&mdsc->snap_rwsem);
4208 close_sessions = true;
4209 if (err == -EIO)
4210 ceph_msg_dump(msg);
4211 goto out_err;
4212 }
4213 downgrade_write(&mdsc->snap_rwsem);
4214 } else {
4215 down_read(&mdsc->snap_rwsem);
4216 }
4217
4218 /* insert trace into our cache */
4219 mutex_lock(&req->r_fill_mutex);
4220
4221 /* disable fs reclaim while we are using current->journal_info
4222 * for our own purposes, or else shrinkers of other
4223 * filesystems might dereference this pointer as a different
4224 * type
4225 */
4226 nofs_flags = memalloc_nofs_save();
4227
4228 current->journal_info = req;
4229 err = ceph_fill_trace(mdsc->fsc->sb, req);
4230 if (err == 0) {
4231 if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
4232 req->r_op == CEPH_MDS_OP_LSSNAP))
4233 err = ceph_readdir_prepopulate(req, req->r_session);
4234 }
4235 current->journal_info = NULL;
4236 memalloc_nofs_restore(nofs_flags);
4237 mutex_unlock(&req->r_fill_mutex);
4238
4239 up_read(&mdsc->snap_rwsem);
4240 if (realm)
4241 ceph_put_snap_realm(mdsc, realm);
4242
4243 if (err == 0) {
4244 if (req->r_target_inode &&
4245 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
4246 struct ceph_inode_info *ci =
4247 ceph_inode(req->r_target_inode);
4248 spin_lock(&ci->i_unsafe_lock);
4249 list_add_tail(&req->r_unsafe_target_item,
4250 &ci->i_unsafe_iops);
4251 spin_unlock(&ci->i_unsafe_lock);
4252 }
4253
4254 ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
4255 }
4256out_err:
4257 mutex_lock(&mdsc->mutex);
4258 if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
4259 if (err) {
4260 req->r_err = err;
4261 } else {
4262 req->r_reply = ceph_msg_get(msg);
4263 set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
4264 }
4265 } else {
4266 doutc(cl, "reply arrived after request %lld was aborted\n", tid);
4267 }
4268 mutex_unlock(&mdsc->mutex);
4269
4270 mutex_unlock(&session->s_mutex);
4271
4272 /* kick calling process */
4273 complete_request(mdsc, req);
4274
4275 ceph_update_metadata_metrics(&mdsc->metric, req->r_start_latency,
4276 req->r_end_latency, err);
4277out:
4278 ceph_mdsc_put_request(req);
4279
4280 /* Defer closing the sessions after s_mutex lock being released */
4281 if (close_sessions)
4282 ceph_mdsc_close_sessions(mdsc);
4283 return;
4284}
4285
4286
4287
4288/*
4289 * handle mds notification that our request has been forwarded.
4290 */
4291static void handle_forward(struct ceph_mds_client *mdsc,
4292 struct ceph_mds_session *session,
4293 struct ceph_msg *msg)
4294{
4295 struct ceph_client *cl = mdsc->fsc->client;
4296 struct ceph_mds_request *req;
4297 u64 tid = le64_to_cpu(msg->hdr.tid);
4298 u32 next_mds;
4299 u32 fwd_seq;
4300 int err = -EINVAL;
4301 void *p = msg->front.iov_base;
4302 void *end = p + msg->front.iov_len;
4303 bool aborted = false;
4304
4305 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
4306 next_mds = ceph_decode_32(&p);
4307 fwd_seq = ceph_decode_32(&p);
4308
4309 mutex_lock(&mdsc->mutex);
4310 req = lookup_get_request(mdsc, tid);
4311 if (!req) {
4312 mutex_unlock(&mdsc->mutex);
4313 doutc(cl, "forward tid %llu to mds%d - req dne\n", tid, next_mds);
4314 return; /* dup reply? */
4315 }
4316
4317 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
4318 doutc(cl, "forward tid %llu aborted, unregistering\n", tid);
4319 __unregister_request(mdsc, req);
4320 } else if (fwd_seq <= req->r_num_fwd || (uint32_t)fwd_seq >= U32_MAX) {
4321 /*
4322 * Avoid infinite retrying after overflow.
4323 *
4324 * The MDS will increase the fwd count and in client side
4325 * if the num_fwd is less than the one saved in request
4326 * that means the MDS is an old version and overflowed of
4327 * 8 bits.
4328 */
4329 mutex_lock(&req->r_fill_mutex);
4330 req->r_err = -EMULTIHOP;
4331 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
4332 mutex_unlock(&req->r_fill_mutex);
4333 aborted = true;
4334 pr_warn_ratelimited_client(cl, "forward tid %llu seq overflow\n",
4335 tid);
4336 } else {
4337 /* resend. forward race not possible; mds would drop */
4338 doutc(cl, "forward tid %llu to mds%d (we resend)\n", tid, next_mds);
4339 BUG_ON(req->r_err);
4340 BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
4341 req->r_attempts = 0;
4342 req->r_num_fwd = fwd_seq;
4343 req->r_resend_mds = next_mds;
4344 put_request_session(req);
4345 __do_request(mdsc, req);
4346 }
4347 mutex_unlock(&mdsc->mutex);
4348
4349 /* kick calling process */
4350 if (aborted)
4351 complete_request(mdsc, req);
4352 ceph_mdsc_put_request(req);
4353 return;
4354
4355bad:
4356 pr_err_client(cl, "decode error err=%d\n", err);
4357 ceph_msg_dump(msg);
4358}
4359
4360static int __decode_session_metadata(void **p, void *end,
4361 bool *blocklisted)
4362{
4363 /* map<string,string> */
4364 u32 n;
4365 bool err_str;
4366 ceph_decode_32_safe(p, end, n, bad);
4367 while (n-- > 0) {
4368 u32 len;
4369 ceph_decode_32_safe(p, end, len, bad);
4370 ceph_decode_need(p, end, len, bad);
4371 err_str = !strncmp(*p, "error_string", len);
4372 *p += len;
4373 ceph_decode_32_safe(p, end, len, bad);
4374 ceph_decode_need(p, end, len, bad);
4375 /*
4376 * Match "blocklisted (blacklisted)" from newer MDSes,
4377 * or "blacklisted" from older MDSes.
4378 */
4379 if (err_str && strnstr(*p, "blacklisted", len))
4380 *blocklisted = true;
4381 *p += len;
4382 }
4383 return 0;
4384bad:
4385 return -1;
4386}
4387
4388/*
4389 * handle a mds session control message
4390 */
4391static void handle_session(struct ceph_mds_session *session,
4392 struct ceph_msg *msg)
4393{
4394 struct ceph_mds_client *mdsc = session->s_mdsc;
4395 struct ceph_client *cl = mdsc->fsc->client;
4396 int mds = session->s_mds;
4397 int msg_version = le16_to_cpu(msg->hdr.version);
4398 void *p = msg->front.iov_base;
4399 void *end = p + msg->front.iov_len;
4400 struct ceph_mds_session_head *h;
4401 struct ceph_mds_cap_auth *cap_auths = NULL;
4402 u32 op, cap_auths_num = 0;
4403 u64 seq, features = 0;
4404 int wake = 0;
4405 bool blocklisted = false;
4406 u32 i;
4407
4408
4409 /* decode */
4410 ceph_decode_need(&p, end, sizeof(*h), bad);
4411 h = p;
4412 p += sizeof(*h);
4413
4414 op = le32_to_cpu(h->op);
4415 seq = le64_to_cpu(h->seq);
4416
4417 if (msg_version >= 3) {
4418 u32 len;
4419 /* version >= 2 and < 5, decode metadata, skip otherwise
4420 * as it's handled via flags.
4421 */
4422 if (msg_version >= 5)
4423 ceph_decode_skip_map(&p, end, string, string, bad);
4424 else if (__decode_session_metadata(&p, end, &blocklisted) < 0)
4425 goto bad;
4426
4427 /* version >= 3, feature bits */
4428 ceph_decode_32_safe(&p, end, len, bad);
4429 if (len) {
4430 ceph_decode_64_safe(&p, end, features, bad);
4431 p += len - sizeof(features);
4432 }
4433 }
4434
4435 if (msg_version >= 5) {
4436 u32 flags, len;
4437
4438 /* version >= 4 */
4439 ceph_decode_skip_16(&p, end, bad); /* struct_v, struct_cv */
4440 ceph_decode_32_safe(&p, end, len, bad); /* len */
4441 ceph_decode_skip_n(&p, end, len, bad); /* metric_spec */
4442
4443 /* version >= 5, flags */
4444 ceph_decode_32_safe(&p, end, flags, bad);
4445 if (flags & CEPH_SESSION_BLOCKLISTED) {
4446 pr_warn_client(cl, "mds%d session blocklisted\n",
4447 session->s_mds);
4448 blocklisted = true;
4449 }
4450 }
4451
4452 if (msg_version >= 6) {
4453 ceph_decode_32_safe(&p, end, cap_auths_num, bad);
4454 doutc(cl, "cap_auths_num %d\n", cap_auths_num);
4455
4456 if (cap_auths_num && op != CEPH_SESSION_OPEN) {
4457 WARN_ON_ONCE(op != CEPH_SESSION_OPEN);
4458 goto skip_cap_auths;
4459 }
4460
4461 cap_auths = kzalloc_objs(struct ceph_mds_cap_auth,
4462 cap_auths_num);
4463 if (!cap_auths) {
4464 pr_err_client(cl, "No memory for cap_auths\n");
4465 return;
4466 }
4467
4468 for (i = 0; i < cap_auths_num; i++) {
4469 u32 _len, j;
4470
4471 /* struct_v, struct_compat, and struct_len in MDSCapAuth */
4472 ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad);
4473
4474 /* struct_v, struct_compat, and struct_len in MDSCapMatch */
4475 ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad);
4476 ceph_decode_64_safe(&p, end, cap_auths[i].match.uid, bad);
4477 ceph_decode_32_safe(&p, end, _len, bad);
4478 if (_len) {
4479 cap_auths[i].match.gids = kcalloc(_len, sizeof(u32),
4480 GFP_KERNEL);
4481 if (!cap_auths[i].match.gids) {
4482 pr_err_client(cl, "No memory for gids\n");
4483 goto fail;
4484 }
4485
4486 cap_auths[i].match.num_gids = _len;
4487 for (j = 0; j < _len; j++)
4488 ceph_decode_32_safe(&p, end,
4489 cap_auths[i].match.gids[j],
4490 bad);
4491 }
4492
4493 ceph_decode_32_safe(&p, end, _len, bad);
4494 if (_len) {
4495 cap_auths[i].match.path = kcalloc(_len + 1, sizeof(char),
4496 GFP_KERNEL);
4497 if (!cap_auths[i].match.path) {
4498 pr_err_client(cl, "No memory for path\n");
4499 goto fail;
4500 }
4501 ceph_decode_copy_safe(&p, end,
4502 cap_auths[i].match.path,
4503 _len, bad);
4504
4505 /* Remove the tailing '/' */
4506 while (_len && cap_auths[i].match.path[_len - 1] == '/') {
4507 cap_auths[i].match.path[_len - 1] = '\0';
4508 _len -= 1;
4509 }
4510 }
4511
4512 ceph_decode_32_safe(&p, end, _len, bad);
4513 if (_len) {
4514 cap_auths[i].match.fs_name = kcalloc(_len + 1, sizeof(char),
4515 GFP_KERNEL);
4516 if (!cap_auths[i].match.fs_name) {
4517 pr_err_client(cl, "No memory for fs_name\n");
4518 goto fail;
4519 }
4520 ceph_decode_copy_safe(&p, end,
4521 cap_auths[i].match.fs_name,
4522 _len, bad);
4523 }
4524
4525 ceph_decode_8_safe(&p, end, cap_auths[i].match.root_squash, bad);
4526 ceph_decode_8_safe(&p, end, cap_auths[i].readable, bad);
4527 ceph_decode_8_safe(&p, end, cap_auths[i].writeable, bad);
4528 doutc(cl, "uid %lld, num_gids %u, path %s, fs_name %s, root_squash %d, readable %d, writeable %d\n",
4529 cap_auths[i].match.uid, cap_auths[i].match.num_gids,
4530 cap_auths[i].match.path, cap_auths[i].match.fs_name,
4531 cap_auths[i].match.root_squash,
4532 cap_auths[i].readable, cap_auths[i].writeable);
4533 }
4534 }
4535
4536skip_cap_auths:
4537 mutex_lock(&mdsc->mutex);
4538 if (op == CEPH_SESSION_OPEN) {
4539 if (mdsc->s_cap_auths) {
4540 for (i = 0; i < mdsc->s_cap_auths_num; i++) {
4541 kfree(mdsc->s_cap_auths[i].match.gids);
4542 kfree(mdsc->s_cap_auths[i].match.path);
4543 kfree(mdsc->s_cap_auths[i].match.fs_name);
4544 }
4545 kfree(mdsc->s_cap_auths);
4546 }
4547 mdsc->s_cap_auths_num = cap_auths_num;
4548 mdsc->s_cap_auths = cap_auths;
4549
4550 session->s_features = features;
4551 if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT,
4552 &session->s_features))
4553 ceph_metric_bind_session(mdsc, session);
4554 }
4555 if (op == CEPH_SESSION_CLOSE) {
4556 ceph_get_mds_session(session);
4557 __unregister_session(mdsc, session);
4558 }
4559 /* FIXME: this ttl calculation is generous */
4560 session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
4561 mutex_unlock(&mdsc->mutex);
4562
4563 mutex_lock(&session->s_mutex);
4564
4565 doutc(cl, "mds%d %s %p state %s seq %llu\n", mds,
4566 ceph_session_op_name(op), session,
4567 ceph_session_state_name(session->s_state), seq);
4568
4569 if (session->s_state == CEPH_MDS_SESSION_HUNG) {
4570 session->s_state = CEPH_MDS_SESSION_OPEN;
4571 pr_info_client(cl, "mds%d came back\n", session->s_mds);
4572 }
4573
4574 switch (op) {
4575 case CEPH_SESSION_OPEN:
4576 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
4577 pr_info_client(cl, "mds%d reconnect success\n",
4578 session->s_mds);
4579
4580 if (test_bit(CEPHFS_FEATURE_SUBVOLUME_METRICS,
4581 &session->s_features))
4582 ceph_subvolume_metrics_enable(&mdsc->subvol_metrics, true);
4583 else
4584 ceph_subvolume_metrics_enable(&mdsc->subvol_metrics, false);
4585 if (session->s_state == CEPH_MDS_SESSION_OPEN) {
4586 pr_notice_client(cl, "mds%d is already opened\n",
4587 session->s_mds);
4588 } else {
4589 session->s_state = CEPH_MDS_SESSION_OPEN;
4590 renewed_caps(mdsc, session, 0);
4591 if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT,
4592 &session->s_features))
4593 metric_schedule_delayed(&mdsc->metric);
4594 }
4595
4596 /*
4597 * The connection maybe broken and the session in client
4598 * side has been reinitialized, need to update the seq
4599 * anyway.
4600 */
4601 if (!session->s_seq && seq)
4602 session->s_seq = seq;
4603
4604 wake = 1;
4605 if (mdsc->stopping)
4606 __close_session(mdsc, session);
4607 break;
4608
4609 case CEPH_SESSION_RENEWCAPS:
4610 if (session->s_renew_seq == seq)
4611 renewed_caps(mdsc, session, 1);
4612 break;
4613
4614 case CEPH_SESSION_CLOSE:
4615 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
4616 pr_info_client(cl, "mds%d reconnect denied\n",
4617 session->s_mds);
4618 session->s_state = CEPH_MDS_SESSION_CLOSED;
4619 cleanup_session_requests(mdsc, session);
4620 remove_session_caps(session);
4621 wake = 2; /* for good measure */
4622 wake_up_all(&mdsc->session_close_wq);
4623 break;
4624
4625 case CEPH_SESSION_STALE:
4626 pr_info_client(cl, "mds%d caps went stale, renewing\n",
4627 session->s_mds);
4628 atomic_inc(&session->s_cap_gen);
4629 session->s_cap_ttl = jiffies - 1;
4630 send_renew_caps(mdsc, session);
4631 break;
4632
4633 case CEPH_SESSION_RECALL_STATE:
4634 ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
4635 break;
4636
4637 case CEPH_SESSION_FLUSHMSG:
4638 /* flush cap releases */
4639 spin_lock(&session->s_cap_lock);
4640 if (session->s_num_cap_releases)
4641 ceph_flush_session_cap_releases(mdsc, session);
4642 spin_unlock(&session->s_cap_lock);
4643
4644 send_flushmsg_ack(mdsc, session, seq);
4645 break;
4646
4647 case CEPH_SESSION_FORCE_RO:
4648 doutc(cl, "force_session_readonly %p\n", session);
4649 spin_lock(&session->s_cap_lock);
4650 session->s_readonly = true;
4651 spin_unlock(&session->s_cap_lock);
4652 wake_up_session_caps(session, FORCE_RO);
4653 break;
4654
4655 case CEPH_SESSION_REJECT:
4656 WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING &&
4657 session->s_state != CEPH_MDS_SESSION_RECONNECTING);
4658 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
4659 pr_info_client(cl, "mds%d reconnect rejected\n",
4660 session->s_mds);
4661 else
4662 pr_info_client(cl, "mds%d rejected session\n",
4663 session->s_mds);
4664 session->s_state = CEPH_MDS_SESSION_REJECTED;
4665 cleanup_session_requests(mdsc, session);
4666 remove_session_caps(session);
4667 if (blocklisted)
4668 mdsc->fsc->blocklisted = true;
4669 wake = 2; /* for good measure */
4670 break;
4671
4672 default:
4673 pr_err_client(cl, "bad op %d mds%d\n", op, mds);
4674 WARN_ON(1);
4675 }
4676
4677 mutex_unlock(&session->s_mutex);
4678 if (wake) {
4679 mutex_lock(&mdsc->mutex);
4680 __wake_requests(mdsc, &session->s_waiting);
4681 if (wake == 2)
4682 kick_requests(mdsc, mds);
4683 mutex_unlock(&mdsc->mutex);
4684 }
4685 if (op == CEPH_SESSION_CLOSE)
4686 ceph_put_mds_session(session);
4687 return;
4688
4689bad:
4690 pr_err_client(cl, "corrupt message mds%d len %d\n", mds,
4691 (int)msg->front.iov_len);
4692 ceph_msg_dump(msg);
4693fail:
4694 for (i = 0; i < cap_auths_num; i++) {
4695 kfree(cap_auths[i].match.gids);
4696 kfree(cap_auths[i].match.path);
4697 kfree(cap_auths[i].match.fs_name);
4698 }
4699 kfree(cap_auths);
4700 return;
4701}
4702
4703void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)
4704{
4705 struct ceph_client *cl = req->r_mdsc->fsc->client;
4706 int dcaps;
4707
4708 dcaps = xchg(&req->r_dir_caps, 0);
4709 if (dcaps) {
4710 doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
4711 ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);
4712 }
4713}
4714
4715void ceph_mdsc_release_dir_caps_async(struct ceph_mds_request *req)
4716{
4717 struct ceph_client *cl = req->r_mdsc->fsc->client;
4718 int dcaps;
4719
4720 dcaps = xchg(&req->r_dir_caps, 0);
4721 if (dcaps) {
4722 doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
4723 ceph_put_cap_refs_async(ceph_inode(req->r_parent), dcaps);
4724 }
4725}
4726
4727/*
4728 * called under session->mutex.
4729 */
4730static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
4731 struct ceph_mds_session *session)
4732{
4733 struct ceph_mds_request *req, *nreq;
4734 struct rb_node *p;
4735
4736 doutc(mdsc->fsc->client, "mds%d\n", session->s_mds);
4737
4738 mutex_lock(&mdsc->mutex);
4739 list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)
4740 __send_request(session, req, true);
4741
4742 /*
4743 * also re-send old requests when MDS enters reconnect stage. So that MDS
4744 * can process completed request in clientreplay stage.
4745 */
4746 p = rb_first(&mdsc->request_tree);
4747 while (p) {
4748 req = rb_entry(p, struct ceph_mds_request, r_node);
4749 p = rb_next(p);
4750 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
4751 continue;
4752 if (req->r_attempts == 0)
4753 continue; /* only old requests */
4754 if (!req->r_session)
4755 continue;
4756 if (req->r_session->s_mds != session->s_mds)
4757 continue;
4758
4759 ceph_mdsc_release_dir_caps_async(req);
4760
4761 __send_request(session, req, true);
4762 }
4763 mutex_unlock(&mdsc->mutex);
4764}
4765
4766static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)
4767{
4768 struct ceph_msg *reply;
4769 struct ceph_pagelist *_pagelist;
4770 struct page *page;
4771 __le32 *addr;
4772 int err = -ENOMEM;
4773
4774 if (!recon_state->allow_multi)
4775 return -ENOSPC;
4776
4777 /* can't handle message that contains both caps and realm */
4778 BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);
4779
4780 /* pre-allocate new pagelist */
4781 _pagelist = ceph_pagelist_alloc(GFP_NOFS);
4782 if (!_pagelist)
4783 return -ENOMEM;
4784
4785 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
4786 if (!reply)
4787 goto fail_msg;
4788
4789 /* placeholder for nr_caps */
4790 err = ceph_pagelist_encode_32(_pagelist, 0);
4791 if (err < 0)
4792 goto fail;
4793
4794 if (recon_state->nr_caps) {
4795 /* currently encoding caps */
4796 err = ceph_pagelist_encode_32(recon_state->pagelist, 0);
4797 if (err)
4798 goto fail;
4799 } else {
4800 /* placeholder for nr_realms (currently encoding relams) */
4801 err = ceph_pagelist_encode_32(_pagelist, 0);
4802 if (err < 0)
4803 goto fail;
4804 }
4805
4806 err = ceph_pagelist_encode_8(recon_state->pagelist, 1);
4807 if (err)
4808 goto fail;
4809
4810 page = list_first_entry(&recon_state->pagelist->head, struct page, lru);
4811 addr = kmap_atomic(page);
4812 if (recon_state->nr_caps) {
4813 /* currently encoding caps */
4814 *addr = cpu_to_le32(recon_state->nr_caps);
4815 } else {
4816 /* currently encoding relams */
4817 *(addr + 1) = cpu_to_le32(recon_state->nr_realms);
4818 }
4819 kunmap_atomic(addr);
4820
4821 reply->hdr.version = cpu_to_le16(5);
4822 reply->hdr.compat_version = cpu_to_le16(4);
4823
4824 reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);
4825 ceph_msg_data_add_pagelist(reply, recon_state->pagelist);
4826
4827 ceph_con_send(&recon_state->session->s_con, reply);
4828 ceph_pagelist_release(recon_state->pagelist);
4829
4830 recon_state->pagelist = _pagelist;
4831 recon_state->nr_caps = 0;
4832 recon_state->nr_realms = 0;
4833 recon_state->msg_version = 5;
4834 return 0;
4835fail:
4836 ceph_msg_put(reply);
4837fail_msg:
4838 ceph_pagelist_release(_pagelist);
4839 return err;
4840}
4841
4842static struct dentry* d_find_primary(struct inode *inode)
4843{
4844 struct dentry *alias, *dn = NULL;
4845
4846 if (hlist_empty(&inode->i_dentry))
4847 return NULL;
4848
4849 spin_lock(&inode->i_lock);
4850 if (hlist_empty(&inode->i_dentry))
4851 goto out_unlock;
4852
4853 if (S_ISDIR(inode->i_mode)) {
4854 alias = hlist_entry(inode->i_dentry.first, struct dentry, d_alias);
4855 if (!IS_ROOT(alias))
4856 dn = dget(alias);
4857 goto out_unlock;
4858 }
4859
4860 for_each_alias(alias, inode) {
4861 spin_lock(&alias->d_lock);
4862 if (!d_unhashed(alias) &&
4863 (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {
4864 dn = dget_dlock(alias);
4865 }
4866 spin_unlock(&alias->d_lock);
4867 if (dn)
4868 break;
4869 }
4870out_unlock:
4871 spin_unlock(&inode->i_lock);
4872 return dn;
4873}
4874
4875/*
4876 * Encode information about a cap for a reconnect with the MDS.
4877 */
4878static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
4879{
4880 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
4881 struct ceph_client *cl = ceph_inode_to_client(inode);
4882 union {
4883 struct ceph_mds_cap_reconnect v2;
4884 struct ceph_mds_cap_reconnect_v1 v1;
4885 } rec;
4886 struct ceph_inode_info *ci = ceph_inode(inode);
4887 struct ceph_reconnect_state *recon_state = arg;
4888 struct ceph_pagelist *pagelist = recon_state->pagelist;
4889 struct dentry *dentry;
4890 struct ceph_cap *cap;
4891 struct ceph_path_info path_info = {0};
4892 int err;
4893 u64 snap_follows;
4894
4895 dentry = d_find_primary(inode);
4896 if (dentry) {
4897 /* set pathbase to parent dir when msg_version >= 2 */
4898 char *path = ceph_mdsc_build_path(mdsc, dentry, &path_info,
4899 recon_state->msg_version >= 2);
4900 dput(dentry);
4901 if (IS_ERR(path)) {
4902 err = PTR_ERR(path);
4903 goto out_err;
4904 }
4905 }
4906
4907 spin_lock(&ci->i_ceph_lock);
4908 cap = __get_cap_for_mds(ci, mds);
4909 if (!cap) {
4910 spin_unlock(&ci->i_ceph_lock);
4911 err = 0;
4912 goto out_err;
4913 }
4914 doutc(cl, " adding %p ino %llx.%llx cap %p %lld %s\n", inode,
4915 ceph_vinop(inode), cap, cap->cap_id,
4916 ceph_cap_string(cap->issued));
4917
4918 cap->seq = 0; /* reset cap seq */
4919 cap->issue_seq = 0; /* and issue_seq */
4920 cap->mseq = 0; /* and migrate_seq */
4921 cap->cap_gen = atomic_read(&cap->session->s_cap_gen);
4922
4923 /*
4924 * Note: CEPH_I_ERROR_FILELOCK is not set during reconnect.
4925 * Instead, locks are submitted for best-effort MDS reclaim
4926 * via the flock_len field below. If reclaim fails (e.g.,
4927 * another client grabbed a conflicting lock), future lock
4928 * operations will fail and set the error flag at that point.
4929 */
4930
4931 /* These are lost when the session goes away */
4932 if (S_ISDIR(inode->i_mode)) {
4933 if (cap->issued & CEPH_CAP_DIR_CREATE) {
4934 ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
4935 memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
4936 }
4937 cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;
4938 }
4939
4940 if (recon_state->msg_version >= 2) {
4941 rec.v2.cap_id = cpu_to_le64(cap->cap_id);
4942 rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
4943 rec.v2.issued = cpu_to_le32(cap->issued);
4944 rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
4945 rec.v2.pathbase = cpu_to_le64(path_info.vino.ino);
4946 rec.v2.flock_len = cpu_to_le32(
4947 test_bit(CEPH_I_ERROR_FILELOCK_BIT,
4948 &ci->i_ceph_flags) ? 0 : 1);
4949 } else {
4950 struct timespec64 ts;
4951
4952 rec.v1.cap_id = cpu_to_le64(cap->cap_id);
4953 rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
4954 rec.v1.issued = cpu_to_le32(cap->issued);
4955 rec.v1.size = cpu_to_le64(i_size_read(inode));
4956 ts = inode_get_mtime(inode);
4957 ceph_encode_timespec64(&rec.v1.mtime, &ts);
4958 ts = inode_get_atime(inode);
4959 ceph_encode_timespec64(&rec.v1.atime, &ts);
4960 rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
4961 rec.v1.pathbase = cpu_to_le64(path_info.vino.ino);
4962 }
4963
4964 if (list_empty(&ci->i_cap_snaps)) {
4965 snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0;
4966 } else {
4967 struct ceph_cap_snap *capsnap =
4968 list_first_entry(&ci->i_cap_snaps,
4969 struct ceph_cap_snap, ci_item);
4970 snap_follows = capsnap->follows;
4971 }
4972 spin_unlock(&ci->i_ceph_lock);
4973
4974 if (recon_state->msg_version >= 2) {
4975 int num_fcntl_locks, num_flock_locks;
4976 struct ceph_filelock *flocks = NULL;
4977 size_t struct_len, total_len = sizeof(u64);
4978 u8 struct_v = 0;
4979
4980encode_again:
4981 if (rec.v2.flock_len) {
4982 ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
4983 } else {
4984 num_fcntl_locks = 0;
4985 num_flock_locks = 0;
4986 }
4987 if (num_fcntl_locks + num_flock_locks > 0) {
4988 flocks = kmalloc_objs(struct ceph_filelock,
4989 num_fcntl_locks + num_flock_locks,
4990 GFP_NOFS);
4991 if (!flocks) {
4992 err = -ENOMEM;
4993 goto out_err;
4994 }
4995 err = ceph_encode_locks_to_buffer(inode, flocks,
4996 num_fcntl_locks,
4997 num_flock_locks);
4998 if (err) {
4999 kfree(flocks);
5000 flocks = NULL;
5001 if (err == -ENOSPC)
5002 goto encode_again;
5003 goto out_err;
5004 }
5005 } else {
5006 kfree(flocks);
5007 flocks = NULL;
5008 }
5009
5010 if (recon_state->msg_version >= 3) {
5011 /* version, compat_version and struct_len */
5012 total_len += 2 * sizeof(u8) + sizeof(u32);
5013 struct_v = 2;
5014 }
5015 /*
5016 * number of encoded locks is stable, so copy to pagelist
5017 */
5018 struct_len = 2 * sizeof(u32) +
5019 (num_fcntl_locks + num_flock_locks) *
5020 sizeof(struct ceph_filelock);
5021 rec.v2.flock_len = cpu_to_le32(struct_len);
5022
5023 struct_len += sizeof(u32) + path_info.pathlen + sizeof(rec.v2);
5024
5025 if (struct_v >= 2)
5026 struct_len += sizeof(u64); /* snap_follows */
5027
5028 total_len += struct_len;
5029
5030 if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {
5031 err = send_reconnect_partial(recon_state);
5032 if (err)
5033 goto out_freeflocks;
5034 pagelist = recon_state->pagelist;
5035 }
5036
5037 err = ceph_pagelist_reserve(pagelist, total_len);
5038 if (err)
5039 goto out_freeflocks;
5040
5041 ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
5042 if (recon_state->msg_version >= 3) {
5043 ceph_pagelist_encode_8(pagelist, struct_v);
5044 ceph_pagelist_encode_8(pagelist, 1);
5045 ceph_pagelist_encode_32(pagelist, struct_len);
5046 }
5047 ceph_pagelist_encode_string(pagelist, (char *)path_info.path, path_info.pathlen);
5048 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
5049 ceph_locks_to_pagelist(flocks, pagelist,
5050 num_fcntl_locks, num_flock_locks);
5051 if (struct_v >= 2)
5052 ceph_pagelist_encode_64(pagelist, snap_follows);
5053out_freeflocks:
5054 kfree(flocks);
5055 } else {
5056 err = ceph_pagelist_reserve(pagelist,
5057 sizeof(u64) + sizeof(u32) +
5058 path_info.pathlen + sizeof(rec.v1));
5059 if (err)
5060 goto out_err;
5061
5062 ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
5063 ceph_pagelist_encode_string(pagelist, (char *)path_info.path, path_info.pathlen);
5064 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
5065 }
5066
5067out_err:
5068 ceph_mdsc_free_path_info(&path_info);
5069 if (!err)
5070 recon_state->nr_caps++;
5071 return err;
5072}
5073
5074static int encode_snap_realms(struct ceph_mds_client *mdsc,
5075 struct ceph_reconnect_state *recon_state)
5076{
5077 struct rb_node *p;
5078 struct ceph_pagelist *pagelist = recon_state->pagelist;
5079 struct ceph_client *cl = mdsc->fsc->client;
5080 int err = 0;
5081
5082 if (recon_state->msg_version >= 4) {
5083 err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);
5084 if (err < 0)
5085 goto fail;
5086 }
5087
5088 /*
5089 * snaprealms. we provide mds with the ino, seq (version), and
5090 * parent for all of our realms. If the mds has any newer info,
5091 * it will tell us.
5092 */
5093 for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
5094 struct ceph_snap_realm *realm =
5095 rb_entry(p, struct ceph_snap_realm, node);
5096 struct ceph_mds_snaprealm_reconnect sr_rec;
5097
5098 if (recon_state->msg_version >= 4) {
5099 size_t need = sizeof(u8) * 2 + sizeof(u32) +
5100 sizeof(sr_rec);
5101
5102 if (pagelist->length + need > RECONNECT_MAX_SIZE) {
5103 err = send_reconnect_partial(recon_state);
5104 if (err)
5105 goto fail;
5106 pagelist = recon_state->pagelist;
5107 }
5108
5109 err = ceph_pagelist_reserve(pagelist, need);
5110 if (err)
5111 goto fail;
5112
5113 ceph_pagelist_encode_8(pagelist, 1);
5114 ceph_pagelist_encode_8(pagelist, 1);
5115 ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));
5116 }
5117
5118 doutc(cl, " adding snap realm %llx seq %lld parent %llx\n",
5119 realm->ino, realm->seq, realm->parent_ino);
5120 sr_rec.ino = cpu_to_le64(realm->ino);
5121 sr_rec.seq = cpu_to_le64(realm->seq);
5122 sr_rec.parent = cpu_to_le64(realm->parent_ino);
5123
5124 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
5125 if (err)
5126 goto fail;
5127
5128 recon_state->nr_realms++;
5129 }
5130fail:
5131 return err;
5132}
5133
5134
5135/*
5136 * If an MDS fails and recovers, clients need to reconnect in order to
5137 * reestablish shared state. This includes all caps issued through
5138 * this session _and_ the snap_realm hierarchy. Because it's not
5139 * clear which snap realms the mds cares about, we send everything we
5140 * know about.. that ensures we'll then get any new info the
5141 * recovering MDS might have.
5142 *
5143 * This is a relatively heavyweight operation, but it's rare.
5144 */
5145static int send_mds_reconnect(struct ceph_mds_client *mdsc,
5146 struct ceph_mds_session *session)
5147{
5148 struct ceph_client *cl = mdsc->fsc->client;
5149 struct ceph_msg *reply;
5150 int mds = session->s_mds;
5151 int err = -ENOMEM;
5152 int old_state;
5153 struct ceph_reconnect_state recon_state = {
5154 .session = session,
5155 };
5156 LIST_HEAD(dispose);
5157
5158 recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
5159 if (!recon_state.pagelist)
5160 goto fail_nopagelist;
5161
5162 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
5163 if (!reply)
5164 goto fail_nomsg;
5165
5166 mutex_lock(&session->s_mutex);
5167
5168 /* Serialized by s_mutex against concurrent ceph_get_deleg_ino(). */
5169 xa_destroy(&session->s_delegated_inos);
5170 atomic_set(&session->s_num_deleg_inos, 0);
5171 if (session->s_state == CEPH_MDS_SESSION_CLOSED ||
5172 session->s_state == CEPH_MDS_SESSION_REJECTED) {
5173 pr_info_client(cl, "mds%d skipping reconnect, session %s\n",
5174 mds,
5175 ceph_session_state_name(session->s_state));
5176 mutex_unlock(&session->s_mutex);
5177 ceph_msg_put(reply);
5178 err = -ESTALE;
5179 goto fail_return;
5180 }
5181
5182 /* s_mutex -> mdsc->mutex matches cleanup_session_requests() order. */
5183 mutex_lock(&mdsc->mutex);
5184 if (mds >= mdsc->max_sessions || mdsc->sessions[mds] != session) {
5185 mutex_unlock(&mdsc->mutex);
5186 pr_info_client(cl,
5187 "mds%d skipping reconnect, session unregistered\n",
5188 mds);
5189 mutex_unlock(&session->s_mutex);
5190 ceph_msg_put(reply);
5191 err = -ENOENT;
5192 goto fail_return;
5193 }
5194 mutex_unlock(&mdsc->mutex);
5195
5196 pr_info_client(cl, "mds%d reconnect start\n", mds);
5197 old_state = session->s_state;
5198 session->s_state = CEPH_MDS_SESSION_RECONNECTING;
5199 session->s_seq = 0;
5200
5201 doutc(cl, "session %p state %s\n", session,
5202 ceph_session_state_name(session->s_state));
5203
5204 atomic_inc(&session->s_cap_gen);
5205
5206 spin_lock(&session->s_cap_lock);
5207 /* don't know if session is readonly */
5208 session->s_readonly = 0;
5209 /*
5210 * notify __ceph_remove_cap() that we are composing cap reconnect.
5211 * If a cap get released before being added to the cap reconnect,
5212 * __ceph_remove_cap() should skip queuing cap release.
5213 */
5214 session->s_cap_reconnect = 1;
5215 /* drop old cap expires; we're about to reestablish that state */
5216 detach_cap_releases(session, &dispose);
5217 spin_unlock(&session->s_cap_lock);
5218 dispose_cap_releases(mdsc, &dispose);
5219
5220 /* trim unused caps to reduce MDS's cache rejoin time */
5221 if (mdsc->fsc->sb->s_root)
5222 shrink_dcache_parent(mdsc->fsc->sb->s_root);
5223
5224 ceph_con_close(&session->s_con);
5225 ceph_con_open(&session->s_con,
5226 CEPH_ENTITY_TYPE_MDS, mds,
5227 ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
5228
5229 /* replay unsafe requests */
5230 replay_unsafe_requests(mdsc, session);
5231
5232 ceph_early_kick_flushing_caps(mdsc, session);
5233
5234 down_read(&mdsc->snap_rwsem);
5235
5236 /* placeholder for nr_caps */
5237 err = ceph_pagelist_encode_32(recon_state.pagelist, 0);
5238 if (err)
5239 goto fail_clear_cap_reconnect;
5240
5241 if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {
5242 recon_state.msg_version = 3;
5243 recon_state.allow_multi = true;
5244 } else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {
5245 recon_state.msg_version = 3;
5246 } else {
5247 recon_state.msg_version = 2;
5248 }
5249 /* traverse this session's caps */
5250 err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);
5251
5252 spin_lock(&session->s_cap_lock);
5253 session->s_cap_reconnect = 0;
5254 spin_unlock(&session->s_cap_lock);
5255
5256 if (err < 0)
5257 goto fail;
5258
5259 /* check if all realms can be encoded into current message */
5260 if (mdsc->num_snap_realms) {
5261 size_t total_len =
5262 recon_state.pagelist->length +
5263 mdsc->num_snap_realms *
5264 sizeof(struct ceph_mds_snaprealm_reconnect);
5265 if (recon_state.msg_version >= 4) {
5266 /* number of realms */
5267 total_len += sizeof(u32);
5268 /* version, compat_version and struct_len */
5269 total_len += mdsc->num_snap_realms *
5270 (2 * sizeof(u8) + sizeof(u32));
5271 }
5272 if (total_len > RECONNECT_MAX_SIZE) {
5273 if (!recon_state.allow_multi) {
5274 err = -ENOSPC;
5275 goto fail;
5276 }
5277 if (recon_state.nr_caps) {
5278 err = send_reconnect_partial(&recon_state);
5279 if (err)
5280 goto fail;
5281 }
5282 recon_state.msg_version = 5;
5283 }
5284 }
5285
5286 err = encode_snap_realms(mdsc, &recon_state);
5287 if (err < 0)
5288 goto fail;
5289
5290 if (recon_state.msg_version >= 5) {
5291 err = ceph_pagelist_encode_8(recon_state.pagelist, 0);
5292 if (err < 0)
5293 goto fail;
5294 }
5295
5296 if (recon_state.nr_caps || recon_state.nr_realms) {
5297 struct page *page =
5298 list_first_entry(&recon_state.pagelist->head,
5299 struct page, lru);
5300 __le32 *addr = kmap_atomic(page);
5301 if (recon_state.nr_caps) {
5302 WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);
5303 *addr = cpu_to_le32(recon_state.nr_caps);
5304 } else if (recon_state.msg_version >= 4) {
5305 *(addr + 1) = cpu_to_le32(recon_state.nr_realms);
5306 }
5307 kunmap_atomic(addr);
5308 }
5309
5310 reply->hdr.version = cpu_to_le16(recon_state.msg_version);
5311 if (recon_state.msg_version >= 4)
5312 reply->hdr.compat_version = cpu_to_le16(4);
5313
5314 reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);
5315 ceph_msg_data_add_pagelist(reply, recon_state.pagelist);
5316
5317 ceph_con_send(&session->s_con, reply);
5318
5319 mutex_unlock(&session->s_mutex);
5320
5321 mutex_lock(&mdsc->mutex);
5322 __wake_requests(mdsc, &session->s_waiting);
5323 mutex_unlock(&mdsc->mutex);
5324
5325 up_read(&mdsc->snap_rwsem);
5326 ceph_pagelist_release(recon_state.pagelist);
5327 return 0;
5328
5329fail_clear_cap_reconnect:
5330 spin_lock(&session->s_cap_lock);
5331 session->s_cap_reconnect = 0;
5332 spin_unlock(&session->s_cap_lock);
5333fail:
5334 ceph_msg_put(reply);
5335 up_read(&mdsc->snap_rwsem);
5336 /*
5337 * Restore prior session state so map-driven reconnect logic
5338 * (check_new_map) can retry. Without this, a transient build
5339 * failure strands the session in RECONNECTING indefinitely.
5340 */
5341 session->s_state = old_state;
5342 mutex_unlock(&session->s_mutex);
5343fail_nomsg:
5344 ceph_pagelist_release(recon_state.pagelist);
5345fail_nopagelist:
5346 pr_err_client(cl, "error %d preparing reconnect for mds%d\n",
5347 err, mds);
5348 return err;
5349
5350fail_return:
5351 /*
5352 * Early-exit path for expected concurrent-teardown races
5353 * (-ESTALE for closed/rejected sessions, -ENOENT for
5354 * unregistered sessions). Skip the pr_err_client diagnostic
5355 * since these are not genuine reconnect build failures.
5356 */
5357 ceph_pagelist_release(recon_state.pagelist);
5358 return err;
5359}
5360
5361const char *ceph_reset_phase_name(enum ceph_client_reset_phase phase)
5362{
5363 switch (phase) {
5364 case CEPH_CLIENT_RESET_IDLE: return "idle";
5365 case CEPH_CLIENT_RESET_QUIESCING: return "quiescing";
5366 case CEPH_CLIENT_RESET_DRAINING: return "draining";
5367 case CEPH_CLIENT_RESET_TEARDOWN: return "teardown";
5368 default: return "unknown";
5369 }
5370}
5371
5372/**
5373 * ceph_mdsc_wait_for_reset - wait for an active reset to complete
5374 * @mdsc: MDS client
5375 *
5376 * Returns 0 if reset completed successfully or no reset was active.
5377 * Returns -EAGAIN if reset completed with an error, signalling the
5378 * caller to retry. The internal error (e.g. -ENOMEM) is not propagated
5379 * because callers like open() or flock() have no way to act on
5380 * work-function internals. The detailed error is available via debugfs
5381 * reset/status and tracepoints.
5382 * Returns -ETIMEDOUT if we timed out waiting.
5383 * Returns -ERESTARTSYS if interrupted by signal.
5384 */
5385int ceph_mdsc_wait_for_reset(struct ceph_mds_client *mdsc)
5386{
5387 struct ceph_client_reset_state *st = &mdsc->reset_state;
5388 struct ceph_client *cl = mdsc->fsc->client;
5389 unsigned long deadline = jiffies + CEPH_CLIENT_RESET_WAIT_TIMEOUT_SEC * HZ;
5390 int blocked_count;
5391 long remaining;
5392 long wait_ret;
5393 int ret;
5394
5395 if (ceph_reset_is_idle(st))
5396 return 0;
5397
5398 blocked_count = atomic_inc_return(&st->blocked_requests);
5399 doutc(cl, "request blocked during reset, %d total blocked\n",
5400 blocked_count);
5401 trace_ceph_client_reset_blocked(mdsc, blocked_count);
5402
5403retry:
5404 remaining = max_t(long, deadline - jiffies, 1);
5405 wait_ret = wait_event_interruptible_timeout(st->blocked_wq,
5406 ceph_reset_is_idle(st),
5407 remaining);
5408
5409 if (wait_ret == 0) {
5410 atomic_dec(&st->blocked_requests);
5411 pr_warn_client(cl, "timed out waiting for reset to complete\n");
5412 trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT);
5413 return -ETIMEDOUT;
5414 }
5415 if (wait_ret < 0) {
5416 atomic_dec(&st->blocked_requests);
5417 trace_ceph_client_reset_unblocked(mdsc, (int)wait_ret);
5418 return (int)wait_ret; /* -ERESTARTSYS */
5419 }
5420
5421 /*
5422 * Verify phase is still IDLE under the lock. If another reset
5423 * was scheduled between the wake-up and this check, loop back
5424 * and wait for it to finish rather than returning a stale result.
5425 */
5426 spin_lock(&st->lock);
5427 if (st->phase != CEPH_CLIENT_RESET_IDLE) {
5428 spin_unlock(&st->lock);
5429 if (time_before(jiffies, deadline))
5430 goto retry;
5431 atomic_dec(&st->blocked_requests);
5432 trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT);
5433 return -ETIMEDOUT;
5434 }
5435 ret = st->last_errno;
5436 spin_unlock(&st->lock);
5437
5438 atomic_dec(&st->blocked_requests);
5439 trace_ceph_client_reset_unblocked(mdsc, ret);
5440 return ret ? -EAGAIN : 0;
5441}
5442
5443static void ceph_mdsc_reset_complete(struct ceph_mds_client *mdsc, int ret)
5444{
5445 struct ceph_client_reset_state *st = &mdsc->reset_state;
5446
5447 spin_lock(&st->lock);
5448 /*
5449 * If destroy already marked us as shut down, it owns the
5450 * final bookkeeping and waiter wakeup. Just bail so we
5451 * don't overwrite its state.
5452 */
5453 if (st->shutdown) {
5454 spin_unlock(&st->lock);
5455 return;
5456 }
5457 st->last_finish = jiffies;
5458 st->last_errno = ret;
5459 st->phase = CEPH_CLIENT_RESET_IDLE;
5460 if (ret)
5461 st->failure_count++;
5462 else
5463 st->success_count++;
5464 spin_unlock(&st->lock);
5465
5466 /* Wake up all requests that were blocked waiting for reset */
5467 wake_up_all(&st->blocked_wq);
5468
5469 trace_ceph_client_reset_complete(mdsc, ret);
5470}
5471
5472static void ceph_mdsc_reset_workfn(struct work_struct *work)
5473{
5474 struct ceph_mds_client *mdsc =
5475 container_of(work, struct ceph_mds_client, reset_work);
5476 struct ceph_client_reset_state *st = &mdsc->reset_state;
5477 struct ceph_client *cl = mdsc->fsc->client;
5478 struct ceph_mds_session **sessions = NULL;
5479 char reason[CEPH_CLIENT_RESET_REASON_LEN];
5480 unsigned long drain_deadline;
5481 int max_sessions, i, n = 0, torn_down = 0;
5482 int ret = 0;
5483
5484 spin_lock(&st->lock);
5485 strscpy(reason, st->last_reason, sizeof(reason));
5486 spin_unlock(&st->lock);
5487
5488 mutex_lock(&mdsc->mutex);
5489 max_sessions = mdsc->max_sessions;
5490 if (max_sessions <= 0) {
5491 mutex_unlock(&mdsc->mutex);
5492 goto out_complete;
5493 }
5494
5495 sessions = kzalloc_objs(*sessions, max_sessions);
5496 if (!sessions) {
5497 mutex_unlock(&mdsc->mutex);
5498 ret = -ENOMEM;
5499 pr_err_client(cl,
5500 "manual session reset failed to allocate session array\n");
5501 ceph_mdsc_reset_complete(mdsc, ret);
5502 return;
5503 }
5504
5505 for (i = 0; i < max_sessions; i++) {
5506 struct ceph_mds_session *session = mdsc->sessions[i];
5507
5508 if (!session)
5509 continue;
5510
5511 /*
5512 * Read session state without s_mutex to avoid nesting
5513 * mdsc->mutex -> s_mutex, which would invert the
5514 * s_mutex -> mdsc->mutex order used by
5515 * cleanup_session_requests(). s_state is an int
5516 * so loads are atomic; the teardown loop below
5517 * handles races with concurrent state transitions.
5518 */
5519 switch (READ_ONCE(session->s_state)) {
5520 case CEPH_MDS_SESSION_OPEN:
5521 case CEPH_MDS_SESSION_HUNG:
5522 case CEPH_MDS_SESSION_OPENING:
5523 case CEPH_MDS_SESSION_RESTARTING:
5524 case CEPH_MDS_SESSION_RECONNECTING:
5525 case CEPH_MDS_SESSION_CLOSING:
5526 sessions[n++] = ceph_get_mds_session(session);
5527 break;
5528 default:
5529 pr_info_client(cl,
5530 "mds%d in state %s, skipping reset\n",
5531 session->s_mds,
5532 ceph_session_state_name(session->s_state));
5533 break;
5534 }
5535 }
5536 mutex_unlock(&mdsc->mutex);
5537
5538 pr_info_client(cl,
5539 "manual session reset executing (sessions=%d, reason=\"%s\")\n",
5540 n, reason);
5541
5542 if (n == 0) {
5543 kfree(sessions);
5544 goto out_complete;
5545 }
5546
5547 spin_lock(&st->lock);
5548 if (st->shutdown) {
5549 spin_unlock(&st->lock);
5550 goto out_sessions;
5551 }
5552 st->phase = CEPH_CLIENT_RESET_DRAINING;
5553 spin_unlock(&st->lock);
5554
5555 /*
5556 * Best-effort drain: flush dirty state while sessions are still
5557 * alive. New requests are blocked while phase != IDLE.
5558 * The sessions are functional, so non-stuck state drains normally.
5559 * Stuck state (the cause of the stalemate the operator is trying
5560 * to break) will not drain -- that is expected, and we proceed to
5561 * forced teardown after the timeout.
5562 *
5563 * Four things are drained:
5564 * 1. MDS journal -- send_flush_mdlog asks each MDS to journal
5565 * pending unsafe operations (creates, renames, setattrs).
5566 * 2. Unsafe requests -- bounded wait for each unsafe write
5567 * request to reach safe status via r_safe_completion.
5568 * 3. Dirty caps -- ceph_flush_dirty_caps triggers cap flush on
5569 * all sessions. Non-stuck caps flush in milliseconds.
5570 * 4. Cap releases -- push pending cap release messages.
5571 *
5572 * The unsafe-request wait and cap-flush wait below provide
5573 * the bounded drain window during which all categories can
5574 * make progress.
5575 */
5576 for (i = 0; i < n; i++)
5577 send_flush_mdlog(sessions[i]);
5578
5579 /*
5580 * Both drain legs (unsafe requests and cap flushes) share a
5581 * single deadline so the total drain time is bounded at
5582 * CEPH_CLIENT_RESET_DRAIN_SEC.
5583 */
5584 drain_deadline = jiffies + CEPH_CLIENT_RESET_DRAIN_SEC * HZ;
5585
5586 /*
5587 * Wait for unsafe write requests (creates, renames, setattrs)
5588 * to reach safe status. Uses the same pattern as
5589 * flush_mdlog_and_wait_mdsc_unsafe_requests() but bounded by
5590 * the shared drain deadline. Requests that do not complete within
5591 * the window are force-dropped during teardown.
5592 */
5593 {
5594 struct ceph_mds_request *req;
5595 struct rb_node *rn;
5596 u64 last_tid;
5597
5598 mutex_lock(&mdsc->mutex);
5599 last_tid = mdsc->last_tid;
5600 mutex_unlock(&mdsc->mutex);
5601
5602 mutex_lock(&mdsc->mutex);
5603 rn = rb_first(&mdsc->request_tree);
5604 while (rn) {
5605 req = rb_entry(rn, struct ceph_mds_request, r_node);
5606 if (req->r_tid > last_tid)
5607 break;
5608 if (req->r_op == CEPH_MDS_OP_SETFILELOCK ||
5609 !(req->r_op & CEPH_MDS_OP_WRITE)) {
5610 rn = rb_next(rn);
5611 continue;
5612 }
5613 ceph_mdsc_get_request(req);
5614 mutex_unlock(&mdsc->mutex);
5615
5616 wait_for_completion_timeout(&req->r_safe_completion,
5617 max_t(long, drain_deadline - jiffies, 1));
5618
5619 mutex_lock(&mdsc->mutex);
5620 ceph_mdsc_put_request(req);
5621 if (time_after(jiffies, drain_deadline))
5622 break;
5623 rn = rb_first(&mdsc->request_tree);
5624 }
5625 mutex_unlock(&mdsc->mutex);
5626
5627 if (time_after_eq(jiffies, drain_deadline))
5628 WRITE_ONCE(st->drain_timed_out, true);
5629 }
5630
5631 ceph_flush_dirty_caps(mdsc);
5632 ceph_flush_cap_releases(mdsc);
5633
5634 spin_lock(&mdsc->cap_dirty_lock);
5635 if (!list_empty(&mdsc->cap_flush_list)) {
5636 struct ceph_cap_flush *cf =
5637 list_last_entry(&mdsc->cap_flush_list,
5638 struct ceph_cap_flush, g_list);
5639 u64 want_flush = mdsc->last_cap_flush_tid;
5640 long drain_ret;
5641
5642 /*
5643 * Setting wake on the last entry is sufficient: flush
5644 * entries complete in order, so when this entry finishes
5645 * all earlier ones are already done.
5646 */
5647 cf->wake = true;
5648 spin_unlock(&mdsc->cap_dirty_lock);
5649 pr_info_client(cl,
5650 "draining (want_flush=%llu, %d sessions)\n",
5651 want_flush, n);
5652 drain_ret = wait_event_timeout(mdsc->cap_flushing_wq,
5653 check_caps_flush(mdsc,
5654 want_flush),
5655 max_t(long,
5656 drain_deadline - jiffies,
5657 1));
5658 if (drain_ret == 0) {
5659 pr_info_client(cl,
5660 "drain timed out, proceeding with forced teardown\n");
5661 WRITE_ONCE(st->drain_timed_out, true);
5662 } else {
5663 pr_info_client(cl, "drain completed successfully\n");
5664 }
5665 } else {
5666 spin_unlock(&mdsc->cap_dirty_lock);
5667 }
5668
5669 spin_lock(&st->lock);
5670 if (st->shutdown) {
5671 spin_unlock(&st->lock);
5672 goto out_sessions;
5673 }
5674 st->phase = CEPH_CLIENT_RESET_TEARDOWN;
5675 spin_unlock(&st->lock);
5676
5677 /*
5678 * Ask each MDS to close the session before we tear it down
5679 * locally. Without this the MDS sees only a connection drop and
5680 * waits for the client to reconnect (up to session_autoclose
5681 * seconds) before evicting the session and releasing locks.
5682 *
5683 * Reuse the normal close machinery so the session state/sequence
5684 * snapshot is serialized under s_mutex and a racing s_seq bump
5685 * retransmits REQUEST_CLOSE while the session remains CLOSING.
5686 * We send all close requests first, then yield briefly to let the
5687 * network stack transmit them before __unregister_session()
5688 * closes the connections.
5689 */
5690 for (i = 0; i < n; i++) {
5691 int err;
5692
5693 mutex_lock(&sessions[i]->s_mutex);
5694 err = __close_session(mdsc, sessions[i]);
5695 mutex_unlock(&sessions[i]->s_mutex);
5696 if (err < 0)
5697 pr_warn_client(cl,
5698 "mds%d failed to queue close request before reset: %d\n",
5699 sessions[i]->s_mds, err);
5700 }
5701 /*
5702 * Best-effort grace period: yield briefly so the network stack
5703 * can transmit the queued REQUEST_CLOSE messages before we tear
5704 * down connections. Not a correctness requirement -- the MDS
5705 * will still evict via session_autoclose if it never receives
5706 * the close request.
5707 *
5708 * Event-based waiting is not viable here: there is no completion
5709 * event for "message left the NIC," and waiting for the MDS
5710 * SESSION_CLOSE response would re-create the stalemate that the
5711 * reset is meant to break.
5712 */
5713 if (n > 0)
5714 msleep(CEPH_CLIENT_RESET_CLOSE_GRACE_MS);
5715
5716 /*
5717 * Tear down each session: close the connection, remove all
5718 * caps, clean up requests, then kick pending requests so they
5719 * re-open a fresh session on the next attempt.
5720 *
5721 * This is modeled on the check_new_map() forced-close path
5722 * for stopped MDS ranks - a proven pattern for hard session
5723 * teardown. We do NOT attempt send_mds_reconnect() because
5724 * the MDS only accepts reconnects during its own RECONNECT
5725 * phase (after MDS restart), not from an active client.
5726 *
5727 * Any state that did not drain (caps that didn't flush, unsafe
5728 * requests that the MDS didn't journal) is force-dropped here.
5729 * This is intentional: that state is stuck and is the reason
5730 * the operator triggered the reset.
5731 */
5732 for (i = 0; i < n; i++) {
5733 int mds = sessions[i]->s_mds;
5734
5735 pr_info_client(cl, "mds%d resetting session\n", mds);
5736
5737 mutex_lock(&mdsc->mutex);
5738 if (mds >= mdsc->max_sessions ||
5739 mdsc->sessions[mds] != sessions[i]) {
5740 pr_info_client(cl,
5741 "mds%d session already torn down, skipping\n",
5742 mds);
5743 mutex_unlock(&mdsc->mutex);
5744 ceph_put_mds_session(sessions[i]);
5745 sessions[i] = NULL;
5746 continue;
5747 }
5748 sessions[i]->s_state = CEPH_MDS_SESSION_CLOSED;
5749 __unregister_session(mdsc, sessions[i]);
5750 __wake_requests(mdsc, &sessions[i]->s_waiting);
5751 mutex_unlock(&mdsc->mutex);
5752
5753 mutex_lock(&sessions[i]->s_mutex);
5754 cleanup_session_requests(mdsc, sessions[i]);
5755 remove_session_caps(sessions[i]);
5756 mutex_unlock(&sessions[i]->s_mutex);
5757
5758 wake_up_all(&mdsc->session_close_wq);
5759
5760 ceph_put_mds_session(sessions[i]);
5761
5762 mutex_lock(&mdsc->mutex);
5763 kick_requests(mdsc, mds);
5764 mutex_unlock(&mdsc->mutex);
5765
5766 torn_down++;
5767 pr_info_client(cl, "mds%d session reset complete\n", mds);
5768 }
5769
5770 kfree(sessions);
5771
5772 spin_lock(&st->lock);
5773 st->sessions_reset = torn_down;
5774 spin_unlock(&st->lock);
5775
5776out_complete:
5777 ceph_mdsc_reset_complete(mdsc, ret);
5778 return;
5779
5780out_sessions:
5781 /* shutdown == true: ceph_mdsc_destroy() owns the final transition. */
5782 for (i = 0; i < n; i++)
5783 ceph_put_mds_session(sessions[i]);
5784 kfree(sessions);
5785}
5786
5787int ceph_mdsc_schedule_reset(struct ceph_mds_client *mdsc,
5788 const char *reason)
5789{
5790 struct ceph_client_reset_state *st = &mdsc->reset_state;
5791 struct ceph_fs_client *fsc = mdsc->fsc;
5792 const char *msg = (reason && reason[0]) ? reason : "manual";
5793 int mount_state;
5794
5795 mount_state = READ_ONCE(fsc->mount_state);
5796 if (mount_state != CEPH_MOUNT_MOUNTED) {
5797 pr_warn_client(fsc->client,
5798 "reset rejected: mount_state=%d (not mounted)\n",
5799 mount_state);
5800 return -EINVAL;
5801 }
5802
5803 spin_lock(&st->lock);
5804 if (st->phase != CEPH_CLIENT_RESET_IDLE) {
5805 spin_unlock(&st->lock);
5806 return -EBUSY;
5807 }
5808
5809 st->phase = CEPH_CLIENT_RESET_QUIESCING;
5810 st->last_start = jiffies;
5811 st->last_errno = 0;
5812 st->drain_timed_out = false;
5813 st->sessions_reset = 0;
5814 st->trigger_count++;
5815 strscpy(st->last_reason, msg, sizeof(st->last_reason));
5816 spin_unlock(&st->lock);
5817
5818 if (WARN_ON_ONCE(!queue_work(system_dfl_wq, &mdsc->reset_work))) {
5819 spin_lock(&st->lock);
5820 st->phase = CEPH_CLIENT_RESET_IDLE;
5821 st->last_errno = -EALREADY;
5822 st->last_finish = jiffies;
5823 st->failure_count++;
5824 spin_unlock(&st->lock);
5825 wake_up_all(&st->blocked_wq);
5826 return -EALREADY;
5827 }
5828
5829 pr_info_client(mdsc->fsc->client,
5830 "manual session reset scheduled (reason=\"%s\")\n",
5831 msg);
5832 trace_ceph_client_reset_schedule(mdsc, msg);
5833 return 0;
5834}
5835
5836
5837/*
5838 * compare old and new mdsmaps, kicking requests
5839 * and closing out old connections as necessary
5840 *
5841 * called under mdsc->mutex.
5842 */
5843static void check_new_map(struct ceph_mds_client *mdsc,
5844 struct ceph_mdsmap *newmap,
5845 struct ceph_mdsmap *oldmap)
5846{
5847 int i, j, err;
5848 int oldstate, newstate;
5849 struct ceph_mds_session *s;
5850 unsigned long targets[DIV_ROUND_UP(CEPH_MAX_MDS, sizeof(unsigned long))] = {0};
5851 struct ceph_client *cl = mdsc->fsc->client;
5852
5853 doutc(cl, "new %u old %u\n", newmap->m_epoch, oldmap->m_epoch);
5854
5855 if (newmap->m_info) {
5856 for (i = 0; i < newmap->possible_max_rank; i++) {
5857 for (j = 0; j < newmap->m_info[i].num_export_targets; j++)
5858 set_bit(newmap->m_info[i].export_targets[j], targets);
5859 }
5860 }
5861
5862 for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {
5863 if (!mdsc->sessions[i])
5864 continue;
5865 s = mdsc->sessions[i];
5866 oldstate = ceph_mdsmap_get_state(oldmap, i);
5867 newstate = ceph_mdsmap_get_state(newmap, i);
5868
5869 doutc(cl, "mds%d state %s%s -> %s%s (session %s)\n",
5870 i, ceph_mds_state_name(oldstate),
5871 ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
5872 ceph_mds_state_name(newstate),
5873 ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
5874 ceph_session_state_name(s->s_state));
5875
5876 if (i >= newmap->possible_max_rank) {
5877 /* force close session for stopped mds */
5878 ceph_get_mds_session(s);
5879 __unregister_session(mdsc, s);
5880 __wake_requests(mdsc, &s->s_waiting);
5881 mutex_unlock(&mdsc->mutex);
5882
5883 mutex_lock(&s->s_mutex);
5884 cleanup_session_requests(mdsc, s);
5885 remove_session_caps(s);
5886 mutex_unlock(&s->s_mutex);
5887
5888 ceph_put_mds_session(s);
5889
5890 mutex_lock(&mdsc->mutex);
5891 kick_requests(mdsc, i);
5892 continue;
5893 }
5894
5895 if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
5896 ceph_mdsmap_get_addr(newmap, i),
5897 sizeof(struct ceph_entity_addr))) {
5898 /* just close it */
5899 ceph_get_mds_session(s);
5900 mutex_unlock(&mdsc->mutex);
5901 mutex_lock(&s->s_mutex);
5902 mutex_lock(&mdsc->mutex);
5903 ceph_put_mds_session(s);
5904 ceph_con_close(&s->s_con);
5905 mutex_unlock(&s->s_mutex);
5906 s->s_state = CEPH_MDS_SESSION_RESTARTING;
5907 } else if (oldstate == newstate) {
5908 continue; /* nothing new with this mds */
5909 }
5910
5911 /*
5912 * send reconnect?
5913 */
5914 if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
5915 newstate >= CEPH_MDS_STATE_RECONNECT) {
5916 int rc;
5917
5918 ceph_get_mds_session(s);
5919 mutex_unlock(&mdsc->mutex);
5920 clear_bit(i, targets);
5921 rc = send_mds_reconnect(mdsc, s);
5922 if (rc)
5923 pr_warn_client(cl,
5924 "mds%d reconnect failed: %d\n",
5925 i, rc);
5926 mutex_lock(&mdsc->mutex);
5927 ceph_put_mds_session(s);
5928 }
5929
5930 /*
5931 * kick request on any mds that has gone active.
5932 */
5933 if (oldstate < CEPH_MDS_STATE_ACTIVE &&
5934 newstate >= CEPH_MDS_STATE_ACTIVE) {
5935 if (oldstate != CEPH_MDS_STATE_CREATING &&
5936 oldstate != CEPH_MDS_STATE_STARTING)
5937 pr_info_client(cl, "mds%d recovery completed\n",
5938 s->s_mds);
5939 kick_requests(mdsc, i);
5940 ceph_get_mds_session(s);
5941 mutex_unlock(&mdsc->mutex);
5942 mutex_lock(&s->s_mutex);
5943 mutex_lock(&mdsc->mutex);
5944 ceph_put_mds_session(s);
5945 ceph_kick_flushing_caps(mdsc, s);
5946 mutex_unlock(&s->s_mutex);
5947 wake_up_session_caps(s, RECONNECT);
5948 }
5949 }
5950
5951 /*
5952 * Only open and reconnect sessions that don't exist yet.
5953 */
5954 for (i = 0; i < newmap->possible_max_rank; i++) {
5955 /*
5956 * In case the import MDS is crashed just after
5957 * the EImportStart journal is flushed, so when
5958 * a standby MDS takes over it and is replaying
5959 * the EImportStart journal the new MDS daemon
5960 * will wait the client to reconnect it, but the
5961 * client may never register/open the session yet.
5962 *
5963 * Will try to reconnect that MDS daemon if the
5964 * rank number is in the export targets array and
5965 * is the up:reconnect state.
5966 */
5967 newstate = ceph_mdsmap_get_state(newmap, i);
5968 if (!test_bit(i, targets) || newstate != CEPH_MDS_STATE_RECONNECT)
5969 continue;
5970
5971 /*
5972 * The session maybe registered and opened by some
5973 * requests which were choosing random MDSes during
5974 * the mdsc->mutex's unlock/lock gap below in rare
5975 * case. But the related MDS daemon will just queue
5976 * that requests and be still waiting for the client's
5977 * reconnection request in up:reconnect state.
5978 */
5979 s = __ceph_lookup_mds_session(mdsc, i);
5980 if (likely(!s)) {
5981 s = __open_export_target_session(mdsc, i);
5982 if (IS_ERR(s)) {
5983 err = PTR_ERR(s);
5984 pr_err_client(cl,
5985 "failed to open export target session, err %d\n",
5986 err);
5987 continue;
5988 }
5989 }
5990 doutc(cl, "send reconnect to export target mds.%d\n", i);
5991 mutex_unlock(&mdsc->mutex);
5992 err = send_mds_reconnect(mdsc, s);
5993 if (err)
5994 pr_warn_client(cl,
5995 "mds%d export target reconnect failed: %d\n",
5996 i, err);
5997 ceph_put_mds_session(s);
5998 mutex_lock(&mdsc->mutex);
5999 }
6000
6001 for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {
6002 s = mdsc->sessions[i];
6003 if (!s)
6004 continue;
6005 if (!ceph_mdsmap_is_laggy(newmap, i))
6006 continue;
6007 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
6008 s->s_state == CEPH_MDS_SESSION_HUNG ||
6009 s->s_state == CEPH_MDS_SESSION_CLOSING) {
6010 doutc(cl, " connecting to export targets of laggy mds%d\n", i);
6011 __open_export_target_sessions(mdsc, s);
6012 }
6013 }
6014}
6015
6016
6017
6018/*
6019 * leases
6020 */
6021
6022/*
6023 * caller must hold session s_mutex, dentry->d_lock
6024 */
6025void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
6026{
6027 struct ceph_dentry_info *di = ceph_dentry(dentry);
6028
6029 ceph_put_mds_session(di->lease_session);
6030 di->lease_session = NULL;
6031}
6032
6033static void handle_lease(struct ceph_mds_client *mdsc,
6034 struct ceph_mds_session *session,
6035 struct ceph_msg *msg)
6036{
6037 struct ceph_client *cl = mdsc->fsc->client;
6038 struct super_block *sb = mdsc->fsc->sb;
6039 struct inode *inode;
6040 struct dentry *parent, *dentry;
6041 struct ceph_dentry_info *di;
6042 int mds = session->s_mds;
6043 struct ceph_mds_lease *h = msg->front.iov_base;
6044 u32 seq;
6045 struct ceph_vino vino;
6046 struct qstr dname;
6047 int release = 0;
6048
6049 doutc(cl, "from mds%d\n", mds);
6050
6051 if (!ceph_inc_mds_stopping_blocker(mdsc, session))
6052 return;
6053
6054 /* decode */
6055 if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
6056 goto bad;
6057 vino.ino = le64_to_cpu(h->ino);
6058 vino.snap = CEPH_NOSNAP;
6059 seq = le32_to_cpu(h->seq);
6060 dname.len = get_unaligned_le32(h + 1);
6061 if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len)
6062 goto bad;
6063 dname.name = (void *)(h + 1) + sizeof(u32);
6064
6065 /* lookup inode */
6066 inode = ceph_find_inode(sb, vino);
6067 doutc(cl, "%s, ino %llx %p %.*s\n", ceph_lease_op_name(h->action),
6068 vino.ino, inode, dname.len, dname.name);
6069
6070 mutex_lock(&session->s_mutex);
6071 if (!inode) {
6072 doutc(cl, "no inode %llx\n", vino.ino);
6073 goto release;
6074 }
6075
6076 /* dentry */
6077 parent = d_find_alias(inode);
6078 if (!parent) {
6079 doutc(cl, "no parent dentry on inode %p\n", inode);
6080 WARN_ON(1);
6081 goto release; /* hrm... */
6082 }
6083 dname.hash = full_name_hash(parent, dname.name, dname.len);
6084 dentry = d_lookup(parent, &dname);
6085 dput(parent);
6086 if (!dentry)
6087 goto release;
6088
6089 spin_lock(&dentry->d_lock);
6090 di = ceph_dentry(dentry);
6091 switch (h->action) {
6092 case CEPH_MDS_LEASE_REVOKE:
6093 if (di->lease_session == session) {
6094 if (ceph_seq_cmp(di->lease_seq, seq) > 0)
6095 h->seq = cpu_to_le32(di->lease_seq);
6096 __ceph_mdsc_drop_dentry_lease(dentry);
6097 }
6098 release = 1;
6099 break;
6100
6101 case CEPH_MDS_LEASE_RENEW:
6102 if (di->lease_session == session &&
6103 di->lease_gen == atomic_read(&session->s_cap_gen) &&
6104 di->lease_renew_from &&
6105 di->lease_renew_after == 0) {
6106 unsigned long duration =
6107 msecs_to_jiffies(le32_to_cpu(h->duration_ms));
6108
6109 di->lease_seq = seq;
6110 di->time = di->lease_renew_from + duration;
6111 di->lease_renew_after = di->lease_renew_from +
6112 (duration >> 1);
6113 di->lease_renew_from = 0;
6114 }
6115 break;
6116 }
6117 spin_unlock(&dentry->d_lock);
6118 dput(dentry);
6119
6120 if (!release)
6121 goto out;
6122
6123release:
6124 /* let's just reuse the same message */
6125 h->action = CEPH_MDS_LEASE_REVOKE_ACK;
6126 ceph_msg_get(msg);
6127 ceph_con_send(&session->s_con, msg);
6128
6129out:
6130 mutex_unlock(&session->s_mutex);
6131 iput(inode);
6132
6133 ceph_dec_mds_stopping_blocker(mdsc);
6134 return;
6135
6136bad:
6137 ceph_dec_mds_stopping_blocker(mdsc);
6138
6139 pr_err_client(cl, "corrupt lease message\n");
6140 ceph_msg_dump(msg);
6141}
6142
6143void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
6144 struct dentry *dentry, char action,
6145 u32 seq)
6146{
6147 struct ceph_client *cl = session->s_mdsc->fsc->client;
6148 struct ceph_msg *msg;
6149 struct ceph_mds_lease *lease;
6150 struct inode *dir;
6151 int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;
6152
6153 doutc(cl, "identry %p %s to mds%d\n", dentry, ceph_lease_op_name(action),
6154 session->s_mds);
6155
6156 msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
6157 if (!msg)
6158 return;
6159 lease = msg->front.iov_base;
6160 lease->action = action;
6161 lease->seq = cpu_to_le32(seq);
6162
6163 spin_lock(&dentry->d_lock);
6164 dir = d_inode(dentry->d_parent);
6165 lease->ino = cpu_to_le64(ceph_ino(dir));
6166 lease->first = lease->last = cpu_to_le64(ceph_snap(dir));
6167
6168 put_unaligned_le32(dentry->d_name.len, lease + 1);
6169 memcpy((void *)(lease + 1) + 4,
6170 dentry->d_name.name, dentry->d_name.len);
6171 spin_unlock(&dentry->d_lock);
6172
6173 ceph_con_send(&session->s_con, msg);
6174}
6175
6176/*
6177 * lock unlock the session, to wait ongoing session activities
6178 */
6179static void lock_unlock_session(struct ceph_mds_session *s)
6180{
6181 mutex_lock(&s->s_mutex);
6182 mutex_unlock(&s->s_mutex);
6183}
6184
6185static void maybe_recover_session(struct ceph_mds_client *mdsc)
6186{
6187 struct ceph_client *cl = mdsc->fsc->client;
6188 struct ceph_fs_client *fsc = mdsc->fsc;
6189
6190 if (!ceph_test_mount_opt(fsc, CLEANRECOVER))
6191 return;
6192
6193 if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)
6194 return;
6195
6196 if (!READ_ONCE(fsc->blocklisted))
6197 return;
6198
6199 pr_info_client(cl, "auto reconnect after blocklisted\n");
6200 ceph_force_reconnect(fsc->sb);
6201}
6202
6203bool check_session_state(struct ceph_mds_session *s)
6204{
6205 struct ceph_client *cl = s->s_mdsc->fsc->client;
6206
6207 switch (s->s_state) {
6208 case CEPH_MDS_SESSION_OPEN:
6209 if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
6210 s->s_state = CEPH_MDS_SESSION_HUNG;
6211 pr_info_client(cl, "mds%d hung\n", s->s_mds);
6212 }
6213 break;
6214 case CEPH_MDS_SESSION_CLOSING:
6215 case CEPH_MDS_SESSION_NEW:
6216 case CEPH_MDS_SESSION_RESTARTING:
6217 case CEPH_MDS_SESSION_CLOSED:
6218 case CEPH_MDS_SESSION_REJECTED:
6219 return false;
6220 }
6221
6222 return true;
6223}
6224
6225/*
6226 * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,
6227 * then we need to retransmit that request.
6228 */
6229void inc_session_sequence(struct ceph_mds_session *s)
6230{
6231 struct ceph_client *cl = s->s_mdsc->fsc->client;
6232
6233 lockdep_assert_held(&s->s_mutex);
6234
6235 s->s_seq++;
6236
6237 if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
6238 int ret;
6239
6240 doutc(cl, "resending session close request for mds%d\n", s->s_mds);
6241 ret = request_close_session(s);
6242 if (ret < 0)
6243 pr_err_client(cl, "unable to close session to mds%d: %d\n",
6244 s->s_mds, ret);
6245 }
6246}
6247
6248/*
6249 * delayed work -- periodically trim expired leases, renew caps with mds. If
6250 * the @delay parameter is set to 0 or if it's more than 5 secs, the default
6251 * workqueue delay value of 5 secs will be used.
6252 */
6253static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)
6254{
6255 unsigned long max_delay = HZ * 5;
6256
6257 /* 5 secs default delay */
6258 if (!delay || (delay > max_delay))
6259 delay = max_delay;
6260 schedule_delayed_work(&mdsc->delayed_work,
6261 round_jiffies_relative(delay));
6262}
6263
6264static void delayed_work(struct work_struct *work)
6265{
6266 struct ceph_mds_client *mdsc =
6267 container_of(work, struct ceph_mds_client, delayed_work.work);
6268 unsigned long delay;
6269 int renew_interval;
6270 int renew_caps;
6271 int i;
6272
6273 doutc(mdsc->fsc->client, "mdsc delayed_work\n");
6274
6275 if (mdsc->stopping >= CEPH_MDSC_STOPPING_FLUSHED)
6276 return;
6277
6278 mutex_lock(&mdsc->mutex);
6279 renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
6280 renew_caps = time_after_eq(jiffies, HZ*renew_interval +
6281 mdsc->last_renew_caps);
6282 if (renew_caps)
6283 mdsc->last_renew_caps = jiffies;
6284
6285 for (i = 0; i < mdsc->max_sessions; i++) {
6286 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
6287 if (!s)
6288 continue;
6289
6290 if (!check_session_state(s)) {
6291 ceph_put_mds_session(s);
6292 continue;
6293 }
6294 mutex_unlock(&mdsc->mutex);
6295
6296 ceph_flush_session_cap_releases(mdsc, s);
6297
6298 mutex_lock(&s->s_mutex);
6299 if (renew_caps)
6300 send_renew_caps(mdsc, s);
6301 else
6302 ceph_con_keepalive(&s->s_con);
6303 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
6304 s->s_state == CEPH_MDS_SESSION_HUNG)
6305 ceph_send_cap_releases(mdsc, s);
6306 mutex_unlock(&s->s_mutex);
6307 ceph_put_mds_session(s);
6308
6309 mutex_lock(&mdsc->mutex);
6310 }
6311 mutex_unlock(&mdsc->mutex);
6312
6313 delay = ceph_check_delayed_caps(mdsc);
6314
6315 ceph_queue_cap_reclaim_work(mdsc);
6316
6317 ceph_trim_snapid_map(mdsc);
6318
6319 maybe_recover_session(mdsc);
6320
6321 schedule_delayed(mdsc, delay);
6322}
6323
6324int ceph_mdsc_init(struct ceph_fs_client *fsc)
6325
6326{
6327 struct ceph_mds_client *mdsc;
6328 int err;
6329
6330 mdsc = kzalloc_obj(struct ceph_mds_client, GFP_NOFS);
6331 if (!mdsc)
6332 return -ENOMEM;
6333 mdsc->fsc = fsc;
6334 mutex_init(&mdsc->mutex);
6335 mdsc->mdsmap = kzalloc_obj(*mdsc->mdsmap, GFP_NOFS);
6336 if (!mdsc->mdsmap) {
6337 err = -ENOMEM;
6338 goto err_mdsc;
6339 }
6340
6341 init_completion(&mdsc->safe_umount_waiters);
6342 spin_lock_init(&mdsc->stopping_lock);
6343 atomic_set(&mdsc->stopping_blockers, 0);
6344 init_completion(&mdsc->stopping_waiter);
6345 atomic64_set(&mdsc->dirty_folios, 0);
6346 init_waitqueue_head(&mdsc->flush_end_wq);
6347 init_waitqueue_head(&mdsc->session_close_wq);
6348 INIT_LIST_HEAD(&mdsc->waiting_for_map);
6349 mdsc->quotarealms_inodes = RB_ROOT;
6350 mutex_init(&mdsc->quotarealms_inodes_mutex);
6351 init_rwsem(&mdsc->snap_rwsem);
6352 mdsc->snap_realms = RB_ROOT;
6353 INIT_LIST_HEAD(&mdsc->snap_empty);
6354 spin_lock_init(&mdsc->snap_empty_lock);
6355 mdsc->request_tree = RB_ROOT;
6356 INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
6357 mdsc->last_renew_caps = jiffies;
6358 INIT_LIST_HEAD(&mdsc->cap_delay_list);
6359#ifdef CONFIG_DEBUG_FS
6360 INIT_LIST_HEAD(&mdsc->cap_wait_list);
6361#endif
6362 spin_lock_init(&mdsc->cap_delay_lock);
6363 INIT_LIST_HEAD(&mdsc->cap_unlink_delay_list);
6364 INIT_LIST_HEAD(&mdsc->snap_flush_list);
6365 spin_lock_init(&mdsc->snap_flush_lock);
6366 mdsc->last_cap_flush_tid = 1;
6367 INIT_LIST_HEAD(&mdsc->cap_flush_list);
6368 INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
6369 spin_lock_init(&mdsc->cap_dirty_lock);
6370 init_waitqueue_head(&mdsc->cap_flushing_wq);
6371 INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);
6372 INIT_WORK(&mdsc->cap_unlink_work, ceph_cap_unlink_work);
6373 err = ceph_metric_init(&mdsc->metric);
6374 if (err)
6375 goto err_mdsmap;
6376 ceph_subvolume_metrics_init(&mdsc->subvol_metrics);
6377 mutex_init(&mdsc->subvol_metrics_last_mutex);
6378 mdsc->subvol_metrics_last = NULL;
6379 mdsc->subvol_metrics_last_nr = 0;
6380 mdsc->subvol_metrics_sent = 0;
6381 mdsc->subvol_metrics_nonzero_sends = 0;
6382
6383 spin_lock_init(&mdsc->dentry_list_lock);
6384 INIT_LIST_HEAD(&mdsc->dentry_leases);
6385 INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
6386
6387 spin_lock_init(&mdsc->reset_state.lock);
6388 init_waitqueue_head(&mdsc->reset_state.blocked_wq);
6389 atomic_set(&mdsc->reset_state.blocked_requests, 0);
6390 INIT_WORK(&mdsc->reset_work, ceph_mdsc_reset_workfn);
6391
6392 ceph_caps_init(mdsc);
6393 ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
6394
6395 spin_lock_init(&mdsc->snapid_map_lock);
6396 mdsc->snapid_map_tree = RB_ROOT;
6397 INIT_LIST_HEAD(&mdsc->snapid_map_lru);
6398
6399 init_rwsem(&mdsc->pool_perm_rwsem);
6400 mdsc->pool_perm_tree = RB_ROOT;
6401
6402 strscpy(mdsc->nodename, utsname()->nodename,
6403 sizeof(mdsc->nodename));
6404
6405 fsc->mdsc = mdsc;
6406 return 0;
6407
6408err_mdsmap:
6409 kfree(mdsc->mdsmap);
6410err_mdsc:
6411 kfree(mdsc);
6412 return err;
6413}
6414
6415/*
6416 * Wait for safe replies on open mds requests. If we time out, drop
6417 * all requests from the tree to avoid dangling dentry refs.
6418 */
6419static void wait_requests(struct ceph_mds_client *mdsc)
6420{
6421 struct ceph_client *cl = mdsc->fsc->client;
6422 struct ceph_options *opts = mdsc->fsc->client->options;
6423 struct ceph_mds_request *req;
6424
6425 mutex_lock(&mdsc->mutex);
6426 if (__get_oldest_req(mdsc)) {
6427 mutex_unlock(&mdsc->mutex);
6428
6429 doutc(cl, "waiting for requests\n");
6430 wait_for_completion_timeout(&mdsc->safe_umount_waiters,
6431 ceph_timeout_jiffies(opts->mount_timeout));
6432
6433 /* tear down remaining requests */
6434 mutex_lock(&mdsc->mutex);
6435 while ((req = __get_oldest_req(mdsc))) {
6436 doutc(cl, "timed out on tid %llu\n", req->r_tid);
6437 list_del_init(&req->r_wait);
6438 __unregister_request(mdsc, req);
6439 }
6440 }
6441 mutex_unlock(&mdsc->mutex);
6442 doutc(cl, "done\n");
6443}
6444
6445void send_flush_mdlog(struct ceph_mds_session *s)
6446{
6447 struct ceph_client *cl = s->s_mdsc->fsc->client;
6448 struct ceph_msg *msg;
6449
6450 /*
6451 * Pre-luminous MDS crashes when it sees an unknown session request
6452 */
6453 if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS))
6454 return;
6455
6456 mutex_lock(&s->s_mutex);
6457 doutc(cl, "request mdlog flush to mds%d (%s)s seq %lld\n",
6458 s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);
6459 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG,
6460 s->s_seq);
6461 if (!msg) {
6462 pr_err_client(cl, "failed to request mdlog flush to mds%d (%s) seq %lld\n",
6463 s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);
6464 } else {
6465 ceph_con_send(&s->s_con, msg);
6466 }
6467 mutex_unlock(&s->s_mutex);
6468}
6469
6470static int ceph_mds_auth_match(struct ceph_mds_client *mdsc,
6471 struct ceph_mds_cap_auth *auth,
6472 const struct cred *cred,
6473 char *tpath)
6474{
6475 u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid);
6476 u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid);
6477 struct ceph_client *cl = mdsc->fsc->client;
6478 const char *fs_name = mdsc->mdsmap->m_fs_name;
6479 const char *spath = mdsc->fsc->mount_options->server_path;
6480 bool gid_matched = false;
6481 u32 gid, tlen, len;
6482 int i, j;
6483
6484 doutc(cl, "fsname check fs_name=%s match.fs_name=%s\n",
6485 fs_name, auth->match.fs_name ? auth->match.fs_name : "");
6486
6487 if (!ceph_namespace_match(auth->match.fs_name, fs_name)) {
6488 /* fsname mismatch, try next one */
6489 return 0;
6490 }
6491
6492 doutc(cl, "match.uid %lld\n", auth->match.uid);
6493 if (auth->match.uid != MDS_AUTH_UID_ANY) {
6494 if (auth->match.uid != caller_uid)
6495 return 0;
6496 if (auth->match.num_gids) {
6497 for (i = 0; i < auth->match.num_gids; i++) {
6498 if (caller_gid == auth->match.gids[i])
6499 gid_matched = true;
6500 }
6501 if (!gid_matched && cred->group_info->ngroups) {
6502 for (i = 0; i < cred->group_info->ngroups; i++) {
6503 gid = from_kgid(&init_user_ns,
6504 cred->group_info->gid[i]);
6505 for (j = 0; j < auth->match.num_gids; j++) {
6506 if (gid == auth->match.gids[j]) {
6507 gid_matched = true;
6508 break;
6509 }
6510 }
6511 if (gid_matched)
6512 break;
6513 }
6514 }
6515 if (!gid_matched)
6516 return 0;
6517 }
6518 }
6519
6520 /* path match */
6521 if (auth->match.path) {
6522 if (!tpath)
6523 return 0;
6524
6525 tlen = strlen(tpath);
6526 len = strlen(auth->match.path);
6527 if (len) {
6528 char *_tpath = tpath;
6529 bool free_tpath = false;
6530 int m, n;
6531
6532 doutc(cl, "server path %s, tpath %s, match.path %s\n",
6533 spath, tpath, auth->match.path);
6534 if (spath && (m = strlen(spath)) != 1) {
6535 /* mount path + '/' + tpath + an extra space */
6536 n = m + 1 + tlen + 1;
6537 _tpath = kmalloc(n, GFP_NOFS);
6538 if (!_tpath)
6539 return -ENOMEM;
6540 /* remove the leading '/' */
6541 snprintf(_tpath, n, "%s/%s", spath + 1, tpath);
6542 free_tpath = true;
6543 tlen = strlen(_tpath);
6544 }
6545
6546 /*
6547 * Please note the tailing '/' for match.path has already
6548 * been removed when parsing.
6549 *
6550 * Remove the tailing '/' for the target path.
6551 */
6552 while (tlen && _tpath[tlen - 1] == '/') {
6553 _tpath[tlen - 1] = '\0';
6554 tlen -= 1;
6555 }
6556 doutc(cl, "_tpath %s\n", _tpath);
6557
6558 /*
6559 * In case first == _tpath && tlen == len:
6560 * match.path=/foo --> /foo _path=/foo --> match
6561 * match.path=/foo/ --> /foo _path=/foo --> match
6562 *
6563 * In case first == _tmatch.path && tlen > len:
6564 * match.path=/foo/ --> /foo _path=/foo/ --> match
6565 * match.path=/foo --> /foo _path=/foo/ --> match
6566 * match.path=/foo/ --> /foo _path=/foo/d --> match
6567 * match.path=/foo --> /foo _path=/food --> mismatch
6568 *
6569 * All the other cases --> mismatch
6570 */
6571 bool path_matched = true;
6572 char *first = strstr(_tpath, auth->match.path);
6573 if (first != _tpath ||
6574 (tlen > len && _tpath[len] != '/')) {
6575 path_matched = false;
6576 }
6577
6578 if (free_tpath)
6579 kfree(_tpath);
6580
6581 if (!path_matched)
6582 return 0;
6583 }
6584 }
6585
6586 doutc(cl, "matched\n");
6587 return 1;
6588}
6589
6590int ceph_mds_check_access(struct ceph_mds_client *mdsc, char *tpath, int mask)
6591{
6592 const struct cred *cred = get_current_cred();
6593 u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid);
6594 u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid);
6595 struct ceph_mds_cap_auth *rw_perms_s = NULL;
6596 struct ceph_client *cl = mdsc->fsc->client;
6597 bool root_squash_perms = true;
6598 int i, err;
6599
6600 doutc(cl, "tpath '%s', mask %d, caller_uid %d, caller_gid %d\n",
6601 tpath, mask, caller_uid, caller_gid);
6602
6603 mutex_lock(&mdsc->mutex);
6604 for (i = 0; i < mdsc->s_cap_auths_num; i++) {
6605 struct ceph_mds_cap_auth *s = &mdsc->s_cap_auths[i];
6606
6607 err = ceph_mds_auth_match(mdsc, s, cred, tpath);
6608 if (err < 0) {
6609 mutex_unlock(&mdsc->mutex);
6610 put_cred(cred);
6611 return err;
6612 } else if (err > 0) {
6613 /* always follow the last auth caps' permission */
6614 root_squash_perms = true;
6615 rw_perms_s = NULL;
6616 if ((mask & MAY_WRITE) && s->writeable &&
6617 s->match.root_squash && (!caller_uid || !caller_gid))
6618 root_squash_perms = false;
6619
6620 if (((mask & MAY_WRITE) && !s->writeable) ||
6621 ((mask & MAY_READ) && !s->readable))
6622 rw_perms_s = s;
6623 }
6624 }
6625
6626 put_cred(cred);
6627
6628 doutc(cl, "root_squash_perms %d, rw_perms_s %p\n", root_squash_perms,
6629 rw_perms_s);
6630 if (root_squash_perms && rw_perms_s == NULL) {
6631 mutex_unlock(&mdsc->mutex);
6632 doutc(cl, "access allowed\n");
6633 return 0;
6634 }
6635
6636 if (!root_squash_perms) {
6637 doutc(cl, "root_squash is enabled and user(%d %d) isn't allowed to write",
6638 caller_uid, caller_gid);
6639 }
6640 if (rw_perms_s) {
6641 doutc(cl, "mds auth caps readable/writeable %d/%d while request r/w %d/%d",
6642 rw_perms_s->readable, rw_perms_s->writeable,
6643 !!(mask & MAY_READ), !!(mask & MAY_WRITE));
6644 }
6645 doutc(cl, "access denied\n");
6646 mutex_unlock(&mdsc->mutex);
6647 return -EACCES;
6648}
6649
6650/*
6651 * called before mount is ro, and before dentries are torn down.
6652 * (hmm, does this still race with new lookups?)
6653 */
6654void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
6655{
6656 doutc(mdsc->fsc->client, "begin\n");
6657 mdsc->stopping = CEPH_MDSC_STOPPING_BEGIN;
6658
6659 ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true);
6660 ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false);
6661 ceph_flush_dirty_caps(mdsc);
6662 wait_requests(mdsc);
6663
6664 /*
6665 * wait for reply handlers to drop their request refs and
6666 * their inode/dcache refs
6667 */
6668 ceph_msgr_flush();
6669
6670 ceph_cleanup_quotarealms_inodes(mdsc);
6671 doutc(mdsc->fsc->client, "done\n");
6672}
6673
6674/*
6675 * flush the mdlog and wait for all write mds requests to flush.
6676 */
6677static void flush_mdlog_and_wait_mdsc_unsafe_requests(struct ceph_mds_client *mdsc,
6678 u64 want_tid)
6679{
6680 struct ceph_client *cl = mdsc->fsc->client;
6681 struct ceph_mds_request *req = NULL, *nextreq;
6682 struct ceph_mds_session *last_session = NULL;
6683 struct rb_node *n;
6684
6685 mutex_lock(&mdsc->mutex);
6686 doutc(cl, "want %lld\n", want_tid);
6687restart:
6688 req = __get_oldest_req(mdsc);
6689 while (req && req->r_tid <= want_tid) {
6690 /* find next request */
6691 n = rb_next(&req->r_node);
6692 if (n)
6693 nextreq = rb_entry(n, struct ceph_mds_request, r_node);
6694 else
6695 nextreq = NULL;
6696 if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
6697 (req->r_op & CEPH_MDS_OP_WRITE)) {
6698 struct ceph_mds_session *s = req->r_session;
6699
6700 if (!s) {
6701 req = nextreq;
6702 continue;
6703 }
6704
6705 /* write op */
6706 ceph_mdsc_get_request(req);
6707 if (nextreq)
6708 ceph_mdsc_get_request(nextreq);
6709 s = ceph_get_mds_session(s);
6710 mutex_unlock(&mdsc->mutex);
6711
6712 /* send flush mdlog request to MDS */
6713 if (last_session != s) {
6714 send_flush_mdlog(s);
6715 ceph_put_mds_session(last_session);
6716 last_session = s;
6717 } else {
6718 ceph_put_mds_session(s);
6719 }
6720 doutc(cl, "wait on %llu (want %llu)\n",
6721 req->r_tid, want_tid);
6722 wait_for_completion(&req->r_safe_completion);
6723
6724 mutex_lock(&mdsc->mutex);
6725 ceph_mdsc_put_request(req);
6726 if (!nextreq)
6727 break; /* next dne before, so we're done! */
6728 if (RB_EMPTY_NODE(&nextreq->r_node)) {
6729 /* next request was removed from tree */
6730 ceph_mdsc_put_request(nextreq);
6731 goto restart;
6732 }
6733 ceph_mdsc_put_request(nextreq); /* won't go away */
6734 }
6735 req = nextreq;
6736 }
6737 mutex_unlock(&mdsc->mutex);
6738 ceph_put_mds_session(last_session);
6739 doutc(cl, "done\n");
6740}
6741
6742void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
6743{
6744 struct ceph_client *cl = mdsc->fsc->client;
6745 u64 want_tid, want_flush;
6746
6747 if (READ_ONCE(mdsc->fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN)
6748 return;
6749
6750 doutc(cl, "sync\n");
6751 mutex_lock(&mdsc->mutex);
6752 want_tid = mdsc->last_tid;
6753 mutex_unlock(&mdsc->mutex);
6754
6755 ceph_flush_dirty_caps(mdsc);
6756 ceph_flush_cap_releases(mdsc);
6757 spin_lock(&mdsc->cap_dirty_lock);
6758 want_flush = mdsc->last_cap_flush_tid;
6759 if (!list_empty(&mdsc->cap_flush_list)) {
6760 struct ceph_cap_flush *cf =
6761 list_last_entry(&mdsc->cap_flush_list,
6762 struct ceph_cap_flush, g_list);
6763 cf->wake = true;
6764 }
6765 spin_unlock(&mdsc->cap_dirty_lock);
6766
6767 doutc(cl, "sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
6768
6769 flush_mdlog_and_wait_mdsc_unsafe_requests(mdsc, want_tid);
6770 wait_caps_flush(mdsc, want_flush);
6771}
6772
6773/*
6774 * true if all sessions are closed, or we force unmount
6775 */
6776static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped)
6777{
6778 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
6779 return true;
6780 return atomic_read(&mdsc->num_sessions) <= skipped;
6781}
6782
6783/*
6784 * called after sb is ro or when metadata corrupted.
6785 */
6786void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
6787{
6788 struct ceph_options *opts = mdsc->fsc->client->options;
6789 struct ceph_client *cl = mdsc->fsc->client;
6790 struct ceph_mds_session *session;
6791 int i;
6792 int skipped = 0;
6793
6794 doutc(cl, "begin\n");
6795
6796 /* close sessions */
6797 mutex_lock(&mdsc->mutex);
6798 for (i = 0; i < mdsc->max_sessions; i++) {
6799 session = __ceph_lookup_mds_session(mdsc, i);
6800 if (!session)
6801 continue;
6802 mutex_unlock(&mdsc->mutex);
6803 mutex_lock(&session->s_mutex);
6804 if (__close_session(mdsc, session) <= 0)
6805 skipped++;
6806 mutex_unlock(&session->s_mutex);
6807 ceph_put_mds_session(session);
6808 mutex_lock(&mdsc->mutex);
6809 }
6810 mutex_unlock(&mdsc->mutex);
6811
6812 doutc(cl, "waiting for sessions to close\n");
6813 wait_event_timeout(mdsc->session_close_wq,
6814 done_closing_sessions(mdsc, skipped),
6815 ceph_timeout_jiffies(opts->mount_timeout));
6816
6817 /* tear down remaining sessions */
6818 mutex_lock(&mdsc->mutex);
6819 for (i = 0; i < mdsc->max_sessions; i++) {
6820 if (mdsc->sessions[i]) {
6821 session = ceph_get_mds_session(mdsc->sessions[i]);
6822 __unregister_session(mdsc, session);
6823 mutex_unlock(&mdsc->mutex);
6824 mutex_lock(&session->s_mutex);
6825 remove_session_caps(session);
6826 mutex_unlock(&session->s_mutex);
6827 ceph_put_mds_session(session);
6828 mutex_lock(&mdsc->mutex);
6829 }
6830 }
6831 WARN_ON(!list_empty(&mdsc->cap_delay_list));
6832 mutex_unlock(&mdsc->mutex);
6833
6834 ceph_cleanup_snapid_map(mdsc);
6835 ceph_cleanup_global_and_empty_realms(mdsc);
6836
6837 cancel_work_sync(&mdsc->cap_reclaim_work);
6838 cancel_work_sync(&mdsc->cap_unlink_work);
6839 cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
6840
6841 doutc(cl, "done\n");
6842}
6843
6844void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
6845{
6846 struct ceph_mds_session *session;
6847 int mds;
6848
6849 doutc(mdsc->fsc->client, "force umount\n");
6850
6851 mutex_lock(&mdsc->mutex);
6852 for (mds = 0; mds < mdsc->max_sessions; mds++) {
6853 session = __ceph_lookup_mds_session(mdsc, mds);
6854 if (!session)
6855 continue;
6856
6857 if (session->s_state == CEPH_MDS_SESSION_REJECTED)
6858 __unregister_session(mdsc, session);
6859 __wake_requests(mdsc, &session->s_waiting);
6860 mutex_unlock(&mdsc->mutex);
6861
6862 mutex_lock(&session->s_mutex);
6863 __close_session(mdsc, session);
6864 if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
6865 cleanup_session_requests(mdsc, session);
6866 remove_session_caps(session);
6867 }
6868 mutex_unlock(&session->s_mutex);
6869 ceph_put_mds_session(session);
6870
6871 mutex_lock(&mdsc->mutex);
6872 kick_requests(mdsc, mds);
6873 }
6874 __wake_requests(mdsc, &mdsc->waiting_for_map);
6875 mutex_unlock(&mdsc->mutex);
6876}
6877
6878static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
6879{
6880 doutc(mdsc->fsc->client, "stop\n");
6881 /*
6882 * Make sure the delayed work stopped before releasing
6883 * the resources.
6884 *
6885 * Because the cancel_delayed_work_sync() will only
6886 * guarantee that the work finishes executing. But the
6887 * delayed work will re-arm itself again after that.
6888 */
6889 flush_delayed_work(&mdsc->delayed_work);
6890
6891 if (mdsc->mdsmap)
6892 ceph_mdsmap_destroy(mdsc->mdsmap);
6893 kfree(mdsc->sessions);
6894 ceph_caps_finalize(mdsc);
6895
6896 if (mdsc->s_cap_auths) {
6897 int i;
6898
6899 for (i = 0; i < mdsc->s_cap_auths_num; i++) {
6900 kfree(mdsc->s_cap_auths[i].match.gids);
6901 kfree(mdsc->s_cap_auths[i].match.path);
6902 kfree(mdsc->s_cap_auths[i].match.fs_name);
6903 }
6904 kfree(mdsc->s_cap_auths);
6905 }
6906
6907 ceph_pool_perm_destroy(mdsc);
6908}
6909
6910void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
6911{
6912 struct ceph_mds_client *mdsc = fsc->mdsc;
6913 doutc(fsc->client, "%p\n", mdsc);
6914
6915 if (!mdsc)
6916 return;
6917
6918 /* flush out any connection work with references to us */
6919 ceph_msgr_flush();
6920
6921 /*
6922 * Mark reset as failed and wake any blocked waiters before
6923 * cancelling, so unmount doesn't stall on blocked_wq timeout
6924 * if cancel_work_sync() prevents the work from running.
6925 */
6926 spin_lock(&mdsc->reset_state.lock);
6927 mdsc->reset_state.shutdown = true;
6928 if (mdsc->reset_state.phase != CEPH_CLIENT_RESET_IDLE) {
6929 mdsc->reset_state.phase = CEPH_CLIENT_RESET_IDLE;
6930 mdsc->reset_state.last_errno = -ESHUTDOWN;
6931 mdsc->reset_state.last_finish = jiffies;
6932 mdsc->reset_state.failure_count++;
6933 }
6934 spin_unlock(&mdsc->reset_state.lock);
6935 wake_up_all(&mdsc->reset_state.blocked_wq);
6936
6937 cancel_work_sync(&mdsc->reset_work);
6938 ceph_mdsc_stop(mdsc);
6939
6940 ceph_metric_destroy(&mdsc->metric);
6941 ceph_subvolume_metrics_destroy(&mdsc->subvol_metrics);
6942 kfree(mdsc->subvol_metrics_last);
6943
6944 fsc->mdsc = NULL;
6945 kfree(mdsc);
6946 doutc(fsc->client, "%p done\n", mdsc);
6947}
6948
6949void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
6950{
6951 struct ceph_fs_client *fsc = mdsc->fsc;
6952 struct ceph_client *cl = fsc->client;
6953 const char *mds_namespace = fsc->mount_options->mds_namespace;
6954 void *p = msg->front.iov_base;
6955 void *end = p + msg->front.iov_len;
6956 u32 epoch;
6957 u32 num_fs;
6958 u32 mount_fscid = (u32)-1;
6959 int err = -EINVAL;
6960
6961 ceph_decode_need(&p, end, sizeof(u32), bad);
6962 epoch = ceph_decode_32(&p);
6963
6964 doutc(cl, "epoch %u\n", epoch);
6965
6966 /* struct_v, struct_cv, map_len, epoch, legacy_client_fscid */
6967 ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 3, bad);
6968
6969 ceph_decode_32_safe(&p, end, num_fs, bad);
6970 while (num_fs-- > 0) {
6971 void *info_p, *info_end;
6972 u32 info_len;
6973 u32 fscid, namelen;
6974
6975 ceph_decode_need(&p, end, 2 + sizeof(u32), bad);
6976 p += 2; // info_v, info_cv
6977 info_len = ceph_decode_32(&p);
6978 ceph_decode_need(&p, end, info_len, bad);
6979 info_p = p;
6980 info_end = p + info_len;
6981 p = info_end;
6982
6983 ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad);
6984 fscid = ceph_decode_32(&info_p);
6985 namelen = ceph_decode_32(&info_p);
6986 ceph_decode_need(&info_p, info_end, namelen, bad);
6987
6988 if (mds_namespace &&
6989 strlen(mds_namespace) == namelen &&
6990 !strncmp(mds_namespace, (char *)info_p, namelen)) {
6991 mount_fscid = fscid;
6992 break;
6993 }
6994 }
6995
6996 ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch);
6997 if (mount_fscid != (u32)-1) {
6998 fsc->client->monc.fs_cluster_id = mount_fscid;
6999 ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP,
7000 0, true);
7001 ceph_monc_renew_subs(&fsc->client->monc);
7002 } else {
7003 err = -ENOENT;
7004 goto err_out;
7005 }
7006 return;
7007
7008bad:
7009 pr_err_client(cl, "error decoding fsmap %d. Shutting down mount.\n",
7010 err);
7011 ceph_umount_begin(mdsc->fsc->sb);
7012 ceph_msg_dump(msg);
7013err_out:
7014 mutex_lock(&mdsc->mutex);
7015 mdsc->mdsmap_err = err;
7016 __wake_requests(mdsc, &mdsc->waiting_for_map);
7017 mutex_unlock(&mdsc->mutex);
7018}
7019
7020/*
7021 * handle mds map update.
7022 */
7023void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
7024{
7025 struct ceph_client *cl = mdsc->fsc->client;
7026 u32 epoch;
7027 u32 maplen;
7028 void *p = msg->front.iov_base;
7029 void *end = p + msg->front.iov_len;
7030 struct ceph_mdsmap *newmap, *oldmap;
7031 struct ceph_fsid fsid;
7032 int err = -EINVAL;
7033
7034 ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
7035 ceph_decode_copy(&p, &fsid, sizeof(fsid));
7036 if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
7037 return;
7038 epoch = ceph_decode_32(&p);
7039 maplen = ceph_decode_32(&p);
7040 doutc(cl, "epoch %u len %d\n", epoch, (int)maplen);
7041
7042 /* do we need it? */
7043 mutex_lock(&mdsc->mutex);
7044 if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
7045 doutc(cl, "epoch %u <= our %u\n", epoch, mdsc->mdsmap->m_epoch);
7046 mutex_unlock(&mdsc->mutex);
7047 return;
7048 }
7049
7050 newmap = ceph_mdsmap_decode(mdsc, &p, end, ceph_msgr2(mdsc->fsc->client));
7051 if (IS_ERR(newmap)) {
7052 err = PTR_ERR(newmap);
7053 goto bad_unlock;
7054 }
7055
7056 /* swap into place */
7057 if (mdsc->mdsmap) {
7058 oldmap = mdsc->mdsmap;
7059 mdsc->mdsmap = newmap;
7060 check_new_map(mdsc, newmap, oldmap);
7061 ceph_mdsmap_destroy(oldmap);
7062 } else {
7063 mdsc->mdsmap = newmap; /* first mds map */
7064 }
7065 mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size,
7066 MAX_LFS_FILESIZE);
7067
7068 __wake_requests(mdsc, &mdsc->waiting_for_map);
7069 ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP,
7070 mdsc->mdsmap->m_epoch);
7071
7072 mutex_unlock(&mdsc->mutex);
7073 schedule_delayed(mdsc, 0);
7074 return;
7075
7076bad_unlock:
7077 mutex_unlock(&mdsc->mutex);
7078bad:
7079 pr_err_client(cl, "error decoding mdsmap %d. Shutting down mount.\n",
7080 err);
7081 ceph_umount_begin(mdsc->fsc->sb);
7082 ceph_msg_dump(msg);
7083 return;
7084}
7085
7086static struct ceph_connection *mds_get_con(struct ceph_connection *con)
7087{
7088 struct ceph_mds_session *s = con->private;
7089
7090 if (ceph_get_mds_session(s))
7091 return con;
7092 return NULL;
7093}
7094
7095static void mds_put_con(struct ceph_connection *con)
7096{
7097 struct ceph_mds_session *s = con->private;
7098
7099 ceph_put_mds_session(s);
7100}
7101
7102/*
7103 * if the client is unresponsive for long enough, the mds will kill
7104 * the session entirely.
7105 */
7106static void mds_peer_reset(struct ceph_connection *con)
7107{
7108 struct ceph_mds_session *s = con->private;
7109 struct ceph_mds_client *mdsc = s->s_mdsc;
7110 int session_state;
7111
7112 pr_warn_client(mdsc->fsc->client, "mds%d closed our session\n",
7113 s->s_mds);
7114
7115 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO ||
7116 ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) < CEPH_MDS_STATE_RECONNECT)
7117 return;
7118
7119 /*
7120 * Only reconnect if MDS is in its RECONNECT phase. An MDS past
7121 * RECONNECT (REJOIN, CLIENTREPLAY, ACTIVE) will reject reconnect
7122 * attempts, so those states fall through to session teardown below.
7123 */
7124 if (ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) == CEPH_MDS_STATE_RECONNECT) {
7125 int rc = send_mds_reconnect(mdsc, s);
7126
7127 if (rc)
7128 pr_warn_client(mdsc->fsc->client,
7129 "mds%d reconnect failed: %d\n",
7130 s->s_mds, rc);
7131 return;
7132 }
7133
7134 /*
7135 * MDS is active (past RECONNECT). It will not accept a
7136 * CLIENT_RECONNECT from us, so tear the session down locally
7137 * and let new requests re-open a fresh session.
7138 *
7139 * Snapshot session state with READ_ONCE, then revalidate under
7140 * mdsc->mutex before acting. The subsequent mdsc->mutex
7141 * section rechecks s_state to catch concurrent transitions, so
7142 * the lockless snapshot here is safe. s->s_mutex is taken
7143 * separately for cleanup after unregistration, which avoids
7144 * introducing a new s->s_mutex + mdsc->mutex nesting.
7145 */
7146 session_state = READ_ONCE(s->s_state);
7147
7148 switch (session_state) {
7149 case CEPH_MDS_SESSION_RESTARTING:
7150 case CEPH_MDS_SESSION_RECONNECTING:
7151 case CEPH_MDS_SESSION_CLOSING:
7152 case CEPH_MDS_SESSION_OPEN:
7153 case CEPH_MDS_SESSION_HUNG:
7154 case CEPH_MDS_SESSION_OPENING:
7155 mutex_lock(&mdsc->mutex);
7156 if (s->s_mds >= mdsc->max_sessions ||
7157 mdsc->sessions[s->s_mds] != s ||
7158 s->s_state != session_state) {
7159 pr_info_client(mdsc->fsc->client,
7160 "mds%d state changed to %s during peer reset\n",
7161 s->s_mds,
7162 ceph_session_state_name(s->s_state));
7163 mutex_unlock(&mdsc->mutex);
7164 return;
7165 }
7166
7167 ceph_get_mds_session(s);
7168 s->s_state = CEPH_MDS_SESSION_CLOSED;
7169 __unregister_session(mdsc, s);
7170 __wake_requests(mdsc, &s->s_waiting);
7171 mutex_unlock(&mdsc->mutex);
7172
7173 mutex_lock(&s->s_mutex);
7174 cleanup_session_requests(mdsc, s);
7175 remove_session_caps(s);
7176 mutex_unlock(&s->s_mutex);
7177
7178 wake_up_all(&mdsc->session_close_wq);
7179
7180 mutex_lock(&mdsc->mutex);
7181 kick_requests(mdsc, s->s_mds);
7182 mutex_unlock(&mdsc->mutex);
7183
7184 ceph_put_mds_session(s);
7185 break;
7186 case CEPH_MDS_SESSION_CLOSED:
7187 case CEPH_MDS_SESSION_REJECTED:
7188 break;
7189 default:
7190 pr_warn_client(mdsc->fsc->client,
7191 "mds%d peer reset in unexpected state %s\n",
7192 s->s_mds,
7193 ceph_session_state_name(session_state));
7194 break;
7195 }
7196}
7197
7198static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
7199{
7200 struct ceph_mds_session *s = con->private;
7201 struct ceph_mds_client *mdsc = s->s_mdsc;
7202 struct ceph_client *cl = mdsc->fsc->client;
7203 int type = le16_to_cpu(msg->hdr.type);
7204
7205 mutex_lock(&mdsc->mutex);
7206 if (__verify_registered_session(mdsc, s) < 0) {
7207 doutc(cl, "dropping tid %llu from unregistered session %d\n",
7208 le64_to_cpu(msg->hdr.tid), s->s_mds);
7209 mutex_unlock(&mdsc->mutex);
7210 goto out;
7211 }
7212 mutex_unlock(&mdsc->mutex);
7213
7214 switch (type) {
7215 case CEPH_MSG_MDS_MAP:
7216 ceph_mdsc_handle_mdsmap(mdsc, msg);
7217 break;
7218 case CEPH_MSG_FS_MAP_USER:
7219 ceph_mdsc_handle_fsmap(mdsc, msg);
7220 break;
7221 case CEPH_MSG_CLIENT_SESSION:
7222 handle_session(s, msg);
7223 break;
7224 case CEPH_MSG_CLIENT_REPLY:
7225 handle_reply(s, msg);
7226 break;
7227 case CEPH_MSG_CLIENT_REQUEST_FORWARD:
7228 handle_forward(mdsc, s, msg);
7229 break;
7230 case CEPH_MSG_CLIENT_CAPS:
7231 ceph_handle_caps(s, msg);
7232 break;
7233 case CEPH_MSG_CLIENT_SNAP:
7234 ceph_handle_snap(mdsc, s, msg);
7235 break;
7236 case CEPH_MSG_CLIENT_LEASE:
7237 handle_lease(mdsc, s, msg);
7238 break;
7239 case CEPH_MSG_CLIENT_QUOTA:
7240 ceph_handle_quota(mdsc, s, msg);
7241 break;
7242
7243 default:
7244 pr_err_client(cl, "received unknown message type %d %s\n",
7245 type, ceph_msg_type_name(type));
7246 }
7247out:
7248 ceph_msg_put(msg);
7249}
7250
7251/*
7252 * authentication
7253 */
7254
7255/*
7256 * Note: returned pointer is the address of a structure that's
7257 * managed separately. Caller must *not* attempt to free it.
7258 */
7259static struct ceph_auth_handshake *
7260mds_get_authorizer(struct ceph_connection *con, int *proto, int force_new)
7261{
7262 struct ceph_mds_session *s = con->private;
7263 struct ceph_mds_client *mdsc = s->s_mdsc;
7264 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
7265 struct ceph_auth_handshake *auth = &s->s_auth;
7266 int ret;
7267
7268 ret = __ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
7269 force_new, proto, NULL, NULL);
7270 if (ret)
7271 return ERR_PTR(ret);
7272
7273 return auth;
7274}
7275
7276static int mds_add_authorizer_challenge(struct ceph_connection *con,
7277 void *challenge_buf, int challenge_buf_len)
7278{
7279 struct ceph_mds_session *s = con->private;
7280 struct ceph_mds_client *mdsc = s->s_mdsc;
7281 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
7282
7283 return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer,
7284 challenge_buf, challenge_buf_len);
7285}
7286
7287static int mds_verify_authorizer_reply(struct ceph_connection *con)
7288{
7289 struct ceph_mds_session *s = con->private;
7290 struct ceph_mds_client *mdsc = s->s_mdsc;
7291 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
7292 struct ceph_auth_handshake *auth = &s->s_auth;
7293
7294 return ceph_auth_verify_authorizer_reply(ac, auth->authorizer,
7295 auth->authorizer_reply_buf, auth->authorizer_reply_buf_len,
7296 NULL, NULL, NULL, NULL);
7297}
7298
7299static int mds_invalidate_authorizer(struct ceph_connection *con)
7300{
7301 struct ceph_mds_session *s = con->private;
7302 struct ceph_mds_client *mdsc = s->s_mdsc;
7303 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
7304
7305 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
7306
7307 return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
7308}
7309
7310static int mds_get_auth_request(struct ceph_connection *con,
7311 void *buf, int *buf_len,
7312 void **authorizer, int *authorizer_len)
7313{
7314 struct ceph_mds_session *s = con->private;
7315 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
7316 struct ceph_auth_handshake *auth = &s->s_auth;
7317 int ret;
7318
7319 ret = ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
7320 buf, buf_len);
7321 if (ret)
7322 return ret;
7323
7324 *authorizer = auth->authorizer_buf;
7325 *authorizer_len = auth->authorizer_buf_len;
7326 return 0;
7327}
7328
7329static int mds_handle_auth_reply_more(struct ceph_connection *con,
7330 void *reply, int reply_len,
7331 void *buf, int *buf_len,
7332 void **authorizer, int *authorizer_len)
7333{
7334 struct ceph_mds_session *s = con->private;
7335 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
7336 struct ceph_auth_handshake *auth = &s->s_auth;
7337 int ret;
7338
7339 ret = ceph_auth_handle_svc_reply_more(ac, auth, reply, reply_len,
7340 buf, buf_len);
7341 if (ret)
7342 return ret;
7343
7344 *authorizer = auth->authorizer_buf;
7345 *authorizer_len = auth->authorizer_buf_len;
7346 return 0;
7347}
7348
7349static int mds_handle_auth_done(struct ceph_connection *con,
7350 u64 global_id, void *reply, int reply_len,
7351 u8 *session_key, int *session_key_len,
7352 u8 *con_secret, int *con_secret_len)
7353{
7354 struct ceph_mds_session *s = con->private;
7355 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
7356 struct ceph_auth_handshake *auth = &s->s_auth;
7357
7358 return ceph_auth_handle_svc_reply_done(ac, auth, reply, reply_len,
7359 session_key, session_key_len,
7360 con_secret, con_secret_len);
7361}
7362
7363static int mds_handle_auth_bad_method(struct ceph_connection *con,
7364 int used_proto, int result,
7365 const int *allowed_protos, int proto_cnt,
7366 const int *allowed_modes, int mode_cnt)
7367{
7368 struct ceph_mds_session *s = con->private;
7369 struct ceph_mon_client *monc = &s->s_mdsc->fsc->client->monc;
7370 int ret;
7371
7372 if (ceph_auth_handle_bad_authorizer(monc->auth, CEPH_ENTITY_TYPE_MDS,
7373 used_proto, result,
7374 allowed_protos, proto_cnt,
7375 allowed_modes, mode_cnt)) {
7376 ret = ceph_monc_validate_auth(monc);
7377 if (ret)
7378 return ret;
7379 }
7380
7381 return -EACCES;
7382}
7383
7384static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
7385 struct ceph_msg_header *hdr, int *skip)
7386{
7387 struct ceph_msg *msg;
7388 int type = (int) le16_to_cpu(hdr->type);
7389 int front_len = (int) le32_to_cpu(hdr->front_len);
7390
7391 if (con->in_msg)
7392 return con->in_msg;
7393
7394 *skip = 0;
7395 msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
7396 if (!msg) {
7397 pr_err("unable to allocate msg type %d len %d\n",
7398 type, front_len);
7399 return NULL;
7400 }
7401
7402 return msg;
7403}
7404
7405static int mds_sign_message(struct ceph_msg *msg)
7406{
7407 struct ceph_mds_session *s = msg->con->private;
7408 struct ceph_auth_handshake *auth = &s->s_auth;
7409
7410 return ceph_auth_sign_message(auth, msg);
7411}
7412
7413static int mds_check_message_signature(struct ceph_msg *msg)
7414{
7415 struct ceph_mds_session *s = msg->con->private;
7416 struct ceph_auth_handshake *auth = &s->s_auth;
7417
7418 return ceph_auth_check_message_signature(auth, msg);
7419}
7420
7421static const struct ceph_connection_operations mds_con_ops = {
7422 .get = mds_get_con,
7423 .put = mds_put_con,
7424 .alloc_msg = mds_alloc_msg,
7425 .dispatch = mds_dispatch,
7426 .peer_reset = mds_peer_reset,
7427 .get_authorizer = mds_get_authorizer,
7428 .add_authorizer_challenge = mds_add_authorizer_challenge,
7429 .verify_authorizer_reply = mds_verify_authorizer_reply,
7430 .invalidate_authorizer = mds_invalidate_authorizer,
7431 .sign_message = mds_sign_message,
7432 .check_message_signature = mds_check_message_signature,
7433 .get_auth_request = mds_get_auth_request,
7434 .handle_auth_reply_more = mds_handle_auth_reply_more,
7435 .handle_auth_done = mds_handle_auth_done,
7436 .handle_auth_bad_method = mds_handle_auth_bad_method,
7437};
7438
7439/* eof */