1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
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>
11 #include "mds_client.h"
14 * Directory operations: readdir, lookup, create, link, unlink,
19 * Ceph MDS operations are specified in terms of a base ino and
20 * relative path. Thus, the client can specify an operation on a
21 * specific inode (e.g., a getattr due to fstat(2)), or as a path
22 * relative to, say, the root directory.
24 * Normally, we limit ourselves to strict inode ops (no path component)
25 * or dentry operations (a single path component relative to an ino). The
26 * exception to this is open_root_dentry(), which will open the mount
30 const struct dentry_operations ceph_dentry_ops
;
32 static bool __dentry_lease_is_valid(struct ceph_dentry_info
*di
);
33 static int __dir_lease_try_check(const struct dentry
*dentry
);
36 * Initialize ceph dentry state.
38 static int ceph_d_init(struct dentry
*dentry
)
40 struct ceph_dentry_info
*di
;
41 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dentry
->d_sb
);
43 di
= kmem_cache_zalloc(ceph_dentry_cachep
, GFP_KERNEL
);
45 return -ENOMEM
; /* oh well */
48 di
->lease_session
= NULL
;
50 dentry
->d_fsdata
= di
;
51 INIT_LIST_HEAD(&di
->lease_list
);
53 atomic64_inc(&mdsc
->metric
.total_dentries
);
59 * for f_pos for readdir:
61 * (0xff << 52) | ((24 bits hash) << 28) |
62 * (the nth entry has hash collision);
64 * ((frag value) << 28) | (the nth entry in frag);
66 #define OFFSET_BITS 28
67 #define OFFSET_MASK ((1 << OFFSET_BITS) - 1)
68 #define HASH_ORDER (0xffull << (OFFSET_BITS + 24))
69 loff_t
ceph_make_fpos(unsigned high
, unsigned off
, bool hash_order
)
71 loff_t fpos
= ((loff_t
)high
<< 28) | (loff_t
)off
;
77 static bool is_hash_order(loff_t p
)
79 return (p
& HASH_ORDER
) == HASH_ORDER
;
82 static unsigned fpos_frag(loff_t p
)
84 return p
>> OFFSET_BITS
;
87 static unsigned fpos_hash(loff_t p
)
89 return ceph_frag_value(fpos_frag(p
));
92 static unsigned fpos_off(loff_t p
)
94 return p
& OFFSET_MASK
;
97 static int fpos_cmp(loff_t l
, loff_t r
)
99 int v
= ceph_frag_compare(fpos_frag(l
), fpos_frag(r
));
102 return (int)(fpos_off(l
) - fpos_off(r
));
106 * make note of the last dentry we read, so we can
107 * continue at the same lexicographical point,
108 * regardless of what dir changes take place on the
111 static int note_last_dentry(struct ceph_dir_file_info
*dfi
, const char *name
,
112 int len
, unsigned next_offset
)
114 char *buf
= kmalloc(len
+1, GFP_KERNEL
);
117 kfree(dfi
->last_name
);
118 dfi
->last_name
= buf
;
119 memcpy(dfi
->last_name
, name
, len
);
120 dfi
->last_name
[len
] = 0;
121 dfi
->next_offset
= next_offset
;
122 dout("note_last_dentry '%s'\n", dfi
->last_name
);
127 static struct dentry
*
128 __dcache_find_get_entry(struct dentry
*parent
, u64 idx
,
129 struct ceph_readdir_cache_control
*cache_ctl
)
131 struct inode
*dir
= d_inode(parent
);
132 struct dentry
*dentry
;
133 unsigned idx_mask
= (PAGE_SIZE
/ sizeof(struct dentry
*)) - 1;
134 loff_t ptr_pos
= idx
* sizeof(struct dentry
*);
135 pgoff_t ptr_pgoff
= ptr_pos
>> PAGE_SHIFT
;
137 if (ptr_pos
>= i_size_read(dir
))
140 if (!cache_ctl
->page
|| ptr_pgoff
!= page_index(cache_ctl
->page
)) {
141 ceph_readdir_cache_release(cache_ctl
);
142 cache_ctl
->page
= find_lock_page(&dir
->i_data
, ptr_pgoff
);
143 if (!cache_ctl
->page
) {
144 dout(" page %lu not found\n", ptr_pgoff
);
145 return ERR_PTR(-EAGAIN
);
147 /* reading/filling the cache are serialized by
148 i_mutex, no need to use page lock */
149 unlock_page(cache_ctl
->page
);
150 cache_ctl
->dentries
= kmap(cache_ctl
->page
);
153 cache_ctl
->index
= idx
& idx_mask
;
156 spin_lock(&parent
->d_lock
);
157 /* check i_size again here, because empty directory can be
158 * marked as complete while not holding the i_mutex. */
159 if (ceph_dir_is_complete_ordered(dir
) && ptr_pos
< i_size_read(dir
))
160 dentry
= cache_ctl
->dentries
[cache_ctl
->index
];
163 spin_unlock(&parent
->d_lock
);
164 if (dentry
&& !lockref_get_not_dead(&dentry
->d_lockref
))
167 return dentry
? : ERR_PTR(-EAGAIN
);
171 * When possible, we try to satisfy a readdir by peeking at the
172 * dcache. We make this work by carefully ordering dentries on
173 * d_child when we initially get results back from the MDS, and
174 * falling back to a "normal" sync readdir if any dentries in the dir
177 * Complete dir indicates that we have all dentries in the dir. It is
178 * defined IFF we hold CEPH_CAP_FILE_SHARED (which will be revoked by
179 * the MDS if/when the directory is modified).
181 static int __dcache_readdir(struct file
*file
, struct dir_context
*ctx
,
184 struct ceph_dir_file_info
*dfi
= file
->private_data
;
185 struct dentry
*parent
= file
->f_path
.dentry
;
186 struct inode
*dir
= d_inode(parent
);
187 struct dentry
*dentry
, *last
= NULL
;
188 struct ceph_dentry_info
*di
;
189 struct ceph_readdir_cache_control cache_ctl
= {};
193 dout("__dcache_readdir %p v%u at %llx\n", dir
, (unsigned)shared_gen
, ctx
->pos
);
195 /* search start position */
197 u64 count
= div_u64(i_size_read(dir
), sizeof(struct dentry
*));
199 u64 step
= count
>> 1;
200 dentry
= __dcache_find_get_entry(parent
, idx
+ step
,
203 /* use linar search */
207 if (IS_ERR(dentry
)) {
208 err
= PTR_ERR(dentry
);
211 di
= ceph_dentry(dentry
);
212 spin_lock(&dentry
->d_lock
);
213 if (fpos_cmp(di
->offset
, ctx
->pos
) < 0) {
219 spin_unlock(&dentry
->d_lock
);
223 dout("__dcache_readdir %p cache idx %llu\n", dir
, idx
);
228 bool emit_dentry
= false;
229 dentry
= __dcache_find_get_entry(parent
, idx
++, &cache_ctl
);
231 dfi
->file_info
.flags
|= CEPH_F_ATEND
;
235 if (IS_ERR(dentry
)) {
236 err
= PTR_ERR(dentry
);
240 spin_lock(&dentry
->d_lock
);
241 di
= ceph_dentry(dentry
);
242 if (d_unhashed(dentry
) ||
243 d_really_is_negative(dentry
) ||
244 di
->lease_shared_gen
!= shared_gen
) {
245 spin_unlock(&dentry
->d_lock
);
250 if (fpos_cmp(ctx
->pos
, di
->offset
) <= 0) {
251 __ceph_dentry_dir_lease_touch(di
);
254 spin_unlock(&dentry
->d_lock
);
257 dout(" %llx dentry %p %pd %p\n", di
->offset
,
258 dentry
, dentry
, d_inode(dentry
));
259 ctx
->pos
= di
->offset
;
260 if (!dir_emit(ctx
, dentry
->d_name
.name
,
261 dentry
->d_name
.len
, ceph_present_inode(d_inode(dentry
)),
262 d_inode(dentry
)->i_mode
>> 12)) {
277 ceph_readdir_cache_release(&cache_ctl
);
280 di
= ceph_dentry(last
);
281 ret
= note_last_dentry(dfi
, last
->d_name
.name
, last
->d_name
.len
,
282 fpos_off(di
->offset
) + 1);
286 /* last_name no longer match cache index */
287 if (dfi
->readdir_cache_idx
>= 0) {
288 dfi
->readdir_cache_idx
= -1;
289 dfi
->dir_release_count
= 0;
295 static bool need_send_readdir(struct ceph_dir_file_info
*dfi
, loff_t pos
)
297 if (!dfi
->last_readdir
)
299 if (is_hash_order(pos
))
300 return !ceph_frag_contains_value(dfi
->frag
, fpos_hash(pos
));
302 return dfi
->frag
!= fpos_frag(pos
);
305 static int ceph_readdir(struct file
*file
, struct dir_context
*ctx
)
307 struct ceph_dir_file_info
*dfi
= file
->private_data
;
308 struct inode
*inode
= file_inode(file
);
309 struct ceph_inode_info
*ci
= ceph_inode(inode
);
310 struct ceph_fs_client
*fsc
= ceph_inode_to_client(inode
);
311 struct ceph_mds_client
*mdsc
= fsc
->mdsc
;
315 struct ceph_mds_reply_info_parsed
*rinfo
;
317 dout("readdir %p file %p pos %llx\n", inode
, file
, ctx
->pos
);
318 if (dfi
->file_info
.flags
& CEPH_F_ATEND
)
321 /* always start with . and .. */
323 dout("readdir off 0 -> '.'\n");
324 if (!dir_emit(ctx
, ".", 1, ceph_present_inode(inode
),
325 inode
->i_mode
>> 12))
331 struct dentry
*dentry
= file
->f_path
.dentry
;
333 spin_lock(&dentry
->d_lock
);
334 ino
= ceph_present_inode(dentry
->d_parent
->d_inode
);
335 spin_unlock(&dentry
->d_lock
);
337 dout("readdir off 1 -> '..'\n");
338 if (!dir_emit(ctx
, "..", 2, ino
, inode
->i_mode
>> 12))
343 spin_lock(&ci
->i_ceph_lock
);
344 /* request Fx cap. if have Fx, we don't need to release Fs cap
345 * for later create/unlink. */
346 __ceph_touch_fmode(ci
, mdsc
, CEPH_FILE_MODE_WR
);
347 /* can we use the dcache? */
348 if (ceph_test_mount_opt(fsc
, DCACHE
) &&
349 !ceph_test_mount_opt(fsc
, NOASYNCREADDIR
) &&
350 ceph_snap(inode
) != CEPH_SNAPDIR
&&
351 __ceph_dir_is_complete_ordered(ci
) &&
352 __ceph_caps_issued_mask_metric(ci
, CEPH_CAP_FILE_SHARED
, 1)) {
353 int shared_gen
= atomic_read(&ci
->i_shared_gen
);
355 spin_unlock(&ci
->i_ceph_lock
);
356 err
= __dcache_readdir(file
, ctx
, shared_gen
);
360 spin_unlock(&ci
->i_ceph_lock
);
363 /* proceed with a normal readdir */
365 /* do we have the correct frag content buffered? */
366 if (need_send_readdir(dfi
, ctx
->pos
)) {
367 struct ceph_mds_request
*req
;
368 int op
= ceph_snap(inode
) == CEPH_SNAPDIR
?
369 CEPH_MDS_OP_LSSNAP
: CEPH_MDS_OP_READDIR
;
371 /* discard old result, if any */
372 if (dfi
->last_readdir
) {
373 ceph_mdsc_put_request(dfi
->last_readdir
);
374 dfi
->last_readdir
= NULL
;
377 if (is_hash_order(ctx
->pos
)) {
378 /* fragtree isn't always accurate. choose frag
379 * based on previous reply when possible. */
380 if (frag
== (unsigned)-1)
381 frag
= ceph_choose_frag(ci
, fpos_hash(ctx
->pos
),
384 frag
= fpos_frag(ctx
->pos
);
387 dout("readdir fetching %llx.%llx frag %x offset '%s'\n",
388 ceph_vinop(inode
), frag
, dfi
->last_name
);
389 req
= ceph_mdsc_create_request(mdsc
, op
, USE_AUTH_MDS
);
392 err
= ceph_alloc_readdir_reply_buffer(req
, inode
);
394 ceph_mdsc_put_request(req
);
397 /* hints to request -> mds selection code */
398 req
->r_direct_mode
= USE_AUTH_MDS
;
399 if (op
== CEPH_MDS_OP_READDIR
) {
400 req
->r_direct_hash
= ceph_frag_value(frag
);
401 __set_bit(CEPH_MDS_R_DIRECT_IS_HASH
, &req
->r_req_flags
);
402 req
->r_inode_drop
= CEPH_CAP_FILE_EXCL
;
404 if (dfi
->last_name
) {
405 req
->r_path2
= kstrdup(dfi
->last_name
, GFP_KERNEL
);
407 ceph_mdsc_put_request(req
);
410 } else if (is_hash_order(ctx
->pos
)) {
411 req
->r_args
.readdir
.offset_hash
=
412 cpu_to_le32(fpos_hash(ctx
->pos
));
415 req
->r_dir_release_cnt
= dfi
->dir_release_count
;
416 req
->r_dir_ordered_cnt
= dfi
->dir_ordered_count
;
417 req
->r_readdir_cache_idx
= dfi
->readdir_cache_idx
;
418 req
->r_readdir_offset
= dfi
->next_offset
;
419 req
->r_args
.readdir
.frag
= cpu_to_le32(frag
);
420 req
->r_args
.readdir
.flags
=
421 cpu_to_le16(CEPH_READDIR_REPLY_BITFLAGS
);
423 req
->r_inode
= inode
;
425 req
->r_dentry
= dget(file
->f_path
.dentry
);
426 err
= ceph_mdsc_do_request(mdsc
, NULL
, req
);
428 ceph_mdsc_put_request(req
);
431 dout("readdir got and parsed readdir result=%d on "
432 "frag %x, end=%d, complete=%d, hash_order=%d\n",
434 (int)req
->r_reply_info
.dir_end
,
435 (int)req
->r_reply_info
.dir_complete
,
436 (int)req
->r_reply_info
.hash_order
);
438 rinfo
= &req
->r_reply_info
;
439 if (le32_to_cpu(rinfo
->dir_dir
->frag
) != frag
) {
440 frag
= le32_to_cpu(rinfo
->dir_dir
->frag
);
441 if (!rinfo
->hash_order
) {
442 dfi
->next_offset
= req
->r_readdir_offset
;
443 /* adjust ctx->pos to beginning of frag */
444 ctx
->pos
= ceph_make_fpos(frag
,
451 dfi
->last_readdir
= req
;
453 if (test_bit(CEPH_MDS_R_DID_PREPOPULATE
, &req
->r_req_flags
)) {
454 dfi
->readdir_cache_idx
= req
->r_readdir_cache_idx
;
455 if (dfi
->readdir_cache_idx
< 0) {
456 /* preclude from marking dir ordered */
457 dfi
->dir_ordered_count
= 0;
458 } else if (ceph_frag_is_leftmost(frag
) &&
459 dfi
->next_offset
== 2) {
460 /* note dir version at start of readdir so
461 * we can tell if any dentries get dropped */
462 dfi
->dir_release_count
= req
->r_dir_release_cnt
;
463 dfi
->dir_ordered_count
= req
->r_dir_ordered_cnt
;
466 dout("readdir !did_prepopulate\n");
467 /* disable readdir cache */
468 dfi
->readdir_cache_idx
= -1;
469 /* preclude from marking dir complete */
470 dfi
->dir_release_count
= 0;
473 /* note next offset and last dentry name */
474 if (rinfo
->dir_nr
> 0) {
475 struct ceph_mds_reply_dir_entry
*rde
=
476 rinfo
->dir_entries
+ (rinfo
->dir_nr
-1);
477 unsigned next_offset
= req
->r_reply_info
.dir_end
?
478 2 : (fpos_off(rde
->offset
) + 1);
479 err
= note_last_dentry(dfi
, rde
->name
, rde
->name_len
,
483 } else if (req
->r_reply_info
.dir_end
) {
484 dfi
->next_offset
= 2;
489 rinfo
= &dfi
->last_readdir
->r_reply_info
;
490 dout("readdir frag %x num %d pos %llx chunk first %llx\n",
491 dfi
->frag
, rinfo
->dir_nr
, ctx
->pos
,
492 rinfo
->dir_nr
? rinfo
->dir_entries
[0].offset
: 0LL);
495 /* search start position */
496 if (rinfo
->dir_nr
> 0) {
497 int step
, nr
= rinfo
->dir_nr
;
500 if (rinfo
->dir_entries
[i
+ step
].offset
< ctx
->pos
) {
508 for (; i
< rinfo
->dir_nr
; i
++) {
509 struct ceph_mds_reply_dir_entry
*rde
= rinfo
->dir_entries
+ i
;
511 BUG_ON(rde
->offset
< ctx
->pos
);
513 ctx
->pos
= rde
->offset
;
514 dout("readdir (%d/%d) -> %llx '%.*s' %p\n",
515 i
, rinfo
->dir_nr
, ctx
->pos
,
516 rde
->name_len
, rde
->name
, &rde
->inode
.in
);
518 BUG_ON(!rde
->inode
.in
);
520 if (!dir_emit(ctx
, rde
->name
, rde
->name_len
,
521 ceph_present_ino(inode
->i_sb
, le64_to_cpu(rde
->inode
.in
->ino
)),
522 le32_to_cpu(rde
->inode
.in
->mode
) >> 12)) {
523 dout("filldir stopping us...\n");
529 ceph_mdsc_put_request(dfi
->last_readdir
);
530 dfi
->last_readdir
= NULL
;
532 if (dfi
->next_offset
> 2) {
538 if (!ceph_frag_is_rightmost(dfi
->frag
)) {
539 frag
= ceph_frag_next(dfi
->frag
);
540 if (is_hash_order(ctx
->pos
)) {
541 loff_t new_pos
= ceph_make_fpos(ceph_frag_value(frag
),
542 dfi
->next_offset
, true);
543 if (new_pos
> ctx
->pos
)
547 ctx
->pos
= ceph_make_fpos(frag
, dfi
->next_offset
,
549 kfree(dfi
->last_name
);
550 dfi
->last_name
= NULL
;
552 dout("readdir next frag is %x\n", frag
);
555 dfi
->file_info
.flags
|= CEPH_F_ATEND
;
558 * if dir_release_count still matches the dir, no dentries
559 * were released during the whole readdir, and we should have
560 * the complete dir contents in our cache.
562 if (atomic64_read(&ci
->i_release_count
) ==
563 dfi
->dir_release_count
) {
564 spin_lock(&ci
->i_ceph_lock
);
565 if (dfi
->dir_ordered_count
==
566 atomic64_read(&ci
->i_ordered_count
)) {
567 dout(" marking %p complete and ordered\n", inode
);
568 /* use i_size to track number of entries in
570 BUG_ON(dfi
->readdir_cache_idx
< 0);
571 i_size_write(inode
, dfi
->readdir_cache_idx
*
572 sizeof(struct dentry
*));
574 dout(" marking %p complete\n", inode
);
576 __ceph_dir_set_complete(ci
, dfi
->dir_release_count
,
577 dfi
->dir_ordered_count
);
578 spin_unlock(&ci
->i_ceph_lock
);
581 dout("readdir %p file %p done.\n", inode
, file
);
585 static void reset_readdir(struct ceph_dir_file_info
*dfi
)
587 if (dfi
->last_readdir
) {
588 ceph_mdsc_put_request(dfi
->last_readdir
);
589 dfi
->last_readdir
= NULL
;
591 kfree(dfi
->last_name
);
592 dfi
->last_name
= NULL
;
593 dfi
->dir_release_count
= 0;
594 dfi
->readdir_cache_idx
= -1;
595 dfi
->next_offset
= 2; /* compensate for . and .. */
596 dfi
->file_info
.flags
&= ~CEPH_F_ATEND
;
600 * discard buffered readdir content on seekdir(0), or seek to new frag,
601 * or seek prior to current chunk
603 static bool need_reset_readdir(struct ceph_dir_file_info
*dfi
, loff_t new_pos
)
605 struct ceph_mds_reply_info_parsed
*rinfo
;
609 if (is_hash_order(new_pos
)) {
610 /* no need to reset last_name for a forward seek when
611 * dentries are sotred in hash order */
612 } else if (dfi
->frag
!= fpos_frag(new_pos
)) {
615 rinfo
= dfi
->last_readdir
? &dfi
->last_readdir
->r_reply_info
: NULL
;
616 if (!rinfo
|| !rinfo
->dir_nr
)
618 chunk_offset
= rinfo
->dir_entries
[0].offset
;
619 return new_pos
< chunk_offset
||
620 is_hash_order(new_pos
) != is_hash_order(chunk_offset
);
623 static loff_t
ceph_dir_llseek(struct file
*file
, loff_t offset
, int whence
)
625 struct ceph_dir_file_info
*dfi
= file
->private_data
;
626 struct inode
*inode
= file
->f_mapping
->host
;
633 offset
+= file
->f_pos
;
637 retval
= -EOPNOTSUPP
;
643 if (need_reset_readdir(dfi
, offset
)) {
644 dout("dir_llseek dropping %p content\n", file
);
646 } else if (is_hash_order(offset
) && offset
> file
->f_pos
) {
647 /* for hash offset, we don't know if a forward seek
648 * is within same frag */
649 dfi
->dir_release_count
= 0;
650 dfi
->readdir_cache_idx
= -1;
653 if (offset
!= file
->f_pos
) {
654 file
->f_pos
= offset
;
656 dfi
->file_info
.flags
&= ~CEPH_F_ATEND
;
666 * Handle lookups for the hidden .snap directory.
668 int ceph_handle_snapdir(struct ceph_mds_request
*req
,
669 struct dentry
*dentry
, int err
)
671 struct ceph_fs_client
*fsc
= ceph_sb_to_client(dentry
->d_sb
);
672 struct inode
*parent
= d_inode(dentry
->d_parent
); /* we hold i_mutex */
675 if (err
== -ENOENT
&&
676 ceph_snap(parent
) == CEPH_NOSNAP
&&
677 strcmp(dentry
->d_name
.name
,
678 fsc
->mount_options
->snapdir_name
) == 0) {
679 struct inode
*inode
= ceph_get_snapdir(parent
);
680 dout("ENOENT on snapdir %p '%pd', linking to snapdir %p\n",
681 dentry
, dentry
, inode
);
682 BUG_ON(!d_unhashed(dentry
));
683 d_add(dentry
, inode
);
690 * Figure out final result of a lookup/open request.
692 * Mainly, make sure we return the final req->r_dentry (if it already
693 * existed) in place of the original VFS-provided dentry when they
696 * Gracefully handle the case where the MDS replies with -ENOENT and
697 * no trace (which it may do, at its discretion, e.g., if it doesn't
698 * care to issue a lease on the negative dentry).
700 struct dentry
*ceph_finish_lookup(struct ceph_mds_request
*req
,
701 struct dentry
*dentry
, int err
)
703 if (err
== -ENOENT
) {
706 if (!req
->r_reply_info
.head
->is_dentry
) {
707 dout("ENOENT and no trace, dentry %p inode %p\n",
708 dentry
, d_inode(dentry
));
709 if (d_really_is_positive(dentry
)) {
718 dentry
= ERR_PTR(err
);
719 else if (dentry
!= req
->r_dentry
)
720 dentry
= dget(req
->r_dentry
); /* we got spliced */
726 static bool is_root_ceph_dentry(struct inode
*inode
, struct dentry
*dentry
)
728 return ceph_ino(inode
) == CEPH_INO_ROOT
&&
729 strncmp(dentry
->d_name
.name
, ".ceph", 5) == 0;
733 * Look up a single dir entry. If there is a lookup intent, inform
734 * the MDS so that it gets our 'caps wanted' value in a single op.
736 static struct dentry
*ceph_lookup(struct inode
*dir
, struct dentry
*dentry
,
739 struct ceph_fs_client
*fsc
= ceph_sb_to_client(dir
->i_sb
);
740 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dir
->i_sb
);
741 struct ceph_mds_request
*req
;
746 dout("lookup %p dentry %p '%pd'\n",
747 dir
, dentry
, dentry
);
749 if (dentry
->d_name
.len
> NAME_MAX
)
750 return ERR_PTR(-ENAMETOOLONG
);
752 /* can we conclude ENOENT locally? */
753 if (d_really_is_negative(dentry
)) {
754 struct ceph_inode_info
*ci
= ceph_inode(dir
);
755 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
757 spin_lock(&ci
->i_ceph_lock
);
758 dout(" dir %p flags are 0x%lx\n", dir
, ci
->i_ceph_flags
);
759 if (strncmp(dentry
->d_name
.name
,
760 fsc
->mount_options
->snapdir_name
,
761 dentry
->d_name
.len
) &&
762 !is_root_ceph_dentry(dir
, dentry
) &&
763 ceph_test_mount_opt(fsc
, DCACHE
) &&
764 __ceph_dir_is_complete(ci
) &&
765 __ceph_caps_issued_mask_metric(ci
, CEPH_CAP_FILE_SHARED
, 1)) {
766 __ceph_touch_fmode(ci
, mdsc
, CEPH_FILE_MODE_RD
);
767 spin_unlock(&ci
->i_ceph_lock
);
768 dout(" dir %p complete, -ENOENT\n", dir
);
770 di
->lease_shared_gen
= atomic_read(&ci
->i_shared_gen
);
773 spin_unlock(&ci
->i_ceph_lock
);
776 op
= ceph_snap(dir
) == CEPH_SNAPDIR
?
777 CEPH_MDS_OP_LOOKUPSNAP
: CEPH_MDS_OP_LOOKUP
;
778 req
= ceph_mdsc_create_request(mdsc
, op
, USE_ANY_MDS
);
780 return ERR_CAST(req
);
781 req
->r_dentry
= dget(dentry
);
784 mask
= CEPH_STAT_CAP_INODE
| CEPH_CAP_AUTH_SHARED
;
785 if (ceph_security_xattr_wanted(dir
))
786 mask
|= CEPH_CAP_XATTR_SHARED
;
787 req
->r_args
.getattr
.mask
= cpu_to_le32(mask
);
790 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
791 err
= ceph_mdsc_do_request(mdsc
, NULL
, req
);
792 err
= ceph_handle_snapdir(req
, dentry
, err
);
793 dentry
= ceph_finish_lookup(req
, dentry
, err
);
794 ceph_mdsc_put_request(req
); /* will dput(dentry) */
795 dout("lookup result=%p\n", dentry
);
800 * If we do a create but get no trace back from the MDS, follow up with
801 * a lookup (the VFS expects us to link up the provided dentry).
803 int ceph_handle_notrace_create(struct inode
*dir
, struct dentry
*dentry
)
805 struct dentry
*result
= ceph_lookup(dir
, dentry
, 0);
807 if (result
&& !IS_ERR(result
)) {
809 * We created the item, then did a lookup, and found
810 * it was already linked to another inode we already
811 * had in our cache (and thus got spliced). To not
812 * confuse VFS (especially when inode is a directory),
813 * we don't link our dentry to that inode, return an
816 * This event should be rare and it happens only when
817 * we talk to old MDS. Recent MDS does not send traceless
818 * reply for request that creates new inode.
823 return PTR_ERR(result
);
826 static int ceph_mknod(struct inode
*dir
, struct dentry
*dentry
,
827 umode_t mode
, dev_t rdev
)
829 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dir
->i_sb
);
830 struct ceph_mds_request
*req
;
831 struct ceph_acl_sec_ctx as_ctx
= {};
834 if (ceph_snap(dir
) != CEPH_NOSNAP
)
837 if (ceph_quota_is_max_files_exceeded(dir
)) {
842 err
= ceph_pre_init_acls(dir
, &mode
, &as_ctx
);
845 err
= ceph_security_init_secctx(dentry
, mode
, &as_ctx
);
849 dout("mknod in dir %p dentry %p mode 0%ho rdev %d\n",
850 dir
, dentry
, mode
, rdev
);
851 req
= ceph_mdsc_create_request(mdsc
, CEPH_MDS_OP_MKNOD
, USE_AUTH_MDS
);
856 req
->r_dentry
= dget(dentry
);
859 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
860 req
->r_args
.mknod
.mode
= cpu_to_le32(mode
);
861 req
->r_args
.mknod
.rdev
= cpu_to_le32(rdev
);
862 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
| CEPH_CAP_AUTH_EXCL
;
863 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
864 if (as_ctx
.pagelist
) {
865 req
->r_pagelist
= as_ctx
.pagelist
;
866 as_ctx
.pagelist
= NULL
;
868 err
= ceph_mdsc_do_request(mdsc
, dir
, req
);
869 if (!err
&& !req
->r_reply_info
.head
->is_dentry
)
870 err
= ceph_handle_notrace_create(dir
, dentry
);
871 ceph_mdsc_put_request(req
);
874 ceph_init_inode_acls(d_inode(dentry
), &as_ctx
);
877 ceph_release_acl_sec_ctx(&as_ctx
);
881 static int ceph_create(struct inode
*dir
, struct dentry
*dentry
, umode_t mode
,
884 return ceph_mknod(dir
, dentry
, mode
, 0);
887 static int ceph_symlink(struct inode
*dir
, struct dentry
*dentry
,
890 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dir
->i_sb
);
891 struct ceph_mds_request
*req
;
892 struct ceph_acl_sec_ctx as_ctx
= {};
895 if (ceph_snap(dir
) != CEPH_NOSNAP
)
898 if (ceph_quota_is_max_files_exceeded(dir
)) {
903 err
= ceph_security_init_secctx(dentry
, S_IFLNK
| 0777, &as_ctx
);
907 dout("symlink in dir %p dentry %p to '%s'\n", dir
, dentry
, dest
);
908 req
= ceph_mdsc_create_request(mdsc
, CEPH_MDS_OP_SYMLINK
, USE_AUTH_MDS
);
913 req
->r_path2
= kstrdup(dest
, GFP_KERNEL
);
916 ceph_mdsc_put_request(req
);
920 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
921 req
->r_dentry
= dget(dentry
);
923 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
| CEPH_CAP_AUTH_EXCL
;
924 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
925 if (as_ctx
.pagelist
) {
926 req
->r_pagelist
= as_ctx
.pagelist
;
927 as_ctx
.pagelist
= NULL
;
929 err
= ceph_mdsc_do_request(mdsc
, dir
, req
);
930 if (!err
&& !req
->r_reply_info
.head
->is_dentry
)
931 err
= ceph_handle_notrace_create(dir
, dentry
);
932 ceph_mdsc_put_request(req
);
936 ceph_release_acl_sec_ctx(&as_ctx
);
940 static int ceph_mkdir(struct inode
*dir
, struct dentry
*dentry
, umode_t mode
)
942 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dir
->i_sb
);
943 struct ceph_mds_request
*req
;
944 struct ceph_acl_sec_ctx as_ctx
= {};
948 if (ceph_snap(dir
) == CEPH_SNAPDIR
) {
949 /* mkdir .snap/foo is a MKSNAP */
950 op
= CEPH_MDS_OP_MKSNAP
;
951 dout("mksnap dir %p snap '%pd' dn %p\n", dir
,
953 } else if (ceph_snap(dir
) == CEPH_NOSNAP
) {
954 dout("mkdir dir %p dn %p mode 0%ho\n", dir
, dentry
, mode
);
955 op
= CEPH_MDS_OP_MKDIR
;
960 if (op
== CEPH_MDS_OP_MKDIR
&&
961 ceph_quota_is_max_files_exceeded(dir
)) {
967 err
= ceph_pre_init_acls(dir
, &mode
, &as_ctx
);
970 err
= ceph_security_init_secctx(dentry
, mode
, &as_ctx
);
974 req
= ceph_mdsc_create_request(mdsc
, op
, USE_AUTH_MDS
);
980 req
->r_dentry
= dget(dentry
);
983 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
984 req
->r_args
.mkdir
.mode
= cpu_to_le32(mode
);
985 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
| CEPH_CAP_AUTH_EXCL
;
986 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
987 if (as_ctx
.pagelist
) {
988 req
->r_pagelist
= as_ctx
.pagelist
;
989 as_ctx
.pagelist
= NULL
;
991 err
= ceph_mdsc_do_request(mdsc
, dir
, req
);
993 !req
->r_reply_info
.head
->is_target
&&
994 !req
->r_reply_info
.head
->is_dentry
)
995 err
= ceph_handle_notrace_create(dir
, dentry
);
996 ceph_mdsc_put_request(req
);
999 ceph_init_inode_acls(d_inode(dentry
), &as_ctx
);
1002 ceph_release_acl_sec_ctx(&as_ctx
);
1006 static int ceph_link(struct dentry
*old_dentry
, struct inode
*dir
,
1007 struct dentry
*dentry
)
1009 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(dir
->i_sb
);
1010 struct ceph_mds_request
*req
;
1013 if (ceph_snap(dir
) != CEPH_NOSNAP
)
1016 dout("link in dir %p old_dentry %p dentry %p\n", dir
,
1017 old_dentry
, dentry
);
1018 req
= ceph_mdsc_create_request(mdsc
, CEPH_MDS_OP_LINK
, USE_AUTH_MDS
);
1021 return PTR_ERR(req
);
1023 req
->r_dentry
= dget(dentry
);
1024 req
->r_num_caps
= 2;
1025 req
->r_old_dentry
= dget(old_dentry
);
1026 req
->r_parent
= dir
;
1027 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
1028 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
;
1029 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
1030 /* release LINK_SHARED on source inode (mds will lock it) */
1031 req
->r_old_inode_drop
= CEPH_CAP_LINK_SHARED
| CEPH_CAP_LINK_EXCL
;
1032 err
= ceph_mdsc_do_request(mdsc
, dir
, req
);
1035 } else if (!req
->r_reply_info
.head
->is_dentry
) {
1036 ihold(d_inode(old_dentry
));
1037 d_instantiate(dentry
, d_inode(old_dentry
));
1039 ceph_mdsc_put_request(req
);
1043 static void ceph_async_unlink_cb(struct ceph_mds_client
*mdsc
,
1044 struct ceph_mds_request
*req
)
1046 int result
= req
->r_err
? req
->r_err
:
1047 le32_to_cpu(req
->r_reply_info
.head
->result
);
1049 if (result
== -EJUKEBOX
)
1052 /* If op failed, mark everyone involved for errors */
1056 char *path
= ceph_mdsc_build_path(req
->r_dentry
, &pathlen
,
1059 /* mark error on parent + clear complete */
1060 mapping_set_error(req
->r_parent
->i_mapping
, result
);
1061 ceph_dir_clear_complete(req
->r_parent
);
1063 /* drop the dentry -- we don't know its status */
1064 if (!d_unhashed(req
->r_dentry
))
1065 d_drop(req
->r_dentry
);
1067 /* mark inode itself for an error (since metadata is bogus) */
1068 mapping_set_error(req
->r_old_inode
->i_mapping
, result
);
1070 pr_warn("ceph: async unlink failure path=(%llx)%s result=%d!\n",
1071 base
, IS_ERR(path
) ? "<<bad>>" : path
, result
);
1072 ceph_mdsc_free_path(path
, pathlen
);
1075 iput(req
->r_old_inode
);
1076 ceph_mdsc_release_dir_caps(req
);
1079 static int get_caps_for_async_unlink(struct inode
*dir
, struct dentry
*dentry
)
1081 struct ceph_inode_info
*ci
= ceph_inode(dir
);
1082 struct ceph_dentry_info
*di
;
1083 int got
= 0, want
= CEPH_CAP_FILE_EXCL
| CEPH_CAP_DIR_UNLINK
;
1085 spin_lock(&ci
->i_ceph_lock
);
1086 if ((__ceph_caps_issued(ci
, NULL
) & want
) == want
) {
1087 ceph_take_cap_refs(ci
, want
, false);
1090 spin_unlock(&ci
->i_ceph_lock
);
1092 /* If we didn't get anything, return 0 */
1096 spin_lock(&dentry
->d_lock
);
1097 di
= ceph_dentry(dentry
);
1099 * - We are holding Fx, which implies Fs caps.
1100 * - Only support async unlink for primary linkage
1102 if (atomic_read(&ci
->i_shared_gen
) != di
->lease_shared_gen
||
1103 !(di
->flags
& CEPH_DENTRY_PRIMARY_LINK
))
1105 spin_unlock(&dentry
->d_lock
);
1107 /* Do we still want what we've got? */
1111 ceph_put_cap_refs(ci
, got
);
1116 * rmdir and unlink are differ only by the metadata op code
1118 static int ceph_unlink(struct inode
*dir
, struct dentry
*dentry
)
1120 struct ceph_fs_client
*fsc
= ceph_sb_to_client(dir
->i_sb
);
1121 struct ceph_mds_client
*mdsc
= fsc
->mdsc
;
1122 struct inode
*inode
= d_inode(dentry
);
1123 struct ceph_mds_request
*req
;
1124 bool try_async
= ceph_test_mount_opt(fsc
, ASYNC_DIROPS
);
1128 if (ceph_snap(dir
) == CEPH_SNAPDIR
) {
1129 /* rmdir .snap/foo is RMSNAP */
1130 dout("rmsnap dir %p '%pd' dn %p\n", dir
, dentry
, dentry
);
1131 op
= CEPH_MDS_OP_RMSNAP
;
1132 } else if (ceph_snap(dir
) == CEPH_NOSNAP
) {
1133 dout("unlink/rmdir dir %p dn %p inode %p\n",
1134 dir
, dentry
, inode
);
1135 op
= d_is_dir(dentry
) ?
1136 CEPH_MDS_OP_RMDIR
: CEPH_MDS_OP_UNLINK
;
1140 req
= ceph_mdsc_create_request(mdsc
, op
, USE_AUTH_MDS
);
1145 req
->r_dentry
= dget(dentry
);
1146 req
->r_num_caps
= 2;
1147 req
->r_parent
= dir
;
1148 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
;
1149 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
1150 req
->r_inode_drop
= ceph_drop_caps_for_unlink(inode
);
1152 if (try_async
&& op
== CEPH_MDS_OP_UNLINK
&&
1153 (req
->r_dir_caps
= get_caps_for_async_unlink(dir
, dentry
))) {
1154 dout("async unlink on %llu/%.*s caps=%s", ceph_ino(dir
),
1155 dentry
->d_name
.len
, dentry
->d_name
.name
,
1156 ceph_cap_string(req
->r_dir_caps
));
1157 set_bit(CEPH_MDS_R_ASYNC
, &req
->r_req_flags
);
1158 req
->r_callback
= ceph_async_unlink_cb
;
1159 req
->r_old_inode
= d_inode(dentry
);
1160 ihold(req
->r_old_inode
);
1161 err
= ceph_mdsc_submit_request(mdsc
, dir
, req
);
1164 * We have enough caps, so we assume that the unlink
1165 * will succeed. Fix up the target inode and dcache.
1169 } else if (err
== -EJUKEBOX
) {
1171 ceph_mdsc_put_request(req
);
1175 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
1176 err
= ceph_mdsc_do_request(mdsc
, dir
, req
);
1177 if (!err
&& !req
->r_reply_info
.head
->is_dentry
)
1181 ceph_mdsc_put_request(req
);
1186 static int ceph_rename(struct inode
*old_dir
, struct dentry
*old_dentry
,
1187 struct inode
*new_dir
, struct dentry
*new_dentry
,
1190 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(old_dir
->i_sb
);
1191 struct ceph_mds_request
*req
;
1192 int op
= CEPH_MDS_OP_RENAME
;
1198 if (ceph_snap(old_dir
) != ceph_snap(new_dir
))
1200 if (ceph_snap(old_dir
) != CEPH_NOSNAP
) {
1201 if (old_dir
== new_dir
&& ceph_snap(old_dir
) == CEPH_SNAPDIR
)
1202 op
= CEPH_MDS_OP_RENAMESNAP
;
1206 /* don't allow cross-quota renames */
1207 if ((old_dir
!= new_dir
) &&
1208 (!ceph_quota_is_same_realm(old_dir
, new_dir
)))
1211 dout("rename dir %p dentry %p to dir %p dentry %p\n",
1212 old_dir
, old_dentry
, new_dir
, new_dentry
);
1213 req
= ceph_mdsc_create_request(mdsc
, op
, USE_AUTH_MDS
);
1215 return PTR_ERR(req
);
1217 req
->r_dentry
= dget(new_dentry
);
1218 req
->r_num_caps
= 2;
1219 req
->r_old_dentry
= dget(old_dentry
);
1220 req
->r_old_dentry_dir
= old_dir
;
1221 req
->r_parent
= new_dir
;
1222 set_bit(CEPH_MDS_R_PARENT_LOCKED
, &req
->r_req_flags
);
1223 req
->r_old_dentry_drop
= CEPH_CAP_FILE_SHARED
;
1224 req
->r_old_dentry_unless
= CEPH_CAP_FILE_EXCL
;
1225 req
->r_dentry_drop
= CEPH_CAP_FILE_SHARED
;
1226 req
->r_dentry_unless
= CEPH_CAP_FILE_EXCL
;
1227 /* release LINK_RDCACHE on source inode (mds will lock it) */
1228 req
->r_old_inode_drop
= CEPH_CAP_LINK_SHARED
| CEPH_CAP_LINK_EXCL
;
1229 if (d_really_is_positive(new_dentry
)) {
1231 ceph_drop_caps_for_unlink(d_inode(new_dentry
));
1233 err
= ceph_mdsc_do_request(mdsc
, old_dir
, req
);
1234 if (!err
&& !req
->r_reply_info
.head
->is_dentry
) {
1236 * Normally d_move() is done by fill_trace (called by
1237 * do_request, above). If there is no trace, we need
1240 d_move(old_dentry
, new_dentry
);
1242 ceph_mdsc_put_request(req
);
1247 * Move dentry to tail of mdsc->dentry_leases list when lease is updated.
1248 * Leases at front of the list will expire first. (Assume all leases have
1251 * Called under dentry->d_lock.
1253 void __ceph_dentry_lease_touch(struct ceph_dentry_info
*di
)
1255 struct dentry
*dn
= di
->dentry
;
1256 struct ceph_mds_client
*mdsc
;
1258 dout("dentry_lease_touch %p %p '%pd'\n", di
, dn
, dn
);
1260 di
->flags
|= CEPH_DENTRY_LEASE_LIST
;
1261 if (di
->flags
& CEPH_DENTRY_SHRINK_LIST
) {
1262 di
->flags
|= CEPH_DENTRY_REFERENCED
;
1266 mdsc
= ceph_sb_to_client(dn
->d_sb
)->mdsc
;
1267 spin_lock(&mdsc
->dentry_list_lock
);
1268 list_move_tail(&di
->lease_list
, &mdsc
->dentry_leases
);
1269 spin_unlock(&mdsc
->dentry_list_lock
);
1272 static void __dentry_dir_lease_touch(struct ceph_mds_client
* mdsc
,
1273 struct ceph_dentry_info
*di
)
1275 di
->flags
&= ~(CEPH_DENTRY_LEASE_LIST
| CEPH_DENTRY_REFERENCED
);
1278 list_move_tail(&di
->lease_list
, &mdsc
->dentry_dir_leases
);
1282 * When dir lease is used, add dentry to tail of mdsc->dentry_dir_leases
1283 * list if it's not in the list, otherwise set 'referenced' flag.
1285 * Called under dentry->d_lock.
1287 void __ceph_dentry_dir_lease_touch(struct ceph_dentry_info
*di
)
1289 struct dentry
*dn
= di
->dentry
;
1290 struct ceph_mds_client
*mdsc
;
1292 dout("dentry_dir_lease_touch %p %p '%pd' (offset 0x%llx)\n",
1293 di
, dn
, dn
, di
->offset
);
1295 if (!list_empty(&di
->lease_list
)) {
1296 if (di
->flags
& CEPH_DENTRY_LEASE_LIST
) {
1297 /* don't remove dentry from dentry lease list
1298 * if its lease is valid */
1299 if (__dentry_lease_is_valid(di
))
1302 di
->flags
|= CEPH_DENTRY_REFERENCED
;
1307 if (di
->flags
& CEPH_DENTRY_SHRINK_LIST
) {
1308 di
->flags
|= CEPH_DENTRY_REFERENCED
;
1309 di
->flags
&= ~CEPH_DENTRY_LEASE_LIST
;
1313 mdsc
= ceph_sb_to_client(dn
->d_sb
)->mdsc
;
1314 spin_lock(&mdsc
->dentry_list_lock
);
1315 __dentry_dir_lease_touch(mdsc
, di
),
1316 spin_unlock(&mdsc
->dentry_list_lock
);
1319 static void __dentry_lease_unlist(struct ceph_dentry_info
*di
)
1321 struct ceph_mds_client
*mdsc
;
1322 if (di
->flags
& CEPH_DENTRY_SHRINK_LIST
)
1324 if (list_empty(&di
->lease_list
))
1327 mdsc
= ceph_sb_to_client(di
->dentry
->d_sb
)->mdsc
;
1328 spin_lock(&mdsc
->dentry_list_lock
);
1329 list_del_init(&di
->lease_list
);
1330 spin_unlock(&mdsc
->dentry_list_lock
);
1340 struct ceph_lease_walk_control
{
1342 bool expire_dir_lease
;
1343 unsigned long nr_to_scan
;
1344 unsigned long dir_lease_ttl
;
1347 static unsigned long
1348 __dentry_leases_walk(struct ceph_mds_client
*mdsc
,
1349 struct ceph_lease_walk_control
*lwc
,
1350 int (*check
)(struct dentry
*, void*))
1352 struct ceph_dentry_info
*di
, *tmp
;
1353 struct dentry
*dentry
, *last
= NULL
;
1354 struct list_head
* list
;
1356 unsigned long freed
= 0;
1359 list
= lwc
->dir_lease
? &mdsc
->dentry_dir_leases
: &mdsc
->dentry_leases
;
1360 spin_lock(&mdsc
->dentry_list_lock
);
1361 list_for_each_entry_safe(di
, tmp
, list
, lease_list
) {
1362 if (!lwc
->nr_to_scan
)
1366 dentry
= di
->dentry
;
1370 if (!spin_trylock(&dentry
->d_lock
))
1373 if (__lockref_is_dead(&dentry
->d_lockref
)) {
1374 list_del_init(&di
->lease_list
);
1378 ret
= check(dentry
, lwc
);
1380 /* move it into tail of dir lease list */
1381 __dentry_dir_lease_touch(mdsc
, di
);
1387 di
->flags
&= ~CEPH_DENTRY_REFERENCED
;
1388 if (dentry
->d_lockref
.count
> 0) {
1389 /* update_dentry_lease() will re-add
1390 * it to lease list, or
1391 * ceph_d_delete() will return 1 when
1392 * last reference is dropped */
1393 list_del_init(&di
->lease_list
);
1395 di
->flags
|= CEPH_DENTRY_SHRINK_LIST
;
1396 list_move_tail(&di
->lease_list
, &dispose
);
1401 spin_unlock(&dentry
->d_lock
);
1405 spin_unlock(&mdsc
->dentry_list_lock
);
1407 while (!list_empty(&dispose
)) {
1408 di
= list_first_entry(&dispose
, struct ceph_dentry_info
,
1410 dentry
= di
->dentry
;
1411 spin_lock(&dentry
->d_lock
);
1413 list_del_init(&di
->lease_list
);
1414 di
->flags
&= ~CEPH_DENTRY_SHRINK_LIST
;
1415 if (di
->flags
& CEPH_DENTRY_REFERENCED
) {
1416 spin_lock(&mdsc
->dentry_list_lock
);
1417 if (di
->flags
& CEPH_DENTRY_LEASE_LIST
) {
1418 list_add_tail(&di
->lease_list
,
1419 &mdsc
->dentry_leases
);
1421 __dentry_dir_lease_touch(mdsc
, di
);
1423 spin_unlock(&mdsc
->dentry_list_lock
);
1428 spin_unlock(&dentry
->d_lock
);
1429 /* ceph_d_delete() does the trick */
1435 static int __dentry_lease_check(struct dentry
*dentry
, void *arg
)
1437 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
1440 if (__dentry_lease_is_valid(di
))
1442 ret
= __dir_lease_try_check(dentry
);
1450 static int __dir_lease_check(struct dentry
*dentry
, void *arg
)
1452 struct ceph_lease_walk_control
*lwc
= arg
;
1453 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
1455 int ret
= __dir_lease_try_check(dentry
);
1459 if (time_before(jiffies
, di
->time
+ lwc
->dir_lease_ttl
))
1461 /* Move dentry to tail of dir lease list if we don't want
1462 * to delete it. So dentries in the list are checked in a
1463 * round robin manner */
1464 if (!lwc
->expire_dir_lease
)
1466 if (dentry
->d_lockref
.count
> 0 ||
1467 (di
->flags
& CEPH_DENTRY_REFERENCED
))
1469 /* invalidate dir lease */
1470 di
->lease_shared_gen
= 0;
1475 int ceph_trim_dentries(struct ceph_mds_client
*mdsc
)
1477 struct ceph_lease_walk_control lwc
;
1478 unsigned long count
;
1479 unsigned long freed
;
1481 spin_lock(&mdsc
->caps_list_lock
);
1482 if (mdsc
->caps_use_max
> 0 &&
1483 mdsc
->caps_use_count
> mdsc
->caps_use_max
)
1484 count
= mdsc
->caps_use_count
- mdsc
->caps_use_max
;
1487 spin_unlock(&mdsc
->caps_list_lock
);
1489 lwc
.dir_lease
= false;
1490 lwc
.nr_to_scan
= CEPH_CAPS_PER_RELEASE
* 2;
1491 freed
= __dentry_leases_walk(mdsc
, &lwc
, __dentry_lease_check
);
1492 if (!lwc
.nr_to_scan
) /* more invalid leases */
1495 if (lwc
.nr_to_scan
< CEPH_CAPS_PER_RELEASE
)
1496 lwc
.nr_to_scan
= CEPH_CAPS_PER_RELEASE
;
1498 lwc
.dir_lease
= true;
1499 lwc
.expire_dir_lease
= freed
< count
;
1500 lwc
.dir_lease_ttl
= mdsc
->fsc
->mount_options
->caps_wanted_delay_max
* HZ
;
1501 freed
+=__dentry_leases_walk(mdsc
, &lwc
, __dir_lease_check
);
1502 if (!lwc
.nr_to_scan
) /* more to check */
1505 return freed
> 0 ? 1 : 0;
1509 * Ensure a dentry lease will no longer revalidate.
1511 void ceph_invalidate_dentry_lease(struct dentry
*dentry
)
1513 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
1514 spin_lock(&dentry
->d_lock
);
1516 di
->lease_shared_gen
= 0;
1517 di
->flags
&= ~CEPH_DENTRY_PRIMARY_LINK
;
1518 __dentry_lease_unlist(di
);
1519 spin_unlock(&dentry
->d_lock
);
1523 * Check if dentry lease is valid. If not, delete the lease. Try to
1524 * renew if the least is more than half up.
1526 static bool __dentry_lease_is_valid(struct ceph_dentry_info
*di
)
1528 struct ceph_mds_session
*session
;
1533 session
= di
->lease_session
;
1538 spin_lock(&session
->s_gen_ttl_lock
);
1539 gen
= session
->s_cap_gen
;
1540 ttl
= session
->s_cap_ttl
;
1541 spin_unlock(&session
->s_gen_ttl_lock
);
1543 if (di
->lease_gen
== gen
&&
1544 time_before(jiffies
, ttl
) &&
1545 time_before(jiffies
, di
->time
))
1552 static int dentry_lease_is_valid(struct dentry
*dentry
, unsigned int flags
)
1554 struct ceph_dentry_info
*di
;
1555 struct ceph_mds_session
*session
= NULL
;
1559 spin_lock(&dentry
->d_lock
);
1560 di
= ceph_dentry(dentry
);
1561 if (di
&& __dentry_lease_is_valid(di
)) {
1564 if (di
->lease_renew_after
&&
1565 time_after(jiffies
, di
->lease_renew_after
)) {
1567 * We should renew. If we're in RCU walk mode
1568 * though, we can't do that so just return
1571 if (flags
& LOOKUP_RCU
) {
1574 session
= ceph_get_mds_session(di
->lease_session
);
1575 seq
= di
->lease_seq
;
1576 di
->lease_renew_after
= 0;
1577 di
->lease_renew_from
= jiffies
;
1581 spin_unlock(&dentry
->d_lock
);
1584 ceph_mdsc_lease_send_msg(session
, dentry
,
1585 CEPH_MDS_LEASE_RENEW
, seq
);
1586 ceph_put_mds_session(session
);
1588 dout("dentry_lease_is_valid - dentry %p = %d\n", dentry
, valid
);
1593 * Called under dentry->d_lock.
1595 static int __dir_lease_try_check(const struct dentry
*dentry
)
1597 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
1599 struct ceph_inode_info
*ci
;
1602 if (!di
->lease_shared_gen
)
1604 if (IS_ROOT(dentry
))
1607 dir
= d_inode(dentry
->d_parent
);
1608 ci
= ceph_inode(dir
);
1610 if (spin_trylock(&ci
->i_ceph_lock
)) {
1611 if (atomic_read(&ci
->i_shared_gen
) == di
->lease_shared_gen
&&
1612 __ceph_caps_issued_mask(ci
, CEPH_CAP_FILE_SHARED
, 0))
1614 spin_unlock(&ci
->i_ceph_lock
);
1620 di
->lease_shared_gen
= 0;
1625 * Check if directory-wide content lease/cap is valid.
1627 static int dir_lease_is_valid(struct inode
*dir
, struct dentry
*dentry
,
1628 struct ceph_mds_client
*mdsc
)
1630 struct ceph_inode_info
*ci
= ceph_inode(dir
);
1634 spin_lock(&ci
->i_ceph_lock
);
1635 valid
= __ceph_caps_issued_mask(ci
, CEPH_CAP_FILE_SHARED
, 1);
1637 __ceph_touch_fmode(ci
, mdsc
, CEPH_FILE_MODE_RD
);
1638 shared_gen
= atomic_read(&ci
->i_shared_gen
);
1640 spin_unlock(&ci
->i_ceph_lock
);
1642 struct ceph_dentry_info
*di
;
1643 spin_lock(&dentry
->d_lock
);
1644 di
= ceph_dentry(dentry
);
1645 if (dir
== d_inode(dentry
->d_parent
) &&
1646 di
&& di
->lease_shared_gen
== shared_gen
)
1647 __ceph_dentry_dir_lease_touch(di
);
1650 spin_unlock(&dentry
->d_lock
);
1652 dout("dir_lease_is_valid dir %p v%u dentry %p = %d\n",
1653 dir
, (unsigned)atomic_read(&ci
->i_shared_gen
), dentry
, valid
);
1658 * Check if cached dentry can be trusted.
1660 static int ceph_d_revalidate(struct dentry
*dentry
, unsigned int flags
)
1663 struct dentry
*parent
;
1664 struct inode
*dir
, *inode
;
1665 struct ceph_mds_client
*mdsc
;
1667 if (flags
& LOOKUP_RCU
) {
1668 parent
= READ_ONCE(dentry
->d_parent
);
1669 dir
= d_inode_rcu(parent
);
1672 inode
= d_inode_rcu(dentry
);
1674 parent
= dget_parent(dentry
);
1675 dir
= d_inode(parent
);
1676 inode
= d_inode(dentry
);
1679 dout("d_revalidate %p '%pd' inode %p offset 0x%llx\n", dentry
,
1680 dentry
, inode
, ceph_dentry(dentry
)->offset
);
1682 mdsc
= ceph_sb_to_client(dir
->i_sb
)->mdsc
;
1684 /* always trust cached snapped dentries, snapdir dentry */
1685 if (ceph_snap(dir
) != CEPH_NOSNAP
) {
1686 dout("d_revalidate %p '%pd' inode %p is SNAPPED\n", dentry
,
1689 } else if (inode
&& ceph_snap(inode
) == CEPH_SNAPDIR
) {
1692 valid
= dentry_lease_is_valid(dentry
, flags
);
1693 if (valid
== -ECHILD
)
1695 if (valid
|| dir_lease_is_valid(dir
, dentry
, mdsc
)) {
1697 valid
= ceph_is_any_caps(inode
);
1704 struct ceph_mds_request
*req
;
1708 if (flags
& LOOKUP_RCU
)
1711 percpu_counter_inc(&mdsc
->metric
.d_lease_mis
);
1713 op
= ceph_snap(dir
) == CEPH_SNAPDIR
?
1714 CEPH_MDS_OP_LOOKUPSNAP
: CEPH_MDS_OP_LOOKUP
;
1715 req
= ceph_mdsc_create_request(mdsc
, op
, USE_ANY_MDS
);
1717 req
->r_dentry
= dget(dentry
);
1718 req
->r_num_caps
= 2;
1719 req
->r_parent
= dir
;
1721 mask
= CEPH_STAT_CAP_INODE
| CEPH_CAP_AUTH_SHARED
;
1722 if (ceph_security_xattr_wanted(dir
))
1723 mask
|= CEPH_CAP_XATTR_SHARED
;
1724 req
->r_args
.getattr
.mask
= cpu_to_le32(mask
);
1726 err
= ceph_mdsc_do_request(mdsc
, NULL
, req
);
1729 if (d_really_is_positive(dentry
) &&
1730 d_inode(dentry
) == req
->r_target_inode
)
1734 if (d_really_is_negative(dentry
))
1740 ceph_mdsc_put_request(req
);
1741 dout("d_revalidate %p lookup result=%d\n",
1745 percpu_counter_inc(&mdsc
->metric
.d_lease_hit
);
1748 dout("d_revalidate %p %s\n", dentry
, valid
? "valid" : "invalid");
1750 ceph_dir_clear_complete(dir
);
1752 if (!(flags
& LOOKUP_RCU
))
1758 * Delete unused dentry that doesn't have valid lease
1760 * Called under dentry->d_lock.
1762 static int ceph_d_delete(const struct dentry
*dentry
)
1764 struct ceph_dentry_info
*di
;
1766 /* won't release caps */
1767 if (d_really_is_negative(dentry
))
1769 if (ceph_snap(d_inode(dentry
)) != CEPH_NOSNAP
)
1772 di
= ceph_dentry(dentry
);
1774 if (__dentry_lease_is_valid(di
))
1776 if (__dir_lease_try_check(dentry
))
1783 * Release our ceph_dentry_info.
1785 static void ceph_d_release(struct dentry
*dentry
)
1787 struct ceph_dentry_info
*di
= ceph_dentry(dentry
);
1788 struct ceph_fs_client
*fsc
= ceph_sb_to_client(dentry
->d_sb
);
1790 dout("d_release %p\n", dentry
);
1792 atomic64_dec(&fsc
->mdsc
->metric
.total_dentries
);
1794 spin_lock(&dentry
->d_lock
);
1795 __dentry_lease_unlist(di
);
1796 dentry
->d_fsdata
= NULL
;
1797 spin_unlock(&dentry
->d_lock
);
1799 if (di
->lease_session
)
1800 ceph_put_mds_session(di
->lease_session
);
1801 kmem_cache_free(ceph_dentry_cachep
, di
);
1805 * When the VFS prunes a dentry from the cache, we need to clear the
1806 * complete flag on the parent directory.
1808 * Called under dentry->d_lock.
1810 static void ceph_d_prune(struct dentry
*dentry
)
1812 struct ceph_inode_info
*dir_ci
;
1813 struct ceph_dentry_info
*di
;
1815 dout("ceph_d_prune %pd %p\n", dentry
, dentry
);
1817 /* do we have a valid parent? */
1818 if (IS_ROOT(dentry
))
1821 /* we hold d_lock, so d_parent is stable */
1822 dir_ci
= ceph_inode(d_inode(dentry
->d_parent
));
1823 if (dir_ci
->i_vino
.snap
== CEPH_SNAPDIR
)
1826 /* who calls d_delete() should also disable dcache readdir */
1827 if (d_really_is_negative(dentry
))
1830 /* d_fsdata does not get cleared until d_release */
1831 if (!d_unhashed(dentry
)) {
1832 __ceph_dir_clear_complete(dir_ci
);
1836 /* Disable dcache readdir just in case that someone called d_drop()
1837 * or d_invalidate(), but MDS didn't revoke CEPH_CAP_FILE_SHARED
1838 * properly (dcache readdir is still enabled) */
1839 di
= ceph_dentry(dentry
);
1840 if (di
->offset
> 0 &&
1841 di
->lease_shared_gen
== atomic_read(&dir_ci
->i_shared_gen
))
1842 __ceph_dir_clear_ordered(dir_ci
);
1846 * read() on a dir. This weird interface hack only works if mounted
1847 * with '-o dirstat'.
1849 static ssize_t
ceph_read_dir(struct file
*file
, char __user
*buf
, size_t size
,
1852 struct ceph_dir_file_info
*dfi
= file
->private_data
;
1853 struct inode
*inode
= file_inode(file
);
1854 struct ceph_inode_info
*ci
= ceph_inode(inode
);
1856 const int bufsize
= 1024;
1858 if (!ceph_test_mount_opt(ceph_sb_to_client(inode
->i_sb
), DIRSTAT
))
1861 if (!dfi
->dir_info
) {
1862 dfi
->dir_info
= kmalloc(bufsize
, GFP_KERNEL
);
1866 snprintf(dfi
->dir_info
, bufsize
,
1869 " subdirs: %20lld\n"
1870 "rentries: %20lld\n"
1872 " rsubdirs: %20lld\n"
1874 "rctime: %10lld.%09ld\n",
1875 ci
->i_files
+ ci
->i_subdirs
,
1878 ci
->i_rfiles
+ ci
->i_rsubdirs
,
1882 ci
->i_rctime
.tv_sec
,
1883 ci
->i_rctime
.tv_nsec
);
1886 if (*ppos
>= dfi
->dir_info_len
)
1888 size
= min_t(unsigned, size
, dfi
->dir_info_len
-*ppos
);
1889 left
= copy_to_user(buf
, dfi
->dir_info
+ *ppos
, size
);
1892 *ppos
+= (size
- left
);
1899 * Return name hash for a given dentry. This is dependent on
1900 * the parent directory's hash function.
1902 unsigned ceph_dentry_hash(struct inode
*dir
, struct dentry
*dn
)
1904 struct ceph_inode_info
*dci
= ceph_inode(dir
);
1907 switch (dci
->i_dir_layout
.dl_dir_hash
) {
1908 case 0: /* for backward compat */
1909 case CEPH_STR_HASH_LINUX
:
1910 return dn
->d_name
.hash
;
1913 spin_lock(&dn
->d_lock
);
1914 hash
= ceph_str_hash(dci
->i_dir_layout
.dl_dir_hash
,
1915 dn
->d_name
.name
, dn
->d_name
.len
);
1916 spin_unlock(&dn
->d_lock
);
1921 const struct file_operations ceph_dir_fops
= {
1922 .read
= ceph_read_dir
,
1923 .iterate
= ceph_readdir
,
1924 .llseek
= ceph_dir_llseek
,
1926 .release
= ceph_release
,
1927 .unlocked_ioctl
= ceph_ioctl
,
1928 .compat_ioctl
= compat_ptr_ioctl
,
1929 .fsync
= ceph_fsync
,
1931 .flock
= ceph_flock
,
1934 const struct file_operations ceph_snapdir_fops
= {
1935 .iterate
= ceph_readdir
,
1936 .llseek
= ceph_dir_llseek
,
1938 .release
= ceph_release
,
1941 const struct inode_operations ceph_dir_iops
= {
1942 .lookup
= ceph_lookup
,
1943 .permission
= ceph_permission
,
1944 .getattr
= ceph_getattr
,
1945 .setattr
= ceph_setattr
,
1946 .listxattr
= ceph_listxattr
,
1947 .get_acl
= ceph_get_acl
,
1948 .set_acl
= ceph_set_acl
,
1949 .mknod
= ceph_mknod
,
1950 .symlink
= ceph_symlink
,
1951 .mkdir
= ceph_mkdir
,
1953 .unlink
= ceph_unlink
,
1954 .rmdir
= ceph_unlink
,
1955 .rename
= ceph_rename
,
1956 .create
= ceph_create
,
1957 .atomic_open
= ceph_atomic_open
,
1960 const struct inode_operations ceph_snapdir_iops
= {
1961 .lookup
= ceph_lookup
,
1962 .permission
= ceph_permission
,
1963 .getattr
= ceph_getattr
,
1964 .mkdir
= ceph_mkdir
,
1965 .rmdir
= ceph_unlink
,
1966 .rename
= ceph_rename
,
1969 const struct dentry_operations ceph_dentry_ops
= {
1970 .d_revalidate
= ceph_d_revalidate
,
1971 .d_delete
= ceph_d_delete
,
1972 .d_release
= ceph_d_release
,
1973 .d_prune
= ceph_d_prune
,
1974 .d_init
= ceph_d_init
,