// SPDX-License-Identifier: GPL-2.0#include <linux/ceph/ceph_debug.h> #include <linux/spinlock.h>#include <linux/namei.h>#include <linux/slab.h>#include <linux/sched.h>#include <linux/xattr.h> #include "super.h"#include "mds_client.h"#include "crypto.h" /* * Directory operations: readdir, lookup, create, link, unlink, * rename, etc. */ /* * Ceph MDS operations are specified in terms of a base ino and * relative path. Thus, the client can specify an operation on a * specific inode (e.g., a getattr due to fstat(2)), or as a path * relative to, say, the root directory. * * Normally, we limit ourselves to strict inode ops (no path component) * or dentry operations (a single path component relative to an ino). The * exception to this is open_root_dentry(), which will open the mount * point by name. */ const struct dentry_operations ceph_dentry_ops; static bool __dentry_lease_is_valid(struct ceph_dentry_info *di);static int __dir_lease_try_check(const struct dentry *dentry); /* * Initialize ceph dentry state. */static int ceph_d_init(struct dentry *dentry){ struct ceph_dentry_info *di; struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dentry->d_sb); di = kmem_cache_zalloc(ceph_dentry_cachep, GFP_KERNEL); if (!di) return -ENOMEM; /* oh well */ di->dentry = dentry; di->lease_session = NULL; di->time = jiffies; dentry->d_fsdata = di; INIT_LIST_HEAD(&di->lease_list); atomic64_inc(&mdsc->metric.total_dentries); return 0;} /* * for f_pos for readdir: * - hash order: * (0xff << 52) | ((24 bits hash) << 28) | * (the nth entry has hash collision); * - frag+name order; * ((frag value) << 28) | (the nth entry in frag); */#define OFFSET_BITS 28#define OFFSET_MASK ((1 << OFFSET_BITS) - 1)#define HASH_ORDER (0xffull << (OFFSET_BITS + 24))loff_t ceph_make_fpos(unsigned high, unsigned off, bool hash_order){ loff_t fpos = ((loff_t)high << 28) | (loff_t)off; if (hash_order) fpos |= HASH_ORDER; return fpos;} static bool is_hash_order(loff_t p){ return (p & HASH_ORDER) == HASH_ORDER;} static unsigned fpos_frag(loff_t p){ return p >> OFFSET_BITS;} static unsigned fpos_hash(loff_t p){ return ceph_frag_value(fpos_frag(p));} static unsigned fpos_off(loff_t p){ return p & OFFSET_MASK;} static int fpos_cmp(loff_t l, loff_t r){ int v = ceph_frag_compare(fpos_frag(l), fpos_frag(r)); if (v) return v; return (int)(fpos_off(l) - fpos_off(r));} /* * make note of the last dentry we read, so we can * continue at the same lexicographical point, * regardless of what dir changes take place on the * server. */static int note_last_dentry(struct ceph_fs_client *fsc, struct ceph_dir_file_info *dfi, const char *name, int len, unsigned next_offset){ char *buf = kmalloc(len+1, GFP_KERNEL); if (!buf) return -ENOMEM; kfree(dfi->last_name); dfi->last_name = buf; memcpy(dfi->last_name, name, len); dfi->last_name[len] = 0; dfi->next_offset = next_offset; doutc(fsc->client, "'%s'\n", dfi->last_name); return 0;} static struct dentry *__dcache_find_get_entry(struct dentry *parent, u64 idx, struct ceph_readdir_cache_control *cache_ctl){ struct inode *dir = d_inode(parent); struct ceph_client *cl = ceph_inode_to_client(dir); struct dentry *dentry; unsigned idx_mask = (PAGE_SIZE / sizeof(struct dentry *)) - 1; loff_t ptr_pos = idx * sizeof(struct dentry *); pgoff_t ptr_pgoff = ptr_pos >> PAGE_SHIFT; if (ptr_pos >= i_size_read(dir)) return NULL; if (!cache_ctl->folio || ptr_pgoff != cache_ctl->folio->index) { ceph_readdir_cache_release(cache_ctl); cache_ctl->folio = filemap_lock_folio(&dir->i_data, ptr_pgoff); if (IS_ERR(cache_ctl->folio)) { cache_ctl->folio = NULL; doutc(cl, " folio %lu not found\n", ptr_pgoff); return ERR_PTR(-EAGAIN); } /* reading/filling the cache are serialized by i_rwsem, no need to use folio lock */ folio_unlock(cache_ctl->folio); cache_ctl->dentries = kmap_local_folio(cache_ctl->folio, 0); } cache_ctl->index = idx & idx_mask; rcu_read_lock(); spin_lock(&parent->d_lock); /* check i_size again here, because empty directory can be * marked as complete while not holding the i_rwsem. */ if (ceph_dir_is_complete_ordered(dir) && ptr_pos < i_size_read(dir)) dentry = cache_ctl->dentries[cache_ctl->index]; else dentry = NULL; spin_unlock(&parent->d_lock); if (dentry && !lockref_get_not_dead(&dentry->d_lockref)) dentry = NULL; rcu_read_unlock(); return dentry ? : ERR_PTR(-EAGAIN);} /* * When possible, we try to satisfy a readdir by peeking at the * dcache. We make this work by carefully ordering dentries on * d_children when we initially get results back from the MDS, and * falling back to a "normal" sync readdir if any dentries in the dir * are dropped. * * Complete dir indicates that we have all dentries in the dir. It is * defined IFF we hold CEPH_CAP_FILE_SHARED (which will be revoked by * the MDS if/when the directory is modified). */static int __dcache_readdir(struct file *file, struct dir_context *ctx, int shared_gen){ struct ceph_dir_file_info *dfi = file->private_data; struct dentry *parent = file->f_path.dentry; struct inode *dir = d_inode(parent); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(dir); struct ceph_client *cl = ceph_inode_to_client(dir); struct dentry *dentry, *last = NULL; struct ceph_dentry_info *di; struct ceph_readdir_cache_control cache_ctl = {}; u64 idx = 0; int err = 0; doutc(cl, "%p %llx.%llx v%u at %llx\n", dir, ceph_vinop(dir), (unsigned)shared_gen, ctx->pos); /* search start position */ if (ctx->pos > 2) { u64 count = div_u64(i_size_read(dir), sizeof(struct dentry *)); while (count > 0) { u64 step = count >> 1; dentry = __dcache_find_get_entry(parent, idx + step, &cache_ctl); if (!dentry) { /* use linear search */ idx = 0; break; } if (IS_ERR(dentry)) { err = PTR_ERR(dentry); goto out; } di = ceph_dentry(dentry); spin_lock(&dentry->d_lock); if (fpos_cmp(di->offset, ctx->pos) < 0) { idx += step + 1; count -= step + 1; } else { count = step; } spin_unlock(&dentry->d_lock); dput(dentry); } doutc(cl, "%p %llx.%llx cache idx %llu\n", dir, ceph_vinop(dir), idx); } for (;;) { bool emit_dentry = false; dentry = __dcache_find_get_entry(parent, idx++, &cache_ctl); if (!dentry) { dfi->file_info.flags |= CEPH_F_ATEND; err = 0; break; } if (IS_ERR(dentry)) { err = PTR_ERR(dentry); goto out; } spin_lock(&dentry->d_lock); di = ceph_dentry(dentry); if (d_unhashed(dentry) || d_really_is_negative(dentry) || di->lease_shared_gen != shared_gen || ((dentry->d_flags & DCACHE_NOKEY_NAME) && fscrypt_has_encryption_key(dir))) { spin_unlock(&dentry->d_lock); dput(dentry); err = -EAGAIN; goto out; } if (fpos_cmp(ctx->pos, di->offset) <= 0) { __ceph_dentry_dir_lease_touch(di); emit_dentry = true; } spin_unlock(&dentry->d_lock); if (emit_dentry) { doutc(cl, " %llx dentry %p %pd %p\n", di->offset, dentry, dentry, d_inode(dentry)); ctx->pos = di->offset; if (!dir_emit(ctx, dentry->d_name.name, dentry->d_name.len, ceph_present_inode(d_inode(dentry)), d_inode(dentry)->i_mode >> 12)) { dput(dentry); err = 0; break; } ctx->pos++; if (last) dput(last); last = dentry; } else { dput(dentry); } }out: ceph_readdir_cache_release(&cache_ctl); if (last) { int ret; di = ceph_dentry(last); ret = note_last_dentry(fsc, dfi, last->d_name.name, last->d_name.len, fpos_off(di->offset) + 1); if (ret < 0) err = ret; dput(last); /* last_name no longer match cache index */ if (dfi->readdir_cache_idx >= 0) { dfi->readdir_cache_idx = -1; dfi->dir_release_count = 0; } } return err;} static bool need_send_readdir(struct ceph_dir_file_info *dfi, loff_t pos){ if (!dfi->last_readdir) return true; if (is_hash_order(pos)) return !ceph_frag_contains_value(dfi->frag, fpos_hash(pos)); else return dfi->frag != fpos_frag(pos);} static int ceph_readdir(struct file *file, struct dir_context *ctx){ struct ceph_dir_file_info *dfi = file->private_data; struct inode *inode = file_inode(file); struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_mds_client *mdsc = fsc->mdsc; struct ceph_client *cl = fsc->client; int i; int err; unsigned frag = -1; struct ceph_mds_reply_info_parsed *rinfo; doutc(cl, "%p %llx.%llx file %p pos %llx\n", inode, ceph_vinop(inode), file, ctx->pos); if (dfi->file_info.flags & CEPH_F_ATEND) return 0; /* always start with . and .. */ if (ctx->pos == 0) { doutc(cl, "%p %llx.%llx off 0 -> '.'\n", inode, ceph_vinop(inode)); if (!dir_emit(ctx, ".", 1, ceph_present_inode(inode), inode->i_mode >> 12)) return 0; ctx->pos = 1; } if (ctx->pos == 1) { u64 ino; struct dentry *dentry = file->f_path.dentry; spin_lock(&dentry->d_lock); ino = ceph_present_inode(dentry->d_parent->d_inode); spin_unlock(&dentry->d_lock); doutc(cl, "%p %llx.%llx off 1 -> '..'\n", inode, ceph_vinop(inode)); if (!dir_emit(ctx, "..", 2, ino, inode->i_mode >> 12)) return 0; ctx->pos = 2; } err = ceph_fscrypt_prepare_readdir(inode); if (err < 0) return err; spin_lock(&ci->i_ceph_lock); /* request Fx cap. if have Fx, we don't need to release Fs cap * for later create/unlink. */ __ceph_touch_fmode(ci, mdsc, CEPH_FILE_MODE_WR); /* can we use the dcache? */ if (ceph_test_mount_opt(fsc, DCACHE) && !ceph_test_mount_opt(fsc, NOASYNCREADDIR) && ceph_snap(inode) != CEPH_SNAPDIR && __ceph_dir_is_complete_ordered(ci) && __ceph_caps_issued_mask_metric(ci, CEPH_CAP_FILE_SHARED, 1)) { int shared_gen = atomic_read(&ci->i_shared_gen); spin_unlock(&ci->i_ceph_lock); err = __dcache_readdir(file, ctx, shared_gen); if (err != -EAGAIN) return err; } else { spin_unlock(&ci->i_ceph_lock); } /* proceed with a normal readdir */more: /* do we have the correct frag content buffered? */ if (need_send_readdir(dfi, ctx->pos)) { struct ceph_mds_request *req; int op = ceph_snap(inode) == CEPH_SNAPDIR ? CEPH_MDS_OP_LSSNAP : CEPH_MDS_OP_READDIR; /* discard old result, if any */ if (dfi->last_readdir) { ceph_mdsc_put_request(dfi->last_readdir); dfi->last_readdir = NULL; } if (is_hash_order(ctx->pos)) { /* fragtree isn't always accurate. choose frag * based on previous reply when possible. */ if (frag == (unsigned)-1) frag = ceph_choose_frag(ci, fpos_hash(ctx->pos), NULL, NULL); } else { frag = fpos_frag(ctx->pos); } doutc(cl, "fetching %p %llx.%llx frag %x offset '%s'\n", inode, ceph_vinop(inode), frag, dfi->last_name); req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS); if (IS_ERR(req)) return PTR_ERR(req); err = ceph_alloc_readdir_reply_buffer(req, inode); if (err) { ceph_mdsc_put_request(req); return err; } /* hints to request -> mds selection code */ req->r_direct_mode = USE_AUTH_MDS; if (op == CEPH_MDS_OP_READDIR) { req->r_direct_hash = ceph_frag_value(frag); __set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags); req->r_inode_drop = CEPH_CAP_FILE_EXCL; } if (dfi->last_name) { int len = strlen(dfi->last_name); req->r_path2 = kzalloc(NAME_MAX + 1, GFP_KERNEL); if (!req->r_path2) { ceph_mdsc_put_request(req); return -ENOMEM; } memcpy(req->r_path2, dfi->last_name, len); err = ceph_encode_encrypted_dname(inode, req->r_path2, len); if (err < 0) { ceph_mdsc_put_request(req); return err; } } else if (is_hash_order(ctx->pos)) { req->r_args.readdir.offset_hash = cpu_to_le32(fpos_hash(ctx->pos)); } req->r_dir_release_cnt = dfi->dir_release_count; req->r_dir_ordered_cnt = dfi->dir_ordered_count; req->r_readdir_cache_idx = dfi->readdir_cache_idx; req->r_readdir_offset = dfi->next_offset; req->r_args.readdir.frag = cpu_to_le32(frag); req->r_args.readdir.flags = cpu_to_le16(CEPH_READDIR_REPLY_BITFLAGS); req->r_inode = inode; ihold(inode); req->r_dentry = dget(file->f_path.dentry); err = ceph_mdsc_do_request(mdsc, NULL, req); if (err < 0) { ceph_mdsc_put_request(req); return err; } doutc(cl, "%p %llx.%llx got and parsed readdir result=%d" "on frag %x, end=%d, complete=%d, hash_order=%d\n", inode, ceph_vinop(inode), err, frag, (int)req->r_reply_info.dir_end, (int)req->r_reply_info.dir_complete, (int)req->r_reply_info.hash_order); rinfo = &req->r_reply_info; if (le32_to_cpu(rinfo->dir_dir->frag) != frag) { frag = le32_to_cpu(rinfo->dir_dir->frag); if (!rinfo->hash_order) { dfi->next_offset = req->r_readdir_offset; /* adjust ctx->pos to beginning of frag */ ctx->pos = ceph_make_fpos(frag, dfi->next_offset, false); } } dfi->frag = frag; dfi->last_readdir = req; if (test_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags)) { dfi->readdir_cache_idx = req->r_readdir_cache_idx; if (dfi->readdir_cache_idx < 0) { /* preclude from marking dir ordered */ dfi->dir_ordered_count = 0; } else if (ceph_frag_is_leftmost(frag) && dfi->next_offset == 2) { /* note dir version at start of readdir so * we can tell if any dentries get dropped */ dfi->dir_release_count = req->r_dir_release_cnt; dfi->dir_ordered_count = req->r_dir_ordered_cnt; } } else { doutc(cl, "%p %llx.%llx !did_prepopulate\n", inode, ceph_vinop(inode)); /* disable readdir cache */ dfi->readdir_cache_idx = -1; /* preclude from marking dir complete */Lines 1–500 of 2289. Load more
// SPDX-License-Identifier: GPL-2.0 #include <linux/ceph/ceph_debug.h> #include <linux/spinlock.h> #include <linux/namei.h> #include <linux/slab.h> #include <linux/sched.h> #include <linux/xattr.h> #include "super.h" #include "mds_client.h" #include "crypto.h" /* * Directory operations: readdir, lookup, create, link, unlink, * rename, etc. */ /* * Ceph MDS operations are specified in terms of a base ino and * relative path. Thus, the client can specify an operation on a * specific inode (e.g., a getattr due to fstat(2)), or as a path * relative to, say, the root directory. * * Normally, we limit ourselves to strict inode ops (no path component) * or dentry operations (a single path component relative to an ino). The * exception to this is open_root_dentry(), which will open the mount * point by name. */ const struct dentry_operations ceph_dentry_ops; static bool __dentry_lease_is_valid(struct ceph_dentry_info *di); static int __dir_lease_try_check(const struct dentry *dentry); /* * Initialize ceph dentry state. */ static int ceph_d_init(struct dentry *dentry) { struct ceph_dentry_info *di; struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dentry->d_sb); di = kmem_cache_zalloc(ceph_dentry_cachep, GFP_KERNEL); if (!di) return -ENOMEM; /* oh well */ di->dentry = dentry; di->lease_session = NULL; di->time = jiffies; dentry->d_fsdata = di; INIT_LIST_HEAD(&di->lease_list); atomic64_inc(&mdsc->metric.total_dentries); return 0; } /* * for f_pos for readdir: * - hash order: * (0xff << 52) | ((24 bits hash) << 28) | * (the nth entry has hash collision); * - frag+name order; * ((frag value) << 28) | (the nth entry in frag); */ #define OFFSET_BITS 28 #define OFFSET_MASK ((1 << OFFSET_BITS) - 1) #define HASH_ORDER (0xffull << (OFFSET_BITS + 24)) loff_t ceph_make_fpos(unsigned high, unsigned off, bool hash_order) { loff_t fpos = ((loff_t)high << 28) | (loff_t)off; if (hash_order) fpos |= HASH_ORDER; return fpos; } static bool is_hash_order(loff_t p) { return (p & HASH_ORDER) == HASH_ORDER; } static unsigned fpos_frag(loff_t p) { return p >> OFFSET_BITS; } static unsigned fpos_hash(loff_t p) { return ceph_frag_value(fpos_frag(p)); } static unsigned fpos_off(loff_t p) { return p & OFFSET_MASK; } static int fpos_cmp(loff_t l, loff_t r) { int v = ceph_frag_compare(fpos_frag(l), fpos_frag(r)); if (v) return v; return (int)(fpos_off(l) - fpos_off(r)); } /* * make note of the last dentry we read, so we can * continue at the same lexicographical point, * regardless of what dir changes take place on the * server. */ static int note_last_dentry(struct ceph_fs_client *fsc, struct ceph_dir_file_info *dfi, const char *name, int len, unsigned next_offset) { char *buf = kmalloc(len+1, GFP_KERNEL); if (!buf) return -ENOMEM; kfree(dfi->last_name); dfi->last_name = buf; memcpy(dfi->last_name, name, len); dfi->last_name[len] = 0; dfi->next_offset = next_offset; doutc(fsc->client, "'%s'\n", dfi->last_name); return 0; } static struct dentry * __dcache_find_get_entry(struct dentry *parent, u64 idx, struct ceph_readdir_cache_control *cache_ctl) { struct inode *dir = d_inode(parent); struct ceph_client *cl = ceph_inode_to_client(dir); struct dentry *dentry; unsigned idx_mask = (PAGE_SIZE / sizeof(struct dentry *)) - 1; loff_t ptr_pos = idx * sizeof(struct dentry *); pgoff_t ptr_pgoff = ptr_pos >> PAGE_SHIFT; if (ptr_pos >= i_size_read(dir)) return NULL; if (!cache_ctl->folio || ptr_pgoff != cache_ctl->folio->index) { ceph_readdir_cache_release(cache_ctl); cache_ctl->folio = filemap_lock_folio(&dir->i_data, ptr_pgoff); if (IS_ERR(cache_ctl->folio)) { cache_ctl->folio = NULL; doutc(cl, " folio %lu not found\n", ptr_pgoff); return ERR_PTR(-EAGAIN); } /* reading/filling the cache are serialized by i_rwsem, no need to use folio lock */ folio_unlock(cache_ctl->folio); cache_ctl->dentries = kmap_local_folio(cache_ctl->folio, 0); } cache_ctl->index = idx & idx_mask; rcu_read_lock(); spin_lock(&parent->d_lock); /* check i_size again here, because empty directory can be * marked as complete while not holding the i_rwsem. */ if (ceph_dir_is_complete_ordered(dir) && ptr_pos < i_size_read(dir)) dentry = cache_ctl->dentries[cache_ctl->index]; else dentry = NULL; spin_unlock(&parent->d_lock); if (dentry && !lockref_get_not_dead(&dentry->d_lockref)) dentry = NULL; rcu_read_unlock(); return dentry ? : ERR_PTR(-EAGAIN); } /* * When possible, we try to satisfy a readdir by peeking at the * dcache. We make this work by carefully ordering dentries on * d_children when we initially get results back from the MDS, and * falling back to a "normal" sync readdir if any dentries in the dir * are dropped. * * Complete dir indicates that we have all dentries in the dir. It is * defined IFF we hold CEPH_CAP_FILE_SHARED (which will be revoked by * the MDS if/when the directory is modified). */ static int __dcache_readdir(struct file *file, struct dir_context *ctx, int shared_gen) { struct ceph_dir_file_info *dfi = file->private_data; struct dentry *parent = file->f_path.dentry; struct inode *dir = d_inode(parent); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(dir); struct ceph_client *cl = ceph_inode_to_client(dir); struct dentry *dentry, *last = NULL; struct ceph_dentry_info *di; struct ceph_readdir_cache_control cache_ctl = {}; u64 idx = 0; int err = 0; doutc(cl, "%p %llx.%llx v%u at %llx\n", dir, ceph_vinop(dir), (unsigned)shared_gen, ctx->pos); /* search start position */ if (ctx->pos > 2) { u64 count = div_u64(i_size_read(dir), sizeof(struct dentry *)); while (count > 0) { u64 step = count >> 1; dentry = __dcache_find_get_entry(parent, idx + step, &cache_ctl); if (!dentry) { /* use linear search */ idx = 0; break; } if (IS_ERR(dentry)) { err = PTR_ERR(dentry); goto out; } di = ceph_dentry(dentry); spin_lock(&dentry->d_lock); if (fpos_cmp(di->offset, ctx->pos) < 0) { idx += step + 1; count -= step + 1; } else { count = step; } spin_unlock(&dentry->d_lock); dput(dentry); } doutc(cl, "%p %llx.%llx cache idx %llu\n", dir, ceph_vinop(dir), idx); } for (;;) { bool emit_dentry = false; dentry = __dcache_find_get_entry(parent, idx++, &cache_ctl); if (!dentry) { dfi->file_info.flags |= CEPH_F_ATEND; err = 0; break; } if (IS_ERR(dentry)) { err = PTR_ERR(dentry); goto out; } spin_lock(&dentry->d_lock); di = ceph_dentry(dentry); if (d_unhashed(dentry) || d_really_is_negative(dentry) || di->lease_shared_gen != shared_gen || ((dentry->d_flags & DCACHE_NOKEY_NAME) && fscrypt_has_encryption_key(dir))) { spin_unlock(&dentry->d_lock); dput(dentry); err = -EAGAIN; goto out; } if (fpos_cmp(ctx->pos, di->offset) <= 0) { __ceph_dentry_dir_lease_touch(di); emit_dentry = true; } spin_unlock(&dentry->d_lock); if (emit_dentry) { doutc(cl, " %llx dentry %p %pd %p\n", di->offset, dentry, dentry, d_inode(dentry)); ctx->pos = di->offset; if (!dir_emit(ctx, dentry->d_name.name, dentry->d_name.len, ceph_present_inode(d_inode(dentry)), d_inode(dentry)->i_mode >> 12)) { dput(dentry); err = 0; break; } ctx->pos++; if (last) dput(last); last = dentry; } else { dput(dentry); } } out: ceph_readdir_cache_release(&cache_ctl); if (last) { int ret; di = ceph_dentry(last); ret = note_last_dentry(fsc, dfi, last->d_name.name, last->d_name.len, fpos_off(di->offset) + 1); if (ret < 0) err = ret; dput(last); /* last_name no longer match cache index */ if (dfi->readdir_cache_idx >= 0) { dfi->readdir_cache_idx = -1; dfi->dir_release_count = 0; } } return err; } static bool need_send_readdir(struct ceph_dir_file_info *dfi, loff_t pos) { if (!dfi->last_readdir) return true; if (is_hash_order(pos)) return !ceph_frag_contains_value(dfi->frag, fpos_hash(pos)); else return dfi->frag != fpos_frag(pos); } static int ceph_readdir(struct file *file, struct dir_context *ctx) { struct ceph_dir_file_info *dfi = file->private_data; struct inode *inode = file_inode(file); struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_mds_client *mdsc = fsc->mdsc; struct ceph_client *cl = fsc->client; int i; int err; unsigned frag = -1; struct ceph_mds_reply_info_parsed *rinfo; doutc(cl, "%p %llx.%llx file %p pos %llx\n", inode, ceph_vinop(inode), file, ctx->pos); if (dfi->file_info.flags & CEPH_F_ATEND) return 0; /* always start with . and .. */ if (ctx->pos == 0) { doutc(cl, "%p %llx.%llx off 0 -> '.'\n", inode, ceph_vinop(inode)); if (!dir_emit(ctx, ".", 1, ceph_present_inode(inode), inode->i_mode >> 12)) return 0; ctx->pos = 1; } if (ctx->pos == 1) { u64 ino; struct dentry *dentry = file->f_path.dentry; spin_lock(&dentry->d_lock); ino = ceph_present_inode(dentry->d_parent->d_inode); spin_unlock(&dentry->d_lock); doutc(cl, "%p %llx.%llx off 1 -> '..'\n", inode, ceph_vinop(inode)); if (!dir_emit(ctx, "..", 2, ino, inode->i_mode >> 12)) return 0; ctx->pos = 2; } err = ceph_fscrypt_prepare_readdir(inode); if (err < 0) return err; spin_lock(&ci->i_ceph_lock); /* request Fx cap. if have Fx, we don't need to release Fs cap * for later create/unlink. */ __ceph_touch_fmode(ci, mdsc, CEPH_FILE_MODE_WR); /* can we use the dcache? */ if (ceph_test_mount_opt(fsc, DCACHE) && !ceph_test_mount_opt(fsc, NOASYNCREADDIR) && ceph_snap(inode) != CEPH_SNAPDIR && __ceph_dir_is_complete_ordered(ci) && __ceph_caps_issued_mask_metric(ci, CEPH_CAP_FILE_SHARED, 1)) { int shared_gen = atomic_read(&ci->i_shared_gen); spin_unlock(&ci->i_ceph_lock); err = __dcache_readdir(file, ctx, shared_gen); if (err != -EAGAIN) return err; } else { spin_unlock(&ci->i_ceph_lock); } /* proceed with a normal readdir */ more: /* do we have the correct frag content buffered? */ if (need_send_readdir(dfi, ctx->pos)) { struct ceph_mds_request *req; int op = ceph_snap(inode) == CEPH_SNAPDIR ? CEPH_MDS_OP_LSSNAP : CEPH_MDS_OP_READDIR; /* discard old result, if any */ if (dfi->last_readdir) { ceph_mdsc_put_request(dfi->last_readdir); dfi->last_readdir = NULL; } if (is_hash_order(ctx->pos)) { /* fragtree isn't always accurate. choose frag * based on previous reply when possible. */ if (frag == (unsigned)-1) frag = ceph_choose_frag(ci, fpos_hash(ctx->pos), NULL, NULL); } else { frag = fpos_frag(ctx->pos); } doutc(cl, "fetching %p %llx.%llx frag %x offset '%s'\n", inode, ceph_vinop(inode), frag, dfi->last_name); req = ceph_mdsc_create_request(mdsc, op, USE_AUTH_MDS); if (IS_ERR(req)) return PTR_ERR(req); err = ceph_alloc_readdir_reply_buffer(req, inode); if (err) { ceph_mdsc_put_request(req); return err; } /* hints to request -> mds selection code */ req->r_direct_mode = USE_AUTH_MDS; if (op == CEPH_MDS_OP_READDIR) { req->r_direct_hash = ceph_frag_value(frag); __set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags); req->r_inode_drop = CEPH_CAP_FILE_EXCL; } if (dfi->last_name) { int len = strlen(dfi->last_name); req->r_path2 = kzalloc(NAME_MAX + 1, GFP_KERNEL); if (!req->r_path2) { ceph_mdsc_put_request(req); return -ENOMEM; } memcpy(req->r_path2, dfi->last_name, len); err = ceph_encode_encrypted_dname(inode, req->r_path2, len); if (err < 0) { ceph_mdsc_put_request(req); return err; } } else if (is_hash_order(ctx->pos)) { req->r_args.readdir.offset_hash = cpu_to_le32(fpos_hash(ctx->pos)); } req->r_dir_release_cnt = dfi->dir_release_count; req->r_dir_ordered_cnt = dfi->dir_ordered_count; req->r_readdir_cache_idx = dfi->readdir_cache_idx; req->r_readdir_offset = dfi->next_offset; req->r_args.readdir.frag = cpu_to_le32(frag); req->r_args.readdir.flags = cpu_to_le16(CEPH_READDIR_REPLY_BITFLAGS); req->r_inode = inode; ihold(inode); req->r_dentry = dget(file->f_path.dentry); err = ceph_mdsc_do_request(mdsc, NULL, req); if (err < 0) { ceph_mdsc_put_request(req); return err; } doutc(cl, "%p %llx.%llx got and parsed readdir result=%d" "on frag %x, end=%d, complete=%d, hash_order=%d\n", inode, ceph_vinop(inode), err, frag, (int)req->r_reply_info.dir_end, (int)req->r_reply_info.dir_complete, (int)req->r_reply_info.hash_order); rinfo = &req->r_reply_info; if (le32_to_cpu(rinfo->dir_dir->frag) != frag) { frag = le32_to_cpu(rinfo->dir_dir->frag); if (!rinfo->hash_order) { dfi->next_offset = req->r_readdir_offset; /* adjust ctx->pos to beginning of frag */ ctx->pos = ceph_make_fpos(frag, dfi->next_offset, false); } } dfi->frag = frag; dfi->last_readdir = req; if (test_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags)) { dfi->readdir_cache_idx = req->r_readdir_cache_idx; if (dfi->readdir_cache_idx < 0) { /* preclude from marking dir ordered */ dfi->dir_ordered_count = 0; } else if (ceph_frag_is_leftmost(frag) && dfi->next_offset == 2) { /* note dir version at start of readdir so * we can tell if any dentries get dropped */ dfi->dir_release_count = req->r_dir_release_cnt; dfi->dir_ordered_count = req->r_dir_ordered_cnt; } } else { doutc(cl, "%p %llx.%llx !did_prepopulate\n", inode, ceph_vinop(inode)); /* disable readdir cache */ dfi->readdir_cache_idx = -1; /* preclude from marking dir complete */