projects / linux-block.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Merge tag 'devprop-5.19-rc1' of git://git.kernel.org/pub/scm/linux/kernel/git/rafael...
[linux-block.git] / fs / ceph / mds_client.c
CommitLineData
b2441318 1// SPDX-License-Identifier: GPL-2.0
3d14c5d2 2#include <linux/ceph/ceph_debug.h>
2f2dc053 3
496e5955 4#include <linux/fs.h>
2f2dc053 5#include <linux/wait.h>
5a0e3ad6 6#include <linux/slab.h>
54008399 7#include <linux/gfp.h>
2f2dc053 8#include <linux/sched.h>
3d14c5d2
YS		 9#include <linux/debugfs.h>
10#include <linux/seq_file.h>
3e0708b9 11#include <linux/ratelimit.h>
9ba1e224 12#include <linux/bits.h>
70c94820 13#include <linux/ktime.h>
d517b398 14#include <linux/bitmap.h>
2f2dc053 15
2f2dc053 16#include "super.h"
3d14c5d2
YS		 17#include "mds_client.h"
18
1fe60e51 19#include <linux/ceph/ceph_features.h>
3d14c5d2
YS		 20#include <linux/ceph/messenger.h>
21#include <linux/ceph/decode.h>
22#include <linux/ceph/pagelist.h>
23#include <linux/ceph/auth.h>
24#include <linux/ceph/debugfs.h>
2f2dc053 25
81c5a148
YZ		 26#define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)
27
2f2dc053
SW		 28/*
29 * A cluster of MDS (metadata server) daemons is responsible for
30 * managing the file system namespace (the directory hierarchy and
31 * inodes) and for coordinating shared access to storage. Metadata is
32 * partitioning hierarchically across a number of servers, and that
33 * partition varies over time as the cluster adjusts the distribution
34 * in order to balance load.
35 *
36 * The MDS client is primarily responsible to managing synchronous
37 * metadata requests for operations like open, unlink, and so forth.
38 * If there is a MDS failure, we find out about it when we (possibly
39 * request and) receive a new MDS map, and can resubmit affected
40 * requests.
41 *
42 * For the most part, though, we take advantage of a lossless
43 * communications channel to the MDS, and do not need to worry about
44 * timing out or resubmitting requests.
45 *
46 * We maintain a stateful "session" with each MDS we interact with.
47 * Within each session, we sent periodic heartbeat messages to ensure
48 * any capabilities or leases we have been issues remain valid. If
49 * the session times out and goes stale, our leases and capabilities
50 * are no longer valid.
51 */
52
20cb34ae 53struct ceph_reconnect_state {
81c5a148
YZ		 54 struct ceph_mds_session *session;
55 int nr_caps, nr_realms;
20cb34ae 56 struct ceph_pagelist *pagelist;
121f22a1 57 unsigned msg_version;
81c5a148 58 bool allow_multi;
20cb34ae
SW		 59};
60
2f2dc053
SW		 61static void __wake_requests(struct ceph_mds_client *mdsc,
62 struct list_head *head);
e3ec8d68 63static void ceph_cap_release_work(struct work_struct *work);
37c4efc1 64static void ceph_cap_reclaim_work(struct work_struct *work);
2f2dc053 65
9e32789f 66static const struct ceph_connection_operations mds_con_ops;
2f2dc053
SW		 67
68
69/*
70 * mds reply parsing
71 */
72
b37fe1f9
YZ		 73static int parse_reply_info_quota(void **p, void *end,
74 struct ceph_mds_reply_info_in *info)
75{
76 u8 struct_v, struct_compat;
77 u32 struct_len;
78
79 ceph_decode_8_safe(p, end, struct_v, bad);
80 ceph_decode_8_safe(p, end, struct_compat, bad);
81 /* struct_v is expected to be >= 1. we only
82 * understand encoding with struct_compat == 1. */
83 if (!struct_v || struct_compat != 1)
84 goto bad;
85 ceph_decode_32_safe(p, end, struct_len, bad);
86 ceph_decode_need(p, end, struct_len, bad);
87 end = *p + struct_len;
88 ceph_decode_64_safe(p, end, info->max_bytes, bad);
89 ceph_decode_64_safe(p, end, info->max_files, bad);
90 *p = end;
91 return 0;
92bad:
93 return -EIO;
94}
95
2f2dc053
SW		 96/*
97 * parse individual inode info
98 */
99static int parse_reply_info_in(void **p, void *end,
14303d20 100 struct ceph_mds_reply_info_in *info,
12b4629a 101 u64 features)
2f2dc053 102{
b37fe1f9
YZ		 103 int err = 0;
104 u8 struct_v = 0;
2f2dc053 105
b37fe1f9
YZ		 106 if (features == (u64)-1) {
107 u32 struct_len;
108 u8 struct_compat;
109 ceph_decode_8_safe(p, end, struct_v, bad);
110 ceph_decode_8_safe(p, end, struct_compat, bad);
111 /* struct_v is expected to be >= 1. we only understand
112 * encoding with struct_compat == 1. */
113 if (!struct_v || struct_compat != 1)
114 goto bad;
115 ceph_decode_32_safe(p, end, struct_len, bad);
116 ceph_decode_need(p, end, struct_len, bad);
117 end = *p + struct_len;
118 }
119
120 ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);
2f2dc053
SW		 121 info->in = *p;
122 *p += sizeof(struct ceph_mds_reply_inode) +
123 sizeof(*info->in->fragtree.splits) *
124 le32_to_cpu(info->in->fragtree.nsplits);
125
126 ceph_decode_32_safe(p, end, info->symlink_len, bad);
127 ceph_decode_need(p, end, info->symlink_len, bad);
128 info->symlink = *p;
129 *p += info->symlink_len;
130
23c625ce
ID		 131 ceph_decode_copy_safe(p, end, &info->dir_layout,
132 sizeof(info->dir_layout), bad);
2f2dc053
SW		 133 ceph_decode_32_safe(p, end, info->xattr_len, bad);
134 ceph_decode_need(p, end, info->xattr_len, bad);
135 info->xattr_data = *p;
136 *p += info->xattr_len;
fb01d1f8 137
b37fe1f9
YZ		 138 if (features == (u64)-1) {
139 /* inline data */
fb01d1f8
YZ		 140 ceph_decode_64_safe(p, end, info->inline_version, bad);
141 ceph_decode_32_safe(p, end, info->inline_len, bad);
142 ceph_decode_need(p, end, info->inline_len, bad);
143 info->inline_data = *p;
144 *p += info->inline_len;
b37fe1f9
YZ		 145 /* quota */
146 err = parse_reply_info_quota(p, end, info);
147 if (err < 0)
148 goto out_bad;
149 /* pool namespace */
150 ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
151 if (info->pool_ns_len > 0) {
152 ceph_decode_need(p, end, info->pool_ns_len, bad);
153 info->pool_ns_data = *p;
154 *p += info->pool_ns_len;
155 }
245ce991
JL		 156
157 /* btime */
158 ceph_decode_need(p, end, sizeof(info->btime), bad);
159 ceph_decode_copy(p, &info->btime, sizeof(info->btime));
160
161 /* change attribute */
a35ead31 162 ceph_decode_64_safe(p, end, info->change_attr, bad);
fb01d1f8 163
08796873
YZ		 164 /* dir pin */
165 if (struct_v >= 2) {
166 ceph_decode_32_safe(p, end, info->dir_pin, bad);
167 } else {
168 info->dir_pin = -ENODATA;
169 }
170
193e7b37
DD		 171 /* snapshot birth time, remains zero for v<=2 */
172 if (struct_v >= 3) {
173 ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
174 ceph_decode_copy(p, &info->snap_btime,
175 sizeof(info->snap_btime));
176 } else {
177 memset(&info->snap_btime, 0, sizeof(info->snap_btime));
178 }
179
e7f72952
YC		 180 /* snapshot count, remains zero for v<=3 */
181 if (struct_v >= 4) {
182 ceph_decode_64_safe(p, end, info->rsnaps, bad);
183 } else {
184 info->rsnaps = 0;
185 }
186
b37fe1f9
YZ		 187 *p = end;
188 } else {
189 if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
190 ceph_decode_64_safe(p, end, info->inline_version, bad);
191 ceph_decode_32_safe(p, end, info->inline_len, bad);
192 ceph_decode_need(p, end, info->inline_len, bad);
193 info->inline_data = *p;
194 *p += info->inline_len;
195 } else
196 info->inline_version = CEPH_INLINE_NONE;
197
198 if (features & CEPH_FEATURE_MDS_QUOTA) {
199 err = parse_reply_info_quota(p, end, info);
200 if (err < 0)
201 goto out_bad;
202 } else {
203 info->max_bytes = 0;
204 info->max_files = 0;
205 }
206
207 info->pool_ns_len = 0;
208 info->pool_ns_data = NULL;
209 if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
210 ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
211 if (info->pool_ns_len > 0) {
212 ceph_decode_need(p, end, info->pool_ns_len, bad);
213 info->pool_ns_data = *p;
214 *p += info->pool_ns_len;
215 }
216 }
08796873 217
245ce991
JL		 218 if (features & CEPH_FEATURE_FS_BTIME) {
219 ceph_decode_need(p, end, sizeof(info->btime), bad);
220 ceph_decode_copy(p, &info->btime, sizeof(info->btime));
a35ead31 221 ceph_decode_64_safe(p, end, info->change_attr, bad);
245ce991
JL		 222 }
223
08796873 224 info->dir_pin = -ENODATA;
e7f72952 225 /* info->snap_btime and info->rsnaps remain zero */
b37fe1f9
YZ		 226 }
227 return 0;
228bad:
229 err = -EIO;
230out_bad:
231 return err;
232}
233
234static int parse_reply_info_dir(void **p, void *end,
235 struct ceph_mds_reply_dirfrag **dirfrag,
236 u64 features)
237{
238 if (features == (u64)-1) {
fb18a575
LH		 239 u8 struct_v, struct_compat;
240 u32 struct_len;
fb18a575
LH		 241 ceph_decode_8_safe(p, end, struct_v, bad);
242 ceph_decode_8_safe(p, end, struct_compat, bad);
b37fe1f9
YZ		 243 /* struct_v is expected to be >= 1. we only understand
244 * encoding whose struct_compat == 1. */
245 if (!struct_v || struct_compat != 1)
fb18a575
LH		 246 goto bad;
247 ceph_decode_32_safe(p, end, struct_len, bad);
248 ceph_decode_need(p, end, struct_len, bad);
b37fe1f9 249 end = *p + struct_len;
fb18a575
LH		 250 }
251
b37fe1f9
YZ		 252 ceph_decode_need(p, end, sizeof(**dirfrag), bad);
253 *dirfrag = *p;
254 *p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist);
255 if (unlikely(*p > end))
256 goto bad;
257 if (features == (u64)-1)
258 *p = end;
259 return 0;
260bad:
261 return -EIO;
262}
263
264static int parse_reply_info_lease(void **p, void *end,
265 struct ceph_mds_reply_lease **lease,
266 u64 features)
267{
268 if (features == (u64)-1) {
269 u8 struct_v, struct_compat;
270 u32 struct_len;
271 ceph_decode_8_safe(p, end, struct_v, bad);
272 ceph_decode_8_safe(p, end, struct_compat, bad);
273 /* struct_v is expected to be >= 1. we only understand
274 * encoding whose struct_compat == 1. */
275 if (!struct_v || struct_compat != 1)
276 goto bad;
277 ceph_decode_32_safe(p, end, struct_len, bad);
278 ceph_decode_need(p, end, struct_len, bad);
279 end = *p + struct_len;
5ea5c5e0
YZ		 280 }
281
b37fe1f9
YZ		 282 ceph_decode_need(p, end, sizeof(**lease), bad);
283 *lease = *p;
284 *p += sizeof(**lease);
285 if (features == (u64)-1)
286 *p = end;
2f2dc053
SW		 287 return 0;
288bad:
b37fe1f9 289 return -EIO;
2f2dc053
SW		 290}
291
292/*
293 * parse a normal reply, which may contain a (dir+)dentry and/or a
294 * target inode.
295 */
296static int parse_reply_info_trace(void **p, void *end,
14303d20 297 struct ceph_mds_reply_info_parsed *info,
12b4629a 298 u64 features)
2f2dc053
SW		 299{
300 int err;
301
302 if (info->head->is_dentry) {
14303d20 303 err = parse_reply_info_in(p, end, &info->diri, features);
2f2dc053
SW		 304 if (err < 0)
305 goto out_bad;
306
b37fe1f9
YZ		 307 err = parse_reply_info_dir(p, end, &info->dirfrag, features);
308 if (err < 0)
309 goto out_bad;
2f2dc053
SW		 310
311 ceph_decode_32_safe(p, end, info->dname_len, bad);
312 ceph_decode_need(p, end, info->dname_len, bad);
313 info->dname = *p;
314 *p += info->dname_len;
b37fe1f9
YZ		 315
316 err = parse_reply_info_lease(p, end, &info->dlease, features);
317 if (err < 0)
318 goto out_bad;
2f2dc053
SW		 319 }
320
321 if (info->head->is_target) {
14303d20 322 err = parse_reply_info_in(p, end, &info->targeti, features);
2f2dc053
SW		 323 if (err < 0)
324 goto out_bad;
325 }
326
327 if (unlikely(*p != end))
328 goto bad;
329 return 0;
330
331bad:
332 err = -EIO;
333out_bad:
334 pr_err("problem parsing mds trace %d\n", err);
335 return err;
336}
337
338/*
339 * parse readdir results
340 */
b37fe1f9 341static int parse_reply_info_readdir(void **p, void *end,
14303d20 342 struct ceph_mds_reply_info_parsed *info,
12b4629a 343 u64 features)
2f2dc053
SW		 344{
345 u32 num, i = 0;
346 int err;
347
b37fe1f9
YZ		 348 err = parse_reply_info_dir(p, end, &info->dir_dir, features);
349 if (err < 0)
350 goto out_bad;
2f2dc053
SW		 351
352 ceph_decode_need(p, end, sizeof(num) + 2, bad);
c89136ea 353 num = ceph_decode_32(p);
956d39d6
YZ		 354 {
355 u16 flags = ceph_decode_16(p);
356 info->dir_end = !!(flags & CEPH_READDIR_FRAG_END);
357 info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE);
f3c4ebe6 358 info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER);
79162547 359 info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH);
956d39d6 360 }
2f2dc053
SW		 361 if (num == 0)
362 goto done;
363
2a5beea3
YZ		 364 BUG_ON(!info->dir_entries);
365 if ((unsigned long)(info->dir_entries + num) >
366 (unsigned long)info->dir_entries + info->dir_buf_size) {
54008399
YZ		 367 pr_err("dir contents are larger than expected\n");
368 WARN_ON(1);
369 goto bad;
370 }
2f2dc053 371
54008399 372 info->dir_nr = num;
2f2dc053 373 while (num) {
2a5beea3 374 struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
2f2dc053 375 /* dentry */
b37fe1f9 376 ceph_decode_32_safe(p, end, rde->name_len, bad);
2a5beea3
YZ		 377 ceph_decode_need(p, end, rde->name_len, bad);
378 rde->name = *p;
379 *p += rde->name_len;
380 dout("parsed dir dname '%.*s'\n", rde->name_len, rde->name);
2f2dc053 381
b37fe1f9
YZ		 382 /* dentry lease */
383 err = parse_reply_info_lease(p, end, &rde->lease, features);
384 if (err)
385 goto out_bad;
2f2dc053 386 /* inode */
2a5beea3 387 err = parse_reply_info_in(p, end, &rde->inode, features);
2f2dc053
SW		 388 if (err < 0)
389 goto out_bad;
8974eebd
YZ		 390 /* ceph_readdir_prepopulate() will update it */
391 rde->offset = 0;
2f2dc053
SW		 392 i++;
393 num--;
394 }
395
396done:
1d3f8723
JL		 397 /* Skip over any unrecognized fields */
398 *p = end;
2f2dc053
SW		 399 return 0;
400
401bad:
402 err = -EIO;
403out_bad:
404 pr_err("problem parsing dir contents %d\n", err);
405 return err;
406}
407
25933abd
HS		 408/*
409 * parse fcntl F_GETLK results
410 */
411static int parse_reply_info_filelock(void **p, void *end,
14303d20 412 struct ceph_mds_reply_info_parsed *info,
12b4629a 413 u64 features)
25933abd
HS		 414{
415 if (*p + sizeof(*info->filelock_reply) > end)
416 goto bad;
417
418 info->filelock_reply = *p;
25933abd 419
1d3f8723
JL		 420 /* Skip over any unrecognized fields */
421 *p = end;
25933abd 422 return 0;
25933abd
HS		 423bad:
424 return -EIO;
425}
426
d4846487
JL		 427
428#if BITS_PER_LONG == 64
429
430#define DELEGATED_INO_AVAILABLE xa_mk_value(1)
431
432static int ceph_parse_deleg_inos(void **p, void *end,
433 struct ceph_mds_session *s)
434{
435 u32 sets;
436
437 ceph_decode_32_safe(p, end, sets, bad);
438 dout("got %u sets of delegated inodes\n", sets);
439 while (sets--) {
440 u64 start, len, ino;
441
442 ceph_decode_64_safe(p, end, start, bad);
443 ceph_decode_64_safe(p, end, len, bad);
d4f6b31d
JL		 444
445 /* Don't accept a delegation of system inodes */
446 if (start < CEPH_INO_SYSTEM_BASE) {
447 pr_warn_ratelimited("ceph: ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n",
448 start, len);
449 continue;
450 }
d4846487
JL		 451 while (len--) {
452 int err = xa_insert(&s->s_delegated_inos, ino = start++,
453 DELEGATED_INO_AVAILABLE,
454 GFP_KERNEL);
455 if (!err) {
456 dout("added delegated inode 0x%llx\n",
457 start - 1);
458 } else if (err == -EBUSY) {
459 pr_warn("ceph: MDS delegated inode 0x%llx more than once.\n",
460 start - 1);
461 } else {
462 return err;
463 }
464 }
465 }
466 return 0;
467bad:
468 return -EIO;
469}
470
471u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
472{
473 unsigned long ino;
474 void *val;
475
476 xa_for_each(&s->s_delegated_inos, ino, val) {
477 val = xa_erase(&s->s_delegated_inos, ino);
478 if (val == DELEGATED_INO_AVAILABLE)
479 return ino;
480 }
481 return 0;
482}
483
484int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
485{
486 return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
487 GFP_KERNEL);
488}
489#else /* BITS_PER_LONG == 64 */
490/*
491 * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
492 * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
493 * and bottom words?
494 */
495static int ceph_parse_deleg_inos(void **p, void *end,
496 struct ceph_mds_session *s)
497{
498 u32 sets;
499
500 ceph_decode_32_safe(p, end, sets, bad);
501 if (sets)
502 ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
503 return 0;
504bad:
505 return -EIO;
506}
507
508u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
509{
510 return 0;
511}
512
513int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
514{
515 return 0;
516}
517#endif /* BITS_PER_LONG == 64 */
518
6e8575fa
SL		 519/*
520 * parse create results
521 */
522static int parse_reply_info_create(void **p, void *end,
523 struct ceph_mds_reply_info_parsed *info,
d4846487 524 u64 features, struct ceph_mds_session *s)
6e8575fa 525{
d4846487
JL		 526 int ret;
527
b37fe1f9
YZ		 528 if (features == (u64)-1 ||
529 (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {
6e8575fa 530 if (*p == end) {
d4846487 531 /* Malformed reply? */
6e8575fa 532 info->has_create_ino = false;
d4846487 533 } else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
6e8575fa 534 info->has_create_ino = true;
06a1ad43
JL		 535 /* struct_v, struct_compat, and len */
536 ceph_decode_skip_n(p, end, 2 + sizeof(u32), bad);
d4846487
JL		 537 ceph_decode_64_safe(p, end, info->ino, bad);
538 ret = ceph_parse_deleg_inos(p, end, s);
539 if (ret)
540 return ret;
541 } else {
542 /* legacy */
1d3f8723 543 ceph_decode_64_safe(p, end, info->ino, bad);
d4846487 544 info->has_create_ino = true;
6e8575fa 545 }
1d3f8723
JL		 546 } else {
547 if (*p != end)
548 goto bad;
6e8575fa
SL		 549 }
550
1d3f8723
JL		 551 /* Skip over any unrecognized fields */
552 *p = end;
6e8575fa 553 return 0;
6e8575fa
SL		 554bad:
555 return -EIO;
556}
557
6ddf5f16
MC		 558static int parse_reply_info_getvxattr(void **p, void *end,
559 struct ceph_mds_reply_info_parsed *info,
560 u64 features)
561{
562 u32 value_len;
563
564 ceph_decode_skip_8(p, end, bad); /* skip current version: 1 */
565 ceph_decode_skip_8(p, end, bad); /* skip first version: 1 */
566 ceph_decode_skip_32(p, end, bad); /* skip payload length */
567
568 ceph_decode_32_safe(p, end, value_len, bad);
569
570 if (value_len == end - *p) {
571 info->xattr_info.xattr_value = *p;
572 info->xattr_info.xattr_value_len = value_len;
573 *p = end;
574 return value_len;
575 }
576bad:
577 return -EIO;
578}
579
25933abd
HS		 580/*
581 * parse extra results
582 */
583static int parse_reply_info_extra(void **p, void *end,
14303d20 584 struct ceph_mds_reply_info_parsed *info,
d4846487 585 u64 features, struct ceph_mds_session *s)
25933abd 586{
6df8c9d8
JL		 587 u32 op = le32_to_cpu(info->head->op);
588
589 if (op == CEPH_MDS_OP_GETFILELOCK)
14303d20 590 return parse_reply_info_filelock(p, end, info, features);
6df8c9d8 591 else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
b37fe1f9 592 return parse_reply_info_readdir(p, end, info, features);
6df8c9d8 593 else if (op == CEPH_MDS_OP_CREATE)
d4846487 594 return parse_reply_info_create(p, end, info, features, s);
6ddf5f16
MC		 595 else if (op == CEPH_MDS_OP_GETVXATTR)
596 return parse_reply_info_getvxattr(p, end, info, features);
6e8575fa
SL		 597 else
598 return -EIO;
25933abd
HS		 599}
600
2f2dc053
SW		 601/*
602 * parse entire mds reply
603 */
d4846487 604static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg,
14303d20 605 struct ceph_mds_reply_info_parsed *info,
12b4629a 606 u64 features)
2f2dc053
SW		 607{
608 void *p, *end;
609 u32 len;
610 int err;
611
612 info->head = msg->front.iov_base;
613 p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
614 end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
615
616 /* trace */
617 ceph_decode_32_safe(&p, end, len, bad);
618 if (len > 0) {
32852a81 619 ceph_decode_need(&p, end, len, bad);
14303d20 620 err = parse_reply_info_trace(&p, p+len, info, features);
2f2dc053
SW		 621 if (err < 0)
622 goto out_bad;
623 }
624
25933abd 625 /* extra */
2f2dc053
SW		 626 ceph_decode_32_safe(&p, end, len, bad);
627 if (len > 0) {
32852a81 628 ceph_decode_need(&p, end, len, bad);
d4846487 629 err = parse_reply_info_extra(&p, p+len, info, features, s);
2f2dc053
SW		 630 if (err < 0)
631 goto out_bad;
632 }
633
634 /* snap blob */
635 ceph_decode_32_safe(&p, end, len, bad);
636 info->snapblob_len = len;
637 info->snapblob = p;
638 p += len;
639
640 if (p != end)
641 goto bad;
642 return 0;
643
644bad:
645 err = -EIO;
646out_bad:
647 pr_err("mds parse_reply err %d\n", err);
648 return err;
649}
650
651static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
652{
2a5beea3 653 if (!info->dir_entries)
54008399 654 return;
2a5beea3 655 free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size));
2f2dc053
SW		 656}
657
658
659/*
660 * sessions
661 */
a687ecaf 662const char *ceph_session_state_name(int s)
2f2dc053
SW		 663{
664 switch (s) {
665 case CEPH_MDS_SESSION_NEW: return "new";
666 case CEPH_MDS_SESSION_OPENING: return "opening";
667 case CEPH_MDS_SESSION_OPEN: return "open";
668 case CEPH_MDS_SESSION_HUNG: return "hung";
669 case CEPH_MDS_SESSION_CLOSING: return "closing";
4d681c2f 670 case CEPH_MDS_SESSION_CLOSED: return "closed";
44ca18f2 671 case CEPH_MDS_SESSION_RESTARTING: return "restarting";
2f2dc053 672 case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
fcff415c 673 case CEPH_MDS_SESSION_REJECTED: return "rejected";
2f2dc053
SW		 674 default: return "???";
675 }
676}
677
5b3248c6 678struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s)
2f2dc053 679{
9f358999 680 if (refcount_inc_not_zero(&s->s_ref))
2f2dc053 681 return s;
9f358999 682 return NULL;
2f2dc053
SW		 683}
684
685void ceph_put_mds_session(struct ceph_mds_session *s)
686{
7e65624d
JL		 687 if (IS_ERR_OR_NULL(s))
688 return;
689
3997c01d 690 if (refcount_dec_and_test(&s->s_ref)) {
6c4a1915 691 if (s->s_auth.authorizer)
6c1ea260 692 ceph_auth_destroy_authorizer(s->s_auth.authorizer);
88828190 693 WARN_ON(mutex_is_locked(&s->s_mutex));
d4846487 694 xa_destroy(&s->s_delegated_inos);
2f2dc053 695 kfree(s);
4e7a5dcd 696 }
2f2dc053
SW		 697}
698
699/*
700 * called under mdsc->mutex
701 */
702struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
703 int mds)
704{
d37b1d99 705 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
2f2dc053 706 return NULL;
5b3248c6 707 return ceph_get_mds_session(mdsc->sessions[mds]);
2f2dc053
SW		 708}
709
710static bool __have_session(struct ceph_mds_client *mdsc, int mds)
711{
98cfda81 712 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
2f2dc053 713 return false;
98cfda81
CX		 714 else
715 return true;
2f2dc053
SW		 716}
717
2600d2dd
SW		 718static int __verify_registered_session(struct ceph_mds_client *mdsc,
719 struct ceph_mds_session *s)
720{
721 if (s->s_mds >= mdsc->max_sessions ||
722 mdsc->sessions[s->s_mds] != s)
723 return -ENOENT;
724 return 0;
725}
726
2f2dc053
SW		 727/*
728 * create+register a new session for given mds.
729 * called under mdsc->mutex.
730 */
731static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
732 int mds)
733{
734 struct ceph_mds_session *s;
735
b38c9eb4 736 if (mds >= mdsc->mdsmap->possible_max_rank)
c338c07c
NY		 737 return ERR_PTR(-EINVAL);
738
2f2dc053 739 s = kzalloc(sizeof(*s), GFP_NOFS);
4736b009
DC		 740 if (!s)
741 return ERR_PTR(-ENOMEM);
47474d0b
CX		 742
743 if (mds >= mdsc->max_sessions) {
744 int newmax = 1 << get_count_order(mds + 1);
745 struct ceph_mds_session **sa;
746
747 dout("%s: realloc to %d\n", __func__, newmax);
748 sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
749 if (!sa)
750 goto fail_realloc;
751 if (mdsc->sessions) {
752 memcpy(sa, mdsc->sessions,
753 mdsc->max_sessions * sizeof(void *));
754 kfree(mdsc->sessions);
755 }
756 mdsc->sessions = sa;
757 mdsc->max_sessions = newmax;
758 }
759
760 dout("%s: mds%d\n", __func__, mds);
2f2dc053
SW		 761 s->s_mdsc = mdsc;
762 s->s_mds = mds;
763 s->s_state = CEPH_MDS_SESSION_NEW;
2f2dc053
SW		 764 mutex_init(&s->s_mutex);
765
b7a9e5dd 766 ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
2f2dc053 767
52d60f8e 768 atomic_set(&s->s_cap_gen, 1);
1ce208a6 769 s->s_cap_ttl = jiffies - 1;
d8fb02ab
AE		 770
771 spin_lock_init(&s->s_cap_lock);
2f2dc053 772 INIT_LIST_HEAD(&s->s_caps);
3997c01d 773 refcount_set(&s->s_ref, 1);
2f2dc053
SW		 774 INIT_LIST_HEAD(&s->s_waiting);
775 INIT_LIST_HEAD(&s->s_unsafe);
d4846487 776 xa_init(&s->s_delegated_inos);
2f2dc053 777 INIT_LIST_HEAD(&s->s_cap_releases);
e3ec8d68
YZ		 778 INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);
779
1cf03a68 780 INIT_LIST_HEAD(&s->s_cap_dirty);
2f2dc053 781 INIT_LIST_HEAD(&s->s_cap_flushing);
2f2dc053 782
2f2dc053 783 mdsc->sessions[mds] = s;
86d8f67b 784 atomic_inc(&mdsc->num_sessions);
3997c01d 785 refcount_inc(&s->s_ref); /* one ref to sessions[], one to caller */
42ce56e5 786
b7a9e5dd
SW		 787 ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
788 ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
42ce56e5 789
2f2dc053 790 return s;
42ce56e5
SW		 791
792fail_realloc:
793 kfree(s);
794 return ERR_PTR(-ENOMEM);
2f2dc053
SW		 795}
796
797/*
798 * called under mdsc->mutex
799 */
2600d2dd 800static void __unregister_session(struct ceph_mds_client *mdsc,
42ce56e5 801 struct ceph_mds_session *s)
2f2dc053 802{
2600d2dd
SW		 803 dout("__unregister_session mds%d %p\n", s->s_mds, s);
804 BUG_ON(mdsc->sessions[s->s_mds] != s);
42ce56e5
SW		 805 mdsc->sessions[s->s_mds] = NULL;
806 ceph_con_close(&s->s_con);
807 ceph_put_mds_session(s);
86d8f67b 808 atomic_dec(&mdsc->num_sessions);
2f2dc053
SW		 809}
810
811/*
812 * drop session refs in request.
813 *
814 * should be last request ref, or hold mdsc->mutex
815 */
816static void put_request_session(struct ceph_mds_request *req)
817{
818 if (req->r_session) {
819 ceph_put_mds_session(req->r_session);
820 req->r_session = NULL;
821 }
822}
823
59b312f3
XL		 824void ceph_mdsc_iterate_sessions(struct ceph_mds_client *mdsc,
825 void (*cb)(struct ceph_mds_session *),
826 bool check_state)
827{
828 int mds;
829
830 mutex_lock(&mdsc->mutex);
831 for (mds = 0; mds < mdsc->max_sessions; ++mds) {
832 struct ceph_mds_session *s;
833
834 s = __ceph_lookup_mds_session(mdsc, mds);
835 if (!s)
836 continue;
837
838 if (check_state && !check_session_state(s)) {
839 ceph_put_mds_session(s);
840 continue;
841 }
842
843 mutex_unlock(&mdsc->mutex);
844 cb(s);
845 ceph_put_mds_session(s);
846 mutex_lock(&mdsc->mutex);
847 }
848 mutex_unlock(&mdsc->mutex);
849}
850
153c8e6b 851void ceph_mdsc_release_request(struct kref *kref)
2f2dc053 852{
153c8e6b
SW		 853 struct ceph_mds_request *req = container_of(kref,
854 struct ceph_mds_request,
855 r_kref);
e64f44a8 856 ceph_mdsc_release_dir_caps_no_check(req);
54008399 857 destroy_reply_info(&req->r_reply_info);
153c8e6b
SW		 858 if (req->r_request)
859 ceph_msg_put(req->r_request);
54008399 860 if (req->r_reply)
153c8e6b 861 ceph_msg_put(req->r_reply);
153c8e6b 862 if (req->r_inode) {
41b02e1f 863 ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
23c2c76e 864 iput(req->r_inode);
153c8e6b 865 }
9c1c2b35 866 if (req->r_parent) {
3dd69aab 867 ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
23c2c76e 868 iput(req->r_parent);
9c1c2b35 869 }
23c2c76e 870 iput(req->r_target_inode);
153c8e6b
SW		 871 if (req->r_dentry)
872 dput(req->r_dentry);
844d87c3
SW		 873 if (req->r_old_dentry)
874 dput(req->r_old_dentry);
875 if (req->r_old_dentry_dir) {
41b02e1f
SW		 876 /*
877 * track (and drop pins for) r_old_dentry_dir
878 * separately, since r_old_dentry's d_parent may have
879 * changed between the dir mutex being dropped and
880 * this request being freed.
881 */
882 ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
883 CEPH_CAP_PIN);
23c2c76e 884 iput(req->r_old_dentry_dir);
2f2dc053 885 }
153c8e6b
SW		 886 kfree(req->r_path1);
887 kfree(req->r_path2);
7fe0cdeb 888 put_cred(req->r_cred);
25e6bae3
YZ		 889 if (req->r_pagelist)
890 ceph_pagelist_release(req->r_pagelist);
153c8e6b 891 put_request_session(req);
37151668 892 ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
428138c9 893 WARN_ON_ONCE(!list_empty(&req->r_wait));
058daab7 894 kmem_cache_free(ceph_mds_request_cachep, req);
2f2dc053
SW		 895}
896
fcd00b68
ID		 897DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)
898
2f2dc053
SW		 899/*
900 * lookup session, bump ref if found.
901 *
902 * called under mdsc->mutex.
903 */
fcd00b68
ID		 904static struct ceph_mds_request *
905lookup_get_request(struct ceph_mds_client *mdsc, u64 tid)
2f2dc053
SW		 906{
907 struct ceph_mds_request *req;
44ca18f2 908
fcd00b68
ID		 909 req = lookup_request(&mdsc->request_tree, tid);
910 if (req)
911 ceph_mdsc_get_request(req);
44ca18f2 912
fcd00b68 913 return req;
2f2dc053
SW		 914}
915
916/*
917 * Register an in-flight request, and assign a tid. Link to directory
918 * are modifying (if any).
919 *
920 * Called under mdsc->mutex.
921 */
922static void __register_request(struct ceph_mds_client *mdsc,
923 struct ceph_mds_request *req,
924 struct inode *dir)
925{
e30ee581
ZZ		 926 int ret = 0;
927
2f2dc053 928 req->r_tid = ++mdsc->last_tid;
e30ee581
ZZ		 929 if (req->r_num_caps) {
930 ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
931 req->r_num_caps);
932 if (ret < 0) {
933 pr_err("__register_request %p "
934 "failed to reserve caps: %d\n", req, ret);
935 /* set req->r_err to fail early from __do_request */
936 req->r_err = ret;
937 return;
938 }
939 }
2f2dc053
SW		 940 dout("__register_request %p tid %lld\n", req, req->r_tid);
941 ceph_mdsc_get_request(req);
fcd00b68 942 insert_request(&mdsc->request_tree, req);
2f2dc053 943
7fe0cdeb 944 req->r_cred = get_current_cred();
cb4276cc 945
e8a7b8b1
YZ		 946 if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
947 mdsc->oldest_tid = req->r_tid;
948
2f2dc053 949 if (dir) {
3db0a2fc
JL		 950 struct ceph_inode_info *ci = ceph_inode(dir);
951
3b663780 952 ihold(dir);
2f2dc053 953 req->r_unsafe_dir = dir;
3db0a2fc
JL		 954 spin_lock(&ci->i_unsafe_lock);
955 list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
956 spin_unlock(&ci->i_unsafe_lock);
2f2dc053
SW		 957 }
958}
959
960static void __unregister_request(struct ceph_mds_client *mdsc,
961 struct ceph_mds_request *req)
962{
963 dout("__unregister_request %p tid %lld\n", req, req->r_tid);
e8a7b8b1 964
df963ea8
JL		 965 /* Never leave an unregistered request on an unsafe list! */
966 list_del_init(&req->r_unsafe_item);
967
e8a7b8b1
YZ		 968 if (req->r_tid == mdsc->oldest_tid) {
969 struct rb_node *p = rb_next(&req->r_node);
970 mdsc->oldest_tid = 0;
971 while (p) {
972 struct ceph_mds_request *next_req =
973 rb_entry(p, struct ceph_mds_request, r_node);
974 if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
975 mdsc->oldest_tid = next_req->r_tid;
976 break;
977 }
978 p = rb_next(p);
979 }
980 }
981
fcd00b68 982 erase_request(&mdsc->request_tree, req);
2f2dc053 983
3db0a2fc 984 if (req->r_unsafe_dir) {
2f2dc053 985 struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
2f2dc053
SW		 986 spin_lock(&ci->i_unsafe_lock);
987 list_del_init(&req->r_unsafe_dir_item);
988 spin_unlock(&ci->i_unsafe_lock);
4c06ace8 989 }
bc2de10d
JL		 990 if (req->r_target_inode &&
991 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
68cd5b4b
YZ		 992 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
993 spin_lock(&ci->i_unsafe_lock);
994 list_del_init(&req->r_unsafe_target_item);
995 spin_unlock(&ci->i_unsafe_lock);
996 }
3b663780 997
4c06ace8 998 if (req->r_unsafe_dir) {
23c2c76e 999 iput(req->r_unsafe_dir);
3b663780 1000 req->r_unsafe_dir = NULL;
2f2dc053 1001 }
94aa8ae1 1002
fc55d2c9
YZ		 1003 complete_all(&req->r_safe_completion);
1004
94aa8ae1 1005 ceph_mdsc_put_request(req);
2f2dc053
SW		 1006}
1007
30c71233
JL		 1008/*
1009 * Walk back up the dentry tree until we hit a dentry representing a
1010 * non-snapshot inode. We do this using the rcu_read_lock (which must be held
1011 * when calling this) to ensure that the objects won't disappear while we're
1012 * working with them. Once we hit a candidate dentry, we attempt to take a
1013 * reference to it, and return that as the result.
1014 */
f1075480
DC		 1015static struct inode *get_nonsnap_parent(struct dentry *dentry)
1016{
1017 struct inode *inode = NULL;
30c71233
JL		 1018
1019 while (dentry && !IS_ROOT(dentry)) {
1020 inode = d_inode_rcu(dentry);
1021 if (!inode || ceph_snap(inode) == CEPH_NOSNAP)
1022 break;
1023 dentry = dentry->d_parent;
1024 }
1025 if (inode)
1026 inode = igrab(inode);
1027 return inode;
1028}
1029
2f2dc053
SW		 1030/*
1031 * Choose mds to send request to next. If there is a hint set in the
1032 * request (e.g., due to a prior forward hint from the mds), use that.
1033 * Otherwise, consult frag tree and/or caps to identify the
1034 * appropriate mds. If all else fails, choose randomly.
1035 *
1036 * Called under mdsc->mutex.
1037 */
1038static int __choose_mds(struct ceph_mds_client *mdsc,
c4853e97
XL		 1039 struct ceph_mds_request *req,
1040 bool *random)
2f2dc053
SW		 1041{
1042 struct inode *inode;
1043 struct ceph_inode_info *ci;
1044 struct ceph_cap *cap;
1045 int mode = req->r_direct_mode;
1046 int mds = -1;
1047 u32 hash = req->r_direct_hash;
bc2de10d 1048 bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
2f2dc053 1049
c4853e97
XL		 1050 if (random)
1051 *random = false;
1052
2f2dc053
SW		 1053 /*
1054 * is there a specific mds we should try? ignore hint if we have
1055 * no session and the mds is not up (active or recovering).
1056 */
1057 if (req->r_resend_mds >= 0 &&
1058 (__have_session(mdsc, req->r_resend_mds) ||
1059 ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
3c802092 1060 dout("%s using resend_mds mds%d\n", __func__,
2f2dc053
SW		 1061 req->r_resend_mds);
1062 return req->r_resend_mds;
1063 }
1064
1065 if (mode == USE_RANDOM_MDS)
1066 goto random;
1067
1068 inode = NULL;
1069 if (req->r_inode) {
5d37ca14
YZ		 1070 if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) {
1071 inode = req->r_inode;
1072 ihold(inode);
1073 } else {
38f340cc
YZ		 1074 /* req->r_dentry is non-null for LSSNAP request */
1075 rcu_read_lock();
1076 inode = get_nonsnap_parent(req->r_dentry);
1077 rcu_read_unlock();
3c802092 1078 dout("%s using snapdir's parent %p\n", __func__, inode);
5d37ca14 1079 }
38f340cc 1080 } else if (req->r_dentry) {
d79698da 1081 /* ignore race with rename; old or new d_parent is okay */
30c71233
JL		 1082 struct dentry *parent;
1083 struct inode *dir;
1084
1085 rcu_read_lock();
41883ba8 1086 parent = READ_ONCE(req->r_dentry->d_parent);
3dd69aab 1087 dir = req->r_parent ? : d_inode_rcu(parent);
eb6bb1c5 1088
30c71233
JL		 1089 if (!dir || dir->i_sb != mdsc->fsc->sb) {
1090 /* not this fs or parent went negative */
2b0143b5 1091 inode = d_inode(req->r_dentry);
30c71233
JL		 1092 if (inode)
1093 ihold(inode);
eb6bb1c5
SW		 1094 } else if (ceph_snap(dir) != CEPH_NOSNAP) {
1095 /* direct snapped/virtual snapdir requests
1096 * based on parent dir inode */
30c71233 1097 inode = get_nonsnap_parent(parent);
3c802092 1098 dout("%s using nonsnap parent %p\n", __func__, inode);
ca18bede 1099 } else {
eb6bb1c5 1100 /* dentry target */
2b0143b5 1101 inode = d_inode(req->r_dentry);
ca18bede
YZ		 1102 if (!inode || mode == USE_AUTH_MDS) {
1103 /* dir + name */
30c71233 1104 inode = igrab(dir);
ca18bede
YZ		 1105 hash = ceph_dentry_hash(dir, req->r_dentry);
1106 is_hash = true;
30c71233
JL		 1107 } else {
1108 ihold(inode);
ca18bede 1109 }
2f2dc053 1110 }
30c71233 1111 rcu_read_unlock();
2f2dc053 1112 }
eb6bb1c5 1113
3c802092
XL		 1114 dout("%s %p is_hash=%d (0x%x) mode %d\n", __func__, inode, (int)is_hash,
1115 hash, mode);
2f2dc053
SW		 1116 if (!inode)
1117 goto random;
1118 ci = ceph_inode(inode);
1119
1120 if (is_hash && S_ISDIR(inode->i_mode)) {
1121 struct ceph_inode_frag frag;
1122 int found;
1123
1124 ceph_choose_frag(ci, hash, &frag, &found);
1125 if (found) {
1126 if (mode == USE_ANY_MDS && frag.ndist > 0) {
1127 u8 r;
1128
1129 /* choose a random replica */
1130 get_random_bytes(&r, 1);
1131 r %= frag.ndist;
1132 mds = frag.dist[r];
3c802092
XL		 1133 dout("%s %p %llx.%llx frag %u mds%d (%d/%d)\n",
1134 __func__, inode, ceph_vinop(inode),
1135 frag.frag, mds, (int)r, frag.ndist);
d66bbd44 1136 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
5d47648f
XL		 1137 CEPH_MDS_STATE_ACTIVE &&
1138 !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))
30c71233 1139 goto out;
2f2dc053
SW		 1140 }
1141
1142 /* since this file/dir wasn't known to be
1143 * replicated, then we want to look for the
1144 * authoritative mds. */
2f2dc053
SW		 1145 if (frag.mds >= 0) {
1146 /* choose auth mds */
1147 mds = frag.mds;
3c802092
XL		 1148 dout("%s %p %llx.%llx frag %u mds%d (auth)\n",
1149 __func__, inode, ceph_vinop(inode),
1150 frag.frag, mds);
d66bbd44 1151 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
5d47648f 1152 CEPH_MDS_STATE_ACTIVE) {
224c7b67 1153 if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,
5d47648f
XL		 1154 mds))
1155 goto out;
1156 }
2f2dc053 1157 }
5d47648f 1158 mode = USE_AUTH_MDS;
2f2dc053
SW		 1159 }
1160 }
1161
be655596 1162 spin_lock(&ci->i_ceph_lock);
2f2dc053
SW		 1163 cap = NULL;
1164 if (mode == USE_AUTH_MDS)
1165 cap = ci->i_auth_cap;
1166 if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
1167 cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
1168 if (!cap) {
be655596 1169 spin_unlock(&ci->i_ceph_lock);
23c2c76e 1170 iput(inode);
2f2dc053
SW		 1171 goto random;
1172 }
1173 mds = cap->session->s_mds;
3c802092 1174 dout("%s %p %llx.%llx mds%d (%scap %p)\n", __func__,
2f2dc053
SW		 1175 inode, ceph_vinop(inode), mds,
1176 cap == ci->i_auth_cap ? "auth " : "", cap);
be655596 1177 spin_unlock(&ci->i_ceph_lock);
30c71233 1178out:
23c2c76e 1179 iput(inode);
2f2dc053
SW		 1180 return mds;
1181
1182random:
c4853e97
XL		 1183 if (random)
1184 *random = true;
1185
2f2dc053 1186 mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
3c802092 1187 dout("%s chose random mds%d\n", __func__, mds);
2f2dc053
SW		 1188 return mds;
1189}
1190
1191
1192/*
1193 * session messages
1194 */
fba97e80 1195struct ceph_msg *ceph_create_session_msg(u32 op, u64 seq)
2f2dc053
SW		 1196{
1197 struct ceph_msg *msg;
1198 struct ceph_mds_session_head *h;
1199
b61c2763
SW		 1200 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
1201 false);
a79832f2 1202 if (!msg) {
fba97e80
XL		 1203 pr_err("ENOMEM creating session %s msg\n",
1204 ceph_session_op_name(op));
a79832f2 1205 return NULL;
2f2dc053
SW		 1206 }
1207 h = msg->front.iov_base;
1208 h->op = cpu_to_le32(op);
1209 h->seq = cpu_to_le64(seq);
dbd0c8bf
JS		 1210
1211 return msg;
1212}
1213
9ba1e224
XL		 1214static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
1215#define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)
b682c6d4 1216static int encode_supported_features(void **p, void *end)
342ce182 1217{
9ba1e224 1218 static const size_t count = ARRAY_SIZE(feature_bits);
342ce182
YZ		 1219
1220 if (count > 0) {
1221 size_t i;
9ba1e224 1222 size_t size = FEATURE_BYTES(count);
342ce182 1223
b682c6d4
XL		 1224 if (WARN_ON_ONCE(*p + 4 + size > end))
1225 return -ERANGE;
1226
342ce182
YZ		 1227 ceph_encode_32(p, size);
1228 memset(*p, 0, size);
1229 for (i = 0; i < count; i++)
9ba1e224 1230 ((unsigned char*)(*p))[i / 8] |= BIT(feature_bits[i] % 8);
342ce182
YZ		 1231 *p += size;
1232 } else {
b682c6d4
XL		 1233 if (WARN_ON_ONCE(*p + 4 > end))
1234 return -ERANGE;
1235
342ce182
YZ		 1236 ceph_encode_32(p, 0);
1237 }
b682c6d4
XL		 1238
1239 return 0;
342ce182
YZ		 1240}
1241
3b4168dd
XL		 1242static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
1243#define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
1244static int encode_metric_spec(void **p, void *end)
1245{
1246 static const size_t count = ARRAY_SIZE(metric_bits);
1247
1248 /* header */
1249 if (WARN_ON_ONCE(*p + 2 > end))
1250 return -ERANGE;
1251
1252 ceph_encode_8(p, 1); /* version */
1253 ceph_encode_8(p, 1); /* compat */
1254
1255 if (count > 0) {
1256 size_t i;
1257 size_t size = METRIC_BYTES(count);
1258
1259 if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
1260 return -ERANGE;
1261
1262 /* metric spec info length */
1263 ceph_encode_32(p, 4 + size);
1264
1265 /* metric spec */
1266 ceph_encode_32(p, size);
1267 memset(*p, 0, size);
1268 for (i = 0; i < count; i++)
1269 ((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8);
1270 *p += size;
1271 } else {
1272 if (WARN_ON_ONCE(*p + 4 + 4 > end))
1273 return -ERANGE;
1274
1275 /* metric spec info length */
1276 ceph_encode_32(p, 4);
1277 /* metric spec */
1278 ceph_encode_32(p, 0);
1279 }
1280
1281 return 0;
1282}
1283
dbd0c8bf
JS		 1284/*
1285 * session message, specialization for CEPH_SESSION_REQUEST_OPEN
1286 * to include additional client metadata fields.
1287 */
1288static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
1289{
1290 struct ceph_msg *msg;
1291 struct ceph_mds_session_head *h;
4a756db2 1292 int i;
342ce182 1293 int extra_bytes = 0;
dbd0c8bf
JS		 1294 int metadata_key_count = 0;
1295 struct ceph_options *opt = mdsc->fsc->client->options;
3f384954 1296 struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;
9ba1e224 1297 size_t size, count;
342ce182 1298 void *p, *end;
b682c6d4 1299 int ret;
dbd0c8bf 1300
a6a5ce4f 1301 const char* metadata[][2] = {
717e6f28
YZ		 1302 {"hostname", mdsc->nodename},
1303 {"kernel_version", init_utsname()->release},
3f384954
YZ		 1304 {"entity_id", opt->name ? : ""},
1305 {"root", fsopt->server_path ? : "/"},
dbd0c8bf
JS		 1306 {NULL, NULL}
1307 };
1308
1309 /* Calculate serialized length of metadata */
342ce182 1310 extra_bytes = 4; /* map length */
d37b1d99 1311 for (i = 0; metadata[i][0]; ++i) {
342ce182 1312 extra_bytes += 8 + strlen(metadata[i][0]) +
dbd0c8bf
JS		 1313 strlen(metadata[i][1]);
1314 metadata_key_count++;
1315 }
9ba1e224 1316
342ce182 1317 /* supported feature */
9ba1e224
XL		 1318 size = 0;
1319 count = ARRAY_SIZE(feature_bits);
1320 if (count > 0)
1321 size = FEATURE_BYTES(count);
1322 extra_bytes += 4 + size;
dbd0c8bf 1323
3b4168dd
XL		 1324 /* metric spec */
1325 size = 0;
1326 count = ARRAY_SIZE(metric_bits);
1327 if (count > 0)
1328 size = METRIC_BYTES(count);
1329 extra_bytes += 2 + 4 + 4 + size;
1330
dbd0c8bf 1331 /* Allocate the message */
342ce182 1332 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,
dbd0c8bf
JS		 1333 GFP_NOFS, false);
1334 if (!msg) {
fba97e80 1335 pr_err("ENOMEM creating session open msg\n");
b682c6d4 1336 return ERR_PTR(-ENOMEM);
dbd0c8bf 1337 }
342ce182
YZ		 1338 p = msg->front.iov_base;
1339 end = p + msg->front.iov_len;
1340
1341 h = p;
dbd0c8bf
JS		 1342 h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
1343 h->seq = cpu_to_le64(seq);
1344
1345 /*
1346 * Serialize client metadata into waiting buffer space, using
1347 * the format that userspace expects for map<string, string>
7cfa0313 1348 *
3b4168dd 1349 * ClientSession messages with metadata are v4
dbd0c8bf 1350 */
3b4168dd 1351 msg->hdr.version = cpu_to_le16(4);
7cfa0313 1352 msg->hdr.compat_version = cpu_to_le16(1);
dbd0c8bf
JS		 1353
1354 /* The write pointer, following the session_head structure */
342ce182 1355 p += sizeof(*h);
dbd0c8bf
JS		 1356
1357 /* Number of entries in the map */
1358 ceph_encode_32(&p, metadata_key_count);
1359
1360 /* Two length-prefixed strings for each entry in the map */
d37b1d99 1361 for (i = 0; metadata[i][0]; ++i) {
dbd0c8bf
JS		 1362 size_t const key_len = strlen(metadata[i][0]);
1363 size_t const val_len = strlen(metadata[i][1]);
1364
1365 ceph_encode_32(&p, key_len);
1366 memcpy(p, metadata[i][0], key_len);
1367 p += key_len;
1368 ceph_encode_32(&p, val_len);
1369 memcpy(p, metadata[i][1], val_len);
1370 p += val_len;
1371 }
1372
b682c6d4
XL		 1373 ret = encode_supported_features(&p, end);
1374 if (ret) {
1375 pr_err("encode_supported_features failed!\n");
1376 ceph_msg_put(msg);
1377 return ERR_PTR(ret);
1378 }
1379
3b4168dd
XL		 1380 ret = encode_metric_spec(&p, end);
1381 if (ret) {
1382 pr_err("encode_metric_spec failed!\n");
1383 ceph_msg_put(msg);
1384 return ERR_PTR(ret);
1385 }
1386
342ce182
YZ		 1387 msg->front.iov_len = p - msg->front.iov_base;
1388 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1389
2f2dc053
SW		 1390 return msg;
1391}
1392
1393/*
1394 * send session open request.
1395 *
1396 * called under mdsc->mutex
1397 */
1398static int __open_session(struct ceph_mds_client *mdsc,
1399 struct ceph_mds_session *session)
1400{
1401 struct ceph_msg *msg;
1402 int mstate;
1403 int mds = session->s_mds;
2f2dc053
SW		 1404
1405 /* wait for mds to go active? */
1406 mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
1407 dout("open_session to mds%d (%s)\n", mds,
1408 ceph_mds_state_name(mstate));
1409 session->s_state = CEPH_MDS_SESSION_OPENING;
1410 session->s_renew_requested = jiffies;
1411
1412 /* send connect message */
dbd0c8bf 1413 msg = create_session_open_msg(mdsc, session->s_seq);
b682c6d4
XL		 1414 if (IS_ERR(msg))
1415 return PTR_ERR(msg);
2f2dc053 1416 ceph_con_send(&session->s_con, msg);
2f2dc053
SW		 1417 return 0;
1418}
1419
ed0552a1
SW		 1420/*
1421 * open sessions for any export targets for the given mds
1422 *
1423 * called under mdsc->mutex
1424 */
5d72d13c
YZ		 1425static struct ceph_mds_session *
1426__open_export_target_session(struct ceph_mds_client *mdsc, int target)
1427{
1428 struct ceph_mds_session *session;
b682c6d4 1429 int ret;
5d72d13c
YZ		 1430
1431 session = __ceph_lookup_mds_session(mdsc, target);
1432 if (!session) {
1433 session = register_session(mdsc, target);
1434 if (IS_ERR(session))
1435 return session;
1436 }
1437 if (session->s_state == CEPH_MDS_SESSION_NEW ||
b682c6d4
XL		 1438 session->s_state == CEPH_MDS_SESSION_CLOSING) {
1439 ret = __open_session(mdsc, session);
1440 if (ret)
1441 return ERR_PTR(ret);
1442 }
5d72d13c
YZ		 1443
1444 return session;
1445}
1446
1447struct ceph_mds_session *
1448ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
1449{
1450 struct ceph_mds_session *session;
1451
1452 dout("open_export_target_session to mds%d\n", target);
1453
1454 mutex_lock(&mdsc->mutex);
1455 session = __open_export_target_session(mdsc, target);
1456 mutex_unlock(&mdsc->mutex);
1457
1458 return session;
1459}
1460
ed0552a1
SW		 1461static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
1462 struct ceph_mds_session *session)
1463{
1464 struct ceph_mds_info *mi;
1465 struct ceph_mds_session *ts;
1466 int i, mds = session->s_mds;
ed0552a1 1467
b38c9eb4 1468 if (mds >= mdsc->mdsmap->possible_max_rank)
ed0552a1 1469 return;
5d72d13c 1470
ed0552a1
SW		 1471 mi = &mdsc->mdsmap->m_info[mds];
1472 dout("open_export_target_sessions for mds%d (%d targets)\n",
1473 session->s_mds, mi->num_export_targets);
1474
1475 for (i = 0; i < mi->num_export_targets; i++) {
5d72d13c 1476 ts = __open_export_target_session(mdsc, mi->export_targets[i]);
7e65624d 1477 ceph_put_mds_session(ts);
ed0552a1
SW		 1478 }
1479}
1480
154f42c2
SW		 1481void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
1482 struct ceph_mds_session *session)
1483{
1484 mutex_lock(&mdsc->mutex);
1485 __open_export_target_sessions(mdsc, session);
1486 mutex_unlock(&mdsc->mutex);
1487}
1488
2f2dc053
SW		 1489/*
1490 * session caps
1491 */
1492
c8a96a31
JL		 1493static void detach_cap_releases(struct ceph_mds_session *session,
1494 struct list_head *target)
2f2dc053 1495{
c8a96a31
JL		 1496 lockdep_assert_held(&session->s_cap_lock);
1497
1498 list_splice_init(&session->s_cap_releases, target);
745a8e3b 1499 session->s_num_cap_releases = 0;
c8a96a31
JL		 1500 dout("dispose_cap_releases mds%d\n", session->s_mds);
1501}
2f2dc053 1502
c8a96a31
JL		 1503static void dispose_cap_releases(struct ceph_mds_client *mdsc,
1504 struct list_head *dispose)
1505{
1506 while (!list_empty(dispose)) {
745a8e3b
YZ		 1507 struct ceph_cap *cap;
1508 /* zero out the in-progress message */
c8a96a31 1509 cap = list_first_entry(dispose, struct ceph_cap, session_caps);
745a8e3b
YZ		 1510 list_del(&cap->session_caps);
1511 ceph_put_cap(mdsc, cap);
2f2dc053 1512 }
2f2dc053
SW		 1513}
1514
1c841a96
YZ		 1515static void cleanup_session_requests(struct ceph_mds_client *mdsc,
1516 struct ceph_mds_session *session)
1517{
1518 struct ceph_mds_request *req;
1519 struct rb_node *p;
1520
1521 dout("cleanup_session_requests mds%d\n", session->s_mds);
1522 mutex_lock(&mdsc->mutex);
1523 while (!list_empty(&session->s_unsafe)) {
1524 req = list_first_entry(&session->s_unsafe,
1525 struct ceph_mds_request, r_unsafe_item);
3e0708b9
YZ		 1526 pr_warn_ratelimited(" dropping unsafe request %llu\n",
1527 req->r_tid);
1bd85aa6
JL		 1528 if (req->r_target_inode)
1529 mapping_set_error(req->r_target_inode->i_mapping, -EIO);
1530 if (req->r_unsafe_dir)
1531 mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);
1c841a96
YZ		 1532 __unregister_request(mdsc, req);
1533 }
1534 /* zero r_attempts, so kick_requests() will re-send requests */
1535 p = rb_first(&mdsc->request_tree);
1536 while (p) {
1537 req = rb_entry(p, struct ceph_mds_request, r_node);
1538 p = rb_next(p);
1539 if (req->r_session &&
1540 req->r_session->s_mds == session->s_mds)
1541 req->r_attempts = 0;
1542 }
1543 mutex_unlock(&mdsc->mutex);
1544}
1545
2f2dc053 1546/*
f818a736
SW		 1547 * Helper to safely iterate over all caps associated with a session, with
1548 * special care taken to handle a racing __ceph_remove_cap().
2f2dc053 1549 *
f818a736 1550 * Caller must hold session s_mutex.
2f2dc053 1551 */
f5d77269
JL		 1552int ceph_iterate_session_caps(struct ceph_mds_session *session,
1553 int (*cb)(struct inode *, struct ceph_cap *,
1554 void *), void *arg)
2f2dc053 1555{
7c1332b8
SW		 1556 struct list_head *p;
1557 struct ceph_cap *cap;
1558 struct inode *inode, *last_inode = NULL;
1559 struct ceph_cap *old_cap = NULL;
2f2dc053
SW		 1560 int ret;
1561
1562 dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
1563 spin_lock(&session->s_cap_lock);
7c1332b8
SW		 1564 p = session->s_caps.next;
1565 while (p != &session->s_caps) {
1566 cap = list_entry(p, struct ceph_cap, session_caps);
2f2dc053 1567 inode = igrab(&cap->ci->vfs_inode);
7c1332b8
SW		 1568 if (!inode) {
1569 p = p->next;
2f2dc053 1570 continue;
7c1332b8
SW		 1571 }
1572 session->s_cap_iterator = cap;
2f2dc053 1573 spin_unlock(&session->s_cap_lock);
7c1332b8
SW		 1574
1575 if (last_inode) {
23c2c76e 1576 iput(last_inode);
7c1332b8
SW		 1577 last_inode = NULL;
1578 }
1579 if (old_cap) {
37151668 1580 ceph_put_cap(session->s_mdsc, old_cap);
7c1332b8
SW		 1581 old_cap = NULL;
1582 }
1583
2f2dc053 1584 ret = cb(inode, cap, arg);
7c1332b8
SW		 1585 last_inode = inode;
1586
2f2dc053 1587 spin_lock(&session->s_cap_lock);
7c1332b8 1588 p = p->next;
d37b1d99 1589 if (!cap->ci) {
7c1332b8
SW		 1590 dout("iterate_session_caps finishing cap %p removal\n",
1591 cap);
1592 BUG_ON(cap->session != session);
745a8e3b 1593 cap->session = NULL;
7c1332b8
SW		 1594 list_del_init(&cap->session_caps);
1595 session->s_nr_caps--;
4f1d756d 1596 atomic64_dec(&session->s_mdsc->metric.total_caps);
e3ec8d68
YZ		 1597 if (cap->queue_release)
1598 __ceph_queue_cap_release(session, cap);
1599 else
745a8e3b 1600 old_cap = cap; /* put_cap it w/o locks held */
7c1332b8 1601 }
5dacf091
SW		 1602 if (ret < 0)
1603 goto out;
2f2dc053 1604 }
5dacf091
SW		 1605 ret = 0;
1606out:
7c1332b8 1607 session->s_cap_iterator = NULL;
2f2dc053 1608 spin_unlock(&session->s_cap_lock);
7c1332b8 1609
23c2c76e 1610 iput(last_inode);
7c1332b8 1611 if (old_cap)
37151668 1612 ceph_put_cap(session->s_mdsc, old_cap);
7c1332b8 1613
5dacf091 1614 return ret;
2f2dc053
SW		 1615}
1616
1617static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
6c99f254 1618 void *arg)
2f2dc053
SW		 1619{
1620 struct ceph_inode_info *ci = ceph_inode(inode);
6c93df5d 1621 bool invalidate = false;
36e6da98 1622 int iputs;
6c99f254 1623
2f2dc053
SW		 1624 dout("removing cap %p, ci is %p, inode is %p\n",
1625 cap, ci, &ci->vfs_inode);
be655596 1626 spin_lock(&ci->i_ceph_lock);
36e6da98 1627 iputs = ceph_purge_inode_cap(inode, cap, &invalidate);
be655596 1628 spin_unlock(&ci->i_ceph_lock);
77310320
YZ		 1629
1630 wake_up_all(&ci->i_cap_wq);
6c93df5d
YZ		 1631 if (invalidate)
1632 ceph_queue_invalidate(inode);
36e6da98 1633 while (iputs--)
a6d37ccd 1634 iput(inode);
2f2dc053
SW		 1635 return 0;
1636}
1637
1638/*
1639 * caller must hold session s_mutex
1640 */
1641static void remove_session_caps(struct ceph_mds_session *session)
1642{
6c93df5d
YZ		 1643 struct ceph_fs_client *fsc = session->s_mdsc->fsc;
1644 struct super_block *sb = fsc->sb;
c8a96a31
JL		 1645 LIST_HEAD(dispose);
1646
2f2dc053 1647 dout("remove_session_caps on %p\n", session);
f5d77269 1648 ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);
6f60f889 1649
c8799fc4
YZ		 1650 wake_up_all(&fsc->mdsc->cap_flushing_wq);
1651
6f60f889
YZ		 1652 spin_lock(&session->s_cap_lock);
1653 if (session->s_nr_caps > 0) {
6f60f889
YZ		 1654 struct inode *inode;
1655 struct ceph_cap *cap, *prev = NULL;
1656 struct ceph_vino vino;
1657 /*
1658 * iterate_session_caps() skips inodes that are being
1659 * deleted, we need to wait until deletions are complete.
1660 * __wait_on_freeing_inode() is designed for the job,
1661 * but it is not exported, so use lookup inode function
1662 * to access it.
1663 */
1664 while (!list_empty(&session->s_caps)) {
1665 cap = list_entry(session->s_caps.next,
1666 struct ceph_cap, session_caps);
1667 if (cap == prev)
1668 break;
1669 prev = cap;
1670 vino = cap->ci->i_vino;
1671 spin_unlock(&session->s_cap_lock);
1672
ed284c49 1673 inode = ceph_find_inode(sb, vino);
23c2c76e 1674 iput(inode);
6f60f889
YZ		 1675
1676 spin_lock(&session->s_cap_lock);
1677 }
1678 }
745a8e3b
YZ		 1679
1680 // drop cap expires and unlock s_cap_lock
c8a96a31 1681 detach_cap_releases(session, &dispose);
6f60f889 1682
2f2dc053 1683 BUG_ON(session->s_nr_caps > 0);
6c99f254 1684 BUG_ON(!list_empty(&session->s_cap_flushing));
c8a96a31
JL		 1685 spin_unlock(&session->s_cap_lock);
1686 dispose_cap_releases(session->s_mdsc, &dispose);
2f2dc053
SW		 1687}
1688
d2f8bb27
YZ		 1689enum {
1690 RECONNECT,
1691 RENEWCAPS,
1692 FORCE_RO,
1693};
1694
2f2dc053
SW		 1695/*
1696 * wake up any threads waiting on this session's caps. if the cap is
1697 * old (didn't get renewed on the client reconnect), remove it now.
1698 *
1699 * caller must hold s_mutex.
1700 */
1701static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
1702 void *arg)
1703{
0dc2570f 1704 struct ceph_inode_info *ci = ceph_inode(inode);
d2f8bb27 1705 unsigned long ev = (unsigned long)arg;
0dc2570f 1706
d2f8bb27 1707 if (ev == RECONNECT) {
be655596 1708 spin_lock(&ci->i_ceph_lock);
0dc2570f
SW		 1709 ci->i_wanted_max_size = 0;
1710 ci->i_requested_max_size = 0;
be655596 1711 spin_unlock(&ci->i_ceph_lock);
d2f8bb27 1712 } else if (ev == RENEWCAPS) {
52d60f8e 1713 if (cap->cap_gen < atomic_read(&cap->session->s_cap_gen)) {
d2f8bb27
YZ		 1714 /* mds did not re-issue stale cap */
1715 spin_lock(&ci->i_ceph_lock);
1716 cap->issued = cap->implemented = CEPH_CAP_PIN;
d2f8bb27
YZ		 1717 spin_unlock(&ci->i_ceph_lock);
1718 }
1719 } else if (ev == FORCE_RO) {
0dc2570f 1720 }
e5360309 1721 wake_up_all(&ci->i_cap_wq);
2f2dc053
SW		 1722 return 0;
1723}
1724
d2f8bb27 1725static void wake_up_session_caps(struct ceph_mds_session *session, int ev)
2f2dc053
SW		 1726{
1727 dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
f5d77269
JL		 1728 ceph_iterate_session_caps(session, wake_up_session_cb,
1729 (void *)(unsigned long)ev);
2f2dc053
SW		 1730}
1731
1732/*
1733 * Send periodic message to MDS renewing all currently held caps. The
1734 * ack will reset the expiration for all caps from this session.
1735 *
1736 * caller holds s_mutex
1737 */
1738static int send_renew_caps(struct ceph_mds_client *mdsc,
1739 struct ceph_mds_session *session)
1740{
1741 struct ceph_msg *msg;
1742 int state;
1743
1744 if (time_after_eq(jiffies, session->s_cap_ttl) &&
1745 time_after_eq(session->s_cap_ttl, session->s_renew_requested))
1746 pr_info("mds%d caps stale\n", session->s_mds);
e4cb4cb8 1747 session->s_renew_requested = jiffies;
2f2dc053
SW		 1748
1749 /* do not try to renew caps until a recovering mds has reconnected
1750 * with its clients. */
1751 state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
1752 if (state < CEPH_MDS_STATE_RECONNECT) {
1753 dout("send_renew_caps ignoring mds%d (%s)\n",
1754 session->s_mds, ceph_mds_state_name(state));
1755 return 0;
1756 }
1757
1758 dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
1759 ceph_mds_state_name(state));
fba97e80
XL		 1760 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
1761 ++session->s_renew_seq);
a79832f2
SW		 1762 if (!msg)
1763 return -ENOMEM;
2f2dc053
SW		 1764 ceph_con_send(&session->s_con, msg);
1765 return 0;
1766}
1767
186e4f7a
YZ		 1768static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
1769 struct ceph_mds_session *session, u64 seq)
1770{
1771 struct ceph_msg *msg;
1772
1773 dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
a687ecaf 1774 session->s_mds, ceph_session_state_name(session->s_state), seq);
fba97e80 1775 msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
186e4f7a
YZ		 1776 if (!msg)
1777 return -ENOMEM;
1778 ceph_con_send(&session->s_con, msg);
1779 return 0;
1780}
1781
1782
2f2dc053
SW		 1783/*
1784 * Note new cap ttl, and any transition from stale -> not stale (fresh?).
0dc2570f
SW		 1785 *
1786 * Called under session->s_mutex
2f2dc053
SW		 1787 */
1788static void renewed_caps(struct ceph_mds_client *mdsc,
1789 struct ceph_mds_session *session, int is_renew)
1790{
1791 int was_stale;
1792 int wake = 0;
1793
1794 spin_lock(&session->s_cap_lock);
1ce208a6 1795 was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
2f2dc053
SW		 1796
1797 session->s_cap_ttl = session->s_renew_requested +
1798 mdsc->mdsmap->m_session_timeout*HZ;
1799
1800 if (was_stale) {
1801 if (time_before(jiffies, session->s_cap_ttl)) {
1802 pr_info("mds%d caps renewed\n", session->s_mds);
1803 wake = 1;
1804 } else {
1805 pr_info("mds%d caps still stale\n", session->s_mds);
1806 }
1807 }
1808 dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
1809 session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
1810 time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
1811 spin_unlock(&session->s_cap_lock);
1812
1813 if (wake)
d2f8bb27 1814 wake_up_session_caps(session, RENEWCAPS);
2f2dc053
SW		 1815}
1816
1817/*
1818 * send a session close request
1819 */
3e699bd8 1820static int request_close_session(struct ceph_mds_session *session)
2f2dc053
SW		 1821{
1822 struct ceph_msg *msg;
2f2dc053
SW		 1823
1824 dout("request_close_session mds%d state %s seq %lld\n",
a687ecaf 1825 session->s_mds, ceph_session_state_name(session->s_state),
2f2dc053 1826 session->s_seq);
fba97e80
XL		 1827 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE,
1828 session->s_seq);
a79832f2
SW		 1829 if (!msg)
1830 return -ENOMEM;
1831 ceph_con_send(&session->s_con, msg);
fcff415c 1832 return 1;
2f2dc053
SW		 1833}
1834
1835/*
1836 * Called with s_mutex held.
1837 */
1838static int __close_session(struct ceph_mds_client *mdsc,
1839 struct ceph_mds_session *session)
1840{
1841 if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
1842 return 0;
1843 session->s_state = CEPH_MDS_SESSION_CLOSING;
3e699bd8 1844 return request_close_session(session);
2f2dc053
SW		 1845}
1846
040d7860
YZ		 1847static bool drop_negative_children(struct dentry *dentry)
1848{
1849 struct dentry *child;
1850 bool all_negative = true;
1851
1852 if (!d_is_dir(dentry))
1853 goto out;
1854
1855 spin_lock(&dentry->d_lock);
1856 list_for_each_entry(child, &dentry->d_subdirs, d_child) {
1857 if (d_really_is_positive(child)) {
1858 all_negative = false;
1859 break;
1860 }
1861 }
1862 spin_unlock(&dentry->d_lock);
1863
1864 if (all_negative)
1865 shrink_dcache_parent(dentry);
1866out:
1867 return all_negative;
1868}
1869
2f2dc053
SW		 1870/*
1871 * Trim old(er) caps.
1872 *
1873 * Because we can't cache an inode without one or more caps, we do
1874 * this indirectly: if a cap is unused, we prune its aliases, at which
1875 * point the inode will hopefully get dropped to.
1876 *
1877 * Yes, this is a bit sloppy. Our only real goal here is to respond to
1878 * memory pressure from the MDS, though, so it needn't be perfect.
1879 */
1880static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
1881{
533a2818 1882 int *remaining = arg;
2f2dc053 1883 struct ceph_inode_info *ci = ceph_inode(inode);
979abfdd 1884 int used, wanted, oissued, mine;
2f2dc053 1885
533a2818 1886 if (*remaining <= 0)
2f2dc053
SW		 1887 return -1;
1888
be655596 1889 spin_lock(&ci->i_ceph_lock);
2f2dc053
SW		 1890 mine = cap->issued | cap->implemented;
1891 used = __ceph_caps_used(ci);
979abfdd 1892 wanted = __ceph_caps_file_wanted(ci);
2f2dc053
SW		 1893 oissued = __ceph_caps_issued_other(ci, cap);
1894
979abfdd 1895 dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
2f2dc053 1896 inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
979abfdd
YZ		 1897 ceph_cap_string(used), ceph_cap_string(wanted));
1898 if (cap == ci->i_auth_cap) {
622f3e25
YZ		 1899 if (ci->i_dirty_caps || ci->i_flushing_caps ||
1900 !list_empty(&ci->i_cap_snaps))
979abfdd
YZ		 1901 goto out;
1902 if ((used | wanted) & CEPH_CAP_ANY_WR)
1903 goto out;
89aa5930
YZ		 1904 /* Note: it's possible that i_filelock_ref becomes non-zero
1905 * after dropping auth caps. It doesn't hurt because reply
1906 * of lock mds request will re-add auth caps. */
1907 if (atomic_read(&ci->i_filelock_ref) > 0)
1908 goto out;
979abfdd 1909 }
5e804ac4
YZ		 1910 /* The inode has cached pages, but it's no longer used.
1911 * we can safely drop it */
525d15e8
YZ		 1912 if (S_ISREG(inode->i_mode) &&
1913 wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
5e804ac4
YZ		 1914 !(oissued & CEPH_CAP_FILE_CACHE)) {
1915 used = 0;
1916 oissued = 0;
1917 }
979abfdd 1918 if ((used | wanted) & ~oissued & mine)
2f2dc053
SW		 1919 goto out; /* we need these caps */
1920
2f2dc053
SW		 1921 if (oissued) {
1922 /* we aren't the only cap.. just remove us */
a76d0a9c 1923 ceph_remove_cap(cap, true);
533a2818 1924 (*remaining)--;
2f2dc053 1925 } else {
040d7860 1926 struct dentry *dentry;
5e804ac4 1927 /* try dropping referring dentries */
be655596 1928 spin_unlock(&ci->i_ceph_lock);
040d7860
YZ		 1929 dentry = d_find_any_alias(inode);
1930 if (dentry && drop_negative_children(dentry)) {
1931 int count;
1932 dput(dentry);
1933 d_prune_aliases(inode);
1934 count = atomic_read(&inode->i_count);
1935 if (count == 1)
533a2818 1936 (*remaining)--;
040d7860
YZ		 1937 dout("trim_caps_cb %p cap %p pruned, count now %d\n",
1938 inode, cap, count);
1939 } else {
1940 dput(dentry);
1941 }
2f2dc053
SW		 1942 return 0;
1943 }
1944
1945out:
be655596 1946 spin_unlock(&ci->i_ceph_lock);
2f2dc053
SW		 1947 return 0;
1948}
1949
1950/*
1951 * Trim session cap count down to some max number.
1952 */
e30ee581
ZZ		 1953int ceph_trim_caps(struct ceph_mds_client *mdsc,
1954 struct ceph_mds_session *session,
1955 int max_caps)
2f2dc053
SW		 1956{
1957 int trim_caps = session->s_nr_caps - max_caps;
1958
1959 dout("trim_caps mds%d start: %d / %d, trim %d\n",
1960 session->s_mds, session->s_nr_caps, max_caps, trim_caps);
1961 if (trim_caps > 0) {
533a2818
JL		 1962 int remaining = trim_caps;
1963
1964 ceph_iterate_session_caps(session, trim_caps_cb, &remaining);
2f2dc053
SW		 1965 dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
1966 session->s_mds, session->s_nr_caps, max_caps,
533a2818 1967 trim_caps - remaining);
2f2dc053 1968 }
a56371d9 1969
e3ec8d68 1970 ceph_flush_cap_releases(mdsc, session);
2f2dc053
SW		 1971 return 0;
1972}
1973
8310b089
YZ		 1974static int check_caps_flush(struct ceph_mds_client *mdsc,
1975 u64 want_flush_tid)
1976{
8310b089
YZ		 1977 int ret = 1;
1978
1979 spin_lock(&mdsc->cap_dirty_lock);
e4500b5e
YZ		 1980 if (!list_empty(&mdsc->cap_flush_list)) {
1981 struct ceph_cap_flush *cf =
1982 list_first_entry(&mdsc->cap_flush_list,
1983 struct ceph_cap_flush, g_list);
1984 if (cf->tid <= want_flush_tid) {
1985 dout("check_caps_flush still flushing tid "
1986 "%llu <= %llu\n", cf->tid, want_flush_tid);
1987 ret = 0;
1988 }
8310b089
YZ		 1989 }
1990 spin_unlock(&mdsc->cap_dirty_lock);
1991 return ret;
d3383a8e
YZ		 1992}
1993
2f2dc053
SW		 1994/*
1995 * flush all dirty inode data to disk.
1996 *
8310b089 1997 * returns true if we've flushed through want_flush_tid
2f2dc053 1998 */
affbc19a 1999static void wait_caps_flush(struct ceph_mds_client *mdsc,
0e294387 2000 u64 want_flush_tid)
2f2dc053 2001{
0e294387 2002 dout("check_caps_flush want %llu\n", want_flush_tid);
8310b089
YZ		 2003
2004 wait_event(mdsc->cap_flushing_wq,
2005 check_caps_flush(mdsc, want_flush_tid));
2006
2007 dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid);
2f2dc053
SW		 2008}
2009
2010/*
2011 * called under s_mutex
2012 */
e3ec8d68
YZ		 2013static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
2014 struct ceph_mds_session *session)
2f2dc053 2015{
745a8e3b
YZ		 2016 struct ceph_msg *msg = NULL;
2017 struct ceph_mds_cap_release *head;
2018 struct ceph_mds_cap_item *item;
92475f05 2019 struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
745a8e3b
YZ		 2020 struct ceph_cap *cap;
2021 LIST_HEAD(tmp_list);
2022 int num_cap_releases;
92475f05
JL		 2023 __le32 barrier, *cap_barrier;
2024
2025 down_read(&osdc->lock);
2026 barrier = cpu_to_le32(osdc->epoch_barrier);
2027 up_read(&osdc->lock);
2f2dc053 2028
0f8605f2 2029 spin_lock(&session->s_cap_lock);
745a8e3b
YZ		 2030again:
2031 list_splice_init(&session->s_cap_releases, &tmp_list);
2032 num_cap_releases = session->s_num_cap_releases;
2033 session->s_num_cap_releases = 0;
2f2dc053 2034 spin_unlock(&session->s_cap_lock);
e01a5946 2035
745a8e3b
YZ		 2036 while (!list_empty(&tmp_list)) {
2037 if (!msg) {
2038 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
09cbfeaf 2039 PAGE_SIZE, GFP_NOFS, false);
745a8e3b
YZ		 2040 if (!msg)
2041 goto out_err;
2042 head = msg->front.iov_base;
2043 head->num = cpu_to_le32(0);
2044 msg->front.iov_len = sizeof(*head);
92475f05
JL		 2045
2046 msg->hdr.version = cpu_to_le16(2);
2047 msg->hdr.compat_version = cpu_to_le16(1);
745a8e3b 2048 }
92475f05 2049
745a8e3b
YZ		 2050 cap = list_first_entry(&tmp_list, struct ceph_cap,
2051 session_caps);
2052 list_del(&cap->session_caps);
2053 num_cap_releases--;
e01a5946 2054
00bd8edb 2055 head = msg->front.iov_base;
4198aba4
JL		 2056 put_unaligned_le32(get_unaligned_le32(&head->num) + 1,
2057 &head->num);
745a8e3b
YZ		 2058 item = msg->front.iov_base + msg->front.iov_len;
2059 item->ino = cpu_to_le64(cap->cap_ino);
2060 item->cap_id = cpu_to_le64(cap->cap_id);
2061 item->migrate_seq = cpu_to_le32(cap->mseq);
2062 item->seq = cpu_to_le32(cap->issue_seq);
2063 msg->front.iov_len += sizeof(*item);
2064
2065 ceph_put_cap(mdsc, cap);
2066
2067 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
92475f05
JL		 2068 // Append cap_barrier field
2069 cap_barrier = msg->front.iov_base + msg->front.iov_len;
2070 *cap_barrier = barrier;
2071 msg->front.iov_len += sizeof(*cap_barrier);
2072
745a8e3b
YZ		 2073 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2074 dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
2075 ceph_con_send(&session->s_con, msg);
2076 msg = NULL;
2077 }
00bd8edb 2078 }
e01a5946 2079
745a8e3b 2080 BUG_ON(num_cap_releases != 0);
e01a5946 2081
745a8e3b
YZ		 2082 spin_lock(&session->s_cap_lock);
2083 if (!list_empty(&session->s_cap_releases))
2084 goto again;
2085 spin_unlock(&session->s_cap_lock);
2086
2087 if (msg) {
92475f05
JL		 2088 // Append cap_barrier field
2089 cap_barrier = msg->front.iov_base + msg->front.iov_len;
2090 *cap_barrier = barrier;
2091 msg->front.iov_len += sizeof(*cap_barrier);
2092
745a8e3b
YZ		 2093 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2094 dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
2095 ceph_con_send(&session->s_con, msg);
e01a5946 2096 }
745a8e3b
YZ		 2097 return;
2098out_err:
2099 pr_err("send_cap_releases mds%d, failed to allocate message\n",
2100 session->s_mds);
2101 spin_lock(&session->s_cap_lock);
2102 list_splice(&tmp_list, &session->s_cap_releases);
2103 session->s_num_cap_releases += num_cap_releases;
2104 spin_unlock(&session->s_cap_lock);
e01a5946
SW		 2105}
2106
e3ec8d68
YZ		 2107static void ceph_cap_release_work(struct work_struct *work)
2108{
2109 struct ceph_mds_session *session =
2110 container_of(work, struct ceph_mds_session, s_cap_release_work);
2111
2112 mutex_lock(&session->s_mutex);
2113 if (session->s_state == CEPH_MDS_SESSION_OPEN ||
2114 session->s_state == CEPH_MDS_SESSION_HUNG)
2115 ceph_send_cap_releases(session->s_mdsc, session);
2116 mutex_unlock(&session->s_mutex);
2117 ceph_put_mds_session(session);
2118}
2119
2120void ceph_flush_cap_releases(struct ceph_mds_client *mdsc,
2121 struct ceph_mds_session *session)
2122{
2123 if (mdsc->stopping)
2124 return;
2125
5b3248c6 2126 ceph_get_mds_session(session);
e3ec8d68
YZ		 2127 if (queue_work(mdsc->fsc->cap_wq,
2128 &session->s_cap_release_work)) {
2129 dout("cap release work queued\n");
2130 } else {
2131 ceph_put_mds_session(session);
2132 dout("failed to queue cap release work\n");
2133 }
2134}
2135
2136/*
2137 * caller holds session->s_cap_lock
2138 */
2139void __ceph_queue_cap_release(struct ceph_mds_session *session,
2140 struct ceph_cap *cap)
2141{
2142 list_add_tail(&cap->session_caps, &session->s_cap_releases);
2143 session->s_num_cap_releases++;
2144
2145 if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))
2146 ceph_flush_cap_releases(session->s_mdsc, session);
2147}
2148
37c4efc1
YZ		 2149static void ceph_cap_reclaim_work(struct work_struct *work)
2150{
2151 struct ceph_mds_client *mdsc =
2152 container_of(work, struct ceph_mds_client, cap_reclaim_work);
2153 int ret = ceph_trim_dentries(mdsc);
2154 if (ret == -EAGAIN)
2155 ceph_queue_cap_reclaim_work(mdsc);
2156}
2157
2158void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)
2159{
2160 if (mdsc->stopping)
2161 return;
2162
2163 if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {
2164 dout("caps reclaim work queued\n");
2165 } else {
2166 dout("failed to queue caps release work\n");
2167 }
2168}
2169
fe33032d
YZ		 2170void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)
2171{
2172 int val;
2173 if (!nr)
2174 return;
2175 val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);
bba1560b 2176 if ((val % CEPH_CAPS_PER_RELEASE) < nr) {
fe33032d
YZ		 2177 atomic_set(&mdsc->cap_reclaim_pending, 0);
2178 ceph_queue_cap_reclaim_work(mdsc);
2179 }
2180}
2181
2f2dc053
SW		 2182/*
2183 * requests
2184 */
2185
54008399
YZ		 2186int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
2187 struct inode *dir)
2188{
2189 struct ceph_inode_info *ci = ceph_inode(dir);
2190 struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
2191 struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
2a5beea3 2192 size_t size = sizeof(struct ceph_mds_reply_dir_entry);
ad8c28a9
JL		 2193 unsigned int num_entries;
2194 int order;
54008399
YZ		 2195
2196 spin_lock(&ci->i_ceph_lock);
2197 num_entries = ci->i_files + ci->i_subdirs;
2198 spin_unlock(&ci->i_ceph_lock);
ad8c28a9 2199 num_entries = max(num_entries, 1U);
54008399
YZ		 2200 num_entries = min(num_entries, opt->max_readdir);
2201
2202 order = get_order(size * num_entries);
2203 while (order >= 0) {
2a5beea3 2204 rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL |
2941bf53
XL		 2205 __GFP_NOWARN |
2206 __GFP_ZERO,
2a5beea3
YZ		 2207 order);
2208 if (rinfo->dir_entries)
54008399
YZ		 2209 break;
2210 order--;
2211 }
2a5beea3 2212 if (!rinfo->dir_entries)
54008399
YZ		 2213 return -ENOMEM;
2214
2215 num_entries = (PAGE_SIZE << order) / size;
2216 num_entries = min(num_entries, opt->max_readdir);
2217
2218 rinfo->dir_buf_size = PAGE_SIZE << order;
2219 req->r_num_caps = num_entries + 1;
2220 req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
2221 req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
2222 return 0;
2223}
2224
2f2dc053
SW		 2225/*
2226 * Create an mds request.
2227 */
2228struct ceph_mds_request *
2229ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
2230{
058daab7 2231 struct ceph_mds_request *req;
2f2dc053 2232
058daab7 2233 req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);
2f2dc053
SW		 2234 if (!req)
2235 return ERR_PTR(-ENOMEM);
2236
b4556396 2237 mutex_init(&req->r_fill_mutex);
37151668 2238 req->r_mdsc = mdsc;
2f2dc053 2239 req->r_started = jiffies;
70c94820 2240 req->r_start_latency = ktime_get();
2f2dc053
SW		 2241 req->r_resend_mds = -1;
2242 INIT_LIST_HEAD(&req->r_unsafe_dir_item);
68cd5b4b 2243 INIT_LIST_HEAD(&req->r_unsafe_target_item);
2f2dc053 2244 req->r_fmode = -1;
153c8e6b 2245 kref_init(&req->r_kref);
fcd00b68 2246 RB_CLEAR_NODE(&req->r_node);
2f2dc053
SW		 2247 INIT_LIST_HEAD(&req->r_wait);
2248 init_completion(&req->r_completion);
2249 init_completion(&req->r_safe_completion);
2250 INIT_LIST_HEAD(&req->r_unsafe_item);
2251
668c9a61 2252 ktime_get_coarse_real_ts64(&req->r_stamp);
b8e69066 2253
2f2dc053
SW		 2254 req->r_op = op;
2255 req->r_direct_mode = mode;
2256 return req;
2257}
2258
2259/*
44ca18f2 2260 * return oldest (lowest) request, tid in request tree, 0 if none.
2f2dc053
SW		 2261 *
2262 * called under mdsc->mutex.
2263 */
44ca18f2
SW		 2264static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
2265{
2266 if (RB_EMPTY_ROOT(&mdsc->request_tree))
2267 return NULL;
2268 return rb_entry(rb_first(&mdsc->request_tree),
2269 struct ceph_mds_request, r_node);
2270}
2271
e8a7b8b1 2272static inline u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
2f2dc053 2273{
e8a7b8b1 2274 return mdsc->oldest_tid;
2f2dc053
SW		 2275}
2276
2277/*
2278 * Build a dentry's path. Allocate on heap; caller must kfree. Based
2279 * on build_path_from_dentry in fs/cifs/dir.c.
2280 *
2281 * If @stop_on_nosnap, generate path relative to the first non-snapped
2282 * inode.
2283 *
2284 * Encode hidden .snap dirs as a double /, i.e.
2285 * foo/.snap/bar -> foo//bar
2286 */
69a10fb3 2287char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *pbase,
2f2dc053
SW		 2288 int stop_on_nosnap)
2289{
2290 struct dentry *temp;
2291 char *path;
f77f21bb 2292 int pos;
1b71fe2e 2293 unsigned seq;
69a10fb3 2294 u64 base;
2f2dc053 2295
d37b1d99 2296 if (!dentry)
2f2dc053
SW		 2297 return ERR_PTR(-EINVAL);
2298
f77f21bb 2299 path = __getname();
d37b1d99 2300 if (!path)
2f2dc053 2301 return ERR_PTR(-ENOMEM);
f77f21bb
JL		 2302retry:
2303 pos = PATH_MAX - 1;
2304 path[pos] = '\0';
2305
2306 seq = read_seqbegin(&rename_lock);
1b71fe2e 2307 rcu_read_lock();
f77f21bb
JL		 2308 temp = dentry;
2309 for (;;) {
1b71fe2e 2310 struct inode *inode;
2f2dc053 2311
1b71fe2e 2312 spin_lock(&temp->d_lock);
2b0143b5 2313 inode = d_inode(temp);
2f2dc053 2314 if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
104648ad 2315 dout("build_path path+%d: %p SNAPDIR\n",
2f2dc053 2316 pos, temp);
d6b8bd67 2317 } else if (stop_on_nosnap && inode && dentry != temp &&
2f2dc053 2318 ceph_snap(inode) == CEPH_NOSNAP) {
9d5a09e6 2319 spin_unlock(&temp->d_lock);
d6b8bd67 2320 pos++; /* get rid of any prepended '/' */
2f2dc053
SW		 2321 break;
2322 } else {
2323 pos -= temp->d_name.len;
1b71fe2e
AV		 2324 if (pos < 0) {
2325 spin_unlock(&temp->d_lock);
2f2dc053 2326 break;
1b71fe2e 2327 }
f77f21bb 2328 memcpy(path + pos, temp->d_name.name, temp->d_name.len);
2f2dc053 2329 }
1b71fe2e 2330 spin_unlock(&temp->d_lock);
41883ba8 2331 temp = READ_ONCE(temp->d_parent);
f77f21bb
JL		 2332
2333 /* Are we at the root? */
2334 if (IS_ROOT(temp))
2335 break;
2336
2337 /* Are we out of buffer? */
2338 if (--pos < 0)
2339 break;
2340
2341 path[pos] = '/';
2f2dc053 2342 }
69a10fb3 2343 base = ceph_ino(d_inode(temp));
1b71fe2e 2344 rcu_read_unlock();
f5946bcc
JL		 2345
2346 if (read_seqretry(&rename_lock, seq))
2347 goto retry;
2348
2349 if (pos < 0) {
2350 /*
2351 * A rename didn't occur, but somehow we didn't end up where
2352 * we thought we would. Throw a warning and try again.
2353 */
2354 pr_warn("build_path did not end path lookup where "
2355 "expected, pos is %d\n", pos);
2f2dc053
SW		 2356 goto retry;
2357 }
2358
69a10fb3 2359 *pbase = base;
f77f21bb 2360 *plen = PATH_MAX - 1 - pos;
104648ad 2361 dout("build_path on %p %d built %llx '%.*s'\n",
f77f21bb
JL		 2362 dentry, d_count(dentry), base, *plen, path + pos);
2363 return path + pos;
2f2dc053
SW		 2364}
2365
fd36a717 2366static int build_dentry_path(struct dentry *dentry, struct inode *dir,
2f2dc053 2367 const char **ppath, int *ppathlen, u64 *pino,
1bcb3440 2368 bool *pfreepath, bool parent_locked)
2f2dc053
SW		 2369{
2370 char *path;
2371
c6b0b656 2372 rcu_read_lock();
fd36a717
JL		 2373 if (!dir)
2374 dir = d_inode_rcu(dentry->d_parent);
964fff74 2375 if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP) {
c6b0b656
JL		 2376 *pino = ceph_ino(dir);
2377 rcu_read_unlock();
964fff74
JL		 2378 *ppath = dentry->d_name.name;
2379 *ppathlen = dentry->d_name.len;
2f2dc053
SW		 2380 return 0;
2381 }
c6b0b656 2382 rcu_read_unlock();
2f2dc053
SW		 2383 path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
2384 if (IS_ERR(path))
2385 return PTR_ERR(path);
2386 *ppath = path;
1bcb3440 2387 *pfreepath = true;
2f2dc053
SW		 2388 return 0;
2389}
2390
2391static int build_inode_path(struct inode *inode,
2392 const char **ppath, int *ppathlen, u64 *pino,
1bcb3440 2393 bool *pfreepath)
2f2dc053
SW		 2394{
2395 struct dentry *dentry;
2396 char *path;
2397
2398 if (ceph_snap(inode) == CEPH_NOSNAP) {
2399 *pino = ceph_ino(inode);
2400 *ppathlen = 0;
2401 return 0;
2402 }
2403 dentry = d_find_alias(inode);
2404 path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
2405 dput(dentry);
2406 if (IS_ERR(path))
2407 return PTR_ERR(path);
2408 *ppath = path;
1bcb3440 2409 *pfreepath = true;
2f2dc053
SW		 2410 return 0;
2411}
2412
2413/*
2414 * request arguments may be specified via an inode *, a dentry *, or
2415 * an explicit ino+path.
2416 */
2417static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
fd36a717
JL		 2418 struct inode *rdiri, const char *rpath,
2419 u64 rino, const char **ppath, int *pathlen,
1bcb3440 2420 u64 *ino, bool *freepath, bool parent_locked)
2f2dc053
SW		 2421{
2422 int r = 0;
2423
2424 if (rinode) {
2425 r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
2426 dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
2427 ceph_snap(rinode));
2428 } else if (rdentry) {
fd36a717 2429 r = build_dentry_path(rdentry, rdiri, ppath, pathlen, ino,
1bcb3440 2430 freepath, parent_locked);
2f2dc053
SW		 2431 dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
2432 *ppath);
795858db 2433 } else if (rpath || rino) {
2f2dc053
SW		 2434 *ino = rino;
2435 *ppath = rpath;
b000056a 2436 *pathlen = rpath ? strlen(rpath) : 0;
2f2dc053
SW		 2437 dout(" path %.*s\n", *pathlen, rpath);
2438 }
2439
2440 return r;
2441}
2442
60267ba3
ID		 2443static void encode_timestamp_and_gids(void **p,
2444 const struct ceph_mds_request *req)
2445{
2446 struct ceph_timespec ts;
2447 int i;
2448
2449 ceph_encode_timespec64(&ts, &req->r_stamp);
2450 ceph_encode_copy(p, &ts, sizeof(ts));
2451
2452 /* gid_list */
2453 ceph_encode_32(p, req->r_cred->group_info->ngroups);
2454 for (i = 0; i < req->r_cred->group_info->ngroups; i++)
2455 ceph_encode_64(p, from_kgid(&init_user_ns,
2456 req->r_cred->group_info->gid[i]));
2457}
2458
2f2dc053
SW		 2459/*
2460 * called under mdsc->mutex
2461 */
4f1ddb1e 2462static struct ceph_msg *create_request_message(struct ceph_mds_session *session,
2f2dc053 2463 struct ceph_mds_request *req,
4f1ddb1e 2464 bool drop_cap_releases)
2f2dc053 2465{
4f1ddb1e
JL		 2466 int mds = session->s_mds;
2467 struct ceph_mds_client *mdsc = session->s_mdsc;
2f2dc053 2468 struct ceph_msg *msg;
4f1ddb1e 2469 struct ceph_mds_request_head_old *head;
2f2dc053
SW		 2470 const char *path1 = NULL;
2471 const char *path2 = NULL;
2472 u64 ino1 = 0, ino2 = 0;
2473 int pathlen1 = 0, pathlen2 = 0;
1bcb3440 2474 bool freepath1 = false, freepath2 = false;
60267ba3 2475 int len;
2f2dc053
SW		 2476 u16 releases;
2477 void *p, *end;
2478 int ret;
4f1ddb1e 2479 bool legacy = !(session->s_con.peer_features & CEPH_FEATURE_FS_BTIME);
2f2dc053
SW		 2480
2481 ret = set_request_path_attr(req->r_inode, req->r_dentry,
3dd69aab 2482 req->r_parent, req->r_path1, req->r_ino1.ino,
1bcb3440
JL		 2483 &path1, &pathlen1, &ino1, &freepath1,
2484 test_bit(CEPH_MDS_R_PARENT_LOCKED,
2485 &req->r_req_flags));
2f2dc053
SW		 2486 if (ret < 0) {
2487 msg = ERR_PTR(ret);
2488 goto out;
2489 }
2490
1bcb3440 2491 /* If r_old_dentry is set, then assume that its parent is locked */
2f2dc053 2492 ret = set_request_path_attr(NULL, req->r_old_dentry,
fd36a717 2493 req->r_old_dentry_dir,
2f2dc053 2494 req->r_path2, req->r_ino2.ino,
1bcb3440 2495 &path2, &pathlen2, &ino2, &freepath2, true);
2f2dc053
SW		 2496 if (ret < 0) {
2497 msg = ERR_PTR(ret);
2498 goto out_free1;
2499 }
2500
60267ba3 2501 len = legacy ? sizeof(*head) : sizeof(struct ceph_mds_request_head);
4f1ddb1e 2502 len += pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64)) +
777d738a 2503 sizeof(struct ceph_timespec);
60267ba3 2504 len += sizeof(u32) + (sizeof(u64) * req->r_cred->group_info->ngroups);
2f2dc053
SW		 2505
2506 /* calculate (max) length for cap releases */
2507 len += sizeof(struct ceph_mds_request_release) *
2508 (!!req->r_inode_drop + !!req->r_dentry_drop +
2509 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
4f1ddb1e 2510
2f2dc053 2511 if (req->r_dentry_drop)
c1dfc277 2512 len += pathlen1;
2f2dc053 2513 if (req->r_old_dentry_drop)
c1dfc277 2514 len += pathlen2;
2f2dc053 2515
0d9c1ab3 2516 msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);
a79832f2
SW		 2517 if (!msg) {
2518 msg = ERR_PTR(-ENOMEM);
2f2dc053 2519 goto out_free2;
a79832f2 2520 }
2f2dc053 2521
6df058c0
SW		 2522 msg->hdr.tid = cpu_to_le64(req->r_tid);
2523
4f1ddb1e 2524 /*
60267ba3 2525 * The old ceph_mds_request_head didn't contain a version field, and
4f1ddb1e
JL		 2526 * one was added when we moved the message version from 3->4.
2527 */
2528 if (legacy) {
2529 msg->hdr.version = cpu_to_le16(3);
2530 head = msg->front.iov_base;
2531 p = msg->front.iov_base + sizeof(*head);
2532 } else {
2533 struct ceph_mds_request_head *new_head = msg->front.iov_base;
2534
2535 msg->hdr.version = cpu_to_le16(4);
2536 new_head->version = cpu_to_le16(CEPH_MDS_REQUEST_HEAD_VERSION);
2537 head = (struct ceph_mds_request_head_old *)&new_head->oldest_client_tid;
2538 p = msg->front.iov_base + sizeof(*new_head);
2539 }
2540
2f2dc053
SW		 2541 end = msg->front.iov_base + msg->front.iov_len;
2542
2543 head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
2544 head->op = cpu_to_le32(req->r_op);
7fe0cdeb
JL		 2545 head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns,
2546 req->r_cred->fsuid));
2547 head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns,
2548 req->r_cred->fsgid));
6deb8008 2549 head->ino = cpu_to_le64(req->r_deleg_ino);
2f2dc053
SW		 2550 head->args = req->r_args;
2551
2552 ceph_encode_filepath(&p, end, ino1, path1);
2553 ceph_encode_filepath(&p, end, ino2, path2);
2554
e979cf50
SW		 2555 /* make note of release offset, in case we need to replay */
2556 req->r_request_release_offset = p - msg->front.iov_base;
2557
2f2dc053
SW		 2558 /* cap releases */
2559 releases = 0;
2560 if (req->r_inode_drop)
2561 releases += ceph_encode_inode_release(&p,
2b0143b5 2562 req->r_inode ? req->r_inode : d_inode(req->r_dentry),
719a2514
YZ		 2563 mds, req->r_inode_drop, req->r_inode_unless,
2564 req->r_op == CEPH_MDS_OP_READDIR);
2f2dc053
SW		 2565 if (req->r_dentry_drop)
2566 releases += ceph_encode_dentry_release(&p, req->r_dentry,
3dd69aab 2567 req->r_parent, mds, req->r_dentry_drop,
ca6c8ae0 2568 req->r_dentry_unless);
2f2dc053
SW		 2569 if (req->r_old_dentry_drop)
2570 releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
ca6c8ae0
JL		 2571 req->r_old_dentry_dir, mds,
2572 req->r_old_dentry_drop,
2573 req->r_old_dentry_unless);
2f2dc053
SW		 2574 if (req->r_old_inode_drop)
2575 releases += ceph_encode_inode_release(&p,
2b0143b5 2576 d_inode(req->r_old_dentry),
2f2dc053 2577 mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
6e6f0923
YZ		 2578
2579 if (drop_cap_releases) {
2580 releases = 0;
2581 p = msg->front.iov_base + req->r_request_release_offset;
2582 }
2583
2f2dc053
SW		 2584 head->num_releases = cpu_to_le16(releases);
2585
60267ba3 2586 encode_timestamp_and_gids(&p, req);
4f1ddb1e 2587
b682c6d4
XL		 2588 if (WARN_ON_ONCE(p > end)) {
2589 ceph_msg_put(msg);
2590 msg = ERR_PTR(-ERANGE);
2591 goto out_free2;
2592 }
2593
2f2dc053
SW		 2594 msg->front.iov_len = p - msg->front.iov_base;
2595 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2596
25e6bae3
YZ		 2597 if (req->r_pagelist) {
2598 struct ceph_pagelist *pagelist = req->r_pagelist;
25e6bae3
YZ		 2599 ceph_msg_data_add_pagelist(msg, pagelist);
2600 msg->hdr.data_len = cpu_to_le32(pagelist->length);
2601 } else {
2602 msg->hdr.data_len = 0;
ebf18f47 2603 }
02afca6c 2604
2f2dc053
SW		 2605 msg->hdr.data_off = cpu_to_le16(0);
2606
2607out_free2:
2608 if (freepath2)
f77f21bb 2609 ceph_mdsc_free_path((char *)path2, pathlen2);
2f2dc053
SW		 2610out_free1:
2611 if (freepath1)
f77f21bb 2612 ceph_mdsc_free_path((char *)path1, pathlen1);
2f2dc053
SW		 2613out:
2614 return msg;
2615}
2616
2617/*
2618 * called under mdsc->mutex if error, under no mutex if
2619 * success.
2620 */
2621static void complete_request(struct ceph_mds_client *mdsc,
2622 struct ceph_mds_request *req)
2623{
70c94820
XL		 2624 req->r_end_latency = ktime_get();
2625
2f2dc053
SW		 2626 if (req->r_callback)
2627 req->r_callback(mdsc, req);
111c7081 2628 complete_all(&req->r_completion);
2f2dc053
SW		 2629}
2630
4f1ddb1e
JL		 2631static struct ceph_mds_request_head_old *
2632find_old_request_head(void *p, u64 features)
2633{
2634 bool legacy = !(features & CEPH_FEATURE_FS_BTIME);
2635 struct ceph_mds_request_head *new_head;
2636
2637 if (legacy)
2638 return (struct ceph_mds_request_head_old *)p;
2639 new_head = (struct ceph_mds_request_head *)p;
2640 return (struct ceph_mds_request_head_old *)&new_head->oldest_client_tid;
2641}
2642
2f2dc053
SW		 2643/*
2644 * called under mdsc->mutex
2645 */
396bd62c 2646static int __prepare_send_request(struct ceph_mds_session *session,
2f2dc053 2647 struct ceph_mds_request *req,
396bd62c 2648 bool drop_cap_releases)
2f2dc053 2649{
396bd62c
JL		 2650 int mds = session->s_mds;
2651 struct ceph_mds_client *mdsc = session->s_mdsc;
4f1ddb1e 2652 struct ceph_mds_request_head_old *rhead;
2f2dc053
SW		 2653 struct ceph_msg *msg;
2654 int flags = 0;
2655
2f2dc053 2656 req->r_attempts++;
e55b71f8
GF		 2657 if (req->r_inode) {
2658 struct ceph_cap *cap =
2659 ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
2660
2661 if (cap)
2662 req->r_sent_on_mseq = cap->mseq;
2663 else
2664 req->r_sent_on_mseq = -1;
2665 }
2f2dc053
SW		 2666 dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
2667 req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
2668
bc2de10d 2669 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
c5c9a0bf 2670 void *p;
4f1ddb1e 2671
01a92f17
SW		 2672 /*
2673 * Replay. Do not regenerate message (and rebuild
2674 * paths, etc.); just use the original message.
2675 * Rebuilding paths will break for renames because
2676 * d_move mangles the src name.
2677 */
2678 msg = req->r_request;
4f1ddb1e
JL		 2679 rhead = find_old_request_head(msg->front.iov_base,
2680 session->s_con.peer_features);
01a92f17
SW		 2681
2682 flags = le32_to_cpu(rhead->flags);
2683 flags |= CEPH_MDS_FLAG_REPLAY;
2684 rhead->flags = cpu_to_le32(flags);
2685
2686 if (req->r_target_inode)
2687 rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
2688
2689 rhead->num_retry = req->r_attempts - 1;
e979cf50
SW		 2690
2691 /* remove cap/dentry releases from message */
2692 rhead->num_releases = 0;
c5c9a0bf 2693
c5c9a0bf 2694 p = msg->front.iov_base + req->r_request_release_offset;
60267ba3 2695 encode_timestamp_and_gids(&p, req);
c5c9a0bf
YZ		 2696
2697 msg->front.iov_len = p - msg->front.iov_base;
2698 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
01a92f17
SW		 2699 return 0;
2700 }
2701
2f2dc053
SW		 2702 if (req->r_request) {
2703 ceph_msg_put(req->r_request);
2704 req->r_request = NULL;
2705 }
4f1ddb1e 2706 msg = create_request_message(session, req, drop_cap_releases);
2f2dc053 2707 if (IS_ERR(msg)) {
e1518c7c 2708 req->r_err = PTR_ERR(msg);
a79832f2 2709 return PTR_ERR(msg);
2f2dc053
SW		 2710 }
2711 req->r_request = msg;
2712
4f1ddb1e
JL		 2713 rhead = find_old_request_head(msg->front.iov_base,
2714 session->s_con.peer_features);
2f2dc053 2715 rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
bc2de10d 2716 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
2f2dc053 2717 flags |= CEPH_MDS_FLAG_REPLAY;
3bb48b41
JL		 2718 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))
2719 flags |= CEPH_MDS_FLAG_ASYNC;
3dd69aab 2720 if (req->r_parent)
2f2dc053
SW		 2721 flags |= CEPH_MDS_FLAG_WANT_DENTRY;
2722 rhead->flags = cpu_to_le32(flags);
2723 rhead->num_fwd = req->r_num_fwd;
2724 rhead->num_retry = req->r_attempts - 1;
2725
3dd69aab 2726 dout(" r_parent = %p\n", req->r_parent);
2f2dc053
SW		 2727 return 0;
2728}
2729
9cf54563
XL		 2730/*
2731 * called under mdsc->mutex
2732 */
396bd62c 2733static int __send_request(struct ceph_mds_session *session,
9cf54563
XL		 2734 struct ceph_mds_request *req,
2735 bool drop_cap_releases)
2736{
2737 int err;
2738
396bd62c 2739 err = __prepare_send_request(session, req, drop_cap_releases);
9cf54563
XL		 2740 if (!err) {
2741 ceph_msg_get(req->r_request);
2742 ceph_con_send(&session->s_con, req->r_request);
2743 }
2744
2745 return err;
2746}
2747
2f2dc053
SW		 2748/*
2749 * send request, or put it on the appropriate wait list.
2750 */
d5548492 2751static void __do_request(struct ceph_mds_client *mdsc,
2f2dc053
SW		 2752 struct ceph_mds_request *req)
2753{
2754 struct ceph_mds_session *session = NULL;
2755 int mds = -1;
48fec5d0 2756 int err = 0;
c4853e97 2757 bool random;
2f2dc053 2758
bc2de10d
JL		 2759 if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
2760 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
eb1b8af3 2761 __unregister_request(mdsc, req);
d5548492 2762 return;
eb1b8af3 2763 }
2f2dc053
SW		 2764
2765 if (req->r_timeout &&
2766 time_after_eq(jiffies, req->r_started + req->r_timeout)) {
2767 dout("do_request timed out\n");
8ccf7fcc 2768 err = -ETIMEDOUT;
2f2dc053
SW		 2769 goto finish;
2770 }
52953d55 2771 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
48fec5d0
YZ		 2772 dout("do_request forced umount\n");
2773 err = -EIO;
2774 goto finish;
2775 }
52953d55 2776 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) {
e9e427f0
YZ		 2777 if (mdsc->mdsmap_err) {
2778 err = mdsc->mdsmap_err;
2779 dout("do_request mdsmap err %d\n", err);
2780 goto finish;
2781 }
cc8e8342
YZ		 2782 if (mdsc->mdsmap->m_epoch == 0) {
2783 dout("do_request no mdsmap, waiting for map\n");
2784 list_add(&req->r_wait, &mdsc->waiting_for_map);
d5548492 2785 return;
cc8e8342 2786 }
e9e427f0
YZ		 2787 if (!(mdsc->fsc->mount_options->flags &
2788 CEPH_MOUNT_OPT_MOUNTWAIT) &&
2789 !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) {
97820058 2790 err = -EHOSTUNREACH;
e9e427f0
YZ		 2791 goto finish;
2792 }
2793 }
2f2dc053 2794
dc69e2e9
SW		 2795 put_request_session(req);
2796
c4853e97 2797 mds = __choose_mds(mdsc, req, &random);
2f2dc053
SW		 2798 if (mds < 0 ||
2799 ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
3bb48b41
JL		 2800 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
2801 err = -EJUKEBOX;
2802 goto finish;
2803 }
2f2dc053
SW		 2804 dout("do_request no mds or not active, waiting for map\n");
2805 list_add(&req->r_wait, &mdsc->waiting_for_map);
d5548492 2806 return;
2f2dc053
SW		 2807 }
2808
2809 /* get, open session */
2810 session = __ceph_lookup_mds_session(mdsc, mds);
9c423956 2811 if (!session) {
2f2dc053 2812 session = register_session(mdsc, mds);
9c423956
SW		 2813 if (IS_ERR(session)) {
2814 err = PTR_ERR(session);
2815 goto finish;
2816 }
2817 }
5b3248c6 2818 req->r_session = ceph_get_mds_session(session);
dc69e2e9 2819
2f2dc053 2820 dout("do_request mds%d session %p state %s\n", mds, session,
a687ecaf 2821 ceph_session_state_name(session->s_state));
2f2dc053
SW		 2822 if (session->s_state != CEPH_MDS_SESSION_OPEN &&
2823 session->s_state != CEPH_MDS_SESSION_HUNG) {
3bb48b41
JL		 2824 /*
2825 * We cannot queue async requests since the caps and delegated
2826 * inodes are bound to the session. Just return -EJUKEBOX and
2827 * let the caller retry a sync request in that case.
2828 */
2829 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
2830 err = -EJUKEBOX;
2831 goto out_session;
2832 }
4ae3713f
JL		 2833
2834 /*
2835 * If the session has been REJECTED, then return a hard error,
2836 * unless it's a CLEANRECOVER mount, in which case we'll queue
2837 * it to the mdsc queue.
2838 */
2839 if (session->s_state == CEPH_MDS_SESSION_REJECTED) {
2840 if (ceph_test_mount_opt(mdsc->fsc, CLEANRECOVER))
2841 list_add(&req->r_wait, &mdsc->waiting_for_map);
2842 else
2843 err = -EACCES;
2844 goto out_session;
2845 }
2846
2f2dc053 2847 if (session->s_state == CEPH_MDS_SESSION_NEW ||
c4853e97 2848 session->s_state == CEPH_MDS_SESSION_CLOSING) {
b682c6d4
XL		 2849 err = __open_session(mdsc, session);
2850 if (err)
2851 goto out_session;
c4853e97
XL		 2852 /* retry the same mds later */
2853 if (random)
2854 req->r_resend_mds = mds;
2855 }
2f2dc053
SW		 2856 list_add(&req->r_wait, &session->s_waiting);
2857 goto out_session;
2858 }
2859
2860 /* send request */
2f2dc053
SW		 2861 req->r_resend_mds = -1; /* forget any previous mds hint */
2862
2863 if (req->r_request_started == 0) /* note request start time */
2864 req->r_request_started = jiffies;
2865
396bd62c 2866 err = __send_request(session, req, false);
2f2dc053
SW		 2867
2868out_session:
2869 ceph_put_mds_session(session);
48fec5d0
YZ		 2870finish:
2871 if (err) {
2872 dout("__do_request early error %d\n", err);
2873 req->r_err = err;
2874 complete_request(mdsc, req);
2875 __unregister_request(mdsc, req);
2876 }
d5548492 2877 return;
2f2dc053
SW		 2878}
2879
2880/*
2881 * called under mdsc->mutex
2882 */
2883static void __wake_requests(struct ceph_mds_client *mdsc,
2884 struct list_head *head)
2885{
ed75ec2c
YZ		 2886 struct ceph_mds_request *req;
2887 LIST_HEAD(tmp_list);
2888
2889 list_splice_init(head, &tmp_list);
2f2dc053 2890
ed75ec2c
YZ		 2891 while (!list_empty(&tmp_list)) {
2892 req = list_entry(tmp_list.next,
2893 struct ceph_mds_request, r_wait);
2f2dc053 2894 list_del_init(&req->r_wait);
7971bd92 2895 dout(" wake request %p tid %llu\n", req, req->r_tid);
2f2dc053
SW		 2896 __do_request(mdsc, req);
2897 }
2898}
2899
2900/*
2901 * Wake up threads with requests pending for @mds, so that they can
29790f26 2902 * resubmit their requests to a possibly different mds.
2f2dc053 2903 */
29790f26 2904static void kick_requests(struct ceph_mds_client *mdsc, int mds)
2f2dc053 2905{
44ca18f2 2906 struct ceph_mds_request *req;
282c1052 2907 struct rb_node *p = rb_first(&mdsc->request_tree);
2f2dc053
SW		 2908
2909 dout("kick_requests mds%d\n", mds);
282c1052 2910 while (p) {
44ca18f2 2911 req = rb_entry(p, struct ceph_mds_request, r_node);
282c1052 2912 p = rb_next(p);
bc2de10d 2913 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
44ca18f2 2914 continue;
3de22be6
YZ		 2915 if (req->r_attempts > 0)
2916 continue; /* only new requests */
44ca18f2
SW		 2917 if (req->r_session &&
2918 req->r_session->s_mds == mds) {
2919 dout(" kicking tid %llu\n", req->r_tid);
03974e81 2920 list_del_init(&req->r_wait);
44ca18f2 2921 __do_request(mdsc, req);
2f2dc053
SW		 2922 }
2923 }
2924}
2925
86bda539 2926int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,
2f2dc053
SW		 2927 struct ceph_mds_request *req)
2928{
891f3f5a 2929 int err = 0;
86bda539
JL		 2930
2931 /* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
2932 if (req->r_inode)
2933 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
9c1c2b35 2934 if (req->r_parent) {
719a2514
YZ		 2935 struct ceph_inode_info *ci = ceph_inode(req->r_parent);
2936 int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?
2937 CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;
2938 spin_lock(&ci->i_ceph_lock);
2939 ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);
2940 __ceph_touch_fmode(ci, mdsc, fmode);
2941 spin_unlock(&ci->i_ceph_lock);
9c1c2b35 2942 }
86bda539
JL		 2943 if (req->r_old_dentry_dir)
2944 ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
2945 CEPH_CAP_PIN);
2946
891f3f5a
JL		 2947 if (req->r_inode) {
2948 err = ceph_wait_on_async_create(req->r_inode);
2949 if (err) {
2950 dout("%s: wait for async create returned: %d\n",
2951 __func__, err);
2952 return err;
2953 }
2954 }
2955
2956 if (!err && req->r_old_inode) {
2957 err = ceph_wait_on_async_create(req->r_old_inode);
2958 if (err) {
2959 dout("%s: wait for async create returned: %d\n",
2960 __func__, err);
2961 return err;
2962 }
2963 }
2964
86bda539 2965 dout("submit_request on %p for inode %p\n", req, dir);
2f2dc053 2966 mutex_lock(&mdsc->mutex);
86bda539 2967 __register_request(mdsc, req, dir);
2f2dc053 2968 __do_request(mdsc, req);
86bda539 2969 err = req->r_err;
2f2dc053 2970 mutex_unlock(&mdsc->mutex);
86bda539 2971 return err;
2f2dc053
SW		 2972}
2973
9eaa7b79
JL		 2974int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
2975 struct ceph_mds_request *req,
2976 ceph_mds_request_wait_callback_t wait_func)
2f2dc053
SW		 2977{
2978 int err;
2979
e1518c7c 2980 /* wait */
e1518c7c 2981 dout("do_request waiting\n");
9eaa7b79
JL		 2982 if (wait_func) {
2983 err = wait_func(mdsc, req);
e1518c7c 2984 } else {
5be73034
ID		 2985 long timeleft = wait_for_completion_killable_timeout(
2986 &req->r_completion,
2987 ceph_timeout_jiffies(req->r_timeout));
2988 if (timeleft > 0)
2989 err = 0;
2990 else if (!timeleft)
8ccf7fcc 2991 err = -ETIMEDOUT; /* timed out */
5be73034
ID		 2992 else
2993 err = timeleft; /* killed */
e1518c7c
SW		 2994 }
2995 dout("do_request waited, got %d\n", err);
2996 mutex_lock(&mdsc->mutex);
5b1daecd 2997
e1518c7c 2998 /* only abort if we didn't race with a real reply */
bc2de10d 2999 if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
e1518c7c
SW		 3000 err = le32_to_cpu(req->r_reply_info.head->result);
3001 } else if (err < 0) {
3002 dout("aborted request %lld with %d\n", req->r_tid, err);
b4556396
SW		 3003
3004 /*
3005 * ensure we aren't running concurrently with
3006 * ceph_fill_trace or ceph_readdir_prepopulate, which
3007 * rely on locks (dir mutex) held by our caller.
3008 */
3009 mutex_lock(&req->r_fill_mutex);
e1518c7c 3010 req->r_err = err;
bc2de10d 3011 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
b4556396 3012 mutex_unlock(&req->r_fill_mutex);
5b1daecd 3013
3dd69aab 3014 if (req->r_parent &&
167c9e35
SW		 3015 (req->r_op & CEPH_MDS_OP_WRITE))
3016 ceph_invalidate_dir_request(req);
2f2dc053 3017 } else {
e1518c7c 3018 err = req->r_err;
2f2dc053 3019 }
2f2dc053 3020
e1518c7c 3021 mutex_unlock(&mdsc->mutex);
8340f22c
JL		 3022 return err;
3023}
3024
3025/*
3026 * Synchrously perform an mds request. Take care of all of the
3027 * session setup, forwarding, retry details.
3028 */
3029int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
3030 struct inode *dir,
3031 struct ceph_mds_request *req)
3032{
3033 int err;
3034
3035 dout("do_request on %p\n", req);
3036
3037 /* issue */
3038 err = ceph_mdsc_submit_request(mdsc, dir, req);
3039 if (!err)
9eaa7b79 3040 err = ceph_mdsc_wait_request(mdsc, req, NULL);
2f2dc053
SW		 3041 dout("do_request %p done, result %d\n", req, err);
3042 return err;
3043}
3044
167c9e35 3045/*
2f276c51 3046 * Invalidate dir's completeness, dentry lease state on an aborted MDS
167c9e35
SW		 3047 * namespace request.
3048 */
3049void ceph_invalidate_dir_request(struct ceph_mds_request *req)
3050{
8d8f371c
YZ		 3051 struct inode *dir = req->r_parent;
3052 struct inode *old_dir = req->r_old_dentry_dir;
167c9e35 3053
8d8f371c 3054 dout("invalidate_dir_request %p %p (complete, lease(s))\n", dir, old_dir);
167c9e35 3055
8d8f371c
YZ		 3056 ceph_dir_clear_complete(dir);
3057 if (old_dir)
3058 ceph_dir_clear_complete(old_dir);
167c9e35
SW		 3059 if (req->r_dentry)
3060 ceph_invalidate_dentry_lease(req->r_dentry);
3061 if (req->r_old_dentry)
3062 ceph_invalidate_dentry_lease(req->r_old_dentry);
3063}
3064
2f2dc053
SW		 3065/*
3066 * Handle mds reply.
3067 *
3068 * We take the session mutex and parse and process the reply immediately.
3069 * This preserves the logical ordering of replies, capabilities, etc., sent
3070 * by the MDS as they are applied to our local cache.
3071 */
3072static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
3073{
3074 struct ceph_mds_client *mdsc = session->s_mdsc;
3075 struct ceph_mds_request *req;
3076 struct ceph_mds_reply_head *head = msg->front.iov_base;
3077 struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */
982d6011 3078 struct ceph_snap_realm *realm;
2f2dc053
SW		 3079 u64 tid;
3080 int err, result;
2600d2dd 3081 int mds = session->s_mds;
2f2dc053 3082
2f2dc053
SW		 3083 if (msg->front.iov_len < sizeof(*head)) {
3084 pr_err("mdsc_handle_reply got corrupt (short) reply\n");
9ec7cab1 3085 ceph_msg_dump(msg);
2f2dc053
SW		 3086 return;
3087 }
3088
3089 /* get request, session */
6df058c0 3090 tid = le64_to_cpu(msg->hdr.tid);
2f2dc053 3091 mutex_lock(&mdsc->mutex);
fcd00b68 3092 req = lookup_get_request(mdsc, tid);
2f2dc053
SW		 3093 if (!req) {
3094 dout("handle_reply on unknown tid %llu\n", tid);
3095 mutex_unlock(&mdsc->mutex);
3096 return;
3097 }
3098 dout("handle_reply %p\n", req);
2f2dc053
SW		 3099
3100 /* correct session? */
d96d6049 3101 if (req->r_session != session) {
2f2dc053
SW		 3102 pr_err("mdsc_handle_reply got %llu on session mds%d"
3103 " not mds%d\n", tid, session->s_mds,
3104 req->r_session ? req->r_session->s_mds : -1);
3105 mutex_unlock(&mdsc->mutex);
3106 goto out;
3107 }
3108
3109 /* dup? */
bc2de10d
JL		 3110 if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
3111 (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
f3ae1b97 3112 pr_warn("got a dup %s reply on %llu from mds%d\n",
2f2dc053
SW		 3113 head->safe ? "safe" : "unsafe", tid, mds);
3114 mutex_unlock(&mdsc->mutex);
3115 goto out;
3116 }
bc2de10d 3117 if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
f3ae1b97 3118 pr_warn("got unsafe after safe on %llu from mds%d\n",
85792d0d
SW		 3119 tid, mds);
3120 mutex_unlock(&mdsc->mutex);
3121 goto out;
3122 }
2f2dc053
SW		 3123
3124 result = le32_to_cpu(head->result);
3125
2f2dc053 3126 if (head->safe) {
bc2de10d 3127 set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
2f2dc053 3128 __unregister_request(mdsc, req);
2f2dc053 3129
07edc057
XL		 3130 /* last request during umount? */
3131 if (mdsc->stopping && !__get_oldest_req(mdsc))
3132 complete_all(&mdsc->safe_umount_waiters);
3133
bc2de10d 3134 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
2f2dc053
SW		 3135 /*
3136 * We already handled the unsafe response, now do the
3137 * cleanup. No need to examine the response; the MDS
3138 * doesn't include any result info in the safe
3139 * response. And even if it did, there is nothing
3140 * useful we could do with a revised return value.
3141 */
3142 dout("got safe reply %llu, mds%d\n", tid, mds);
2f2dc053 3143
2f2dc053
SW		 3144 mutex_unlock(&mdsc->mutex);
3145 goto out;
3146 }
e1518c7c 3147 } else {
bc2de10d 3148 set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
2f2dc053
SW		 3149 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
3150 }
3151
3152 dout("handle_reply tid %lld result %d\n", tid, result);
3153 rinfo = &req->r_reply_info;
b37fe1f9 3154 if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))
d4846487 3155 err = parse_reply_info(session, msg, rinfo, (u64)-1);
b37fe1f9 3156 else
d4846487 3157 err = parse_reply_info(session, msg, rinfo, session->s_con.peer_features);
2f2dc053
SW		 3158 mutex_unlock(&mdsc->mutex);
3159
bca9fc14
JL		 3160 /* Must find target inode outside of mutexes to avoid deadlocks */
3161 if ((err >= 0) && rinfo->head->is_target) {
3162 struct inode *in;
3163 struct ceph_vino tvino = {
3164 .ino = le64_to_cpu(rinfo->targeti.in->ino),
3165 .snap = le64_to_cpu(rinfo->targeti.in->snapid)
3166 };
3167
3168 in = ceph_get_inode(mdsc->fsc->sb, tvino);
3169 if (IS_ERR(in)) {
3170 err = PTR_ERR(in);
3171 mutex_lock(&session->s_mutex);
3172 goto out_err;
3173 }
3174 req->r_target_inode = in;
3175 }
3176
2f2dc053
SW		 3177 mutex_lock(&session->s_mutex);
3178 if (err < 0) {
25933abd 3179 pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
9ec7cab1 3180 ceph_msg_dump(msg);
2f2dc053
SW		 3181 goto out_err;
3182 }
3183
3184 /* snap trace */
982d6011 3185 realm = NULL;
2f2dc053
SW		 3186 if (rinfo->snapblob_len) {
3187 down_write(&mdsc->snap_rwsem);
3188 ceph_update_snap_trace(mdsc, rinfo->snapblob,
982d6011
YZ		 3189 rinfo->snapblob + rinfo->snapblob_len,
3190 le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
3191 &realm);
2f2dc053
SW		 3192 downgrade_write(&mdsc->snap_rwsem);
3193 } else {
3194 down_read(&mdsc->snap_rwsem);
3195 }
3196
3197 /* insert trace into our cache */
b4556396 3198 mutex_lock(&req->r_fill_mutex);
315f2408 3199 current->journal_info = req;
f5a03b08 3200 err = ceph_fill_trace(mdsc->fsc->sb, req);
2f2dc053 3201 if (err == 0) {
6e8575fa 3202 if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
81c6aea5 3203 req->r_op == CEPH_MDS_OP_LSSNAP))
2f2dc053 3204 ceph_readdir_prepopulate(req, req->r_session);
2f2dc053 3205 }
315f2408 3206 current->journal_info = NULL;
b4556396 3207 mutex_unlock(&req->r_fill_mutex);
2f2dc053
SW		 3208
3209 up_read(&mdsc->snap_rwsem);
982d6011
YZ		 3210 if (realm)
3211 ceph_put_snap_realm(mdsc, realm);
68cd5b4b 3212
fe33032d
YZ		 3213 if (err == 0) {
3214 if (req->r_target_inode &&
3215 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
3216 struct ceph_inode_info *ci =
3217 ceph_inode(req->r_target_inode);
3218 spin_lock(&ci->i_unsafe_lock);
3219 list_add_tail(&req->r_unsafe_target_item,
3220 &ci->i_unsafe_iops);
3221 spin_unlock(&ci->i_unsafe_lock);
3222 }
3223
3224 ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
68cd5b4b 3225 }
2f2dc053 3226out_err:
e1518c7c 3227 mutex_lock(&mdsc->mutex);
bc2de10d 3228 if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
e1518c7c
SW		 3229 if (err) {
3230 req->r_err = err;
3231 } else {
5fdb1389 3232 req->r_reply = ceph_msg_get(msg);
bc2de10d 3233 set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
e1518c7c 3234 }
2f2dc053 3235 } else {
e1518c7c 3236 dout("reply arrived after request %lld was aborted\n", tid);
2f2dc053 3237 }
e1518c7c 3238 mutex_unlock(&mdsc->mutex);
2f2dc053 3239
2f2dc053
SW		 3240 mutex_unlock(&session->s_mutex);
3241
3242 /* kick calling process */
3243 complete_request(mdsc, req);
70c94820 3244
8ae99ae2 3245 ceph_update_metadata_metrics(&mdsc->metric, req->r_start_latency,
70c94820 3246 req->r_end_latency, err);
2f2dc053
SW		 3247out:
3248 ceph_mdsc_put_request(req);
3249 return;
3250}
3251
3252
3253
3254/*
3255 * handle mds notification that our request has been forwarded.
3256 */
2600d2dd
SW		 3257static void handle_forward(struct ceph_mds_client *mdsc,
3258 struct ceph_mds_session *session,
3259 struct ceph_msg *msg)
2f2dc053
SW		 3260{
3261 struct ceph_mds_request *req;
a1ea787c 3262 u64 tid = le64_to_cpu(msg->hdr.tid);
2f2dc053
SW		 3263 u32 next_mds;
3264 u32 fwd_seq;
2f2dc053
SW		 3265 int err = -EINVAL;
3266 void *p = msg->front.iov_base;
3267 void *end = p + msg->front.iov_len;
2f2dc053 3268
a1ea787c 3269 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
c89136ea
SW		 3270 next_mds = ceph_decode_32(&p);
3271 fwd_seq = ceph_decode_32(&p);
2f2dc053
SW		 3272
3273 mutex_lock(&mdsc->mutex);
fcd00b68 3274 req = lookup_get_request(mdsc, tid);
2f2dc053 3275 if (!req) {
2a8e5e36 3276 dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
2f2dc053
SW		 3277 goto out; /* dup reply? */
3278 }
3279
bc2de10d 3280 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
2a8e5e36
SW		 3281 dout("forward tid %llu aborted, unregistering\n", tid);
3282 __unregister_request(mdsc, req);
3283 } else if (fwd_seq <= req->r_num_fwd) {
3284 dout("forward tid %llu to mds%d - old seq %d <= %d\n",
2f2dc053
SW		 3285 tid, next_mds, req->r_num_fwd, fwd_seq);
3286 } else {
3287 /* resend. forward race not possible; mds would drop */
2a8e5e36
SW		 3288 dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
3289 BUG_ON(req->r_err);
bc2de10d 3290 BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
3de22be6 3291 req->r_attempts = 0;
2f2dc053
SW		 3292 req->r_num_fwd = fwd_seq;
3293 req->r_resend_mds = next_mds;
3294 put_request_session(req);
3295 __do_request(mdsc, req);
3296 }
3297 ceph_mdsc_put_request(req);
3298out:
3299 mutex_unlock(&mdsc->mutex);
3300 return;
3301
3302bad:
3303 pr_err("mdsc_handle_forward decode error err=%d\n", err);
3304}
3305
131d7eb4 3306static int __decode_session_metadata(void **p, void *end,
0b98acd6 3307 bool *blocklisted)
84bf3950
YZ		 3308{
3309 /* map<string,string> */
3310 u32 n;
131d7eb4 3311 bool err_str;
84bf3950
YZ		 3312 ceph_decode_32_safe(p, end, n, bad);
3313 while (n-- > 0) {
3314 u32 len;
3315 ceph_decode_32_safe(p, end, len, bad);
3316 ceph_decode_need(p, end, len, bad);
131d7eb4 3317 err_str = !strncmp(*p, "error_string", len);
84bf3950
YZ		 3318 *p += len;
3319 ceph_decode_32_safe(p, end, len, bad);
3320 ceph_decode_need(p, end, len, bad);
4bb926e8
ID		 3321 /*
3322 * Match "blocklisted (blacklisted)" from newer MDSes,
3323 * or "blacklisted" from older MDSes.
3324 */
131d7eb4 3325 if (err_str && strnstr(*p, "blacklisted", len))
0b98acd6 3326 *blocklisted = true;
84bf3950
YZ		 3327 *p += len;
3328 }
3329 return 0;
3330bad:
3331 return -1;
3332}
3333
2f2dc053
SW		 3334/*
3335 * handle a mds session control message
3336 */
3337static void handle_session(struct ceph_mds_session *session,
3338 struct ceph_msg *msg)
3339{
3340 struct ceph_mds_client *mdsc = session->s_mdsc;
84bf3950
YZ		 3341 int mds = session->s_mds;
3342 int msg_version = le16_to_cpu(msg->hdr.version);
3343 void *p = msg->front.iov_base;
3344 void *end = p + msg->front.iov_len;
3345 struct ceph_mds_session_head *h;
2f2dc053 3346 u32 op;
0fa82633 3347 u64 seq, features = 0;
2f2dc053 3348 int wake = 0;
0b98acd6 3349 bool blocklisted = false;
2f2dc053 3350
2f2dc053 3351 /* decode */
84bf3950
YZ		 3352 ceph_decode_need(&p, end, sizeof(*h), bad);
3353 h = p;
3354 p += sizeof(*h);
3355
2f2dc053
SW		 3356 op = le32_to_cpu(h->op);
3357 seq = le64_to_cpu(h->seq);
3358
84bf3950
YZ		 3359 if (msg_version >= 3) {
3360 u32 len;
e1c9788c
KH		 3361 /* version >= 2 and < 5, decode metadata, skip otherwise
3362 * as it's handled via flags.
3363 */
3364 if (msg_version >= 5)
3365 ceph_decode_skip_map(&p, end, string, string, bad);
3366 else if (__decode_session_metadata(&p, end, &blocklisted) < 0)
84bf3950 3367 goto bad;
e1c9788c 3368
84bf3950
YZ		 3369 /* version >= 3, feature bits */
3370 ceph_decode_32_safe(&p, end, len, bad);
02e37571
JL		 3371 if (len) {
3372 ceph_decode_64_safe(&p, end, features, bad);
3373 p += len - sizeof(features);
3374 }
84bf3950
YZ		 3375 }
3376
e1c9788c
KH		 3377 if (msg_version >= 5) {
3378 u32 flags;
3379 /* version >= 4, struct_v, struct_cv, len, metric_spec */
3380 ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 2, bad);
3381 /* version >= 5, flags */
3382 ceph_decode_32_safe(&p, end, flags, bad);
3383 if (flags & CEPH_SESSION_BLOCKLISTED) {
3384 pr_warn("mds%d session blocklisted\n", session->s_mds);
3385 blocklisted = true;
3386 }
3387 }
3388
2f2dc053 3389 mutex_lock(&mdsc->mutex);
0a07fc8c 3390 if (op == CEPH_SESSION_CLOSE) {
5b3248c6 3391 ceph_get_mds_session(session);
2600d2dd 3392 __unregister_session(mdsc, session);
0a07fc8c 3393 }
2f2dc053
SW		 3394 /* FIXME: this ttl calculation is generous */
3395 session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
3396 mutex_unlock(&mdsc->mutex);
3397
3398 mutex_lock(&session->s_mutex);
3399
3400 dout("handle_session mds%d %s %p state %s seq %llu\n",
3401 mds, ceph_session_op_name(op), session,
a687ecaf 3402 ceph_session_state_name(session->s_state), seq);
2f2dc053
SW		 3403
3404 if (session->s_state == CEPH_MDS_SESSION_HUNG) {
3405 session->s_state = CEPH_MDS_SESSION_OPEN;
3406 pr_info("mds%d came back\n", session->s_mds);
3407 }
3408
3409 switch (op) {
3410 case CEPH_SESSION_OPEN:
29790f26
SW		 3411 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
3412 pr_info("mds%d reconnect success\n", session->s_mds);
2f2dc053 3413 session->s_state = CEPH_MDS_SESSION_OPEN;
84bf3950 3414 session->s_features = features;
2f2dc053 3415 renewed_caps(mdsc, session, 0);
18f473b3
XL		 3416 if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT, &session->s_features))
3417 metric_schedule_delayed(&mdsc->metric);
2f2dc053
SW		 3418 wake = 1;
3419 if (mdsc->stopping)
3420 __close_session(mdsc, session);
3421 break;
3422
3423 case CEPH_SESSION_RENEWCAPS:
3424 if (session->s_renew_seq == seq)
3425 renewed_caps(mdsc, session, 1);
3426 break;
3427
3428 case CEPH_SESSION_CLOSE:
29790f26
SW		 3429 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
3430 pr_info("mds%d reconnect denied\n", session->s_mds);
4d681c2f 3431 session->s_state = CEPH_MDS_SESSION_CLOSED;
1c841a96 3432 cleanup_session_requests(mdsc, session);
2f2dc053 3433 remove_session_caps(session);
656e4382 3434 wake = 2; /* for good measure */
f3c60c59 3435 wake_up_all(&mdsc->session_close_wq);
2f2dc053
SW		 3436 break;
3437
3438 case CEPH_SESSION_STALE:
3439 pr_info("mds%d caps went stale, renewing\n",
3440 session->s_mds);
52d60f8e 3441 atomic_inc(&session->s_cap_gen);
1ce208a6 3442 session->s_cap_ttl = jiffies - 1;
2f2dc053
SW		 3443 send_renew_caps(mdsc, session);
3444 break;
3445
3446 case CEPH_SESSION_RECALL_STATE:
e30ee581 3447 ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
2f2dc053
SW		 3448 break;
3449
186e4f7a
YZ		 3450 case CEPH_SESSION_FLUSHMSG:
3451 send_flushmsg_ack(mdsc, session, seq);
3452 break;
3453
03f4fcb0
YZ		 3454 case CEPH_SESSION_FORCE_RO:
3455 dout("force_session_readonly %p\n", session);
3456 spin_lock(&session->s_cap_lock);
3457 session->s_readonly = true;
3458 spin_unlock(&session->s_cap_lock);
d2f8bb27 3459 wake_up_session_caps(session, FORCE_RO);
03f4fcb0
YZ		 3460 break;
3461
fcff415c
YZ		 3462 case CEPH_SESSION_REJECT:
3463 WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING);
3464 pr_info("mds%d rejected session\n", session->s_mds);
3465 session->s_state = CEPH_MDS_SESSION_REJECTED;
3466 cleanup_session_requests(mdsc, session);
3467 remove_session_caps(session);
0b98acd6
ID		 3468 if (blocklisted)
3469 mdsc->fsc->blocklisted = true;
fcff415c
YZ		 3470 wake = 2; /* for good measure */
3471 break;
3472
2f2dc053
SW		 3473 default:
3474 pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
3475 WARN_ON(1);
3476 }
3477
3478 mutex_unlock(&session->s_mutex);
3479 if (wake) {
3480 mutex_lock(&mdsc->mutex);
3481 __wake_requests(mdsc, &session->s_waiting);
656e4382
YZ		 3482 if (wake == 2)
3483 kick_requests(mdsc, mds);
2f2dc053
SW		 3484 mutex_unlock(&mdsc->mutex);
3485 }
0a07fc8c
YZ		 3486 if (op == CEPH_SESSION_CLOSE)
3487 ceph_put_mds_session(session);
2f2dc053
SW		 3488 return;
3489
3490bad:
3491 pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
3492 (int)msg->front.iov_len);
9ec7cab1 3493 ceph_msg_dump(msg);
2f2dc053
SW		 3494 return;
3495}
3496
a25949b9
JL		 3497void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)
3498{
3499 int dcaps;
3500
3501 dcaps = xchg(&req->r_dir_caps, 0);
3502 if (dcaps) {
3503 dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
3504 ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);
3505 }
3506}
3507
e64f44a8
XL		 3508void ceph_mdsc_release_dir_caps_no_check(struct ceph_mds_request *req)
3509{
3510 int dcaps;
3511
3512 dcaps = xchg(&req->r_dir_caps, 0);
3513 if (dcaps) {
3514 dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
3515 ceph_put_cap_refs_no_check_caps(ceph_inode(req->r_parent),
3516 dcaps);
3517 }
3518}
3519
2f2dc053
SW		 3520/*
3521 * called under session->mutex.
3522 */
3523static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
3524 struct ceph_mds_session *session)
3525{
3526 struct ceph_mds_request *req, *nreq;
3de22be6 3527 struct rb_node *p;
2f2dc053
SW		 3528
3529 dout("replay_unsafe_requests mds%d\n", session->s_mds);
3530
3531 mutex_lock(&mdsc->mutex);
9cf54563 3532 list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)
396bd62c 3533 __send_request(session, req, true);
3de22be6
YZ		 3534
3535 /*
3536 * also re-send old requests when MDS enters reconnect stage. So that MDS
3537 * can process completed request in clientreplay stage.
3538 */
3539 p = rb_first(&mdsc->request_tree);
3540 while (p) {
3541 req = rb_entry(p, struct ceph_mds_request, r_node);
3542 p = rb_next(p);
bc2de10d 3543 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
3de22be6
YZ		 3544 continue;
3545 if (req->r_attempts == 0)
3546 continue; /* only old requests */
a25949b9
JL		 3547 if (!req->r_session)
3548 continue;
3549 if (req->r_session->s_mds != session->s_mds)
3550 continue;
3551
e64f44a8 3552 ceph_mdsc_release_dir_caps_no_check(req);
a25949b9 3553
396bd62c 3554 __send_request(session, req, true);
3de22be6 3555 }
2f2dc053
SW		 3556 mutex_unlock(&mdsc->mutex);
3557}
3558
81c5a148
YZ		 3559static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)
3560{
3561 struct ceph_msg *reply;
3562 struct ceph_pagelist *_pagelist;
3563 struct page *page;
3564 __le32 *addr;
3565 int err = -ENOMEM;
3566
3567 if (!recon_state->allow_multi)
3568 return -ENOSPC;
3569
3570 /* can't handle message that contains both caps and realm */
3571 BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);
3572
3573 /* pre-allocate new pagelist */
3574 _pagelist = ceph_pagelist_alloc(GFP_NOFS);
3575 if (!_pagelist)
3576 return -ENOMEM;
3577
3578 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
3579 if (!reply)
3580 goto fail_msg;
3581
3582 /* placeholder for nr_caps */
3583 err = ceph_pagelist_encode_32(_pagelist, 0);
3584 if (err < 0)
3585 goto fail;
3586
3587 if (recon_state->nr_caps) {
3588 /* currently encoding caps */
3589 err = ceph_pagelist_encode_32(recon_state->pagelist, 0);
3590 if (err)
3591 goto fail;
3592 } else {
3593 /* placeholder for nr_realms (currently encoding relams) */
3594 err = ceph_pagelist_encode_32(_pagelist, 0);
3595 if (err < 0)
3596 goto fail;
3597 }
3598
3599 err = ceph_pagelist_encode_8(recon_state->pagelist, 1);
3600 if (err)
3601 goto fail;
3602
3603 page = list_first_entry(&recon_state->pagelist->head, struct page, lru);
3604 addr = kmap_atomic(page);
3605 if (recon_state->nr_caps) {
3606 /* currently encoding caps */
3607 *addr = cpu_to_le32(recon_state->nr_caps);
3608 } else {
3609 /* currently encoding relams */
3610 *(addr + 1) = cpu_to_le32(recon_state->nr_realms);
3611 }
3612 kunmap_atomic(addr);
3613
3614 reply->hdr.version = cpu_to_le16(5);
3615 reply->hdr.compat_version = cpu_to_le16(4);
3616
3617 reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);
3618 ceph_msg_data_add_pagelist(reply, recon_state->pagelist);
3619
3620 ceph_con_send(&recon_state->session->s_con, reply);
3621 ceph_pagelist_release(recon_state->pagelist);
3622
3623 recon_state->pagelist = _pagelist;
3624 recon_state->nr_caps = 0;
3625 recon_state->nr_realms = 0;
3626 recon_state->msg_version = 5;
3627 return 0;
3628fail:
3629 ceph_msg_put(reply);
3630fail_msg:
3631 ceph_pagelist_release(_pagelist);
3632 return err;
3633}
3634
a33f6432
YZ		 3635static struct dentry* d_find_primary(struct inode *inode)
3636{
3637 struct dentry *alias, *dn = NULL;
3638
3639 if (hlist_empty(&inode->i_dentry))
3640 return NULL;
3641
3642 spin_lock(&inode->i_lock);
3643 if (hlist_empty(&inode->i_dentry))
3644 goto out_unlock;
3645
3646 if (S_ISDIR(inode->i_mode)) {
3647 alias = hlist_entry(inode->i_dentry.first, struct dentry, d_u.d_alias);
3648 if (!IS_ROOT(alias))
3649 dn = dget(alias);
3650 goto out_unlock;
3651 }
3652
3653 hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) {
3654 spin_lock(&alias->d_lock);
3655 if (!d_unhashed(alias) &&
3656 (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {
3657 dn = dget_dlock(alias);
3658 }
3659 spin_unlock(&alias->d_lock);
3660 if (dn)
3661 break;
3662 }
3663out_unlock:
3664 spin_unlock(&inode->i_lock);
3665 return dn;
3666}
3667
2f2dc053
SW		 3668/*
3669 * Encode information about a cap for a reconnect with the MDS.
3670 */
a25949b9 3671static int reconnect_caps_cb(struct inode *inode, struct ceph_cap *cap,
2f2dc053
SW		 3672 void *arg)
3673{
20cb34ae
SW		 3674 union {
3675 struct ceph_mds_cap_reconnect v2;
3676 struct ceph_mds_cap_reconnect_v1 v1;
3677 } rec;
b3f8d68f 3678 struct ceph_inode_info *ci = cap->ci;
20cb34ae
SW		 3679 struct ceph_reconnect_state *recon_state = arg;
3680 struct ceph_pagelist *pagelist = recon_state->pagelist;
a33f6432
YZ		 3681 struct dentry *dentry;
3682 char *path;
ee2a095d 3683 int pathlen = 0, err;
a33f6432 3684 u64 pathbase;
3469ed0d 3685 u64 snap_follows;
2f2dc053 3686
2f2dc053
SW		 3687 dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
3688 inode, ceph_vinop(inode), cap, cap->cap_id,
3689 ceph_cap_string(cap->issued));
2f2dc053 3690
a33f6432
YZ		 3691 dentry = d_find_primary(inode);
3692 if (dentry) {
3693 /* set pathbase to parent dir when msg_version >= 2 */
3694 path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase,
3695 recon_state->msg_version >= 2);
3696 dput(dentry);
3697 if (IS_ERR(path)) {
3698 err = PTR_ERR(path);
3699 goto out_err;
3700 }
3701 } else {
3702 path = NULL;
a33f6432
YZ		 3703 pathbase = 0;
3704 }
3705
be655596 3706 spin_lock(&ci->i_ceph_lock);
2f2dc053
SW		 3707 cap->seq = 0; /* reset cap seq */
3708 cap->issue_seq = 0; /* and issue_seq */
667ca05c 3709 cap->mseq = 0; /* and migrate_seq */
52d60f8e 3710 cap->cap_gen = atomic_read(&cap->session->s_cap_gen);
20cb34ae 3711
a25949b9 3712 /* These are lost when the session goes away */
785892fe
JL		 3713 if (S_ISDIR(inode->i_mode)) {
3714 if (cap->issued & CEPH_CAP_DIR_CREATE) {
3715 ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
3716 memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
3717 }
a25949b9 3718 cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;
785892fe 3719 }
a25949b9 3720
121f22a1 3721 if (recon_state->msg_version >= 2) {
20cb34ae
SW		 3722 rec.v2.cap_id = cpu_to_le64(cap->cap_id);
3723 rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
3724 rec.v2.issued = cpu_to_le32(cap->issued);
3725 rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
a33f6432 3726 rec.v2.pathbase = cpu_to_le64(pathbase);
ec1dff25
JL		 3727 rec.v2.flock_len = (__force __le32)
3728 ((ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) ? 0 : 1);
20cb34ae
SW		 3729 } else {
3730 rec.v1.cap_id = cpu_to_le64(cap->cap_id);
3731 rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
3732 rec.v1.issued = cpu_to_le32(cap->issued);
2d6795fb 3733 rec.v1.size = cpu_to_le64(i_size_read(inode));
9bbeab41
AB		 3734 ceph_encode_timespec64(&rec.v1.mtime, &inode->i_mtime);
3735 ceph_encode_timespec64(&rec.v1.atime, &inode->i_atime);
20cb34ae 3736 rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
a33f6432 3737 rec.v1.pathbase = cpu_to_le64(pathbase);
20cb34ae 3738 }
3469ed0d
YZ		 3739
3740 if (list_empty(&ci->i_cap_snaps)) {
92776fd2 3741 snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0;
3469ed0d
YZ		 3742 } else {
3743 struct ceph_cap_snap *capsnap =
3744 list_first_entry(&ci->i_cap_snaps,
3745 struct ceph_cap_snap, ci_item);
3746 snap_follows = capsnap->follows;
20cb34ae 3747 }
be655596 3748 spin_unlock(&ci->i_ceph_lock);
2f2dc053 3749
121f22a1 3750 if (recon_state->msg_version >= 2) {
40819f6f 3751 int num_fcntl_locks, num_flock_locks;
4deb14a2 3752 struct ceph_filelock *flocks = NULL;
81c5a148 3753 size_t struct_len, total_len = sizeof(u64);
121f22a1 3754 u8 struct_v = 0;
39be95e9
JS		 3755
3756encode_again:
b3f8d68f
YZ		 3757 if (rec.v2.flock_len) {
3758 ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
3759 } else {
3760 num_fcntl_locks = 0;
3761 num_flock_locks = 0;
3762 }
4deb14a2 3763 if (num_fcntl_locks + num_flock_locks > 0) {
6da2ec56
KC		 3764 flocks = kmalloc_array(num_fcntl_locks + num_flock_locks,
3765 sizeof(struct ceph_filelock),
3766 GFP_NOFS);
4deb14a2
YZ		 3767 if (!flocks) {
3768 err = -ENOMEM;
5ccedf1c 3769 goto out_err;
4deb14a2
YZ		 3770 }
3771 err = ceph_encode_locks_to_buffer(inode, flocks,
3772 num_fcntl_locks,
3773 num_flock_locks);
3774 if (err) {
3775 kfree(flocks);
3776 flocks = NULL;
3777 if (err == -ENOSPC)
3778 goto encode_again;
5ccedf1c 3779 goto out_err;
4deb14a2
YZ		 3780 }
3781 } else {
39be95e9 3782 kfree(flocks);
4deb14a2 3783 flocks = NULL;
39be95e9 3784 }
121f22a1
YZ		 3785
3786 if (recon_state->msg_version >= 3) {
3787 /* version, compat_version and struct_len */
81c5a148 3788 total_len += 2 * sizeof(u8) + sizeof(u32);
3469ed0d 3789 struct_v = 2;
121f22a1 3790 }
39be95e9
JS		 3791 /*
3792 * number of encoded locks is stable, so copy to pagelist
3793 */
121f22a1
YZ		 3794 struct_len = 2 * sizeof(u32) +
3795 (num_fcntl_locks + num_flock_locks) *
3796 sizeof(struct ceph_filelock);
3797 rec.v2.flock_len = cpu_to_le32(struct_len);
3798
a33f6432 3799 struct_len += sizeof(u32) + pathlen + sizeof(rec.v2);
121f22a1 3800
3469ed0d
YZ		 3801 if (struct_v >= 2)
3802 struct_len += sizeof(u64); /* snap_follows */
3803
121f22a1 3804 total_len += struct_len;
81c5a148
YZ		 3805
3806 if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {
3807 err = send_reconnect_partial(recon_state);
3808 if (err)
3809 goto out_freeflocks;
3810 pagelist = recon_state->pagelist;
5ccedf1c 3811 }
121f22a1 3812
81c5a148
YZ		 3813 err = ceph_pagelist_reserve(pagelist, total_len);
3814 if (err)
3815 goto out_freeflocks;
3816
3817 ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
5ccedf1c
YZ		 3818 if (recon_state->msg_version >= 3) {
3819 ceph_pagelist_encode_8(pagelist, struct_v);
3820 ceph_pagelist_encode_8(pagelist, 1);
3821 ceph_pagelist_encode_32(pagelist, struct_len);
121f22a1 3822 }
a33f6432 3823 ceph_pagelist_encode_string(pagelist, path, pathlen);
5ccedf1c
YZ		 3824 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
3825 ceph_locks_to_pagelist(flocks, pagelist,
3826 num_fcntl_locks, num_flock_locks);
3827 if (struct_v >= 2)
3828 ceph_pagelist_encode_64(pagelist, snap_follows);
81c5a148 3829out_freeflocks:
39be95e9 3830 kfree(flocks);
3612abbd 3831 } else {
5ccedf1c 3832 err = ceph_pagelist_reserve(pagelist,
81c5a148
YZ		 3833 sizeof(u64) + sizeof(u32) +
3834 pathlen + sizeof(rec.v1));
a33f6432
YZ		 3835 if (err)
3836 goto out_err;
5ccedf1c 3837
81c5a148 3838 ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
5ccedf1c
YZ		 3839 ceph_pagelist_encode_string(pagelist, path, pathlen);
3840 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
40819f6f 3841 }
44c99757 3842
5ccedf1c 3843out_err:
a33f6432
YZ		 3844 ceph_mdsc_free_path(path, pathlen);
3845 if (!err)
81c5a148
YZ		 3846 recon_state->nr_caps++;
3847 return err;
3848}
3849
3850static int encode_snap_realms(struct ceph_mds_client *mdsc,
3851 struct ceph_reconnect_state *recon_state)
3852{
3853 struct rb_node *p;
3854 struct ceph_pagelist *pagelist = recon_state->pagelist;
3855 int err = 0;
3856
3857 if (recon_state->msg_version >= 4) {
3858 err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);
3859 if (err < 0)
3860 goto fail;
3861 }
3862
3863 /*
3864 * snaprealms. we provide mds with the ino, seq (version), and
3865 * parent for all of our realms. If the mds has any newer info,
3866 * it will tell us.
3867 */
3868 for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
3869 struct ceph_snap_realm *realm =
3870 rb_entry(p, struct ceph_snap_realm, node);
3871 struct ceph_mds_snaprealm_reconnect sr_rec;
3872
3873 if (recon_state->msg_version >= 4) {
3874 size_t need = sizeof(u8) * 2 + sizeof(u32) +
3875 sizeof(sr_rec);
3876
3877 if (pagelist->length + need > RECONNECT_MAX_SIZE) {
3878 err = send_reconnect_partial(recon_state);
3879 if (err)
3880 goto fail;
3881 pagelist = recon_state->pagelist;
3882 }
3883
3884 err = ceph_pagelist_reserve(pagelist, need);
3885 if (err)
3886 goto fail;
3887
3888 ceph_pagelist_encode_8(pagelist, 1);
3889 ceph_pagelist_encode_8(pagelist, 1);
3890 ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));
3891 }
3892
3893 dout(" adding snap realm %llx seq %lld parent %llx\n",
3894 realm->ino, realm->seq, realm->parent_ino);
3895 sr_rec.ino = cpu_to_le64(realm->ino);
3896 sr_rec.seq = cpu_to_le64(realm->seq);
3897 sr_rec.parent = cpu_to_le64(realm->parent_ino);
3898
3899 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
3900 if (err)
3901 goto fail;
3902
3903 recon_state->nr_realms++;
3904 }
3905fail:
93cea5be 3906 return err;
2f2dc053
SW		 3907}
3908
3909
3910/*
3911 * If an MDS fails and recovers, clients need to reconnect in order to
3912 * reestablish shared state. This includes all caps issued through
3913 * this session _and_ the snap_realm hierarchy. Because it's not
3914 * clear which snap realms the mds cares about, we send everything we
3915 * know about.. that ensures we'll then get any new info the
3916 * recovering MDS might have.
3917 *
3918 * This is a relatively heavyweight operation, but it's rare.
2f2dc053 3919 */
34b6c855
SW		 3920static void send_mds_reconnect(struct ceph_mds_client *mdsc,
3921 struct ceph_mds_session *session)
2f2dc053 3922{
2f2dc053 3923 struct ceph_msg *reply;
34b6c855 3924 int mds = session->s_mds;
9abf82b8 3925 int err = -ENOMEM;
81c5a148
YZ		 3926 struct ceph_reconnect_state recon_state = {
3927 .session = session,
3928 };
c8a96a31 3929 LIST_HEAD(dispose);
2f2dc053 3930
34b6c855 3931 pr_info("mds%d reconnect start\n", mds);
2f2dc053 3932
81c5a148
YZ		 3933 recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
3934 if (!recon_state.pagelist)
93cea5be 3935 goto fail_nopagelist;
93cea5be 3936
0d9c1ab3 3937 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
a79832f2 3938 if (!reply)
93cea5be 3939 goto fail_nomsg;
93cea5be 3940
d4846487
JL		 3941 xa_destroy(&session->s_delegated_inos);
3942
34b6c855
SW		 3943 mutex_lock(&session->s_mutex);
3944 session->s_state = CEPH_MDS_SESSION_RECONNECTING;
3945 session->s_seq = 0;
2f2dc053 3946
2f2dc053 3947 dout("session %p state %s\n", session,
a687ecaf 3948 ceph_session_state_name(session->s_state));
2f2dc053 3949
52d60f8e 3950 atomic_inc(&session->s_cap_gen);
99a9c273
YZ		 3951
3952 spin_lock(&session->s_cap_lock);
03f4fcb0
YZ		 3953 /* don't know if session is readonly */
3954 session->s_readonly = 0;
99a9c273
YZ		 3955 /*
3956 * notify __ceph_remove_cap() that we are composing cap reconnect.
3957 * If a cap get released before being added to the cap reconnect,
3958 * __ceph_remove_cap() should skip queuing cap release.
3959 */
3960 session->s_cap_reconnect = 1;
e01a5946 3961 /* drop old cap expires; we're about to reestablish that state */
c8a96a31
JL		 3962 detach_cap_releases(session, &dispose);
3963 spin_unlock(&session->s_cap_lock);
3964 dispose_cap_releases(mdsc, &dispose);
e01a5946 3965
5d23371f 3966 /* trim unused caps to reduce MDS's cache rejoin time */
c0bd50e2
YZ		 3967 if (mdsc->fsc->sb->s_root)
3968 shrink_dcache_parent(mdsc->fsc->sb->s_root);
5d23371f
YZ		 3969
3970 ceph_con_close(&session->s_con);
3971 ceph_con_open(&session->s_con,
3972 CEPH_ENTITY_TYPE_MDS, mds,
3973 ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
3974
3975 /* replay unsafe requests */
3976 replay_unsafe_requests(mdsc, session);
3977
81c5a148
YZ		 3978 ceph_early_kick_flushing_caps(mdsc, session);
3979
5d23371f
YZ		 3980 down_read(&mdsc->snap_rwsem);
3981
81c5a148
YZ		 3982 /* placeholder for nr_caps */
3983 err = ceph_pagelist_encode_32(recon_state.pagelist, 0);
93cea5be
SW		 3984 if (err)
3985 goto fail;
20cb34ae 3986
81c5a148 3987 if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {
121f22a1 3988 recon_state.msg_version = 3;
81c5a148
YZ		 3989 recon_state.allow_multi = true;
3990 } else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {
3991 recon_state.msg_version = 3;
3992 } else {
23c625ce 3993 recon_state.msg_version = 2;
81c5a148
YZ		 3994 }
3995 /* trsaverse this session's caps */
a25949b9 3996 err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);
2f2dc053 3997
99a9c273
YZ		 3998 spin_lock(&session->s_cap_lock);
3999 session->s_cap_reconnect = 0;
4000 spin_unlock(&session->s_cap_lock);
4001
81c5a148
YZ		 4002 if (err < 0)
4003 goto fail;
2f2dc053 4004
81c5a148
YZ		 4005 /* check if all realms can be encoded into current message */
4006 if (mdsc->num_snap_realms) {
4007 size_t total_len =
4008 recon_state.pagelist->length +
4009 mdsc->num_snap_realms *
4010 sizeof(struct ceph_mds_snaprealm_reconnect);
4011 if (recon_state.msg_version >= 4) {
4012 /* number of realms */
4013 total_len += sizeof(u32);
4014 /* version, compat_version and struct_len */
4015 total_len += mdsc->num_snap_realms *
4016 (2 * sizeof(u8) + sizeof(u32));
4017 }
4018 if (total_len > RECONNECT_MAX_SIZE) {
4019 if (!recon_state.allow_multi) {
4020 err = -ENOSPC;
4021 goto fail;
4022 }
4023 if (recon_state.nr_caps) {
4024 err = send_reconnect_partial(&recon_state);
4025 if (err)
4026 goto fail;
4027 }
4028 recon_state.msg_version = 5;
4029 }
2f2dc053 4030 }
2f2dc053 4031
81c5a148
YZ		 4032 err = encode_snap_realms(mdsc, &recon_state);
4033 if (err < 0)
4034 goto fail;
4035
4036 if (recon_state.msg_version >= 5) {
4037 err = ceph_pagelist_encode_8(recon_state.pagelist, 0);
4038 if (err < 0)
4039 goto fail;
4040 }
44c99757 4041
81c5a148
YZ		 4042 if (recon_state.nr_caps || recon_state.nr_realms) {
4043 struct page *page =
4044 list_first_entry(&recon_state.pagelist->head,
4045 struct page, lru);
44c99757 4046 __le32 *addr = kmap_atomic(page);
81c5a148
YZ		 4047 if (recon_state.nr_caps) {
4048 WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);
4049 *addr = cpu_to_le32(recon_state.nr_caps);
4050 } else if (recon_state.msg_version >= 4) {
4051 *(addr + 1) = cpu_to_le32(recon_state.nr_realms);
4052 }
44c99757 4053 kunmap_atomic(addr);
ebf18f47 4054 }
44c99757 4055
81c5a148
YZ		 4056 reply->hdr.version = cpu_to_le16(recon_state.msg_version);
4057 if (recon_state.msg_version >= 4)
4058 reply->hdr.compat_version = cpu_to_le16(4);
e548e9b9 4059
81c5a148
YZ		 4060 reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);
4061 ceph_msg_data_add_pagelist(reply, recon_state.pagelist);
e548e9b9 4062
2f2dc053
SW		 4063 ceph_con_send(&session->s_con, reply);
4064
9abf82b8
SW		 4065 mutex_unlock(&session->s_mutex);
4066
4067 mutex_lock(&mdsc->mutex);
4068 __wake_requests(mdsc, &session->s_waiting);
4069 mutex_unlock(&mdsc->mutex);
4070
2f2dc053 4071 up_read(&mdsc->snap_rwsem);
81c5a148 4072 ceph_pagelist_release(recon_state.pagelist);
2f2dc053
SW		 4073 return;
4074
93cea5be 4075fail:
2f2dc053 4076 ceph_msg_put(reply);
9abf82b8
SW		 4077 up_read(&mdsc->snap_rwsem);
4078 mutex_unlock(&session->s_mutex);
93cea5be 4079fail_nomsg:
81c5a148 4080 ceph_pagelist_release(recon_state.pagelist);
93cea5be 4081fail_nopagelist:
9abf82b8 4082 pr_err("error %d preparing reconnect for mds%d\n", err, mds);
9abf82b8 4083 return;
2f2dc053
SW		 4084}
4085
4086
4087/*
4088 * compare old and new mdsmaps, kicking requests
4089 * and closing out old connections as necessary
4090 *
4091 * called under mdsc->mutex.
4092 */
4093static void check_new_map(struct ceph_mds_client *mdsc,
4094 struct ceph_mdsmap *newmap,
4095 struct ceph_mdsmap *oldmap)
4096{
d517b398 4097 int i, j, err;
2f2dc053
SW		 4098 int oldstate, newstate;
4099 struct ceph_mds_session *s;
d517b398 4100 unsigned long targets[DIV_ROUND_UP(CEPH_MAX_MDS, sizeof(unsigned long))] = {0};
2f2dc053
SW		 4101
4102 dout("check_new_map new %u old %u\n",
4103 newmap->m_epoch, oldmap->m_epoch);
4104
d517b398
XL		 4105 if (newmap->m_info) {
4106 for (i = 0; i < newmap->possible_max_rank; i++) {
4107 for (j = 0; j < newmap->m_info[i].num_export_targets; j++)
4108 set_bit(newmap->m_info[i].export_targets[j], targets);
4109 }
4110 }
4111
b38c9eb4 4112 for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {
d37b1d99 4113 if (!mdsc->sessions[i])
2f2dc053
SW		 4114 continue;
4115 s = mdsc->sessions[i];
4116 oldstate = ceph_mdsmap_get_state(oldmap, i);
4117 newstate = ceph_mdsmap_get_state(newmap, i);
4118
0deb01c9 4119 dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
2f2dc053 4120 i, ceph_mds_state_name(oldstate),
0deb01c9 4121 ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
2f2dc053 4122 ceph_mds_state_name(newstate),
0deb01c9 4123 ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
a687ecaf 4124 ceph_session_state_name(s->s_state));
2f2dc053 4125
b38c9eb4 4126 if (i >= newmap->possible_max_rank) {
6f0f597b 4127 /* force close session for stopped mds */
5b3248c6 4128 ceph_get_mds_session(s);
6f0f597b
YZ		 4129 __unregister_session(mdsc, s);
4130 __wake_requests(mdsc, &s->s_waiting);
4131 mutex_unlock(&mdsc->mutex);
2827528d 4132
6f0f597b
YZ		 4133 mutex_lock(&s->s_mutex);
4134 cleanup_session_requests(mdsc, s);
4135 remove_session_caps(s);
4136 mutex_unlock(&s->s_mutex);
2827528d 4137
6f0f597b 4138 ceph_put_mds_session(s);
2827528d 4139
6f0f597b
YZ		 4140 mutex_lock(&mdsc->mutex);
4141 kick_requests(mdsc, i);
4142 continue;
4143 }
4144
4145 if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
4146 ceph_mdsmap_get_addr(newmap, i),
4147 sizeof(struct ceph_entity_addr))) {
4148 /* just close it */
4149 mutex_unlock(&mdsc->mutex);
4150 mutex_lock(&s->s_mutex);
4151 mutex_lock(&mdsc->mutex);
4152 ceph_con_close(&s->s_con);
4153 mutex_unlock(&s->s_mutex);
4154 s->s_state = CEPH_MDS_SESSION_RESTARTING;
2f2dc053
SW		 4155 } else if (oldstate == newstate) {
4156 continue; /* nothing new with this mds */
4157 }
4158
4159 /*
4160 * send reconnect?
4161 */
4162 if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
34b6c855
SW		 4163 newstate >= CEPH_MDS_STATE_RECONNECT) {
4164 mutex_unlock(&mdsc->mutex);
d517b398 4165 clear_bit(i, targets);
34b6c855
SW		 4166 send_mds_reconnect(mdsc, s);
4167 mutex_lock(&mdsc->mutex);
4168 }
2f2dc053
SW		 4169
4170 /*
29790f26 4171 * kick request on any mds that has gone active.
2f2dc053
SW		 4172 */
4173 if (oldstate < CEPH_MDS_STATE_ACTIVE &&
4174 newstate >= CEPH_MDS_STATE_ACTIVE) {
29790f26
SW		 4175 if (oldstate != CEPH_MDS_STATE_CREATING &&
4176 oldstate != CEPH_MDS_STATE_STARTING)
4177 pr_info("mds%d recovery completed\n", s->s_mds);
4178 kick_requests(mdsc, i);
ea8412b2 4179 mutex_unlock(&mdsc->mutex);
829ad4db 4180 mutex_lock(&s->s_mutex);
ea8412b2 4181 mutex_lock(&mdsc->mutex);
2f2dc053 4182 ceph_kick_flushing_caps(mdsc, s);
829ad4db 4183 mutex_unlock(&s->s_mutex);
d2f8bb27 4184 wake_up_session_caps(s, RECONNECT);
2f2dc053
SW		 4185 }
4186 }
cb170a22 4187
d517b398
XL		 4188 /*
4189 * Only open and reconnect sessions that don't exist yet.
4190 */
4191 for (i = 0; i < newmap->possible_max_rank; i++) {
4192 /*
4193 * In case the import MDS is crashed just after
4194 * the EImportStart journal is flushed, so when
4195 * a standby MDS takes over it and is replaying
4196 * the EImportStart journal the new MDS daemon
4197 * will wait the client to reconnect it, but the
4198 * client may never register/open the session yet.
4199 *
4200 * Will try to reconnect that MDS daemon if the
4201 * rank number is in the export targets array and
4202 * is the up:reconnect state.
4203 */
4204 newstate = ceph_mdsmap_get_state(newmap, i);
4205 if (!test_bit(i, targets) || newstate != CEPH_MDS_STATE_RECONNECT)
4206 continue;
4207
4208 /*
4209 * The session maybe registered and opened by some
4210 * requests which were choosing random MDSes during
4211 * the mdsc->mutex's unlock/lock gap below in rare
4212 * case. But the related MDS daemon will just queue
4213 * that requests and be still waiting for the client's
4214 * reconnection request in up:reconnect state.
4215 */
4216 s = __ceph_lookup_mds_session(mdsc, i);
4217 if (likely(!s)) {
4218 s = __open_export_target_session(mdsc, i);
4219 if (IS_ERR(s)) {
4220 err = PTR_ERR(s);
4221 pr_err("failed to open export target session, err %d\n",
4222 err);
4223 continue;
4224 }
4225 }
4226 dout("send reconnect to export target mds.%d\n", i);
4227 mutex_unlock(&mdsc->mutex);
4228 send_mds_reconnect(mdsc, s);
4229 ceph_put_mds_session(s);
4230 mutex_lock(&mdsc->mutex);
4231 }
4232
b38c9eb4 4233 for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {
cb170a22
SW		 4234 s = mdsc->sessions[i];
4235 if (!s)
4236 continue;
4237 if (!ceph_mdsmap_is_laggy(newmap, i))
4238 continue;
4239 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
4240 s->s_state == CEPH_MDS_SESSION_HUNG ||
4241 s->s_state == CEPH_MDS_SESSION_CLOSING) {
4242 dout(" connecting to export targets of laggy mds%d\n",
4243 i);
4244 __open_export_target_sessions(mdsc, s);
4245 }
4246 }
2f2dc053
SW		 4247}
4248
4249
4250
4251/*
4252 * leases
4253 */
4254
4255/*
4256 * caller must hold session s_mutex, dentry->d_lock
4257 */
4258void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
4259{
4260 struct ceph_dentry_info *di = ceph_dentry(dentry);
4261
4262 ceph_put_mds_session(di->lease_session);
4263 di->lease_session = NULL;
4264}
4265
2600d2dd
SW		 4266static void handle_lease(struct ceph_mds_client *mdsc,
4267 struct ceph_mds_session *session,
4268 struct ceph_msg *msg)
2f2dc053 4269{
3d14c5d2 4270 struct super_block *sb = mdsc->fsc->sb;
2f2dc053 4271 struct inode *inode;
2f2dc053
SW		 4272 struct dentry *parent, *dentry;
4273 struct ceph_dentry_info *di;
2600d2dd 4274 int mds = session->s_mds;
2f2dc053 4275 struct ceph_mds_lease *h = msg->front.iov_base;
1e5ea23d 4276 u32 seq;
2f2dc053 4277 struct ceph_vino vino;
2f2dc053
SW		 4278 struct qstr dname;
4279 int release = 0;
4280
2f2dc053
SW		 4281 dout("handle_lease from mds%d\n", mds);
4282
4283 /* decode */
4284 if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
4285 goto bad;
4286 vino.ino = le64_to_cpu(h->ino);
4287 vino.snap = CEPH_NOSNAP;
1e5ea23d 4288 seq = le32_to_cpu(h->seq);
0fcf6c02
YZ		 4289 dname.len = get_unaligned_le32(h + 1);
4290 if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len)
2f2dc053 4291 goto bad;
0fcf6c02 4292 dname.name = (void *)(h + 1) + sizeof(u32);
2f2dc053 4293
2f2dc053
SW		 4294 /* lookup inode */
4295 inode = ceph_find_inode(sb, vino);
2f90b852
SW		 4296 dout("handle_lease %s, ino %llx %p %.*s\n",
4297 ceph_lease_op_name(h->action), vino.ino, inode,
1e5ea23d 4298 dname.len, dname.name);
6cd3bcad
YZ		 4299
4300 mutex_lock(&session->s_mutex);
62575e27 4301 inc_session_sequence(session);
6cd3bcad 4302
d37b1d99 4303 if (!inode) {
2f2dc053
SW		 4304 dout("handle_lease no inode %llx\n", vino.ino);
4305 goto release;
4306 }
2f2dc053
SW		 4307
4308 /* dentry */
4309 parent = d_find_alias(inode);
4310 if (!parent) {
4311 dout("no parent dentry on inode %p\n", inode);
4312 WARN_ON(1);
4313 goto release; /* hrm... */
4314 }
8387ff25 4315 dname.hash = full_name_hash(parent, dname.name, dname.len);
2f2dc053
SW		 4316 dentry = d_lookup(parent, &dname);
4317 dput(parent);
4318 if (!dentry)
4319 goto release;
4320
4321 spin_lock(&dentry->d_lock);
4322 di = ceph_dentry(dentry);
4323 switch (h->action) {
4324 case CEPH_MDS_LEASE_REVOKE:
3d8eb7a9 4325 if (di->lease_session == session) {
1e5ea23d
SW		 4326 if (ceph_seq_cmp(di->lease_seq, seq) > 0)
4327 h->seq = cpu_to_le32(di->lease_seq);
2f2dc053
SW		 4328 __ceph_mdsc_drop_dentry_lease(dentry);
4329 }
4330 release = 1;
4331 break;
4332
4333 case CEPH_MDS_LEASE_RENEW:
3d8eb7a9 4334 if (di->lease_session == session &&
52d60f8e 4335 di->lease_gen == atomic_read(&session->s_cap_gen) &&
2f2dc053
SW		 4336 di->lease_renew_from &&
4337 di->lease_renew_after == 0) {
4338 unsigned long duration =
3563dbdd 4339 msecs_to_jiffies(le32_to_cpu(h->duration_ms));
2f2dc053 4340
1e5ea23d 4341 di->lease_seq = seq;
9b16f03c 4342 di->time = di->lease_renew_from + duration;
2f2dc053
SW		 4343 di->lease_renew_after = di->lease_renew_from +
4344 (duration >> 1);
4345 di->lease_renew_from = 0;
4346 }
4347 break;
4348 }
4349 spin_unlock(&dentry->d_lock);
4350 dput(dentry);
4351
4352 if (!release)
4353 goto out;
4354
4355release:
4356 /* let's just reuse the same message */
4357 h->action = CEPH_MDS_LEASE_REVOKE_ACK;
4358 ceph_msg_get(msg);
4359 ceph_con_send(&session->s_con, msg);
4360
4361out:
2f2dc053 4362 mutex_unlock(&session->s_mutex);
23c2c76e 4363 iput(inode);
2f2dc053
SW		 4364 return;
4365
4366bad:
4367 pr_err("corrupt lease message\n");
9ec7cab1 4368 ceph_msg_dump(msg);
2f2dc053
SW		 4369}
4370
4371void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
2f2dc053
SW		 4372 struct dentry *dentry, char action,
4373 u32 seq)
4374{
4375 struct ceph_msg *msg;
4376 struct ceph_mds_lease *lease;
8f2a98ef
YZ		 4377 struct inode *dir;
4378 int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;
2f2dc053 4379
8f2a98ef
YZ		 4380 dout("lease_send_msg identry %p %s to mds%d\n",
4381 dentry, ceph_lease_op_name(action), session->s_mds);
2f2dc053 4382
b61c2763 4383 msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
a79832f2 4384 if (!msg)
2f2dc053
SW		 4385 return;
4386 lease = msg->front.iov_base;
4387 lease->action = action;
2f2dc053 4388 lease->seq = cpu_to_le32(seq);
2f2dc053 4389
8f2a98ef
YZ		 4390 spin_lock(&dentry->d_lock);
4391 dir = d_inode(dentry->d_parent);
4392 lease->ino = cpu_to_le64(ceph_ino(dir));
4393 lease->first = lease->last = cpu_to_le64(ceph_snap(dir));
4394
4395 put_unaligned_le32(dentry->d_name.len, lease + 1);
4396 memcpy((void *)(lease + 1) + 4,
4397 dentry->d_name.name, dentry->d_name.len);
4398 spin_unlock(&dentry->d_lock);
2f2dc053
SW		 4399 /*
4400 * if this is a preemptive lease RELEASE, no need to
4401 * flush request stream, since the actual request will
4402 * soon follow.
4403 */
4404 msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
4405
4406 ceph_con_send(&session->s_con, msg);
4407}
4408
2f2dc053 4409/*
59b312f3 4410 * lock unlock the session, to wait ongoing session activities
2f2dc053 4411 */
59b312f3 4412static void lock_unlock_session(struct ceph_mds_session *s)
2f2dc053 4413{
59b312f3
XL		 4414 mutex_lock(&s->s_mutex);
4415 mutex_unlock(&s->s_mutex);
2f2dc053
SW		 4416}
4417
131d7eb4
YZ		 4418static void maybe_recover_session(struct ceph_mds_client *mdsc)
4419{
4420 struct ceph_fs_client *fsc = mdsc->fsc;
4421
4422 if (!ceph_test_mount_opt(fsc, CLEANRECOVER))
4423 return;
2f2dc053 4424
131d7eb4
YZ		 4425 if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)
4426 return;
4427
0b98acd6 4428 if (!READ_ONCE(fsc->blocklisted))
131d7eb4
YZ		 4429 return;
4430
0b98acd6 4431 pr_info("auto reconnect after blocklisted\n");
131d7eb4
YZ		 4432 ceph_force_reconnect(fsc->sb);
4433}
2f2dc053 4434
3e699bd8
XL		 4435bool check_session_state(struct ceph_mds_session *s)
4436{
62575e27
JL		 4437 switch (s->s_state) {
4438 case CEPH_MDS_SESSION_OPEN:
4439 if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
3e699bd8
XL		 4440 s->s_state = CEPH_MDS_SESSION_HUNG;
4441 pr_info("mds%d hung\n", s->s_mds);
4442 }
62575e27
JL		 4443 break;
4444 case CEPH_MDS_SESSION_CLOSING:
62575e27
JL		 4445 case CEPH_MDS_SESSION_NEW:
4446 case CEPH_MDS_SESSION_RESTARTING:
4447 case CEPH_MDS_SESSION_CLOSED:
4448 case CEPH_MDS_SESSION_REJECTED:
3e699bd8 4449 return false;
62575e27 4450 }
3e699bd8
XL		 4451
4452 return true;
4453}
4454
62575e27
JL		 4455/*
4456 * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,
4457 * then we need to retransmit that request.
4458 */
4459void inc_session_sequence(struct ceph_mds_session *s)
4460{
4461 lockdep_assert_held(&s->s_mutex);
4462
4463 s->s_seq++;
4464
4465 if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
4466 int ret;
4467
4468 dout("resending session close request for mds%d\n", s->s_mds);
4469 ret = request_close_session(s);
4470 if (ret < 0)
4471 pr_err("unable to close session to mds%d: %d\n",
4472 s->s_mds, ret);
4473 }
4474}
4475
2f2dc053 4476/*
bf2ba432
LH		 4477 * delayed work -- periodically trim expired leases, renew caps with mds. If
4478 * the @delay parameter is set to 0 or if it's more than 5 secs, the default
4479 * workqueue delay value of 5 secs will be used.
2f2dc053 4480 */
bf2ba432 4481static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)
2f2dc053 4482{
bf2ba432
LH		 4483 unsigned long max_delay = HZ * 5;
4484
4485 /* 5 secs default delay */
4486 if (!delay || (delay > max_delay))
4487 delay = max_delay;
4488 schedule_delayed_work(&mdsc->delayed_work,
4489 round_jiffies_relative(delay));
2f2dc053
SW		 4490}
4491
4492static void delayed_work(struct work_struct *work)
4493{
2f2dc053
SW		 4494 struct ceph_mds_client *mdsc =
4495 container_of(work, struct ceph_mds_client, delayed_work.work);
bf2ba432 4496 unsigned long delay;
2f2dc053
SW		 4497 int renew_interval;
4498 int renew_caps;
bf2ba432 4499 int i;
2f2dc053
SW		 4500
4501 dout("mdsc delayed_work\n");
75c9627e 4502
fa996773
XL		 4503 if (mdsc->stopping)
4504 return;
4505
2f2dc053
SW		 4506 mutex_lock(&mdsc->mutex);
4507 renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
4508 renew_caps = time_after_eq(jiffies, HZ*renew_interval +
4509 mdsc->last_renew_caps);
4510 if (renew_caps)
4511 mdsc->last_renew_caps = jiffies;
4512
4513 for (i = 0; i < mdsc->max_sessions; i++) {
4514 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
d37b1d99 4515 if (!s)
2f2dc053 4516 continue;
3e699bd8
XL		 4517
4518 if (!check_session_state(s)) {
2f2dc053
SW		 4519 ceph_put_mds_session(s);
4520 continue;
4521 }
4522 mutex_unlock(&mdsc->mutex);
4523
4524 mutex_lock(&s->s_mutex);
4525 if (renew_caps)
4526 send_renew_caps(mdsc, s);
4527 else
4528 ceph_con_keepalive(&s->s_con);
aab53dd9
SW		 4529 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
4530 s->s_state == CEPH_MDS_SESSION_HUNG)
3d7ded4d 4531 ceph_send_cap_releases(mdsc, s);
2f2dc053
SW		 4532 mutex_unlock(&s->s_mutex);
4533 ceph_put_mds_session(s);
4534
4535 mutex_lock(&mdsc->mutex);
4536 }
4537 mutex_unlock(&mdsc->mutex);
4538
bf2ba432 4539 delay = ceph_check_delayed_caps(mdsc);
37c4efc1
YZ		 4540
4541 ceph_queue_cap_reclaim_work(mdsc);
4542
4543 ceph_trim_snapid_map(mdsc);
4544
131d7eb4
YZ		 4545 maybe_recover_session(mdsc);
4546
bf2ba432 4547 schedule_delayed(mdsc, delay);
2f2dc053
SW		 4548}
4549
3d14c5d2 4550int ceph_mdsc_init(struct ceph_fs_client *fsc)
2f2dc053 4551
2f2dc053 4552{
3d14c5d2 4553 struct ceph_mds_client *mdsc;
f9009efa 4554 int err;
3d14c5d2
YS		 4555
4556 mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
4557 if (!mdsc)
4558 return -ENOMEM;
4559 mdsc->fsc = fsc;
2f2dc053
SW		 4560 mutex_init(&mdsc->mutex);
4561 mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
d37b1d99 4562 if (!mdsc->mdsmap) {
f9009efa
XL		 4563 err = -ENOMEM;
4564 goto err_mdsc;
fb3101b6 4565 }
2d06eeb8 4566
2f2dc053 4567 init_completion(&mdsc->safe_umount_waiters);
f3c60c59 4568 init_waitqueue_head(&mdsc->session_close_wq);
2f2dc053 4569 INIT_LIST_HEAD(&mdsc->waiting_for_map);
0c44a8e0
LH		 4570 mdsc->quotarealms_inodes = RB_ROOT;
4571 mutex_init(&mdsc->quotarealms_inodes_mutex);
2f2dc053 4572 init_rwsem(&mdsc->snap_rwsem);
a105f00c 4573 mdsc->snap_realms = RB_ROOT;
2f2dc053
SW		 4574 INIT_LIST_HEAD(&mdsc->snap_empty);
4575 spin_lock_init(&mdsc->snap_empty_lock);
44ca18f2 4576 mdsc->request_tree = RB_ROOT;
2f2dc053
SW		 4577 INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
4578 mdsc->last_renew_caps = jiffies;
4579 INIT_LIST_HEAD(&mdsc->cap_delay_list);
3a3430af 4580 INIT_LIST_HEAD(&mdsc->cap_wait_list);
2f2dc053
SW		 4581 spin_lock_init(&mdsc->cap_delay_lock);
4582 INIT_LIST_HEAD(&mdsc->snap_flush_list);
4583 spin_lock_init(&mdsc->snap_flush_lock);
553adfd9 4584 mdsc->last_cap_flush_tid = 1;
e4500b5e 4585 INIT_LIST_HEAD(&mdsc->cap_flush_list);
db354052 4586 INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
2f2dc053
SW		 4587 spin_lock_init(&mdsc->cap_dirty_lock);
4588 init_waitqueue_head(&mdsc->cap_flushing_wq);
37c4efc1 4589 INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);
f9009efa
XL		 4590 err = ceph_metric_init(&mdsc->metric);
4591 if (err)
4592 goto err_mdsmap;
37c4efc1
YZ		 4593
4594 spin_lock_init(&mdsc->dentry_list_lock);
4595 INIT_LIST_HEAD(&mdsc->dentry_leases);
4596 INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
2d06eeb8 4597
37151668 4598 ceph_caps_init(mdsc);
fe33032d 4599 ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
37151668 4600
75c9627e
YZ		 4601 spin_lock_init(&mdsc->snapid_map_lock);
4602 mdsc->snapid_map_tree = RB_ROOT;
4603 INIT_LIST_HEAD(&mdsc->snapid_map_lru);
4604
10183a69
YZ		 4605 init_rwsem(&mdsc->pool_perm_rwsem);
4606 mdsc->pool_perm_tree = RB_ROOT;
4607
dfeb84d4
YZ		 4608 strscpy(mdsc->nodename, utsname()->nodename,
4609 sizeof(mdsc->nodename));
a7caa88f
XL		 4610
4611 fsc->mdsc = mdsc;
5f44f142 4612 return 0;
f9009efa
XL		 4613
4614err_mdsmap:
4615 kfree(mdsc->mdsmap);
4616err_mdsc:
4617 kfree(mdsc);
4618 return err;
2f2dc053
SW		 4619}
4620
4621/*
4622 * Wait for safe replies on open mds requests. If we time out, drop
4623 * all requests from the tree to avoid dangling dentry refs.
4624 */
4625static void wait_requests(struct ceph_mds_client *mdsc)
4626{
a319bf56 4627 struct ceph_options *opts = mdsc->fsc->client->options;
2f2dc053 4628 struct ceph_mds_request *req;
2f2dc053
SW		 4629
4630 mutex_lock(&mdsc->mutex);
44ca18f2 4631 if (__get_oldest_req(mdsc)) {
2f2dc053 4632 mutex_unlock(&mdsc->mutex);
44ca18f2 4633
2f2dc053
SW		 4634 dout("wait_requests waiting for requests\n");
4635 wait_for_completion_timeout(&mdsc->safe_umount_waiters,
a319bf56 4636 ceph_timeout_jiffies(opts->mount_timeout));
2f2dc053
SW		 4637
4638 /* tear down remaining requests */
44ca18f2
SW		 4639 mutex_lock(&mdsc->mutex);
4640 while ((req = __get_oldest_req(mdsc))) {
2f2dc053
SW		 4641 dout("wait_requests timed out on tid %llu\n",
4642 req->r_tid);
428138c9 4643 list_del_init(&req->r_wait);
44ca18f2 4644 __unregister_request(mdsc, req);
2f2dc053
SW		 4645 }
4646 }
4647 mutex_unlock(&mdsc->mutex);
4648 dout("wait_requests done\n");
4649}
4650
d095559c
XL		 4651void send_flush_mdlog(struct ceph_mds_session *s)
4652{
4653 struct ceph_msg *msg;
4654
4655 /*
4656 * Pre-luminous MDS crashes when it sees an unknown session request
4657 */
4658 if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS))
4659 return;
4660
4661 mutex_lock(&s->s_mutex);
4662 dout("request mdlog flush to mds%d (%s)s seq %lld\n", s->s_mds,
4663 ceph_session_state_name(s->s_state), s->s_seq);
4664 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG,
4665 s->s_seq);
4666 if (!msg) {
4667 pr_err("failed to request mdlog flush to mds%d (%s) seq %lld\n",
4668 s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);
4669 } else {
4670 ceph_con_send(&s->s_con, msg);
4671 }
4672 mutex_unlock(&s->s_mutex);
4673}
4674
2f2dc053
SW		 4675/*
4676 * called before mount is ro, and before dentries are torn down.
4677 * (hmm, does this still race with new lookups?)
4678 */
4679void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
4680{
4681 dout("pre_umount\n");
4682 mdsc->stopping = 1;
4683
d095559c 4684 ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true);
59b312f3 4685 ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false);
afcdaea3 4686 ceph_flush_dirty_caps(mdsc);
2f2dc053 4687 wait_requests(mdsc);
17c688c3
SW		 4688
4689 /*
4690 * wait for reply handlers to drop their request refs and
4691 * their inode/dcache refs
4692 */
4693 ceph_msgr_flush();
0c44a8e0
LH		 4694
4695 ceph_cleanup_quotarealms_inodes(mdsc);
2f2dc053
SW		 4696}
4697
4698/*
4699 * wait for all write mds requests to flush.
4700 */
4701static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
4702{
80fc7314 4703 struct ceph_mds_request *req = NULL, *nextreq;
44ca18f2 4704 struct rb_node *n;
2f2dc053
SW		 4705
4706 mutex_lock(&mdsc->mutex);
4707 dout("wait_unsafe_requests want %lld\n", want_tid);
80fc7314 4708restart:
44ca18f2
SW		 4709 req = __get_oldest_req(mdsc);
4710 while (req && req->r_tid <= want_tid) {
80fc7314
SW		 4711 /* find next request */
4712 n = rb_next(&req->r_node);
4713 if (n)
4714 nextreq = rb_entry(n, struct ceph_mds_request, r_node);
4715 else
4716 nextreq = NULL;
e8a7b8b1
YZ		 4717 if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
4718 (req->r_op & CEPH_MDS_OP_WRITE)) {
44ca18f2
SW		 4719 /* write op */
4720 ceph_mdsc_get_request(req);
80fc7314
SW		 4721 if (nextreq)
4722 ceph_mdsc_get_request(nextreq);
44ca18f2
SW		 4723 mutex_unlock(&mdsc->mutex);
4724 dout("wait_unsafe_requests wait on %llu (want %llu)\n",
4725 req->r_tid, want_tid);
4726 wait_for_completion(&req->r_safe_completion);
4727 mutex_lock(&mdsc->mutex);
44ca18f2 4728 ceph_mdsc_put_request(req);
80fc7314
SW		 4729 if (!nextreq)
4730 break; /* next dne before, so we're done! */
4731 if (RB_EMPTY_NODE(&nextreq->r_node)) {
4732 /* next request was removed from tree */
4733 ceph_mdsc_put_request(nextreq);
4734 goto restart;
4735 }
4736 ceph_mdsc_put_request(nextreq); /* won't go away */
44ca18f2 4737 }
80fc7314 4738 req = nextreq;
2f2dc053
SW		 4739 }
4740 mutex_unlock(&mdsc->mutex);
4741 dout("wait_unsafe_requests done\n");
4742}
4743
4744void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
4745{
0e294387 4746 u64 want_tid, want_flush;
2f2dc053 4747
50c9132d 4748 if (READ_ONCE(mdsc->fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN)
56b7cf95
SW		 4749 return;
4750
2f2dc053
SW		 4751 dout("sync\n");
4752 mutex_lock(&mdsc->mutex);
4753 want_tid = mdsc->last_tid;
2f2dc053 4754 mutex_unlock(&mdsc->mutex);
2f2dc053 4755
afcdaea3 4756 ceph_flush_dirty_caps(mdsc);
d3383a8e 4757 spin_lock(&mdsc->cap_dirty_lock);
8310b089 4758 want_flush = mdsc->last_cap_flush_tid;
c8799fc4
YZ		 4759 if (!list_empty(&mdsc->cap_flush_list)) {
4760 struct ceph_cap_flush *cf =
4761 list_last_entry(&mdsc->cap_flush_list,
4762 struct ceph_cap_flush, g_list);
4763 cf->wake = true;
4764 }
d3383a8e
YZ		 4765 spin_unlock(&mdsc->cap_dirty_lock);
4766
0e294387
YZ		 4767 dout("sync want tid %lld flush_seq %lld\n",
4768 want_tid, want_flush);
2f2dc053
SW		 4769
4770 wait_unsafe_requests(mdsc, want_tid);
0e294387 4771 wait_caps_flush(mdsc, want_flush);
2f2dc053
SW		 4772}
4773
f3c60c59
SW		 4774/*
4775 * true if all sessions are closed, or we force unmount
4776 */
fcff415c 4777static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped)
f3c60c59 4778{
52953d55 4779 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
f3c60c59 4780 return true;
fcff415c 4781 return atomic_read(&mdsc->num_sessions) <= skipped;
f3c60c59 4782}
2f2dc053
SW		 4783
4784/*
4785 * called after sb is ro.
4786 */
4787void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
4788{
a319bf56 4789 struct ceph_options *opts = mdsc->fsc->client->options;
2f2dc053
SW		 4790 struct ceph_mds_session *session;
4791 int i;
fcff415c 4792 int skipped = 0;
2f2dc053
SW		 4793
4794 dout("close_sessions\n");
4795
2f2dc053 4796 /* close sessions */
f3c60c59
SW		 4797 mutex_lock(&mdsc->mutex);
4798 for (i = 0; i < mdsc->max_sessions; i++) {
4799 session = __ceph_lookup_mds_session(mdsc, i);
4800 if (!session)
4801 continue;
2f2dc053 4802 mutex_unlock(&mdsc->mutex);
f3c60c59 4803 mutex_lock(&session->s_mutex);
fcff415c
YZ		 4804 if (__close_session(mdsc, session) <= 0)
4805 skipped++;
f3c60c59
SW		 4806 mutex_unlock(&session->s_mutex);
4807 ceph_put_mds_session(session);
2f2dc053
SW		 4808 mutex_lock(&mdsc->mutex);
4809 }
f3c60c59
SW		 4810 mutex_unlock(&mdsc->mutex);
4811
4812 dout("waiting for sessions to close\n");
fcff415c
YZ		 4813 wait_event_timeout(mdsc->session_close_wq,
4814 done_closing_sessions(mdsc, skipped),
a319bf56 4815 ceph_timeout_jiffies(opts->mount_timeout));
2f2dc053
SW		 4816
4817 /* tear down remaining sessions */
f3c60c59 4818 mutex_lock(&mdsc->mutex);
2f2dc053
SW		 4819 for (i = 0; i < mdsc->max_sessions; i++) {
4820 if (mdsc->sessions[i]) {
5b3248c6 4821 session = ceph_get_mds_session(mdsc->sessions[i]);
2600d2dd 4822 __unregister_session(mdsc, session);
2f2dc053
SW		 4823 mutex_unlock(&mdsc->mutex);
4824 mutex_lock(&session->s_mutex);
4825 remove_session_caps(session);
4826 mutex_unlock(&session->s_mutex);
4827 ceph_put_mds_session(session);
4828 mutex_lock(&mdsc->mutex);
4829 }
4830 }
2f2dc053 4831 WARN_ON(!list_empty(&mdsc->cap_delay_list));
2f2dc053
SW		 4832 mutex_unlock(&mdsc->mutex);
4833
75c9627e 4834 ceph_cleanup_snapid_map(mdsc);
5ed91587 4835 ceph_cleanup_global_and_empty_realms(mdsc);
2f2dc053 4836
37c4efc1 4837 cancel_work_sync(&mdsc->cap_reclaim_work);
2f2dc053
SW		 4838 cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
4839
4840 dout("stopped\n");
4841}
4842
48fec5d0
YZ		 4843void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
4844{
4845 struct ceph_mds_session *session;
4846 int mds;
4847
4848 dout("force umount\n");
4849
4850 mutex_lock(&mdsc->mutex);
4851 for (mds = 0; mds < mdsc->max_sessions; mds++) {
4852 session = __ceph_lookup_mds_session(mdsc, mds);
4853 if (!session)
4854 continue;
d468e729
YZ		 4855
4856 if (session->s_state == CEPH_MDS_SESSION_REJECTED)
4857 __unregister_session(mdsc, session);
4858 __wake_requests(mdsc, &session->s_waiting);
48fec5d0 4859 mutex_unlock(&mdsc->mutex);
d468e729 4860
48fec5d0
YZ		 4861 mutex_lock(&session->s_mutex);
4862 __close_session(mdsc, session);
4863 if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
4864 cleanup_session_requests(mdsc, session);
4865 remove_session_caps(session);
4866 }
4867 mutex_unlock(&session->s_mutex);
4868 ceph_put_mds_session(session);
d468e729 4869
48fec5d0
YZ		 4870 mutex_lock(&mdsc->mutex);
4871 kick_requests(mdsc, mds);
4872 }
4873 __wake_requests(mdsc, &mdsc->waiting_for_map);
4874 mutex_unlock(&mdsc->mutex);
4875}
4876
3d14c5d2 4877static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
2f2dc053
SW		 4878{
4879 dout("stop\n");
fa996773
XL		 4880 /*
4881 * Make sure the delayed work stopped before releasing
4882 * the resources.
4883 *
4884 * Because the cancel_delayed_work_sync() will only
4885 * guarantee that the work finishes executing. But the
4886 * delayed work will re-arm itself again after that.
4887 */
4888 flush_delayed_work(&mdsc->delayed_work);
4889
2f2dc053
SW		 4890 if (mdsc->mdsmap)
4891 ceph_mdsmap_destroy(mdsc->mdsmap);
4892 kfree(mdsc->sessions);
37151668 4893 ceph_caps_finalize(mdsc);
10183a69 4894 ceph_pool_perm_destroy(mdsc);
2f2dc053
SW		 4895}
4896
3d14c5d2
YS		 4897void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
4898{
4899 struct ceph_mds_client *mdsc = fsc->mdsc;
ef550f6f 4900 dout("mdsc_destroy %p\n", mdsc);
ef550f6f 4901
50c55aec
CX		 4902 if (!mdsc)
4903 return;
4904
ef550f6f
SW		 4905 /* flush out any connection work with references to us */
4906 ceph_msgr_flush();
4907
62a65f36
YZ		 4908 ceph_mdsc_stop(mdsc);
4909
f9009efa
XL		 4910 ceph_metric_destroy(&mdsc->metric);
4911
3d14c5d2
YS		 4912 fsc->mdsc = NULL;
4913 kfree(mdsc);
ef550f6f 4914 dout("mdsc_destroy %p done\n", mdsc);
3d14c5d2
YS		 4915}
4916
430afbad
YZ		 4917void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
4918{
4919 struct ceph_fs_client *fsc = mdsc->fsc;
4920 const char *mds_namespace = fsc->mount_options->mds_namespace;
4921 void *p = msg->front.iov_base;
4922 void *end = p + msg->front.iov_len;
4923 u32 epoch;
430afbad
YZ		 4924 u32 num_fs;
4925 u32 mount_fscid = (u32)-1;
430afbad
YZ		 4926 int err = -EINVAL;
4927
4928 ceph_decode_need(&p, end, sizeof(u32), bad);
4929 epoch = ceph_decode_32(&p);
4930
4931 dout("handle_fsmap epoch %u\n", epoch);
4932
06a1ad43
JL		 4933 /* struct_v, struct_cv, map_len, epoch, legacy_client_fscid */
4934 ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 3, bad);
430afbad 4935
06a1ad43 4936 ceph_decode_32_safe(&p, end, num_fs, bad);
430afbad
YZ		 4937 while (num_fs-- > 0) {
4938 void *info_p, *info_end;
4939 u32 info_len;
430afbad
YZ		 4940 u32 fscid, namelen;
4941
4942 ceph_decode_need(&p, end, 2 + sizeof(u32), bad);
06a1ad43 4943 p += 2; // info_v, info_cv
430afbad
YZ		 4944 info_len = ceph_decode_32(&p);
4945 ceph_decode_need(&p, end, info_len, bad);
4946 info_p = p;
4947 info_end = p + info_len;
4948 p = info_end;
4949
4950 ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad);
4951 fscid = ceph_decode_32(&info_p);
4952 namelen = ceph_decode_32(&info_p);
4953 ceph_decode_need(&info_p, info_end, namelen, bad);
4954
4955 if (mds_namespace &&
4956 strlen(mds_namespace) == namelen &&
4957 !strncmp(mds_namespace, (char *)info_p, namelen)) {
4958 mount_fscid = fscid;
4959 break;
4960 }
4961 }
4962
4963 ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch);
4964 if (mount_fscid != (u32)-1) {
4965 fsc->client->monc.fs_cluster_id = mount_fscid;
4966 ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP,
4967 0, true);
4968 ceph_monc_renew_subs(&fsc->client->monc);
4969 } else {
4970 err = -ENOENT;
4971 goto err_out;
4972 }
4973 return;
76bd6ec4 4974
430afbad 4975bad:
631ed4b0
JL		 4976 pr_err("error decoding fsmap %d. Shutting down mount.\n", err);
4977 ceph_umount_begin(mdsc->fsc->sb);
430afbad
YZ		 4978err_out:
4979 mutex_lock(&mdsc->mutex);
76bd6ec4 4980 mdsc->mdsmap_err = err;
430afbad
YZ		 4981 __wake_requests(mdsc, &mdsc->waiting_for_map);
4982 mutex_unlock(&mdsc->mutex);
430afbad 4983}
2f2dc053
SW		 4984
4985/*
4986 * handle mds map update.
4987 */
430afbad 4988void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
2f2dc053
SW		 4989{
4990 u32 epoch;
4991 u32 maplen;
4992 void *p = msg->front.iov_base;
4993 void *end = p + msg->front.iov_len;
4994 struct ceph_mdsmap *newmap, *oldmap;
4995 struct ceph_fsid fsid;
4996 int err = -EINVAL;
4997
4998 ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
4999 ceph_decode_copy(&p, &fsid, sizeof(fsid));
3d14c5d2 5000 if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
0743304d 5001 return;
c89136ea
SW		 5002 epoch = ceph_decode_32(&p);
5003 maplen = ceph_decode_32(&p);
2f2dc053
SW		 5004 dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
5005
5006 /* do we need it? */
2f2dc053
SW		 5007 mutex_lock(&mdsc->mutex);
5008 if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
5009 dout("handle_map epoch %u <= our %u\n",
5010 epoch, mdsc->mdsmap->m_epoch);
5011 mutex_unlock(&mdsc->mutex);
5012 return;
5013 }
5014
cd1a677c 5015 newmap = ceph_mdsmap_decode(&p, end, ceph_msgr2(mdsc->fsc->client));
2f2dc053
SW		 5016 if (IS_ERR(newmap)) {
5017 err = PTR_ERR(newmap);
5018 goto bad_unlock;
5019 }
5020
5021 /* swap into place */
5022 if (mdsc->mdsmap) {
5023 oldmap = mdsc->mdsmap;
5024 mdsc->mdsmap = newmap;
5025 check_new_map(mdsc, newmap, oldmap);
5026 ceph_mdsmap_destroy(oldmap);
5027 } else {
5028 mdsc->mdsmap = newmap; /* first mds map */
5029 }
719784ba
CX		 5030 mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size,
5031 MAX_LFS_FILESIZE);
2f2dc053
SW		 5032
5033 __wake_requests(mdsc, &mdsc->waiting_for_map);
82dcabad
ID		 5034 ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP,
5035 mdsc->mdsmap->m_epoch);
2f2dc053
SW		 5036
5037 mutex_unlock(&mdsc->mutex);
bf2ba432 5038 schedule_delayed(mdsc, 0);
2f2dc053
SW		 5039 return;
5040
5041bad_unlock:
5042 mutex_unlock(&mdsc->mutex);
5043bad:
631ed4b0
JL		 5044 pr_err("error decoding mdsmap %d. Shutting down mount.\n", err);
5045 ceph_umount_begin(mdsc->fsc->sb);
2f2dc053
SW		 5046 return;
5047}
5048
4972cf60 5049static struct ceph_connection *mds_get_con(struct ceph_connection *con)
2f2dc053
SW		 5050{
5051 struct ceph_mds_session *s = con->private;
5052
5b3248c6 5053 if (ceph_get_mds_session(s))
2f2dc053 5054 return con;
2f2dc053
SW		 5055 return NULL;
5056}
5057
4972cf60 5058static void mds_put_con(struct ceph_connection *con)
2f2dc053
SW		 5059{
5060 struct ceph_mds_session *s = con->private;
5061
2f2dc053
SW		 5062 ceph_put_mds_session(s);
5063}
5064
5065/*
5066 * if the client is unresponsive for long enough, the mds will kill
5067 * the session entirely.
5068 */
4972cf60 5069static void mds_peer_reset(struct ceph_connection *con)
2f2dc053
SW		 5070{
5071 struct ceph_mds_session *s = con->private;
7e70f0ed 5072 struct ceph_mds_client *mdsc = s->s_mdsc;
2f2dc053 5073
f3ae1b97 5074 pr_warn("mds%d closed our session\n", s->s_mds);
7e70f0ed 5075 send_mds_reconnect(mdsc, s);
2f2dc053
SW		 5076}
5077
4972cf60 5078static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
2f2dc053
SW		 5079{
5080 struct ceph_mds_session *s = con->private;
5081 struct ceph_mds_client *mdsc = s->s_mdsc;
5082 int type = le16_to_cpu(msg->hdr.type);
5083
2600d2dd
SW		 5084 mutex_lock(&mdsc->mutex);
5085 if (__verify_registered_session(mdsc, s) < 0) {
5086 mutex_unlock(&mdsc->mutex);
5087 goto out;
5088 }
5089 mutex_unlock(&mdsc->mutex);
5090
2f2dc053
SW		 5091 switch (type) {
5092 case CEPH_MSG_MDS_MAP:
430afbad
YZ		 5093 ceph_mdsc_handle_mdsmap(mdsc, msg);
5094 break;
5095 case CEPH_MSG_FS_MAP_USER:
5096 ceph_mdsc_handle_fsmap(mdsc, msg);
2f2dc053
SW		 5097 break;
5098 case CEPH_MSG_CLIENT_SESSION:
5099 handle_session(s, msg);
5100 break;
5101 case CEPH_MSG_CLIENT_REPLY:
5102 handle_reply(s, msg);
5103 break;
5104 case CEPH_MSG_CLIENT_REQUEST_FORWARD:
2600d2dd 5105 handle_forward(mdsc, s, msg);
2f2dc053
SW		 5106 break;
5107 case CEPH_MSG_CLIENT_CAPS:
5108 ceph_handle_caps(s, msg);
5109 break;
5110 case CEPH_MSG_CLIENT_SNAP:
2600d2dd 5111 ceph_handle_snap(mdsc, s, msg);
2f2dc053
SW		 5112 break;
5113 case CEPH_MSG_CLIENT_LEASE:
2600d2dd 5114 handle_lease(mdsc, s, msg);
2f2dc053 5115 break;
fb18a575
LH		 5116 case CEPH_MSG_CLIENT_QUOTA:
5117 ceph_handle_quota(mdsc, s, msg);
5118 break;
2f2dc053
SW		 5119
5120 default:
5121 pr_err("received unknown message type %d %s\n", type,
5122 ceph_msg_type_name(type));
5123 }
2600d2dd 5124out:
2f2dc053
SW		 5125 ceph_msg_put(msg);
5126}
5127
4e7a5dcd
SW		 5128/*
5129 * authentication
5130 */
a3530df3
AE		 5131
5132/*
5133 * Note: returned pointer is the address of a structure that's
5134 * managed separately. Caller must *not* attempt to free it.
5135 */
4972cf60
ID		 5136static struct ceph_auth_handshake *
5137mds_get_authorizer(struct ceph_connection *con, int *proto, int force_new)
4e7a5dcd
SW		 5138{
5139 struct ceph_mds_session *s = con->private;
5140 struct ceph_mds_client *mdsc = s->s_mdsc;
3d14c5d2 5141 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
74f1869f 5142 struct ceph_auth_handshake *auth = &s->s_auth;
ce287162 5143 int ret;
4e7a5dcd 5144
ce287162
ID		 5145 ret = __ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
5146 force_new, proto, NULL, NULL);
5147 if (ret)
5148 return ERR_PTR(ret);
74f1869f 5149
a3530df3 5150 return auth;
4e7a5dcd
SW		 5151}
5152
4972cf60 5153static int mds_add_authorizer_challenge(struct ceph_connection *con,
6daca13d
ID		 5154 void *challenge_buf, int challenge_buf_len)
5155{
5156 struct ceph_mds_session *s = con->private;
5157 struct ceph_mds_client *mdsc = s->s_mdsc;
5158 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
5159
5160 return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer,
5161 challenge_buf, challenge_buf_len);
5162}
4e7a5dcd 5163
4972cf60 5164static int mds_verify_authorizer_reply(struct ceph_connection *con)
4e7a5dcd
SW		 5165{
5166 struct ceph_mds_session *s = con->private;
5167 struct ceph_mds_client *mdsc = s->s_mdsc;
3d14c5d2 5168 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
285ea34f 5169 struct ceph_auth_handshake *auth = &s->s_auth;
4e7a5dcd 5170
285ea34f
ID		 5171 return ceph_auth_verify_authorizer_reply(ac, auth->authorizer,
5172 auth->authorizer_reply_buf, auth->authorizer_reply_buf_len,
5173 NULL, NULL, NULL, NULL);
4e7a5dcd
SW		 5174}
5175
4972cf60 5176static int mds_invalidate_authorizer(struct ceph_connection *con)
9bd2e6f8
SW		 5177{
5178 struct ceph_mds_session *s = con->private;
5179 struct ceph_mds_client *mdsc = s->s_mdsc;
3d14c5d2 5180 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
9bd2e6f8 5181
27859f97 5182 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
9bd2e6f8 5183
3d14c5d2 5184 return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
9bd2e6f8
SW		 5185}
5186
cd1a677c
ID		 5187static int mds_get_auth_request(struct ceph_connection *con,
5188 void *buf, int *buf_len,
5189 void **authorizer, int *authorizer_len)
5190{
5191 struct ceph_mds_session *s = con->private;
5192 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5193 struct ceph_auth_handshake *auth = &s->s_auth;
5194 int ret;
5195
5196 ret = ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
5197 buf, buf_len);
5198 if (ret)
5199 return ret;
5200
5201 *authorizer = auth->authorizer_buf;
5202 *authorizer_len = auth->authorizer_buf_len;
5203 return 0;
5204}
5205
5206static int mds_handle_auth_reply_more(struct ceph_connection *con,
5207 void *reply, int reply_len,
5208 void *buf, int *buf_len,
5209 void **authorizer, int *authorizer_len)
5210{
5211 struct ceph_mds_session *s = con->private;
5212 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5213 struct ceph_auth_handshake *auth = &s->s_auth;
5214 int ret;
5215
5216 ret = ceph_auth_handle_svc_reply_more(ac, auth, reply, reply_len,
5217 buf, buf_len);
5218 if (ret)
5219 return ret;
5220
5221 *authorizer = auth->authorizer_buf;
5222 *authorizer_len = auth->authorizer_buf_len;
5223 return 0;
5224}
5225
5226static int mds_handle_auth_done(struct ceph_connection *con,
5227 u64 global_id, void *reply, int reply_len,
5228 u8 *session_key, int *session_key_len,
5229 u8 *con_secret, int *con_secret_len)
5230{
5231 struct ceph_mds_session *s = con->private;
5232 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5233 struct ceph_auth_handshake *auth = &s->s_auth;
5234
5235 return ceph_auth_handle_svc_reply_done(ac, auth, reply, reply_len,
5236 session_key, session_key_len,
5237 con_secret, con_secret_len);
5238}
5239
5240static int mds_handle_auth_bad_method(struct ceph_connection *con,
5241 int used_proto, int result,
5242 const int *allowed_protos, int proto_cnt,
5243 const int *allowed_modes, int mode_cnt)
5244{
5245 struct ceph_mds_session *s = con->private;
5246 struct ceph_mon_client *monc = &s->s_mdsc->fsc->client->monc;
5247 int ret;
5248
5249 if (ceph_auth_handle_bad_authorizer(monc->auth, CEPH_ENTITY_TYPE_MDS,
5250 used_proto, result,
5251 allowed_protos, proto_cnt,
5252 allowed_modes, mode_cnt)) {
5253 ret = ceph_monc_validate_auth(monc);
5254 if (ret)
5255 return ret;
5256 }
5257
5258 return -EACCES;
5259}
5260
53ded495
AE		 5261static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
5262 struct ceph_msg_header *hdr, int *skip)
5263{
5264 struct ceph_msg *msg;
5265 int type = (int) le16_to_cpu(hdr->type);
5266 int front_len = (int) le32_to_cpu(hdr->front_len);
5267
5268 if (con->in_msg)
5269 return con->in_msg;
5270
5271 *skip = 0;
5272 msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
5273 if (!msg) {
5274 pr_err("unable to allocate msg type %d len %d\n",
5275 type, front_len);
5276 return NULL;
5277 }
53ded495
AE		 5278
5279 return msg;
5280}
5281
79dbd1ba 5282static int mds_sign_message(struct ceph_msg *msg)
33d07337 5283{
79dbd1ba 5284 struct ceph_mds_session *s = msg->con->private;
33d07337 5285 struct ceph_auth_handshake *auth = &s->s_auth;
79dbd1ba 5286
33d07337
YZ		 5287 return ceph_auth_sign_message(auth, msg);
5288}
5289
79dbd1ba 5290static int mds_check_message_signature(struct ceph_msg *msg)
33d07337 5291{
79dbd1ba 5292 struct ceph_mds_session *s = msg->con->private;
33d07337 5293 struct ceph_auth_handshake *auth = &s->s_auth;
79dbd1ba 5294
33d07337
YZ		 5295 return ceph_auth_check_message_signature(auth, msg);
5296}
5297
9e32789f 5298static const struct ceph_connection_operations mds_con_ops = {
4972cf60
ID		 5299 .get = mds_get_con,
5300 .put = mds_put_con,
53ded495 5301 .alloc_msg = mds_alloc_msg,
4972cf60
ID		 5302 .dispatch = mds_dispatch,
5303 .peer_reset = mds_peer_reset,
5304 .get_authorizer = mds_get_authorizer,
5305 .add_authorizer_challenge = mds_add_authorizer_challenge,
5306 .verify_authorizer_reply = mds_verify_authorizer_reply,
5307 .invalidate_authorizer = mds_invalidate_authorizer,
79dbd1ba
ID		 5308 .sign_message = mds_sign_message,
5309 .check_message_signature = mds_check_message_signature,
cd1a677c
ID		 5310 .get_auth_request = mds_get_auth_request,
5311 .handle_auth_reply_more = mds_handle_auth_reply_more,
5312 .handle_auth_done = mds_handle_auth_done,
5313 .handle_auth_bad_method = mds_handle_auth_bad_method,
2f2dc053
SW		 5314};
5315
2f2dc053 5316/* eof */
Linux 6.x block layer and io_uring tree(s)