projects/linux·blob master · refs

2289 lines · 60.0 KiB

1// SPDX-License-Identifier: GPL-2.0
2#include <linux/ceph/ceph_debug.h>
3
4#include <linux/spinlock.h>
5#include <linux/namei.h>
6#include <linux/slab.h>
7#include <linux/sched.h>
8#include <linux/xattr.h>
9
10#include "super.h"
11#include "mds_client.h"
12#include "crypto.h"
13
14/*
15 * Directory operations: readdir, lookup, create, link, unlink,
16 * rename, etc.
17 */
18
19/*
20 * Ceph MDS operations are specified in terms of a base ino and
21 * relative path. Thus, the client can specify an operation on a
22 * specific inode (e.g., a getattr due to fstat(2)), or as a path
23 * relative to, say, the root directory.
24 *
25 * Normally, we limit ourselves to strict inode ops (no path component)
26 * or dentry operations (a single path component relative to an ino). The
27 * exception to this is open_root_dentry(), which will open the mount
28 * point by name.
29 */
30
31const struct dentry_operations ceph_dentry_ops;
32
33static bool __dentry_lease_is_valid(struct ceph_dentry_info *di);
34static int __dir_lease_try_check(const struct dentry *dentry);
35
36/*
37 * Initialize ceph dentry state.
38 */
39static int ceph_d_init(struct dentry *dentry)
40{
41 struct ceph_dentry_info *di;
42 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dentry->d_sb);
43
44 di = kmem_cache_zalloc(ceph_dentry_cachep, GFP_KERNEL);
45 if (!di)
46 return -ENOMEM; /* oh well */
47
48 di->dentry = dentry;
49 di->lease_session = NULL;
50 di->time = jiffies;
51 dentry->d_fsdata = di;
52 INIT_LIST_HEAD(&di->lease_list);
53
54 atomic64_inc(&mdsc->metric.total_dentries);
55
56 return 0;
57}
58
59/*
60 * for f_pos for readdir:
61 * - hash order:
62 * (0xff << 52) | ((24 bits hash) << 28) |
63 * (the nth entry has hash collision);
64 * - frag+name order;
65 * ((frag value) << 28) | (the nth entry in frag);
66 */
67#define OFFSET_BITS 28
68#define OFFSET_MASK ((1 << OFFSET_BITS) - 1)
69#define HASH_ORDER (0xffull << (OFFSET_BITS + 24))
70loff_t ceph_make_fpos(unsigned high, unsigned off, bool hash_order)
71{
72 loff_t fpos = ((loff_t)high << 28) | (loff_t)off;
73 if (hash_order)
74 fpos |= HASH_ORDER;
75 return fpos;
76}
77
78static bool is_hash_order(loff_t p)
79{
80 return (p & HASH_ORDER) == HASH_ORDER;
81}
82
83static unsigned fpos_frag(loff_t p)
84{
85 return p >> OFFSET_BITS;
86}
87
88static unsigned fpos_hash(loff_t p)
89{
90 return ceph_frag_value(fpos_frag(p));
91}
92
93static unsigned fpos_off(loff_t p)
94{
95 return p & OFFSET_MASK;
96}
97
98static int fpos_cmp(loff_t l, loff_t r)
99{
100 int v = ceph_frag_compare(fpos_frag(l), fpos_frag(r));
101 if (v)
102 return v;
103 return (int)(fpos_off(l) - fpos_off(r));
104}
105
106/*
107 * make note of the last dentry we read, so we can
108 * continue at the same lexicographical point,
109 * regardless of what dir changes take place on the
110 * server.
111 */
112static int note_last_dentry(struct ceph_fs_client *fsc,
113 struct ceph_dir_file_info *dfi,
114 const char *name,
115 int len, unsigned next_offset)
116{
117 char *buf = kmalloc(len+1, GFP_KERNEL);
118 if (!buf)
119 return -ENOMEM;
120 kfree(dfi->last_name);
121 dfi->last_name = buf;
122 memcpy(dfi->last_name, name, len);
123 dfi->last_name[len] = 0;
124 dfi->next_offset = next_offset;
125 doutc(fsc->client, "'%s'\n", dfi->last_name);
126 return 0;
127}
128
129
130static struct dentry *
131__dcache_find_get_entry(struct dentry *parent, u64 idx,
132 struct ceph_readdir_cache_control *cache_ctl)
133{
134 struct inode *dir = d_inode(parent);
135 struct ceph_client *cl = ceph_inode_to_client(dir);
136 struct dentry *dentry;
137 unsigned idx_mask = (PAGE_SIZE / sizeof(struct dentry *)) - 1;
138 loff_t ptr_pos = idx * sizeof(struct dentry *);
139 pgoff_t ptr_pgoff = ptr_pos >> PAGE_SHIFT;
140
141 if (ptr_pos >= i_size_read(dir))
142 return NULL;
143
144 if (!cache_ctl->folio || ptr_pgoff != cache_ctl->folio->index) {
145 ceph_readdir_cache_release(cache_ctl);
146 cache_ctl->folio = filemap_lock_folio(&dir->i_data, ptr_pgoff);
147 if (IS_ERR(cache_ctl->folio)) {
148 cache_ctl->folio = NULL;
149 doutc(cl, " folio %lu not found\n", ptr_pgoff);
150 return ERR_PTR(-EAGAIN);
151 }
152 /* reading/filling the cache are serialized by
153 i_rwsem, no need to use folio lock */
154 folio_unlock(cache_ctl->folio);
155 cache_ctl->dentries = kmap_local_folio(cache_ctl->folio, 0);
156 }
157
158 cache_ctl->index = idx & idx_mask;
159
160 rcu_read_lock();
161 spin_lock(&parent->d_lock);
162 /* check i_size again here, because empty directory can be
163 * marked as complete while not holding the i_rwsem. */
164 if (ceph_dir_is_complete_ordered(dir) && ptr_pos < i_size_read(dir))
165 dentry = cache_ctl->dentries[cache_ctl->index];
166 else
167 dentry = NULL;
168 spin_unlock(&parent->d_lock);
169 if (dentry && !lockref_get_not_dead(&dentry->d_lockref))
170 dentry = NULL;
171 rcu_read_unlock();
172 return dentry ? : ERR_PTR(-EAGAIN);
173}
174
175/*
176 * When possible, we try to satisfy a readdir by peeking at the
177 * dcache. We make this work by carefully ordering dentries on
178 * d_children when we initially get results back from the MDS, and
179 * falling back to a "normal" sync readdir if any dentries in the dir
180 * are dropped.
181 *
182 * Complete dir indicates that we have all dentries in the dir. It is
183 * defined IFF we hold CEPH_CAP_FILE_SHARED (which will be revoked by
184 * the MDS if/when the directory is modified).
185 */
186static int __dcache_readdir(struct file *file, struct dir_context *ctx,
187 int shared_gen)
188{
189 struct ceph_dir_file_info *dfi = file->private_data;
190 struct dentry *parent = file->f_path.dentry;
191 struct inode *dir = d_inode(parent);
192 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(dir);
193 struct ceph_client *cl = ceph_inode_to_client(dir);
194 struct dentry *dentry, *last = NULL;
195 struct ceph_dentry_info *di;
196 struct ceph_readdir_cache_control cache_ctl = {};
197 u64 idx = 0;
198 int err = 0;
199
200 doutc(cl, "%p %llx.%llx v%u at %llx\n", dir, ceph_vinop(dir),
201 (unsigned)shared_gen, ctx->pos);
202
203 /* search start position */
204 if (ctx->pos > 2) {
205 u64 count = div_u64(i_size_read(dir), sizeof(struct dentry *));
206 while (count > 0) {
207 u64 step = count >> 1;
208 dentry = __dcache_find_get_entry(parent, idx + step,
209 &cache_ctl);
210 if (!dentry) {
211 /* use linear search */
212 idx = 0;
213 break;
214 }
215 if (IS_ERR(dentry)) {
216 err = PTR_ERR(dentry);
217 goto out;
218 }
219 di = ceph_dentry(dentry);
220 spin_lock(&dentry->d_lock);
221 if (fpos_cmp(di->offset, ctx->pos) < 0) {
222 idx += step + 1;
223 count -= step + 1;
224 } else {
225 count = step;
226 }
227 spin_unlock(&dentry->d_lock);
228 dput(dentry);
229 }
230
231 doutc(cl, "%p %llx.%llx cache idx %llu\n", dir,
232 ceph_vinop(dir), idx);
233 }
234
235
236 for (;;) {
237 bool emit_dentry = false;
238 dentry = __dcache_find_get_entry(parent, idx++, &cache_ctl);
239 if (!dentry) {
240 dfi->file_info.flags |= CEPH_F_ATEND;
241 err = 0;
242 break;
243 }
244 if (IS_ERR(dentry)) {
245 err = PTR_ERR(dentry);
246 goto out;
247 }
248
249 spin_lock(&dentry->d_lock);
250 di = ceph_dentry(dentry);
251 if (d_unhashed(dentry) ||
252 d_really_is_negative(dentry) ||
253 di->lease_shared_gen != shared_gen ||
254 ((dentry->d_flags & DCACHE_NOKEY_NAME) &&
255 fscrypt_has_encryption_key(dir))) {
256 spin_unlock(&dentry->d_lock);
257 dput(dentry);
258 err = -EAGAIN;
259 goto out;
260 }
261 if (fpos_cmp(ctx->pos, di->offset) <= 0) {
262 __ceph_dentry_dir_lease_touch(di);
263 emit_dentry = true;
264 }
265 spin_unlock(&dentry->d_lock);
266
267 if (emit_dentry) {
268 doutc(cl, " %llx dentry %p %pd %p\n", di->offset,
269 dentry, dentry, d_inode(dentry));
270 ctx->pos = di->offset;
271 if (!dir_emit(ctx, dentry->d_name.name,
272 dentry->d_name.len, ceph_present_inode(d_inode(dentry)),
273 d_inode(dentry)->i_mode >> 12)) {
274 dput(dentry);
275 err = 0;
276 break;
277 }
278 ctx->pos++;
279
280 if (last)
281 dput(last);
282 last = dentry;
283 } else {
284 dput(dentry);
285 }
286 }
287out:
288 ceph_readdir_cache_release(&cache_ctl);
289 if (last) {
290 int ret;
291 di = ceph_dentry(last);
292 ret = note_last_dentry(fsc, dfi, last->d_name.name,
293 last->d_name.len,
294 fpos_off(di->offset) + 1);
295 if (ret < 0)
296 err = ret;
297 dput(last);
298 /* last_name no longer match cache index */
299 if (dfi->readdir_cache_idx >= 0) {
300 dfi->readdir_cache_idx = -1;
301 dfi->dir_release_count = 0;
302 }
303 }
304 return err;
305}
306
307static bool need_send_readdir(struct ceph_dir_file_info *dfi, loff_t pos)
308{
309 if (!dfi->last_readdir)
310 return true;
311 if (is_hash_order(pos))
312 return !ceph_frag_contains_value(dfi->frag, fpos_hash(pos));
313 else
314 return dfi->frag != fpos_frag(pos);
315}
316
317static int ceph_readdir(struct file *file, struct dir_context *ctx)
318{
319 struct ceph_dir_file_info *dfi = file->private_data;
320 struct inode *inode = file_inode(file);
321 struct ceph_inode_info *ci = ceph_inode(inode);
322 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
323 struct ceph_mds_client *mdsc = fsc->mdsc;
324 struct ceph_client *cl = fsc->client;
325 int i;
326 int err;
327 unsigned frag = -1;
328 struct ceph_mds_reply_info_parsed *rinfo;
329
330 doutc(cl, "%p %llx.%llx file %p pos %llx\n", inode,
331 ceph_vinop(inode), file, ctx->pos);
332 if (dfi->file_info.flags & CEPH_F_ATEND)
333 return 0;
334
335 /* always start with . and .. */
336 if (ctx->pos == 0) {
337 doutc(cl, "%p %llx.%llx off 0 -> '.'\n", inode,
338 ceph_vinop(inode));
339 if (!dir_emit(ctx, ".", 1, ceph_present_inode(inode),
340 inode->i_mode >> 12))
341 return 0;
342 ctx->pos = 1;
343 }
344 if (ctx->pos == 1) {
345 u64 ino;
346 struct dentry *dentry = file->f_path.dentry;
347
348 spin_lock(&dentry->d_lock);
349 ino = ceph_present_inode(dentry->d_parent->d_inode);
350 spin_unlock(&dentry->d_lock);
351
352 doutc(cl, "%p %llx.%llx off 1 -> '..'\n", inode,
353 ceph_vinop(inode));
354 if (!dir_emit(ctx, "..", 2, ino, inode->i_mode >> 12))
355 return 0;
356 ctx->pos = 2;
357 }
358
359 err = ceph_fscrypt_prepare_readdir(inode);
360 if (err < 0)
361 return err;
362
363 spin_lock(&ci->i_ceph_lock);
364 /* request Fx cap. if have Fx, we don't need to release Fs cap
365 * for later create/unlink. */
366 __ceph_touch_fmode(ci, mdsc, CEPH_FILE_MODE_WR);
367 /* can we use the dcache? */
368 if (ceph_test_mount_opt(fsc, DCACHE) &&
369 !ceph_test_mount_opt(fsc, NOASYNCREADDIR) &&
370 ceph_snap(inode) != CEPH_SNAPDIR &&
371 __ceph_dir_is_complete_ordered(ci) &&
372 __ceph_caps_issued_mask_metric(ci, CEPH_CAP_FILE_SHARED, 1)) {
373 int shared_gen = atomic_read(&ci->i_shared_gen);
374
375 spin_unlock(&ci->i_ceph_lock);
376 err = __dcache_readdir(file, ctx, shared_gen);
377 if (err != -EAGAIN)
378 return err;
379 } else {
380 spin_unlock(&ci->i_ceph_lock);
381 }
382
383 /* proceed with a normal readdir */
384more:
385 /* do we have the correct frag content buffered? */
386 if (need_send_readdir(dfi, ctx->pos)) {
387 struct ceph_mds_request *req;
388 int op = ceph_snap(inode) == CEPH_SNAPDIR ?
389 CEPH_MDS_OP_LSSNAP : CEPH_MDS_OP_READDIR;
390
391 /* discard old result, if any */
392 if (dfi->last_readdir) {
393 ceph_mdsc_put_request(dfi->last_readdir);
394 dfi->last_readdir = NULL;
395 }
396
397 if (is_hash_order(ctx->pos)) {
398 /* fragtree isn't always accurate. choose frag
399 * based on previous reply when possible. */
400 if (frag == (unsigned)-1)
401 frag = ceph_choose_frag(ci, fpos_hash(ctx->pos),
402 NULL, NULL);
403 } else {
404 frag = fpos_frag(ctx->pos);
405 }
406
407 doutc(cl, "fetching %p %llx.%llx frag %x offset '%s'\n",
408 inode, ceph_vinop(inode), frag, dfi->last_name);
409 req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS);
410 if (IS_ERR(req))
411 return PTR_ERR(req);
412
413 err = ceph_alloc_readdir_reply_buffer(req, inode);
414 if (err) {
415 ceph_mdsc_put_request(req);
416 return err;
417 }
418 /* hints to request -> mds selection code */
419 req->r_direct_mode = USE_AUTH_MDS;
420 if (op == CEPH_MDS_OP_READDIR) {
421 req->r_direct_hash = ceph_frag_value(frag);
422 __set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
423 req->r_inode_drop = CEPH_CAP_FILE_EXCL;
424 }
425 if (dfi->last_name) {
426 int len = strlen(dfi->last_name);
427
428 req->r_path2 = kzalloc(NAME_MAX + 1, GFP_KERNEL);
429 if (!req->r_path2) {
430 ceph_mdsc_put_request(req);
431 return -ENOMEM;
432 }
433 memcpy(req->r_path2, dfi->last_name, len);
434
435 err = ceph_encode_encrypted_dname(inode, req->r_path2, len);
436 if (err < 0) {
437 ceph_mdsc_put_request(req);
438 return err;
439 }
440 } else if (is_hash_order(ctx->pos)) {
441 req->r_args.readdir.offset_hash =
442 cpu_to_le32(fpos_hash(ctx->pos));
443 }
444
445 req->r_dir_release_cnt = dfi->dir_release_count;
446 req->r_dir_ordered_cnt = dfi->dir_ordered_count;
447 req->r_readdir_cache_idx = dfi->readdir_cache_idx;
448 req->r_readdir_offset = dfi->next_offset;
449 req->r_args.readdir.frag = cpu_to_le32(frag);
450 req->r_args.readdir.flags =
451 cpu_to_le16(CEPH_READDIR_REPLY_BITFLAGS);
452
453 req->r_inode = inode;
454 ihold(inode);
455 req->r_dentry = dget(file->f_path.dentry);
456 err = ceph_mdsc_do_request(mdsc, NULL, req);
457 if (err < 0) {
458 ceph_mdsc_put_request(req);
459 return err;
460 }
461 doutc(cl, "%p %llx.%llx got and parsed readdir result=%d"
462 "on frag %x, end=%d, complete=%d, hash_order=%d\n",
463 inode, ceph_vinop(inode), err, frag,
464 (int)req->r_reply_info.dir_end,
465 (int)req->r_reply_info.dir_complete,
466 (int)req->r_reply_info.hash_order);
467
468 rinfo = &req->r_reply_info;
469 if (le32_to_cpu(rinfo->dir_dir->frag) != frag) {
470 frag = le32_to_cpu(rinfo->dir_dir->frag);
471 if (!rinfo->hash_order) {
472 dfi->next_offset = req->r_readdir_offset;
473 /* adjust ctx->pos to beginning of frag */
474 ctx->pos = ceph_make_fpos(frag,
475 dfi->next_offset,
476 false);
477 }
478 }
479
480 dfi->frag = frag;
481 dfi->last_readdir = req;
482
483 if (test_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags)) {
484 dfi->readdir_cache_idx = req->r_readdir_cache_idx;
485 if (dfi->readdir_cache_idx < 0) {
486 /* preclude from marking dir ordered */
487 dfi->dir_ordered_count = 0;
488 } else if (ceph_frag_is_leftmost(frag) &&
489 dfi->next_offset == 2) {
490 /* note dir version at start of readdir so
491 * we can tell if any dentries get dropped */
492 dfi->dir_release_count = req->r_dir_release_cnt;
493 dfi->dir_ordered_count = req->r_dir_ordered_cnt;
494 }
495 } else {
496 doutc(cl, "%p %llx.%llx !did_prepopulate\n", inode,
497 ceph_vinop(inode));
498 /* disable readdir cache */
499 dfi->readdir_cache_idx = -1;
500 /* preclude from marking dir complete */
501 dfi->dir_release_count = 0;
502 }
503
504 /* note next offset and last dentry name */
505 if (rinfo->dir_nr > 0) {
506 struct ceph_mds_reply_dir_entry *rde =
507 rinfo->dir_entries + (rinfo->dir_nr-1);
508 unsigned next_offset = req->r_reply_info.dir_end ?
509 2 : (fpos_off(rde->offset) + 1);
510 err = note_last_dentry(fsc, dfi, rde->name,
511 rde->name_len, next_offset);
512 if (err) {
513 ceph_mdsc_put_request(dfi->last_readdir);
514 dfi->last_readdir = NULL;
515 return err;
516 }
517 } else if (req->r_reply_info.dir_end) {
518 dfi->next_offset = 2;
519 /* keep last name */
520 }
521 }
522
523 rinfo = &dfi->last_readdir->r_reply_info;
524 doutc(cl, "%p %llx.%llx frag %x num %d pos %llx chunk first %llx\n",
525 inode, ceph_vinop(inode), dfi->frag, rinfo->dir_nr, ctx->pos,
526 rinfo->dir_nr ? rinfo->dir_entries[0].offset : 0LL);
527
528 i = 0;
529 /* search start position */
530 if (rinfo->dir_nr > 0) {
531 int step, nr = rinfo->dir_nr;
532 while (nr > 0) {
533 step = nr >> 1;
534 if (rinfo->dir_entries[i + step].offset < ctx->pos) {
535 i += step + 1;
536 nr -= step + 1;
537 } else {
538 nr = step;
539 }
540 }
541 }
542 for (; i < rinfo->dir_nr; i++) {
543 struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
544
545 if (rde->offset < ctx->pos) {
546 pr_warn_client(cl,
547 "%p %llx.%llx rde->offset 0x%llx ctx->pos 0x%llx\n",
548 inode, ceph_vinop(inode), rde->offset, ctx->pos);
549 ceph_mdsc_put_request(dfi->last_readdir);
550 dfi->last_readdir = NULL;
551 return -EIO;
552 }
553
554 if (WARN_ON_ONCE(!rde->inode.in)) {
555 ceph_mdsc_put_request(dfi->last_readdir);
556 dfi->last_readdir = NULL;
557 return -EIO;
558 }
559
560 ctx->pos = rde->offset;
561 doutc(cl, "%p %llx.%llx (%d/%d) -> %llx '%.*s' %p\n", inode,
562 ceph_vinop(inode), i, rinfo->dir_nr, ctx->pos,
563 rde->name_len, rde->name, &rde->inode.in);
564
565 if (!dir_emit(ctx, rde->name, rde->name_len,
566 ceph_present_ino(inode->i_sb, le64_to_cpu(rde->inode.in->ino)),
567 le32_to_cpu(rde->inode.in->mode) >> 12)) {
568 /*
569 * NOTE: Here no need to put the 'dfi->last_readdir',
570 * because when dir_emit stops us it's most likely
571 * doesn't have enough memory, etc. So for next readdir
572 * it will continue.
573 */
574 doutc(cl, "filldir stopping us...\n");
575 return 0;
576 }
577
578 /* Reset the lengths to their original allocated vals */
579 ctx->pos++;
580 }
581
582 ceph_mdsc_put_request(dfi->last_readdir);
583 dfi->last_readdir = NULL;
584
585 if (dfi->next_offset > 2) {
586 frag = dfi->frag;
587 goto more;
588 }
589
590 /* more frags? */
591 if (!ceph_frag_is_rightmost(dfi->frag)) {
592 frag = ceph_frag_next(dfi->frag);
593 if (is_hash_order(ctx->pos)) {
594 loff_t new_pos = ceph_make_fpos(ceph_frag_value(frag),
595 dfi->next_offset, true);
596 if (new_pos > ctx->pos)
597 ctx->pos = new_pos;
598 /* keep last_name */
599 } else {
600 ctx->pos = ceph_make_fpos(frag, dfi->next_offset,
601 false);
602 kfree(dfi->last_name);
603 dfi->last_name = NULL;
604 }
605 doutc(cl, "%p %llx.%llx next frag is %x\n", inode,
606 ceph_vinop(inode), frag);
607 goto more;
608 }
609 dfi->file_info.flags |= CEPH_F_ATEND;
610
611 /*
612 * if dir_release_count still matches the dir, no dentries
613 * were released during the whole readdir, and we should have
614 * the complete dir contents in our cache.
615 */
616 if (atomic64_read(&ci->i_release_count) ==
617 dfi->dir_release_count) {
618 spin_lock(&ci->i_ceph_lock);
619 if (dfi->dir_ordered_count ==
620 atomic64_read(&ci->i_ordered_count)) {
621 doutc(cl, " marking %p %llx.%llx complete and ordered\n",
622 inode, ceph_vinop(inode));
623 /* use i_size to track number of entries in
624 * readdir cache */
625 BUG_ON(dfi->readdir_cache_idx < 0);
626 i_size_write(inode, dfi->readdir_cache_idx *
627 sizeof(struct dentry*));
628 } else {
629 doutc(cl, " marking %llx.%llx complete\n",
630 ceph_vinop(inode));
631 }
632 __ceph_dir_set_complete(ci, dfi->dir_release_count,
633 dfi->dir_ordered_count);
634 spin_unlock(&ci->i_ceph_lock);
635 }
636 doutc(cl, "%p %llx.%llx file %p done.\n", inode, ceph_vinop(inode),
637 file);
638 return 0;
639}
640
641static void reset_readdir(struct ceph_dir_file_info *dfi)
642{
643 if (dfi->last_readdir) {
644 ceph_mdsc_put_request(dfi->last_readdir);
645 dfi->last_readdir = NULL;
646 }
647 kfree(dfi->last_name);
648 dfi->last_name = NULL;
649 dfi->dir_release_count = 0;
650 dfi->readdir_cache_idx = -1;
651 dfi->next_offset = 2; /* compensate for . and .. */
652 dfi->file_info.flags &= ~CEPH_F_ATEND;
653}
654
655/*
656 * discard buffered readdir content on seekdir(0), or seek to new frag,
657 * or seek prior to current chunk
658 */
659static bool need_reset_readdir(struct ceph_dir_file_info *dfi, loff_t new_pos)
660{
661 struct ceph_mds_reply_info_parsed *rinfo;
662 loff_t chunk_offset;
663 if (new_pos == 0)
664 return true;
665 if (is_hash_order(new_pos)) {
666 /* no need to reset last_name for a forward seek when
667 * dentries are sorted in hash order */
668 } else if (dfi->frag != fpos_frag(new_pos)) {
669 return true;
670 }
671 rinfo = dfi->last_readdir ? &dfi->last_readdir->r_reply_info : NULL;
672 if (!rinfo || !rinfo->dir_nr)
673 return true;
674 chunk_offset = rinfo->dir_entries[0].offset;
675 return new_pos < chunk_offset ||
676 is_hash_order(new_pos) != is_hash_order(chunk_offset);
677}
678
679static loff_t ceph_dir_llseek(struct file *file, loff_t offset, int whence)
680{
681 struct ceph_dir_file_info *dfi = file->private_data;
682 struct inode *inode = file->f_mapping->host;
683 struct ceph_client *cl = ceph_inode_to_client(inode);
684 loff_t retval;
685
686 inode_lock(inode);
687 retval = -EINVAL;
688 switch (whence) {
689 case SEEK_CUR:
690 offset += file->f_pos;
691 break;
692 case SEEK_SET:
693 break;
694 case SEEK_END:
695 retval = -EOPNOTSUPP;
696 goto out;
697 default:
698 goto out;
699 }
700
701 if (offset >= 0) {
702 if (need_reset_readdir(dfi, offset)) {
703 doutc(cl, "%p %llx.%llx dropping %p content\n",
704 inode, ceph_vinop(inode), file);
705 reset_readdir(dfi);
706 } else if (is_hash_order(offset) && offset > file->f_pos) {
707 /* for hash offset, we don't know if a forward seek
708 * is within same frag */
709 dfi->dir_release_count = 0;
710 dfi->readdir_cache_idx = -1;
711 }
712
713 if (offset != file->f_pos) {
714 file->f_pos = offset;
715 dfi->file_info.flags &= ~CEPH_F_ATEND;
716 }
717 retval = offset;
718 }
719out:
720 inode_unlock(inode);
721 return retval;
722}
723
724/*
725 * Handle lookups for the hidden .snap directory.
726 */
727struct dentry *ceph_handle_snapdir(struct ceph_mds_request *req,
728 struct dentry *dentry)
729{
730 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb);
731 struct inode *parent = d_inode(dentry->d_parent); /* we hold i_rwsem */
732 struct ceph_client *cl = ceph_inode_to_client(parent);
733
734 /* .snap dir? */
735 if (ceph_snap(parent) == CEPH_NOSNAP &&
736 strcmp(dentry->d_name.name, fsc->mount_options->snapdir_name) == 0) {
737 struct dentry *res;
738 struct inode *inode = ceph_get_snapdir(parent);
739
740 res = d_splice_alias(inode, dentry);
741 doutc(cl, "ENOENT on snapdir %p '%pd', linking to "
742 "snapdir %p %llx.%llx. Spliced dentry %p\n",
743 dentry, dentry, inode, ceph_vinop(inode), res);
744 if (res)
745 dentry = res;
746 }
747 return dentry;
748}
749
750/*
751 * Figure out final result of a lookup/open request.
752 *
753 * Mainly, make sure we return the final req->r_dentry (if it already
754 * existed) in place of the original VFS-provided dentry when they
755 * differ.
756 *
757 * Gracefully handle the case where the MDS replies with -ENOENT and
758 * no trace (which it may do, at its discretion, e.g., if it doesn't
759 * care to issue a lease on the negative dentry).
760 */
761struct dentry *ceph_finish_lookup(struct ceph_mds_request *req,
762 struct dentry *dentry, int err)
763{
764 struct ceph_client *cl = req->r_mdsc->fsc->client;
765
766 if (err == -ENOENT) {
767 /* no trace? */
768 err = 0;
769 if (!req->r_reply_info.head->is_dentry) {
770 doutc(cl,
771 "ENOENT and no trace, dentry %p inode %llx.%llx\n",
772 dentry, ceph_vinop(d_inode(dentry)));
773 if (d_really_is_positive(dentry)) {
774 d_drop(dentry);
775 err = -ENOENT;
776 } else {
777 if (d_unhashed(dentry)) {
778 struct inode *parent =
779 d_inode(dentry->d_parent);
780 if (!parent ||
781 ceph_snap(parent) == CEPH_NOSNAP)
782 d_add(dentry, NULL);
783 }
784 }
785 }
786 }
787 if (err)
788 dentry = ERR_PTR(err);
789 else if (dentry != req->r_dentry)
790 dentry = dget(req->r_dentry); /* we got spliced */
791 else
792 dentry = NULL;
793 return dentry;
794}
795
796static bool is_root_ceph_dentry(struct inode *inode, struct dentry *dentry)
797{
798 return ceph_ino(inode) == CEPH_INO_ROOT &&
799 strncmp(dentry->d_name.name, ".ceph", 5) == 0;
800}
801
802/*
803 * Look up a single dir entry. If there is a lookup intent, inform
804 * the MDS so that it gets our 'caps wanted' value in a single op.
805 */
806static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
807 unsigned int flags)
808{
809 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dir->i_sb);
810 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
811 struct ceph_client *cl = fsc->client;
812 struct ceph_mds_request *req;
813 int op;
814 int mask;
815 int err;
816
817 doutc(cl, "%p %llx.%llx/'%pd' dentry %p\n", dir, ceph_vinop(dir),
818 dentry, dentry);
819
820 if (dentry->d_name.len > NAME_MAX)
821 return ERR_PTR(-ENAMETOOLONG);
822
823 if (IS_ENCRYPTED(dir)) {
824 bool had_key = fscrypt_has_encryption_key(dir);
825
826 err = fscrypt_prepare_lookup_partial(dir, dentry);
827 if (err < 0)
828 return ERR_PTR(err);
829
830 /* mark directory as incomplete if it has been unlocked */
831 if (!had_key && fscrypt_has_encryption_key(dir))
832 ceph_dir_clear_complete(dir);
833 }
834
835 /* can we conclude ENOENT locally? */
836 if (d_really_is_negative(dentry)) {
837 struct ceph_inode_info *ci = ceph_inode(dir);
838 struct ceph_dentry_info *di = ceph_dentry(dentry);
839
840 spin_lock(&ci->i_ceph_lock);
841 doutc(cl, " dir %llx.%llx flags are 0x%lx\n",
842 ceph_vinop(dir), ci->i_ceph_flags);
843 if (strncmp(dentry->d_name.name,
844 fsc->mount_options->snapdir_name,
845 dentry->d_name.len) &&
846 !is_root_ceph_dentry(dir, dentry) &&
847 ceph_test_mount_opt(fsc, DCACHE) &&
848 ceph_snap(dir) == CEPH_NOSNAP &&
849 __ceph_dir_is_complete(ci) &&
850 __ceph_caps_issued_mask_metric(ci, CEPH_CAP_FILE_SHARED, 1)) {
851 __ceph_touch_fmode(ci, mdsc, CEPH_FILE_MODE_RD);
852 spin_unlock(&ci->i_ceph_lock);
853 doutc(cl, " dir %llx.%llx complete, -ENOENT\n",
854 ceph_vinop(dir));
855 if (d_unhashed(dentry))
856 d_add(dentry, NULL);
857 di->lease_shared_gen = atomic_read(&ci->i_shared_gen);
858 return NULL;
859 }
860 spin_unlock(&ci->i_ceph_lock);
861 }
862
863 op = ceph_snap(dir) == CEPH_SNAPDIR ?
864 CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_LOOKUP;
865 req = ceph_mdsc_create_request(mdsc, op, USE_ANY_MDS);
866 if (IS_ERR(req))
867 return ERR_CAST(req);
868 req->r_dentry = dget(dentry);
869 req->r_num_caps = 2;
870
871 mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
872 if (ceph_security_xattr_wanted(dir))
873 mask |= CEPH_CAP_XATTR_SHARED;
874 req->r_args.getattr.mask = cpu_to_le32(mask);
875
876 ihold(dir);
877 req->r_parent = dir;
878 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
879 err = ceph_mdsc_do_request(mdsc, NULL, req);
880 if (err == -ENOENT) {
881 struct dentry *res;
882
883 res = ceph_handle_snapdir(req, dentry);
884 if (IS_ERR(res)) {
885 err = PTR_ERR(res);
886 } else {
887 dentry = res;
888 err = 0;
889 }
890 }
891 dentry = ceph_finish_lookup(req, dentry, err);
892 ceph_mdsc_put_request(req); /* will dput(dentry) */
893 doutc(cl, "result=%p\n", dentry);
894 return dentry;
895}
896
897/*
898 * If we do a create but get no trace back from the MDS, follow up with
899 * a lookup (the VFS expects us to link up the provided dentry).
900 */
901int ceph_handle_notrace_create(struct inode *dir, struct dentry *dentry)
902{
903 struct dentry *result = ceph_lookup(dir, dentry, 0);
904
905 if (result && !IS_ERR(result)) {
906 /*
907 * We created the item, then did a lookup, and found
908 * it was already linked to another inode we already
909 * had in our cache (and thus got spliced). To not
910 * confuse VFS (especially when inode is a directory),
911 * we don't link our dentry to that inode, return an
912 * error instead.
913 *
914 * This event should be rare and it happens only when
915 * we talk to old MDS. Recent MDS does not send traceless
916 * reply for request that creates new inode.
917 */
918 d_drop(result);
919 return -ESTALE;
920 }
921 return PTR_ERR(result);
922}
923
924static int ceph_mknod(const struct mnt_idmap *idmap, struct inode *dir,
925 struct dentry *dentry, umode_t mode, dev_t rdev)
926{
927 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
928 struct ceph_client *cl = mdsc->fsc->client;
929 struct ceph_mds_request *req;
930 struct ceph_acl_sec_ctx as_ctx = {};
931 int err;
932
933 if (ceph_snap(dir) != CEPH_NOSNAP)
934 return -EROFS;
935
936 err = ceph_wait_on_conflict_unlink(dentry);
937 if (err)
938 return err;
939
940 if (ceph_quota_is_max_files_exceeded(dir)) {
941 err = -EDQUOT;
942 goto out;
943 }
944
945 doutc(cl, "%p %llx.%llx/'%pd' dentry %p mode 0%ho rdev %d\n",
946 dir, ceph_vinop(dir), dentry, dentry, mode, rdev);
947 req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_MKNOD, USE_AUTH_MDS);
948 if (IS_ERR(req)) {
949 err = PTR_ERR(req);
950 goto out;
951 }
952
953 req->r_new_inode = ceph_new_inode(dir, dentry, &mode, &as_ctx);
954 if (IS_ERR(req->r_new_inode)) {
955 err = PTR_ERR(req->r_new_inode);
956 req->r_new_inode = NULL;
957 goto out_req;
958 }
959
960 if (S_ISREG(mode) && IS_ENCRYPTED(dir))
961 set_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags);
962
963 req->r_dentry = dget(dentry);
964 req->r_num_caps = 2;
965 req->r_parent = dir;
966 ihold(dir);
967 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
968 req->r_mnt_idmap = mnt_idmap_get(idmap);
969 req->r_args.mknod.mode = cpu_to_le32(mode);
970 req->r_args.mknod.rdev = cpu_to_le32(rdev);
971 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL |
972 CEPH_CAP_XATTR_EXCL;
973 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
974
975 ceph_as_ctx_to_req(req, &as_ctx);
976
977 err = ceph_mdsc_do_request(mdsc, dir, req);
978 if (!err && !req->r_reply_info.head->is_dentry)
979 err = ceph_handle_notrace_create(dir, dentry);
980out_req:
981 ceph_mdsc_put_request(req);
982out:
983 if (!err)
984 ceph_init_inode_acls(d_inode(dentry), &as_ctx);
985 else
986 d_drop(dentry);
987 ceph_release_acl_sec_ctx(&as_ctx);
988 return err;
989}
990
991static int ceph_create(const struct mnt_idmap *idmap, struct inode *dir,
992 struct dentry *dentry, umode_t mode)
993{
994 return ceph_mknod(idmap, dir, dentry, mode, 0);
995}
996
997#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
998static int prep_encrypted_symlink_target(struct ceph_mds_request *req,
999 const char *dest)
1000{
1001 int err;
1002 int len = strlen(dest);
1003 struct fscrypt_str osd_link = FSTR_INIT(NULL, 0);
1004
1005 err = fscrypt_prepare_symlink(req->r_parent, dest, len, PATH_MAX,
1006 &osd_link);
1007 if (err)
1008 goto out;
1009
1010 err = fscrypt_encrypt_symlink(req->r_new_inode, dest, len, &osd_link);
1011 if (err)
1012 goto out;
1013
1014 req->r_path2 = kmalloc(BASE64_CHARS(osd_link.len) + 1, GFP_KERNEL);
1015 if (!req->r_path2) {
1016 err = -ENOMEM;
1017 goto out;
1018 }
1019
1020 len = base64_encode(osd_link.name, osd_link.len,
1021 req->r_path2, false, BASE64_IMAP);
1022 req->r_path2[len] = '\0';
1023out:
1024 fscrypt_fname_free_buffer(&osd_link);
1025 return err;
1026}
1027#else
1028static int prep_encrypted_symlink_target(struct ceph_mds_request *req,
1029 const char *dest)
1030{
1031 return -EOPNOTSUPP;
1032}
1033#endif
1034
1035static int ceph_symlink(const struct mnt_idmap *idmap, struct inode *dir,
1036 struct dentry *dentry, const char *dest)
1037{
1038 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
1039 struct ceph_client *cl = mdsc->fsc->client;
1040 struct ceph_mds_request *req;
1041 struct ceph_acl_sec_ctx as_ctx = {};
1042 umode_t mode = S_IFLNK | 0777;
1043 int err;
1044
1045 if (ceph_snap(dir) != CEPH_NOSNAP)
1046 return -EROFS;
1047
1048 err = ceph_wait_on_conflict_unlink(dentry);
1049 if (err)
1050 return err;
1051
1052 if (ceph_quota_is_max_files_exceeded(dir)) {
1053 err = -EDQUOT;
1054 goto out;
1055 }
1056
1057 doutc(cl, "%p %llx.%llx/'%pd' to '%s'\n", dir, ceph_vinop(dir), dentry,
1058 dest);
1059 req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_SYMLINK, USE_AUTH_MDS);
1060 if (IS_ERR(req)) {
1061 err = PTR_ERR(req);
1062 goto out;
1063 }
1064
1065 req->r_new_inode = ceph_new_inode(dir, dentry, &mode, &as_ctx);
1066 if (IS_ERR(req->r_new_inode)) {
1067 err = PTR_ERR(req->r_new_inode);
1068 req->r_new_inode = NULL;
1069 goto out_req;
1070 }
1071
1072 req->r_parent = dir;
1073 ihold(dir);
1074
1075 if (IS_ENCRYPTED(req->r_new_inode)) {
1076 err = prep_encrypted_symlink_target(req, dest);
1077 if (err)
1078 goto out_req;
1079 } else {
1080 req->r_path2 = kstrdup(dest, GFP_KERNEL);
1081 if (!req->r_path2) {
1082 err = -ENOMEM;
1083 goto out_req;
1084 }
1085 }
1086
1087 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
1088 req->r_mnt_idmap = mnt_idmap_get(idmap);
1089 req->r_dentry = dget(dentry);
1090 req->r_num_caps = 2;
1091 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL |
1092 CEPH_CAP_XATTR_EXCL;
1093 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
1094
1095 ceph_as_ctx_to_req(req, &as_ctx);
1096
1097 err = ceph_mdsc_do_request(mdsc, dir, req);
1098 if (!err && !req->r_reply_info.head->is_dentry)
1099 err = ceph_handle_notrace_create(dir, dentry);
1100out_req:
1101 ceph_mdsc_put_request(req);
1102out:
1103 if (err)
1104 d_drop(dentry);
1105 ceph_release_acl_sec_ctx(&as_ctx);
1106 return err;
1107}
1108
1109static struct dentry *ceph_mkdir(const struct mnt_idmap *idmap, struct inode *dir,
1110 struct dentry *dentry, umode_t mode)
1111{
1112 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
1113 struct ceph_client *cl = mdsc->fsc->client;
1114 struct ceph_mds_request *req;
1115 struct ceph_acl_sec_ctx as_ctx = {};
1116 struct dentry *ret;
1117 int err;
1118 int op;
1119
1120 err = ceph_wait_on_conflict_unlink(dentry);
1121 if (err)
1122 return ERR_PTR(err);
1123
1124 if (ceph_snap(dir) == CEPH_SNAPDIR) {
1125 /* mkdir .snap/foo is a MKSNAP */
1126 op = CEPH_MDS_OP_MKSNAP;
1127 doutc(cl, "mksnap %llx.%llx/'%pd' dentry %p\n",
1128 ceph_vinop(dir), dentry, dentry);
1129 } else if (ceph_snap(dir) == CEPH_NOSNAP) {
1130 doutc(cl, "mkdir %llx.%llx/'%pd' dentry %p mode 0%ho\n",
1131 ceph_vinop(dir), dentry, dentry, mode);
1132 op = CEPH_MDS_OP_MKDIR;
1133 } else {
1134 ret = ERR_PTR(-EROFS);
1135 goto out;
1136 }
1137
1138 if (op == CEPH_MDS_OP_MKDIR &&
1139 ceph_quota_is_max_files_exceeded(dir)) {
1140 ret = ERR_PTR(-EDQUOT);
1141 goto out;
1142 }
1143 if ((op == CEPH_MDS_OP_MKSNAP) && IS_ENCRYPTED(dir) &&
1144 !fscrypt_has_encryption_key(dir)) {
1145 ret = ERR_PTR(-ENOKEY);
1146 goto out;
1147 }
1148
1149
1150 req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS);
1151 if (IS_ERR(req)) {
1152 ret = ERR_CAST(req);
1153 goto out;
1154 }
1155
1156 req->r_new_inode = ceph_new_inode(dir, dentry, &mode, &as_ctx);
1157 if (IS_ERR(req->r_new_inode)) {
1158 ret = ERR_CAST(req->r_new_inode);
1159 req->r_new_inode = NULL;
1160 goto out_req;
1161 }
1162
1163 req->r_dentry = dget(dentry);
1164 req->r_num_caps = 2;
1165 req->r_parent = dir;
1166 ihold(dir);
1167 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
1168 if (op == CEPH_MDS_OP_MKDIR)
1169 req->r_mnt_idmap = mnt_idmap_get(idmap);
1170 req->r_args.mkdir.mode = cpu_to_le32(mode);
1171 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL |
1172 CEPH_CAP_XATTR_EXCL;
1173 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
1174
1175 ceph_as_ctx_to_req(req, &as_ctx);
1176
1177 err = ceph_mdsc_do_request(mdsc, dir, req);
1178 if (!err &&
1179 !req->r_reply_info.head->is_target &&
1180 !req->r_reply_info.head->is_dentry)
1181 err = ceph_handle_notrace_create(dir, dentry);
1182 ret = err ? ERR_PTR(err) : NULL;
1183out_req:
1184 if (!IS_ERR(ret) && req->r_dentry != dentry)
1185 /* Some other dentry was spliced in */
1186 ret = dget(req->r_dentry);
1187 ceph_mdsc_put_request(req);
1188out:
1189 if (!IS_ERR(ret)) {
1190 if (ret)
1191 dentry = ret;
1192 ceph_init_inode_acls(d_inode(dentry), &as_ctx);
1193 } else {
1194 d_drop(dentry);
1195 }
1196 ceph_release_acl_sec_ctx(&as_ctx);
1197 return ret;
1198}
1199
1200static int ceph_link(struct dentry *old_dentry, struct inode *dir,
1201 struct dentry *dentry)
1202{
1203 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
1204 struct ceph_client *cl = mdsc->fsc->client;
1205 struct ceph_mds_request *req;
1206 int err;
1207
1208 if (dentry->d_flags & DCACHE_DISCONNECTED)
1209 return -EINVAL;
1210
1211 err = ceph_wait_on_conflict_unlink(dentry);
1212 if (err)
1213 return err;
1214
1215 if (ceph_snap(dir) != CEPH_NOSNAP)
1216 return -EROFS;
1217
1218 err = fscrypt_prepare_link(old_dentry, dir, dentry);
1219 if (err)
1220 return err;
1221
1222 doutc(cl, "%p %llx.%llx/'%pd' to '%pd'\n", dir, ceph_vinop(dir),
1223 old_dentry, dentry);
1224 req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LINK, USE_AUTH_MDS);
1225 if (IS_ERR(req)) {
1226 d_drop(dentry);
1227 return PTR_ERR(req);
1228 }
1229 req->r_dentry = dget(dentry);
1230 req->r_num_caps = 2;
1231 req->r_old_dentry = dget(old_dentry);
1232 /*
1233 * The old_dentry maybe a DCACHE_DISCONNECTED dentry, then we
1234 * will just pass the ino# to MDSs.
1235 */
1236 if (old_dentry->d_flags & DCACHE_DISCONNECTED)
1237 req->r_ino2 = ceph_vino(d_inode(old_dentry));
1238 req->r_parent = dir;
1239 ihold(dir);
1240 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
1241 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_XATTR_EXCL;
1242 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
1243 /* release LINK_SHARED on source inode (mds will lock it) */
1244 req->r_old_inode_drop = CEPH_CAP_LINK_SHARED | CEPH_CAP_LINK_EXCL;
1245 err = ceph_mdsc_do_request(mdsc, dir, req);
1246 if (err) {
1247 d_drop(dentry);
1248 } else if (!req->r_reply_info.head->is_dentry) {
1249 ihold(d_inode(old_dentry));
1250 d_instantiate(dentry, d_inode(old_dentry));
1251 }
1252 ceph_mdsc_put_request(req);
1253 return err;
1254}
1255
1256static void ceph_async_unlink_cb(struct ceph_mds_client *mdsc,
1257 struct ceph_mds_request *req)
1258{
1259 struct dentry *dentry = req->r_dentry;
1260 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb);
1261 struct ceph_client *cl = fsc->client;
1262 struct ceph_dentry_info *di = ceph_dentry(dentry);
1263 int result = req->r_err ? req->r_err :
1264 le32_to_cpu(req->r_reply_info.head->result);
1265
1266 if (!test_bit(CEPH_DENTRY_ASYNC_UNLINK_BIT, &di->flags))
1267 pr_warn_client(cl,
1268 "dentry %p:%pd async unlink bit is not set\n",
1269 dentry, dentry);
1270
1271 spin_lock(&fsc->async_unlink_conflict_lock);
1272 hash_del_rcu(&di->hnode);
1273 spin_unlock(&fsc->async_unlink_conflict_lock);
1274
1275 spin_lock(&dentry->d_lock);
1276 clear_and_wake_up_bit(CEPH_DENTRY_ASYNC_UNLINK_BIT, &di->flags);
1277 spin_unlock(&dentry->d_lock);
1278
1279 synchronize_rcu();
1280
1281 if (result == -EJUKEBOX)
1282 goto out;
1283
1284 /* If op failed, mark everyone involved for errors */
1285 if (result) {
1286 struct ceph_path_info path_info = {0};
1287 char *path = ceph_mdsc_build_path(mdsc, dentry, &path_info, 0);
1288
1289 /* mark error on parent + clear complete */
1290 mapping_set_error(req->r_parent->i_mapping, result);
1291 ceph_dir_clear_complete(req->r_parent);
1292
1293 /* drop the dentry -- we don't know its status */
1294 if (!d_unhashed(dentry))
1295 d_drop(dentry);
1296
1297 /* mark inode itself for an error (since metadata is bogus) */
1298 mapping_set_error(req->r_old_inode->i_mapping, result);
1299
1300 pr_warn_client(cl, "failure path=(%llx)%s result=%d!\n",
1301 path_info.vino.ino, IS_ERR(path) ? "<<bad>>" : path, result);
1302 ceph_mdsc_free_path_info(&path_info);
1303 }
1304out:
1305 iput(req->r_old_inode);
1306 ceph_mdsc_release_dir_caps(req);
1307}
1308
1309static int get_caps_for_async_unlink(struct inode *dir, struct dentry *dentry)
1310{
1311 struct ceph_inode_info *ci = ceph_inode(dir);
1312 struct ceph_dentry_info *di;
1313 int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_UNLINK;
1314
1315 spin_lock(&ci->i_ceph_lock);
1316 if ((__ceph_caps_issued(ci, NULL) & want) == want) {
1317 ceph_take_cap_refs(ci, want, false);
1318 got = want;
1319 }
1320 spin_unlock(&ci->i_ceph_lock);
1321
1322 /* If we didn't get anything, return 0 */
1323 if (!got)
1324 return 0;
1325
1326 spin_lock(&dentry->d_lock);
1327 di = ceph_dentry(dentry);
1328 /*
1329 * - We are holding Fx, which implies Fs caps.
1330 * - Only support async unlink for primary linkage
1331 */
1332 if (atomic_read(&ci->i_shared_gen) != di->lease_shared_gen ||
1333 !(di->flags & CEPH_DENTRY_PRIMARY_LINK))
1334 want = 0;
1335 spin_unlock(&dentry->d_lock);
1336
1337 /* Do we still want what we've got? */
1338 if (want == got)
1339 return got;
1340
1341 ceph_put_cap_refs(ci, got);
1342 return 0;
1343}
1344
1345/*
1346 * rmdir and unlink are differ only by the metadata op code
1347 */
1348static int ceph_unlink(struct inode *dir, struct dentry *dentry)
1349{
1350 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dir->i_sb);
1351 struct ceph_client *cl = fsc->client;
1352 struct ceph_mds_client *mdsc = fsc->mdsc;
1353 struct inode *inode = d_inode(dentry);
1354 struct ceph_inode_info *ci = ceph_inode(inode);
1355 struct ceph_mds_request *req;
1356 bool try_async = ceph_test_mount_opt(fsc, ASYNC_DIROPS);
1357 struct dentry *dn;
1358 int err = -EROFS;
1359 int op;
1360 char *path;
1361
1362 if (ceph_snap(dir) == CEPH_SNAPDIR) {
1363 /* rmdir .snap/foo is RMSNAP */
1364 doutc(cl, "rmsnap %llx.%llx/'%pd' dn\n", ceph_vinop(dir),
1365 dentry);
1366 op = CEPH_MDS_OP_RMSNAP;
1367 } else if (ceph_snap(dir) == CEPH_NOSNAP) {
1368 doutc(cl, "unlink/rmdir %llx.%llx/'%pd' inode %llx.%llx\n",
1369 ceph_vinop(dir), dentry, ceph_vinop(inode));
1370 op = d_is_dir(dentry) ?
1371 CEPH_MDS_OP_RMDIR : CEPH_MDS_OP_UNLINK;
1372 } else
1373 goto out;
1374
1375 dn = d_find_alias(dir);
1376 if (!dn) {
1377 try_async = false;
1378 } else {
1379 struct ceph_path_info path_info = {0};
1380 path = ceph_mdsc_build_path(mdsc, dn, &path_info, 0);
1381 if (IS_ERR(path)) {
1382 try_async = false;
1383 err = 0;
1384 } else {
1385 err = ceph_mds_check_access(mdsc, path, MAY_WRITE);
1386 }
1387 ceph_mdsc_free_path_info(&path_info);
1388 dput(dn);
1389
1390 /* For none EACCES cases will let the MDS do the mds auth check */
1391 if (err == -EACCES) {
1392 return err;
1393 } else if (err < 0) {
1394 try_async = false;
1395 err = 0;
1396 }
1397 }
1398
1399retry:
1400 req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS);
1401 if (IS_ERR(req)) {
1402 err = PTR_ERR(req);
1403 goto out;
1404 }
1405 req->r_dentry = dget(dentry);
1406 req->r_num_caps = 2;
1407 req->r_parent = dir;
1408 ihold(dir);
1409 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_XATTR_EXCL;
1410 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
1411 req->r_inode_drop = ceph_drop_caps_for_unlink(inode);
1412
1413 if (try_async && op == CEPH_MDS_OP_UNLINK &&
1414 (req->r_dir_caps = get_caps_for_async_unlink(dir, dentry))) {
1415 struct ceph_dentry_info *di = ceph_dentry(dentry);
1416
1417 doutc(cl, "async unlink on %llx.%llx/'%pd' caps=%s",
1418 ceph_vinop(dir), dentry,
1419 ceph_cap_string(req->r_dir_caps));
1420 set_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags);
1421 req->r_callback = ceph_async_unlink_cb;
1422 req->r_old_inode = d_inode(dentry);
1423 ihold(req->r_old_inode);
1424
1425 spin_lock(&dentry->d_lock);
1426 di->flags |= CEPH_DENTRY_ASYNC_UNLINK;
1427 spin_unlock(&dentry->d_lock);
1428
1429 spin_lock(&fsc->async_unlink_conflict_lock);
1430 hash_add_rcu(fsc->async_unlink_conflict, &di->hnode,
1431 dentry->d_name.hash);
1432 spin_unlock(&fsc->async_unlink_conflict_lock);
1433
1434 err = ceph_mdsc_submit_request(mdsc, dir, req);
1435 if (!err) {
1436 /*
1437 * We have enough caps, so we assume that the unlink
1438 * will succeed. Fix up the target inode and dcache.
1439 */
1440
1441 /*
1442 * Protect the i_nlink update with i_ceph_lock
1443 * to precent racing against ceph_fill_inode()
1444 * handling our completion on a worker thread
1445 * and don't decrement if i_nlink has already
1446 * been updated to zero by this completion.
1447 */
1448 spin_lock(&ci->i_ceph_lock);
1449 if (inode->i_nlink > 0)
1450 drop_nlink(inode);
1451 spin_unlock(&ci->i_ceph_lock);
1452
1453 d_delete(dentry);
1454 } else {
1455 spin_lock(&fsc->async_unlink_conflict_lock);
1456 hash_del_rcu(&di->hnode);
1457 spin_unlock(&fsc->async_unlink_conflict_lock);
1458
1459 spin_lock(&dentry->d_lock);
1460 di->flags &= ~CEPH_DENTRY_ASYNC_UNLINK;
1461 spin_unlock(&dentry->d_lock);
1462
1463 if (err == -EJUKEBOX) {
1464 try_async = false;
1465 ceph_mdsc_put_request(req);
1466 goto retry;
1467 }
1468 }
1469 } else {
1470 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
1471 err = ceph_mdsc_do_request(mdsc, dir, req);
1472 if (!err && !req->r_reply_info.head->is_dentry)
1473 d_delete(dentry);
1474 }
1475
1476 ceph_mdsc_put_request(req);
1477out:
1478 return err;
1479}
1480
1481static int ceph_rename(const struct mnt_idmap *idmap, struct inode *old_dir,
1482 struct dentry *old_dentry, struct inode *new_dir,
1483 struct dentry *new_dentry, unsigned int flags)
1484{
1485 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(old_dir->i_sb);
1486 struct ceph_client *cl = mdsc->fsc->client;
1487 struct ceph_mds_request *req;
1488 int op = CEPH_MDS_OP_RENAME;
1489 int err;
1490
1491 if (flags)
1492 return -EINVAL;
1493
1494 if (ceph_snap(old_dir) != ceph_snap(new_dir))
1495 return -EXDEV;
1496 if (ceph_snap(old_dir) != CEPH_NOSNAP) {
1497 if (old_dir == new_dir && ceph_snap(old_dir) == CEPH_SNAPDIR)
1498 op = CEPH_MDS_OP_RENAMESNAP;
1499 else
1500 return -EROFS;
1501 }
1502 /* don't allow cross-quota renames */
1503 if ((old_dir != new_dir) &&
1504 (!ceph_quota_is_same_realm(old_dir, new_dir)))
1505 return -EXDEV;
1506
1507 err = ceph_wait_on_conflict_unlink(new_dentry);
1508 if (err)
1509 return err;
1510
1511 err = fscrypt_prepare_rename(old_dir, old_dentry, new_dir, new_dentry,
1512 flags);
1513 if (err)
1514 return err;
1515
1516 doutc(cl, "%llx.%llx/'%pd' to %llx.%llx/'%pd'\n",
1517 ceph_vinop(old_dir), old_dentry, ceph_vinop(new_dir),
1518 new_dentry);
1519 req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS);
1520 if (IS_ERR(req))
1521 return PTR_ERR(req);
1522 ihold(old_dir);
1523 req->r_dentry = dget(new_dentry);
1524 req->r_num_caps = 2;
1525 req->r_old_dentry = dget(old_dentry);
1526 req->r_old_dentry_dir = old_dir;
1527 req->r_parent = new_dir;
1528 ihold(new_dir);
1529 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
1530 req->r_old_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_XATTR_EXCL;
1531 req->r_old_dentry_unless = CEPH_CAP_FILE_EXCL;
1532 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_XATTR_EXCL;
1533 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
1534 /* release LINK_RDCACHE on source inode (mds will lock it) */
1535 req->r_old_inode_drop = CEPH_CAP_LINK_SHARED | CEPH_CAP_LINK_EXCL;
1536 if (d_really_is_positive(new_dentry)) {
1537 req->r_inode_drop =
1538 ceph_drop_caps_for_unlink(d_inode(new_dentry));
1539 }
1540 err = ceph_mdsc_do_request(mdsc, old_dir, req);
1541 if (!err && !req->r_reply_info.head->is_dentry) {
1542 /*
1543 * Normally d_move() is done by fill_trace (called by
1544 * do_request, above). If there is no trace, we need
1545 * to do it here.
1546 */
1547 d_move(old_dentry, new_dentry);
1548 }
1549 ceph_mdsc_put_request(req);
1550 return err;
1551}
1552
1553/*
1554 * Move dentry to tail of mdsc->dentry_leases list when lease is updated.
1555 * Leases at front of the list will expire first. (Assume all leases have
1556 * similar duration)
1557 *
1558 * Called under dentry->d_lock.
1559 */
1560void __ceph_dentry_lease_touch(struct ceph_dentry_info *di)
1561{
1562 struct dentry *dn = di->dentry;
1563 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(dn->d_sb)->mdsc;
1564 struct ceph_client *cl = mdsc->fsc->client;
1565
1566 doutc(cl, "%p %p '%pd'\n", di, dn, dn);
1567
1568 di->flags |= CEPH_DENTRY_LEASE_LIST;
1569 if (di->flags & CEPH_DENTRY_SHRINK_LIST) {
1570 di->flags |= CEPH_DENTRY_REFERENCED;
1571 return;
1572 }
1573
1574 spin_lock(&mdsc->dentry_list_lock);
1575 list_move_tail(&di->lease_list, &mdsc->dentry_leases);
1576 spin_unlock(&mdsc->dentry_list_lock);
1577}
1578
1579static void __dentry_dir_lease_touch(struct ceph_mds_client* mdsc,
1580 struct ceph_dentry_info *di)
1581{
1582 di->flags &= ~(CEPH_DENTRY_LEASE_LIST | CEPH_DENTRY_REFERENCED);
1583 di->lease_gen = 0;
1584 di->time = jiffies;
1585 list_move_tail(&di->lease_list, &mdsc->dentry_dir_leases);
1586}
1587
1588/*
1589 * When dir lease is used, add dentry to tail of mdsc->dentry_dir_leases
1590 * list if it's not in the list, otherwise set 'referenced' flag.
1591 *
1592 * Called under dentry->d_lock.
1593 */
1594void __ceph_dentry_dir_lease_touch(struct ceph_dentry_info *di)
1595{
1596 struct dentry *dn = di->dentry;
1597 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(dn->d_sb)->mdsc;
1598 struct ceph_client *cl = mdsc->fsc->client;
1599
1600 doutc(cl, "%p %p '%pd' (offset 0x%llx)\n", di, dn, dn, di->offset);
1601
1602 if (!list_empty(&di->lease_list)) {
1603 if (di->flags & CEPH_DENTRY_LEASE_LIST) {
1604 /* don't remove dentry from dentry lease list
1605 * if its lease is valid */
1606 if (__dentry_lease_is_valid(di))
1607 return;
1608 } else {
1609 di->flags |= CEPH_DENTRY_REFERENCED;
1610 return;
1611 }
1612 }
1613
1614 if (di->flags & CEPH_DENTRY_SHRINK_LIST) {
1615 di->flags |= CEPH_DENTRY_REFERENCED;
1616 di->flags &= ~CEPH_DENTRY_LEASE_LIST;
1617 return;
1618 }
1619
1620 spin_lock(&mdsc->dentry_list_lock);
1621 __dentry_dir_lease_touch(mdsc, di);
1622 spin_unlock(&mdsc->dentry_list_lock);
1623}
1624
1625static void __dentry_lease_unlist(struct ceph_dentry_info *di)
1626{
1627 struct ceph_mds_client *mdsc;
1628 if (di->flags & CEPH_DENTRY_SHRINK_LIST)
1629 return;
1630 if (list_empty(&di->lease_list))
1631 return;
1632
1633 mdsc = ceph_sb_to_fs_client(di->dentry->d_sb)->mdsc;
1634 spin_lock(&mdsc->dentry_list_lock);
1635 list_del_init(&di->lease_list);
1636 spin_unlock(&mdsc->dentry_list_lock);
1637}
1638
1639enum {
1640 KEEP = 0,
1641 DELETE = 1,
1642 TOUCH = 2,
1643 STOP = 4,
1644};
1645
1646struct ceph_lease_walk_control {
1647 bool dir_lease;
1648 bool expire_dir_lease;
1649 unsigned long nr_to_scan;
1650 unsigned long dir_lease_ttl;
1651};
1652
1653static int __dir_lease_check(const struct dentry *, struct ceph_lease_walk_control *);
1654static int __dentry_lease_check(const struct dentry *);
1655
1656static unsigned long
1657__dentry_leases_walk(struct ceph_mds_client *mdsc,
1658 struct ceph_lease_walk_control *lwc)
1659{
1660 struct ceph_dentry_info *di, *tmp;
1661 struct dentry *dentry, *last = NULL;
1662 struct list_head* list;
1663 LIST_HEAD(dispose);
1664 unsigned long freed = 0;
1665 int ret = 0;
1666
1667 list = lwc->dir_lease ? &mdsc->dentry_dir_leases : &mdsc->dentry_leases;
1668 spin_lock(&mdsc->dentry_list_lock);
1669 list_for_each_entry_safe(di, tmp, list, lease_list) {
1670 if (!lwc->nr_to_scan)
1671 break;
1672 --lwc->nr_to_scan;
1673
1674 dentry = di->dentry;
1675 if (last == dentry)
1676 break;
1677
1678 if (!spin_trylock(&dentry->d_lock))
1679 continue;
1680
1681 if (lockref_is_dead(&dentry->d_lockref)) {
1682 list_del_init(&di->lease_list);
1683 goto next;
1684 }
1685
1686 if (lwc->dir_lease)
1687 ret = __dir_lease_check(dentry, lwc);
1688 else
1689 ret = __dentry_lease_check(dentry);
1690 if (ret & TOUCH) {
1691 /* move it into tail of dir lease list */
1692 __dentry_dir_lease_touch(mdsc, di);
1693 if (!last)
1694 last = dentry;
1695 }
1696 if (ret & DELETE) {
1697 /* stale lease */
1698 di->flags &= ~CEPH_DENTRY_REFERENCED;
1699 if (dentry->d_lockref.count > 0) {
1700 /* update_dentry_lease() will re-add
1701 * it to lease list, or
1702 * ceph_d_delete() will return 1 when
1703 * last reference is dropped */
1704 list_del_init(&di->lease_list);
1705 } else {
1706 di->flags |= CEPH_DENTRY_SHRINK_LIST;
1707 list_move_tail(&di->lease_list, &dispose);
1708 dget_dlock(dentry);
1709 }
1710 }
1711next:
1712 spin_unlock(&dentry->d_lock);
1713 if (ret & STOP)
1714 break;
1715 }
1716 spin_unlock(&mdsc->dentry_list_lock);
1717
1718 while (!list_empty(&dispose)) {
1719 di = list_first_entry(&dispose, struct ceph_dentry_info,
1720 lease_list);
1721 dentry = di->dentry;
1722 spin_lock(&dentry->d_lock);
1723
1724 list_del_init(&di->lease_list);
1725 di->flags &= ~CEPH_DENTRY_SHRINK_LIST;
1726 if (di->flags & CEPH_DENTRY_REFERENCED) {
1727 spin_lock(&mdsc->dentry_list_lock);
1728 if (di->flags & CEPH_DENTRY_LEASE_LIST) {
1729 list_add_tail(&di->lease_list,
1730 &mdsc->dentry_leases);
1731 } else {
1732 __dentry_dir_lease_touch(mdsc, di);
1733 }
1734 spin_unlock(&mdsc->dentry_list_lock);
1735 } else {
1736 freed++;
1737 }
1738
1739 spin_unlock(&dentry->d_lock);
1740 /* ceph_d_delete() does the trick */
1741 dput(dentry);
1742 }
1743 return freed;
1744}
1745
1746static int __dentry_lease_check(const struct dentry *dentry)
1747{
1748 struct ceph_dentry_info *di = ceph_dentry(dentry);
1749 int ret;
1750
1751 if (__dentry_lease_is_valid(di))
1752 return STOP;
1753 ret = __dir_lease_try_check(dentry);
1754 if (ret == -EBUSY)
1755 return KEEP;
1756 if (ret > 0)
1757 return TOUCH;
1758 return DELETE;
1759}
1760
1761static int __dir_lease_check(const struct dentry *dentry,
1762 struct ceph_lease_walk_control *lwc)
1763{
1764 struct ceph_dentry_info *di = ceph_dentry(dentry);
1765
1766 int ret = __dir_lease_try_check(dentry);
1767 if (ret == -EBUSY)
1768 return KEEP;
1769 if (ret > 0) {
1770 if (time_before(jiffies, di->time + lwc->dir_lease_ttl))
1771 return STOP;
1772 if (!lwc->expire_dir_lease)
1773 return KEEP;
1774 /* Move dentry to tail of dir lease list if we don't want
1775 * to delete it. So dentries in the list are checked in a
1776 * round robin manner */
1777 if (dentry->d_lockref.count > 0 ||
1778 (di->flags & CEPH_DENTRY_REFERENCED))
1779 return TOUCH;
1780 /* invalidate dir lease */
1781 di->lease_shared_gen = 0;
1782 }
1783 return DELETE;
1784}
1785
1786int ceph_trim_dentries(struct ceph_mds_client *mdsc)
1787{
1788 struct ceph_lease_walk_control lwc;
1789 unsigned long count;
1790 unsigned long freed;
1791
1792 spin_lock(&mdsc->caps_list_lock);
1793 if (mdsc->caps_use_max > 0 &&
1794 mdsc->caps_use_count > mdsc->caps_use_max)
1795 count = mdsc->caps_use_count - mdsc->caps_use_max;
1796 else
1797 count = 0;
1798 spin_unlock(&mdsc->caps_list_lock);
1799
1800 lwc.dir_lease = false;
1801 lwc.nr_to_scan = CEPH_CAPS_PER_RELEASE * 2;
1802 freed = __dentry_leases_walk(mdsc, &lwc);
1803 if (freed > 0 && !lwc.nr_to_scan) /* more invalid leases */
1804 return -EAGAIN;
1805
1806 if (lwc.nr_to_scan < CEPH_CAPS_PER_RELEASE)
1807 lwc.nr_to_scan = CEPH_CAPS_PER_RELEASE;
1808
1809 lwc.dir_lease = true;
1810 lwc.expire_dir_lease = freed < count;
1811 lwc.dir_lease_ttl = mdsc->fsc->mount_options->caps_wanted_delay_max * HZ;
1812 freed +=__dentry_leases_walk(mdsc, &lwc);
1813 if (freed == 0 && count == 0)
1814 /* no progress possible currently, retry futile */
1815 return 0;
1816
1817 if (!lwc.nr_to_scan) /* more to check */
1818 return -EAGAIN;
1819
1820 return freed > 0 ? 1 : 0;
1821}
1822
1823/*
1824 * Ensure a dentry lease will no longer revalidate.
1825 */
1826void ceph_invalidate_dentry_lease(struct dentry *dentry)
1827{
1828 struct ceph_dentry_info *di = ceph_dentry(dentry);
1829 spin_lock(&dentry->d_lock);
1830 di->time = jiffies;
1831 di->lease_shared_gen = 0;
1832 di->flags &= ~CEPH_DENTRY_PRIMARY_LINK;
1833 __dentry_lease_unlist(di);
1834 spin_unlock(&dentry->d_lock);
1835}
1836
1837/*
1838 * Check if dentry lease is valid. If not, delete the lease. Try to
1839 * renew if the least is more than half up.
1840 */
1841static bool __dentry_lease_is_valid(struct ceph_dentry_info *di)
1842{
1843 struct ceph_mds_session *session;
1844
1845 if (!di->lease_gen)
1846 return false;
1847
1848 session = di->lease_session;
1849 if (session) {
1850 u32 gen;
1851 unsigned long ttl;
1852
1853 gen = atomic_read(&session->s_cap_gen);
1854 ttl = session->s_cap_ttl;
1855
1856 if (di->lease_gen == gen &&
1857 time_before(jiffies, ttl) &&
1858 time_before(jiffies, di->time))
1859 return true;
1860 }
1861 di->lease_gen = 0;
1862 return false;
1863}
1864
1865static int dentry_lease_is_valid(struct dentry *dentry, unsigned int flags)
1866{
1867 struct ceph_dentry_info *di;
1868 struct ceph_mds_session *session = NULL;
1869 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(dentry->d_sb)->mdsc;
1870 struct ceph_client *cl = mdsc->fsc->client;
1871 u32 seq = 0;
1872 int valid = 0;
1873
1874 spin_lock(&dentry->d_lock);
1875 di = ceph_dentry(dentry);
1876 if (di && __dentry_lease_is_valid(di)) {
1877 valid = 1;
1878
1879 if (di->lease_renew_after &&
1880 time_after(jiffies, di->lease_renew_after)) {
1881 /*
1882 * We should renew. If we're in RCU walk mode
1883 * though, we can't do that so just return
1884 * -ECHILD.
1885 */
1886 if (flags & LOOKUP_RCU) {
1887 valid = -ECHILD;
1888 } else {
1889 session = ceph_get_mds_session(di->lease_session);
1890 seq = di->lease_seq;
1891 di->lease_renew_after = 0;
1892 di->lease_renew_from = jiffies;
1893 }
1894 }
1895 }
1896 spin_unlock(&dentry->d_lock);
1897
1898 if (session) {
1899 ceph_mdsc_lease_send_msg(session, dentry,
1900 CEPH_MDS_LEASE_RENEW, seq);
1901 ceph_put_mds_session(session);
1902 }
1903 doutc(cl, "dentry %p = %d\n", dentry, valid);
1904 return valid;
1905}
1906
1907/*
1908 * Called under dentry->d_lock.
1909 */
1910static int __dir_lease_try_check(const struct dentry *dentry)
1911{
1912 struct ceph_dentry_info *di = ceph_dentry(dentry);
1913 struct inode *dir;
1914 struct ceph_inode_info *ci;
1915 int valid = 0;
1916
1917 if (!di->lease_shared_gen)
1918 return 0;
1919 if (IS_ROOT(dentry))
1920 return 0;
1921
1922 dir = d_inode(dentry->d_parent);
1923 ci = ceph_inode(dir);
1924
1925 if (spin_trylock(&ci->i_ceph_lock)) {
1926 if (atomic_read(&ci->i_shared_gen) == di->lease_shared_gen &&
1927 __ceph_caps_issued_mask(ci, CEPH_CAP_FILE_SHARED, 0))
1928 valid = 1;
1929 spin_unlock(&ci->i_ceph_lock);
1930 } else {
1931 valid = -EBUSY;
1932 }
1933
1934 if (!valid)
1935 di->lease_shared_gen = 0;
1936 return valid;
1937}
1938
1939/*
1940 * Check if directory-wide content lease/cap is valid.
1941 */
1942static int dir_lease_is_valid(struct inode *dir, struct dentry *dentry,
1943 struct ceph_mds_client *mdsc)
1944{
1945 struct ceph_inode_info *ci = ceph_inode(dir);
1946 struct ceph_client *cl = mdsc->fsc->client;
1947 int valid;
1948 int shared_gen;
1949
1950 spin_lock(&ci->i_ceph_lock);
1951 valid = __ceph_caps_issued_mask(ci, CEPH_CAP_FILE_SHARED, 1);
1952 if (valid) {
1953 __ceph_touch_fmode(ci, mdsc, CEPH_FILE_MODE_RD);
1954 shared_gen = atomic_read(&ci->i_shared_gen);
1955 }
1956 spin_unlock(&ci->i_ceph_lock);
1957 if (valid) {
1958 struct ceph_dentry_info *di;
1959 spin_lock(&dentry->d_lock);
1960 di = ceph_dentry(dentry);
1961 if (dir == d_inode(dentry->d_parent) &&
1962 di && di->lease_shared_gen == shared_gen)
1963 __ceph_dentry_dir_lease_touch(di);
1964 else
1965 valid = 0;
1966 spin_unlock(&dentry->d_lock);
1967 }
1968 doutc(cl, "dir %p %llx.%llx v%u dentry %p '%pd' = %d\n", dir,
1969 ceph_vinop(dir), (unsigned)atomic_read(&ci->i_shared_gen),
1970 dentry, dentry, valid);
1971 return valid;
1972}
1973
1974/*
1975 * Check if cached dentry can be trusted.
1976 */
1977static int ceph_d_revalidate(struct inode *dir, const struct qstr *name,
1978 struct dentry *dentry, unsigned int flags)
1979{
1980 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(dentry->d_sb)->mdsc;
1981 struct ceph_client *cl = mdsc->fsc->client;
1982 int valid = 0;
1983 struct inode *inode;
1984
1985 valid = fscrypt_d_revalidate(dir, name, dentry, flags);
1986 if (valid <= 0)
1987 return valid;
1988
1989 inode = d_inode_rcu(dentry);
1990
1991 doutc(cl, "%p '%pd' inode %p offset 0x%llx nokey %d\n",
1992 dentry, dentry, inode, ceph_dentry(dentry)->offset,
1993 !!(dentry->d_flags & DCACHE_NOKEY_NAME));
1994
1995 mdsc = ceph_sb_to_fs_client(dir->i_sb)->mdsc;
1996
1997 /* always trust cached snapped dentries, snapdir dentry */
1998 if (ceph_snap(dir) != CEPH_NOSNAP) {
1999 doutc(cl, "%p '%pd' inode %p is SNAPPED\n", dentry,
2000 dentry, inode);
2001 valid = 1;
2002 } else if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
2003 valid = 1;
2004 } else {
2005 valid = dentry_lease_is_valid(dentry, flags);
2006 if (valid == -ECHILD)
2007 return valid;
2008 if (valid || dir_lease_is_valid(dir, dentry, mdsc)) {
2009 if (inode)
2010 valid = ceph_is_any_caps(inode);
2011 else
2012 valid = 1;
2013 }
2014 }
2015
2016 if (!valid) {
2017 struct ceph_mds_request *req;
2018 int op, err;
2019 u32 mask;
2020
2021 if (flags & LOOKUP_RCU)
2022 return -ECHILD;
2023
2024 percpu_counter_inc(&mdsc->metric.d_lease_mis);
2025
2026 op = ceph_snap(dir) == CEPH_SNAPDIR ?
2027 CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_LOOKUP;
2028 req = ceph_mdsc_create_request(mdsc, op, USE_ANY_MDS);
2029 if (!IS_ERR(req)) {
2030 req->r_dentry = dget(dentry);
2031 req->r_num_caps = 2;
2032 req->r_parent = dir;
2033 ihold(dir);
2034
2035 req->r_dname = name;
2036
2037 mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
2038 if (ceph_security_xattr_wanted(dir))
2039 mask |= CEPH_CAP_XATTR_SHARED;
2040 req->r_args.getattr.mask = cpu_to_le32(mask);
2041
2042 err = ceph_mdsc_do_request(mdsc, NULL, req);
2043 switch (err) {
2044 case 0:
2045 if (d_really_is_positive(dentry) &&
2046 d_inode(dentry) == req->r_target_inode)
2047 valid = 1;
2048 break;
2049 case -ENOENT:
2050 if (d_really_is_negative(dentry))
2051 valid = 1;
2052 fallthrough;
2053 default:
2054 break;
2055 }
2056 ceph_mdsc_put_request(req);
2057 doutc(cl, "%p '%pd', lookup result=%d\n", dentry,
2058 dentry, err);
2059 }
2060 } else {
2061 percpu_counter_inc(&mdsc->metric.d_lease_hit);
2062 }
2063
2064 doutc(cl, "%p '%pd' %s\n", dentry, dentry, valid ? "valid" : "invalid");
2065 if (!valid)
2066 ceph_dir_clear_complete(dir);
2067 return valid;
2068}
2069
2070/*
2071 * Delete unused dentry that doesn't have valid lease
2072 *
2073 * Called under dentry->d_lock.
2074 */
2075static int ceph_d_delete(const struct dentry *dentry)
2076{
2077 struct ceph_dentry_info *di;
2078
2079 /* won't release caps */
2080 if (d_really_is_negative(dentry))
2081 return 0;
2082 if (ceph_snap(d_inode(dentry)) != CEPH_NOSNAP)
2083 return 0;
2084 /* valid lease? */
2085 di = ceph_dentry(dentry);
2086 if (di) {
2087 if (__dentry_lease_is_valid(di))
2088 return 0;
2089 if (__dir_lease_try_check(dentry))
2090 return 0;
2091 }
2092 return 1;
2093}
2094
2095/*
2096 * Release our ceph_dentry_info.
2097 */
2098static void ceph_d_release(struct dentry *dentry)
2099{
2100 struct ceph_dentry_info *di = ceph_dentry(dentry);
2101 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb);
2102
2103 doutc(fsc->client, "dentry %p '%pd'\n", dentry, dentry);
2104
2105 atomic64_dec(&fsc->mdsc->metric.total_dentries);
2106
2107 spin_lock(&dentry->d_lock);
2108 __dentry_lease_unlist(di);
2109 dentry->d_fsdata = NULL;
2110 spin_unlock(&dentry->d_lock);
2111
2112 ceph_put_mds_session(di->lease_session);
2113 kmem_cache_free(ceph_dentry_cachep, di);
2114}
2115
2116/*
2117 * When the VFS prunes a dentry from the cache, we need to clear the
2118 * complete flag on the parent directory.
2119 *
2120 * Called under dentry->d_lock.
2121 */
2122static void ceph_d_prune(struct dentry *dentry)
2123{
2124 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dentry->d_sb);
2125 struct ceph_client *cl = mdsc->fsc->client;
2126 struct ceph_inode_info *dir_ci;
2127 struct ceph_dentry_info *di;
2128
2129 doutc(cl, "dentry %p '%pd'\n", dentry, dentry);
2130
2131 /* do we have a valid parent? */
2132 if (IS_ROOT(dentry))
2133 return;
2134
2135 /* we hold d_lock, so d_parent is stable */
2136 dir_ci = ceph_inode(d_inode(dentry->d_parent));
2137 if (dir_ci->i_vino.snap == CEPH_SNAPDIR)
2138 return;
2139
2140 /* who calls d_delete() should also disable dcache readdir */
2141 if (d_really_is_negative(dentry))
2142 return;
2143
2144 /* d_fsdata does not get cleared until d_release */
2145 if (!d_unhashed(dentry)) {
2146 __ceph_dir_clear_complete(dir_ci);
2147 return;
2148 }
2149
2150 /* Disable dcache readdir just in case that someone called d_drop()
2151 * or d_invalidate(), but MDS didn't revoke CEPH_CAP_FILE_SHARED
2152 * properly (dcache readdir is still enabled) */
2153 di = ceph_dentry(dentry);
2154 if (di->offset > 0 &&
2155 di->lease_shared_gen == atomic_read(&dir_ci->i_shared_gen))
2156 __ceph_dir_clear_ordered(dir_ci);
2157}
2158
2159/*
2160 * read() on a dir. This weird interface hack only works if mounted
2161 * with '-o dirstat'.
2162 */
2163static ssize_t ceph_read_dir(struct file *file, char __user *buf, size_t size,
2164 loff_t *ppos)
2165{
2166 struct ceph_dir_file_info *dfi = file->private_data;
2167 struct inode *inode = file_inode(file);
2168 struct ceph_inode_info *ci = ceph_inode(inode);
2169 int left;
2170 const int bufsize = 1024;
2171
2172 if (!ceph_test_mount_opt(ceph_sb_to_fs_client(inode->i_sb), DIRSTAT))
2173 return -EISDIR;
2174
2175 if (!dfi->dir_info) {
2176 dfi->dir_info = kmalloc(bufsize, GFP_KERNEL);
2177 if (!dfi->dir_info)
2178 return -ENOMEM;
2179 dfi->dir_info_len =
2180 snprintf(dfi->dir_info, bufsize,
2181 "entries: %20lld\n"
2182 " files: %20lld\n"
2183 " subdirs: %20lld\n"
2184 "rentries: %20lld\n"
2185 " rfiles: %20lld\n"
2186 " rsubdirs: %20lld\n"
2187 "rbytes: %20lld\n"
2188 "rctime: %ptSp\n",
2189 ci->i_files + ci->i_subdirs,
2190 ci->i_files,
2191 ci->i_subdirs,
2192 ci->i_rfiles + ci->i_rsubdirs,
2193 ci->i_rfiles,
2194 ci->i_rsubdirs,
2195 ci->i_rbytes,
2196 &ci->i_rctime);
2197 }
2198
2199 if (*ppos >= dfi->dir_info_len)
2200 return 0;
2201 size = min_t(unsigned, size, dfi->dir_info_len-*ppos);
2202 left = copy_to_user(buf, dfi->dir_info + *ppos, size);
2203 if (left == size)
2204 return -EFAULT;
2205 *ppos += (size - left);
2206 return size - left;
2207}
2208
2209
2210
2211/*
2212 * Return name hash for a given dentry. This is dependent on
2213 * the parent directory's hash function.
2214 */
2215unsigned ceph_dentry_hash(struct inode *dir, struct dentry *dn)
2216{
2217 struct ceph_inode_info *dci = ceph_inode(dir);
2218 unsigned hash;
2219
2220 switch (dci->i_dir_layout.dl_dir_hash) {
2221 case 0: /* for backward compat */
2222 case CEPH_STR_HASH_LINUX:
2223 return dn->d_name.hash;
2224
2225 default:
2226 spin_lock(&dn->d_lock);
2227 hash = ceph_str_hash(dci->i_dir_layout.dl_dir_hash,
2228 dn->d_name.name, dn->d_name.len);
2229 spin_unlock(&dn->d_lock);
2230 return hash;
2231 }
2232}
2233
2234WRAP_DIR_ITER(ceph_readdir) // FIXME!
2235const struct file_operations ceph_dir_fops = {
2236 .read = ceph_read_dir,
2237 .iterate_shared = shared_ceph_readdir,
2238 .llseek = ceph_dir_llseek,
2239 .open = ceph_open,
2240 .release = ceph_release,
2241 .unlocked_ioctl = ceph_ioctl,
2242 .compat_ioctl = compat_ptr_ioctl,
2243 .fsync = ceph_fsync,
2244 .lock = ceph_lock,
2245 .flock = ceph_flock,
2246};
2247
2248const struct file_operations ceph_snapdir_fops = {
2249 .iterate_shared = shared_ceph_readdir,
2250 .llseek = ceph_dir_llseek,
2251 .open = ceph_open,
2252 .release = ceph_release,
2253};
2254
2255const struct inode_operations ceph_dir_iops = {
2256 .lookup = ceph_lookup,
2257 .permission = ceph_permission,
2258 .getattr = ceph_getattr,
2259 .setattr = ceph_setattr,
2260 .listxattr = ceph_listxattr,
2261 .get_inode_acl = ceph_get_acl,
2262 .set_acl = ceph_set_acl,
2263 .mknod = ceph_mknod,
2264 .symlink = ceph_symlink,
2265 .mkdir = ceph_mkdir,
2266 .link = ceph_link,
2267 .unlink = ceph_unlink,
2268 .rmdir = ceph_unlink,
2269 .rename = ceph_rename,
2270 .create = ceph_create,
2271 .atomic_open = ceph_atomic_open,
2272};
2273
2274const struct inode_operations ceph_snapdir_iops = {
2275 .lookup = ceph_lookup,
2276 .permission = ceph_permission,
2277 .getattr = ceph_getattr,
2278 .mkdir = ceph_mkdir,
2279 .rmdir = ceph_unlink,
2280 .rename = ceph_rename,
2281};
2282
2283const struct dentry_operations ceph_dentry_ops = {
2284 .d_revalidate = ceph_d_revalidate,
2285 .d_delete = ceph_d_delete,
2286 .d_release = ceph_d_release,
2287 .d_prune = ceph_d_prune,
2288 .d_init = ceph_d_init,
2289};