scoutfs/kmod/src/dir.c

/*
 * Copyright (C) 2016 Versity Software, Inc.  All rights reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License v2 as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 */
#include <linux/kernel.h>
#include <linux/fs.h>
#include <linux/slab.h>
#include <linux/crc32c.h>
#include <linux/uio.h>
#include <linux/xattr.h>
#include <linux/namei.h>

#include "format.h"
#include "dir.h"
#include "inode.h"
#include "key.h"
#include "super.h"
#include "btree.h"
#include "trans.h"
#include "name.h"
#include "xattr.h"
#include "kvec.h"
#include "item.h"

/*
 * Directory entries are stored in entries with offsets calculated from
 * the hash of their entry name.
 *
 * Having a single index of items used for both lookup and readdir
 * iteration reduces the storage overhead of directories.  It also
 * avoids having to manage the allocation of readdir positions as
 * directories age and the aggregate create count inches towards the
 * small 31 bit position limit.  The downside is that dirent name
 * operations produce random item access patterns.
 *
 * Hash values are limited to 31 bits primarily to support older
 * deployed protocols that only support 31 bits of file entry offsets,
 * but also to avoid unlikely bugs in programs that store offsets in
 * signed ints.
 *
 * We have to worry about hash collisions.  We linearly probe a fixed
 * number of hash values past the natural value.  In a typical small
 * directory this search will terminate immediately because adjacent
 * items will have distant offset values.  It's only as the directory
 * gets very large that hash values will start to be this dense and
 * sweeping over items in a btree leaf is reasonably efficient.
 *
 * For each directory entry item stored in a directory inode there is a
 * corresponding link backref item stored at the target inode.  This
 * lets us find all the paths that refer to a given inode.  The link
 * backref offset comes from an advancing counter in the inode and the
 * item value contains the dir inode and dirent offset of the referring
 * link.
 */

static unsigned int mode_to_type(umode_t mode)
{
#define S_SHIFT 12
	static unsigned char mode_types[S_IFMT >> S_SHIFT] = {
		[S_IFIFO >> S_SHIFT]	= SCOUTFS_DT_FIFO,
		[S_IFCHR >> S_SHIFT]	= SCOUTFS_DT_CHR,
		[S_IFDIR >> S_SHIFT]	= SCOUTFS_DT_DIR,
		[S_IFBLK >> S_SHIFT]	= SCOUTFS_DT_BLK,
		[S_IFREG >> S_SHIFT]	= SCOUTFS_DT_REG,
		[S_IFLNK >> S_SHIFT]	= SCOUTFS_DT_LNK,
		[S_IFSOCK >> S_SHIFT]	= SCOUTFS_DT_SOCK,
	};

	return mode_types[(mode & S_IFMT) >> S_SHIFT];
#undef S_SHIFT
}

static unsigned int dentry_type(unsigned int type)
{
	static unsigned char types[] = {
		[SCOUTFS_DT_FIFO]	= DT_FIFO,
		[SCOUTFS_DT_CHR]	= DT_CHR,
		[SCOUTFS_DT_DIR]	= DT_DIR,
		[SCOUTFS_DT_BLK]	= DT_BLK,
		[SCOUTFS_DT_REG]	= DT_REG,
		[SCOUTFS_DT_LNK]	= DT_LNK,
		[SCOUTFS_DT_SOCK]	= DT_SOCK,
		[SCOUTFS_DT_WHT]	= DT_WHT,
	};

	if (type < ARRAY_SIZE(types))
		return types[type];

	return DT_UNKNOWN;
}

/*
 * Each dentry stores the values that are needed to build the keys of
 * the items that are removed on unlink so that we don't to search
 * through items on unlink.
 */
struct dentry_info {
	u64 readdir_pos;
};

static struct kmem_cache *dentry_info_cache;

static void scoutfs_d_release(struct dentry *dentry)
{
	struct dentry_info *di = dentry->d_fsdata;

	if (di) {
		kmem_cache_free(dentry_info_cache, di);
		dentry->d_fsdata = NULL;
	}
}

static const struct dentry_operations scoutfs_dentry_ops = {
	.d_release = scoutfs_d_release,
};

static int alloc_dentry_info(struct dentry *dentry)
{
	struct dentry_info *di;

	/* XXX read mb? */
	if (dentry->d_fsdata)
		return 0;

	di = kmem_cache_zalloc(dentry_info_cache, GFP_NOFS);
	if (!di)
		return -ENOMEM;

	spin_lock(&dentry->d_lock);
	if (!dentry->d_fsdata) {
		dentry->d_fsdata = di;
		d_set_d_op(dentry, &scoutfs_dentry_ops);
	}
	spin_unlock(&dentry->d_lock);

	if (di != dentry->d_fsdata)
		kmem_cache_free(dentry_info_cache, di);

	return 0;
}

static void update_dentry_info(struct dentry *dentry,
			       struct scoutfs_dirent *dent)
{
	struct dentry_info *di = dentry->d_fsdata;

	if (WARN_ON_ONCE(di == NULL))
		return;

	di->readdir_pos = le64_to_cpu(dent->readdir_pos);
}

static u64 dentry_info_pos(struct dentry *dentry)
{
	struct dentry_info *di = dentry->d_fsdata;

	if (WARN_ON_ONCE(di == NULL))
		return 0;

	return di->readdir_pos;
}

static struct scoutfs_key_buf *alloc_dirent_key(struct super_block *sb,
						struct inode *dir,
						struct dentry *dentry)
{
	struct scoutfs_dirent_key *dkey;
	struct scoutfs_key_buf *key;

	key = scoutfs_key_alloc(sb, offsetof(struct scoutfs_dirent_key,
					     name[dentry->d_name.len]));
	if (key) {
		dkey = key->data;
		dkey->type = SCOUTFS_DIRENT_KEY;
		dkey->ino = cpu_to_be64(scoutfs_ino(dir));
		memcpy(dkey->name, (void *)dentry->d_name.name,
		       dentry->d_name.len);
	}

	return key;
}

static void init_link_backref_key(struct scoutfs_key_buf *key,
			          struct scoutfs_link_backref_key *lbrkey,
				  u64 ino, u64 dir_ino,
				  char *name, unsigned name_len)
{
	lbrkey->type = SCOUTFS_LINK_BACKREF_KEY;
	lbrkey->ino = cpu_to_be64(ino);
	lbrkey->dir_ino = cpu_to_be64(dir_ino);
	if (name_len)
		memcpy(lbrkey->name, name, name_len);

	scoutfs_key_init(key, lbrkey, offsetof(struct scoutfs_link_backref_key,
					       name[name_len]));
}

static struct scoutfs_key_buf *alloc_link_backref_key(struct super_block *sb,
						      u64 ino, u64 dir_ino,
						      char *name,
						      unsigned name_len)
{
	struct scoutfs_link_backref_key *lbkey;
	struct scoutfs_key_buf *key;

	key = scoutfs_key_alloc(sb, offsetof(struct scoutfs_link_backref_key,
					     name[name_len]));
	if (key) {
		lbkey = key->data;
		init_link_backref_key(key, lbkey, ino, dir_ino,
				      name, name_len);
	}

	return key;
}

static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry,
				     unsigned int flags)
{
	struct super_block *sb = dir->i_sb;
	struct scoutfs_key_buf *key = NULL;
	struct scoutfs_dirent dent;
	SCOUTFS_DECLARE_KVEC(val);
	struct inode *inode;
	u64 ino = 0;
	int ret;

	if (dentry->d_name.len > SCOUTFS_NAME_LEN) {
		ret = -ENAMETOOLONG;
		goto out;
	}

	ret = alloc_dentry_info(dentry);
	if (ret)
		goto out;

	key = alloc_dirent_key(sb, dir, dentry);
	if (!key) {
		ret = -ENOMEM;
		goto out;
	}

	scoutfs_kvec_init(val, &dent, sizeof(dent));

	ret = scoutfs_item_lookup_exact(sb, key, val, sizeof(dent));
	if (ret == -ENOENT) {
		ino = 0;
		ret = 0;
	} else if (ret == 0) {
		ino = le64_to_cpu(dent.ino);
		update_dentry_info(dentry, &dent);
	}

out:
	if (ret < 0)
		inode = ERR_PTR(ret);
	else if (ino == 0)
		inode = NULL;
	else
		inode = scoutfs_iget(sb, ino);

	scoutfs_key_free(sb, key);

	return d_splice_alias(inode, dentry);
}

/* this exists upstream so we can just delete it in a forward port */
static int dir_emit_dots(struct file *file, void *dirent, filldir_t filldir)
{
	struct dentry *dentry = file->f_path.dentry;
	struct inode *inode = dentry->d_inode;
	struct inode *parent = dentry->d_parent->d_inode;

	if (file->f_pos == 0) {
		if (filldir(dirent, ".", 1, 1, scoutfs_ino(inode), DT_DIR))
			return 0;
		file->f_pos = 1;
	}

	if (file->f_pos == 1) {
		if (filldir(dirent, "..", 2, 1, scoutfs_ino(parent), DT_DIR))
			return 0;
		file->f_pos = 2;
	}

	return 1;
}

static void init_readdir_key(struct scoutfs_key_buf *key,
			     struct scoutfs_readdir_key *rkey,
			     struct inode *inode, loff_t pos)
{
	rkey->type = SCOUTFS_READDIR_KEY;
	rkey->ino = cpu_to_be64(scoutfs_ino(inode));
	rkey->pos = cpu_to_be64(pos);

	scoutfs_key_init(key, rkey, sizeof(struct scoutfs_readdir_key));
}

/*
 * readdir simply iterates over the dirent items for the dir inode and
 * uses their offset as the readdir position.
 *
 * It will need to be careful not to read past the region of the dirent
 * hash offset keys that it has access to.
 */
static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir)
{
	struct inode *inode = file_inode(file);
	struct super_block *sb = inode->i_sb;
	struct scoutfs_dirent *dent;
	struct scoutfs_key_buf key;
	struct scoutfs_key_buf last_key;
	struct scoutfs_readdir_key rkey;
	struct scoutfs_readdir_key last_rkey;
	SCOUTFS_DECLARE_KVEC(val);
	unsigned int item_len;
	unsigned int name_len;
	u64 pos;
	int ret;

	if (!dir_emit_dots(file, dirent, filldir))
		return 0;

	init_readdir_key(&last_key, &last_rkey, inode, SCOUTFS_DIRENT_LAST_POS);

	item_len = offsetof(struct scoutfs_dirent, name[SCOUTFS_NAME_LEN]);
	dent = kmalloc(item_len, GFP_KERNEL);
	if (!dent)
		return -ENOMEM;

	for (;;) {
		init_readdir_key(&key, &rkey, inode, file->f_pos);

		scoutfs_kvec_init(val, dent, item_len);
		ret = scoutfs_item_next_same_min(sb, &key, &last_key, val,
				offsetof(struct scoutfs_dirent, name[1]));
		if (ret < 0) {
			if (ret == -ENOENT)
				ret = 0;
			break;
		}

		name_len = ret - sizeof(struct scoutfs_dirent);
		pos = be64_to_cpu(rkey.pos);

		if (filldir(dirent, dent->name, name_len, pos,
			    le64_to_cpu(dent->ino), dentry_type(dent->type))) {
			ret = 0;
			break;
		}

		file->f_pos = pos + 1;
	}

	kfree(dent);
	return ret;
}

static int add_entry_items(struct inode *dir, struct dentry *dentry,
			   struct inode *inode)
{
	struct scoutfs_inode_info *si = SCOUTFS_I(dir);
	struct dentry_info *di = dentry->d_fsdata;
	struct super_block *sb = dir->i_sb;
	struct scoutfs_key_buf *ent_key = NULL;
	struct scoutfs_key_buf *lb_key = NULL;
	struct scoutfs_key_buf *del_keys[3];
	struct scoutfs_key_buf rdir_key;
	struct scoutfs_readdir_key rkey;
	struct scoutfs_dirent dent;
	SCOUTFS_DECLARE_KVEC(val);
	int del = 0;
	u64 pos;
	int ret;
	int err;

	/* caller should have allocated the dentry info */
	if (WARN_ON_ONCE(di == NULL))
		return -EINVAL;

	if (dentry->d_name.len > SCOUTFS_NAME_LEN)
		return -ENAMETOOLONG;

	ret = scoutfs_dirty_inode_item(dir);
	if (ret)
		return ret;

	/* initialize the dent */
	pos = si->next_readdir_pos++;
	dent.ino = cpu_to_le64(scoutfs_ino(inode));
	dent.readdir_pos = cpu_to_le64(pos);
	dent.type = mode_to_type(inode->i_mode);

	/* dirent item for lookup */
	ent_key = alloc_dirent_key(sb, dir, dentry);
	if (!ent_key)
		return -ENOMEM;

	scoutfs_kvec_init(val, &dent, sizeof(dent));

	ret = scoutfs_item_create(sb, ent_key, val);
	if (ret)
		goto out;
	del_keys[del++] = ent_key;

	/* readdir item for .. readdir */
	init_readdir_key(&rdir_key, &rkey, dir, pos);
	scoutfs_kvec_init(val, &dent, sizeof(dent),
			  (void *)dentry->d_name.name, dentry->d_name.len);

	ret = scoutfs_item_create(sb, &rdir_key, val);
	if (ret)
		goto out;
	del_keys[del++] = &rdir_key;

	/* link backref item for inode to path resolution */
	lb_key = alloc_link_backref_key(sb, scoutfs_ino(inode),
					scoutfs_ino(dir),
					(void *)dentry->d_name.name,
					dentry->d_name.len);
	if (!lb_key) {
		ret = -ENOMEM;
		goto out;
	}

	ret = scoutfs_item_create(sb, lb_key, NULL);
	if (ret)
		goto out;
	del_keys[del++] = lb_key;

	update_dentry_info(dentry, &dent);
	ret = 0;
out:
	while (ret < 0 && --del >= 0) {
		err = scoutfs_item_delete(sb, del_keys[del]);
		/* can always delete dirty while holding */
		BUG_ON(err);
	}

	scoutfs_key_free(sb, ent_key);
	scoutfs_key_free(sb, lb_key);

	return ret;
}

static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
		       dev_t rdev)
{
	struct super_block *sb = dir->i_sb;
	struct inode *inode;
	int ret;

	ret = alloc_dentry_info(dentry);
	if (ret)
		return ret;

	ret = scoutfs_hold_trans(sb);
	if (ret)
		return ret;

	inode = scoutfs_new_inode(sb, dir, mode, rdev);
	if (IS_ERR(inode)) {
		ret = PTR_ERR(inode);
		goto out;
	}

	ret = add_entry_items(dir, dentry, inode);
	if (ret)
		goto out;

	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
	inode->i_mtime = inode->i_atime = inode->i_ctime = dir->i_mtime;

	if (S_ISDIR(mode)) {
		inc_nlink(inode);
		inc_nlink(dir);
	}

	scoutfs_update_inode_item(inode);
	scoutfs_update_inode_item(dir);

	insert_inode_hash(inode);
	d_instantiate(dentry, inode);
out:
	/* XXX delete the inode item here */
	if (ret && !IS_ERR_OR_NULL(inode))
		iput(inode);
	scoutfs_release_trans(sb);
	return ret;
}

/* XXX hmm, do something with excl? */
static int scoutfs_create(struct inode *dir, struct dentry *dentry,
			  umode_t mode, bool excl)
{
	return scoutfs_mknod(dir, dentry, mode | S_IFREG, 0);
}

static int scoutfs_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
{
	return scoutfs_mknod(dir, dentry, mode | S_IFDIR, 0);
}

static int scoutfs_link(struct dentry *old_dentry,
			struct inode *dir, struct dentry *dentry)
{
	struct inode *inode = old_dentry->d_inode;
	struct super_block *sb = dir->i_sb;
	int ret;

	if (inode->i_nlink >= SCOUTFS_LINK_MAX)
		return -EMLINK;

	ret = alloc_dentry_info(dentry);
	if (ret)
		return ret;

	ret = scoutfs_hold_trans(sb);
	if (ret)
		return ret;

	ret = add_entry_items(dir, dentry, inode);
	if (ret)
		goto out;

	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
	inode->i_ctime = dir->i_mtime;
	inc_nlink(inode);

	scoutfs_update_inode_item(inode);
	scoutfs_update_inode_item(dir);

	atomic_inc(&inode->i_count);
	d_instantiate(dentry, inode);
out:
	scoutfs_release_trans(sb);
	return ret;
}

/*
 * Unlink removes the entry from its item and removes the item if ours
 * was the only remaining entry.
 */
static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
{
	struct super_block *sb = dir->i_sb;
	struct inode *inode = dentry->d_inode;
	struct timespec ts = current_kernel_time();
	struct scoutfs_key_buf *keys[3] = {NULL,};
	struct scoutfs_key_buf rdir_key;
	struct scoutfs_readdir_key rkey;
	int ret = 0;

	if (S_ISDIR(inode->i_mode) && i_size_read(inode))
		return -ENOTEMPTY;

	ret = scoutfs_hold_trans(sb);
	if (ret)
		return ret;

	ret = scoutfs_dirty_inode_item(dir) ?:
	      scoutfs_dirty_inode_item(inode);
	if (ret)
		goto out;

	keys[0] = alloc_dirent_key(sb, dir, dentry);
	if (!keys[0]) {
		ret = -ENOMEM;
		goto out;
	}

	init_readdir_key(&rdir_key, &rkey, dir, dentry_info_pos(dentry));
	keys[1] = &rdir_key;

	keys[2] = alloc_link_backref_key(sb, scoutfs_ino(inode),
					 scoutfs_ino(dir),
					 (void *)dentry->d_name.name,
					 dentry->d_name.len);
	if (!keys[2]) {
		ret = -ENOMEM;
		goto out;
	}

	ret = scoutfs_item_delete_many(sb, keys, ARRAY_SIZE(keys));
	if (ret)
		goto out;

	if ((inode->i_nlink == 1) ||
	    (S_ISDIR(inode->i_mode) && inode->i_nlink == 2)) {
		/*
		 * Insert the orphan item before we modify any inode
		 * metadata so we can gracefully exit should it
		 * fail.
		 */
		ret = scoutfs_orphan_inode(inode);
		if (ret)
			goto out;
	}

	dir->i_ctime = ts;
	dir->i_mtime = ts;
	i_size_write(dir, i_size_read(dir) - dentry->d_name.len);

	inode->i_ctime = ts;
	drop_nlink(inode);
	if (S_ISDIR(inode->i_mode)) {
		drop_nlink(dir);
		drop_nlink(inode);
	}
	scoutfs_update_inode_item(inode);
	scoutfs_update_inode_item(dir);

out:
	scoutfs_key_free(sb, keys[0]);
	scoutfs_key_free(sb, keys[2]);
	scoutfs_release_trans(sb);
	return ret;
}

/*
 * Full a buffer with the null terminated symlink, point nd at it, and
 * return it so put_link can free it once the vfs is done.
 *
 * We chose to pay the runtime cost of per-call allocation and copy
 * overhead instead of wiring up symlinks to the page cache, storing
 * each small link in a full page, and later having to reclaim them.
 */
static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
{
	struct inode *inode = dentry->d_inode;
	struct super_block *sb = inode->i_sb;
	struct scoutfs_btree_root *meta = SCOUTFS_META(sb);
	loff_t size = i_size_read(inode);
	struct scoutfs_btree_val val;
	struct scoutfs_key key;
	char *path;
	int bytes;
	int off;
	int ret;
	int k;

	/* update for kvec items */
	return ERR_PTR(-EINVAL);

	/* XXX corruption */
	if (size == 0 || size > SCOUTFS_SYMLINK_MAX_SIZE)
		return ERR_PTR(-EIO);

	/* unlikely, but possible I suppose */
	if (size > PATH_MAX)
		return ERR_PTR(-ENAMETOOLONG);

	path = kmalloc(size, GFP_NOFS);
	if (!path)
		return ERR_PTR(-ENOMEM);

	for (off = 0, k = 0; off < size ; k++) {
		scoutfs_set_key(&key, scoutfs_ino(inode),
				SCOUTFS_SYMLINK_KEY, k);
		bytes = min_t(int, size - off, SCOUTFS_MAX_ITEM_LEN);
		scoutfs_btree_init_val(&val, path + off, bytes);
		val.check_size_eq = 1;

		ret = scoutfs_btree_lookup(sb, meta, &key, &val);
		if (ret < 0) {
			/* XXX corruption */
			if (ret == -ENOENT)
				ret = -EIO;
			break;
		}

		off += bytes;
		ret = 0;
	}

	/* XXX corruption */
	if (ret == 0 && (off != size || path[off - 1] != '\0'))
		ret = -EIO;

	if (ret) {
		kfree(path);
		path = ERR_PTR(ret);
	} else {
		nd_set_link(nd, path);
	}

	return path;
}

static void scoutfs_put_link(struct dentry *dentry, struct nameidata *nd,
			     void *cookie)
{
	if (!IS_ERR_OR_NULL(cookie))
		kfree(cookie);
}

const struct inode_operations scoutfs_symlink_iops = {
	.readlink       = generic_readlink,
	.follow_link    = scoutfs_follow_link,
	.put_link       = scoutfs_put_link,
	.setxattr	= scoutfs_setxattr,
	.getxattr	= scoutfs_getxattr,
	.listxattr	= scoutfs_listxattr,
	.removexattr	= scoutfs_removexattr,
};

/*
 * Symlink target paths can be annoyingly huge.  We don't want large
 * items gumming up the btree so we store relatively rare large paths in
 * multiple items.
 */
static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
			   const char *symname)
{
	struct super_block *sb = dir->i_sb;
	struct scoutfs_btree_root *meta = SCOUTFS_META(sb);
	struct scoutfs_btree_val val;
	struct inode *inode = NULL;
	struct scoutfs_key key;
	const int name_len = strlen(symname) + 1;
	int off;
	int bytes;
	int ret;
	int k = 0;

	/* update for kvec items */
	return -EINVAL;

	/* path_max includes null as does our value for nd_set_link */
	if (name_len > PATH_MAX || name_len > SCOUTFS_SYMLINK_MAX_SIZE)
		return -ENAMETOOLONG;

	ret = alloc_dentry_info(dentry);
	if (ret)
		return ret;

	ret = scoutfs_hold_trans(sb);
	if (ret)
		return ret;

	inode = scoutfs_new_inode(sb, dir, S_IFLNK|S_IRWXUGO, 0);
	if (IS_ERR(inode)) {
		ret = PTR_ERR(inode);
		goto out;
	}

	for (k = 0, off = 0; off < name_len; off += bytes, k++) {
		scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_SYMLINK_KEY,
				k);
		bytes = min(name_len - off, SCOUTFS_MAX_ITEM_LEN);

		scoutfs_btree_init_val(&val, (char *)symname + off, bytes);

		ret = scoutfs_btree_insert(sb, meta, &key, &val);
		if (ret)
			goto out;
	}

	ret = add_entry_items(dir, dentry, inode);
	if (ret)
		goto out;

	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
	dir->i_mtime = dir->i_ctime = CURRENT_TIME;

	inode->i_ctime = dir->i_mtime;
	i_size_write(inode, name_len);

	scoutfs_update_inode_item(inode);
	scoutfs_update_inode_item(dir);

	insert_inode_hash(inode);
	/* XXX need to set i_op/fop before here for sec callbacks */
	d_instantiate(dentry, inode);
out:
	if (ret < 0) {
		if (!IS_ERR_OR_NULL(inode))
			iput(inode);

		while (k--) {
			scoutfs_set_key(&key, scoutfs_ino(inode),
					SCOUTFS_SYMLINK_KEY, k);
			scoutfs_btree_delete(sb, meta, &key);
		}
	}

	scoutfs_release_trans(sb);
	return ret;
}

/*
 * Delete all the symlink items.  There should only ever be a handful of
 * these that contain the target path of the symlink.
 */
int scoutfs_symlink_drop(struct super_block *sb, u64 ino)
{
	struct scoutfs_btree_root *meta = SCOUTFS_META(sb);
	struct scoutfs_key key;
	int ret;
	int nr;
	int k;

	nr = DIV_ROUND_UP(SCOUTFS_SYMLINK_MAX_SIZE, SCOUTFS_MAX_ITEM_LEN);

	for (k = 0; k < nr; k++) {
		scoutfs_set_key(&key, ino, SCOUTFS_SYMLINK_KEY, k);

		ret = scoutfs_btree_delete(sb, meta, &key);
		if (ret < 0) {
			if (ret == -ENOENT)
				ret = 0;
			break;
		}
	}

	return ret;
}

/*
 * Find the next link backref key for the given ino starting from the
 * given dir inode and null terminated name.  If we find a backref item
 * we add an allocated copy of it to the head of the caller's list.
 *
 * Returns 0 if we added an entry, -ENOENT if we didn't, and -errno for
 * search errors.
 */
static int add_next_linkref(struct super_block *sb, u64 ino,
			    u64 dir_ino, char *name, unsigned int name_len,
			    struct list_head *list)
{
	struct scoutfs_link_backref_key last_lbkey;
	struct scoutfs_link_backref_entry *ent;
	struct scoutfs_key_buf last;
	struct scoutfs_key_buf key;
	int len;
	int ret;

	ent = kmalloc(offsetof(struct scoutfs_link_backref_entry,
			       lbkey.name[SCOUTFS_NAME_LEN + 1]), GFP_KERNEL);
	if (!ent)
		return -ENOMEM;

	INIT_LIST_HEAD(&ent->head);

	/* put search key in ent */
	init_link_backref_key(&key, &ent->lbkey, ino, dir_ino, name, name_len);
	/* we actually have room for a full backref item */
	scoutfs_key_init_buf_len(&key, key.data, key.key_len,
				 offsetof(struct scoutfs_link_backref_key,
					  name[SCOUTFS_NAME_LEN + 1]));

	/* small last key to avoid full name copy, XXX enforce no U64_MAX ino */
	init_link_backref_key(&last, &last_lbkey, ino, U64_MAX, NULL, 0);

	/* next backref key is now in ent */
	ret = scoutfs_item_next(sb, &key, &last, NULL);
	trace_printk("ino %llu dir_ino %llu ret %d key_len %u\n",
		      ino, dir_ino, ret, key.key_len);
	if (ret < 0)
		goto out;


	len = (int)key.key_len - sizeof(struct scoutfs_link_backref_key);
	/* XXX corruption */
	if (len < 1 || len > SCOUTFS_NAME_LEN) {
		ret = -EIO;
		goto out;
	}

	ent->name_len = len;
	list_add(&ent->head, list);
	ret = 0;
out:
	if (list_empty(&ent->head))
		kfree(ent);
	return ret;
}

static u64 first_backref_dir_ino(struct list_head *list)
{
	struct scoutfs_link_backref_entry *ent;

	ent = list_first_entry(list, struct scoutfs_link_backref_entry, head);
	return be64_to_cpu(ent->lbkey.dir_ino);
}

void scoutfs_dir_free_backref_path(struct super_block *sb,
				   struct list_head *list)
{
	struct scoutfs_link_backref_entry *ent;
	struct scoutfs_link_backref_entry *pos;

	list_for_each_entry_safe(ent, pos, list, head) {
		list_del_init(&ent->head);
		kfree(ent);
	}
}

/*
 * Give the caller the next path from the root to the inode by walking
 * backref items from the dir and name position, putting the backref keys
 * we find in the caller's list.
 *
 * Return 0 if we found a path, -ENOENT if we didn't, and -errno on error.
 *
 * If parents get unlinked while we're searching we can fail to make it
 * up to the root.  We restart the search in that case.  Parent dirs
 * couldn't have been unlinked while they still had entries and we won't
 * see links to the inode that have been unlinked.
 *
 * XXX Each path component traversal is consistent but that doesn't mean
 * that the total traversed path is consistent.  If renames hit dirs
 * that have been visited and then dirs to be visited we can return a
 * path that was never present in the system:
 *
 * path to inode     mv performed           built up path
 * ----
 * a/b/c/d/e/f
 *                                          d/e/f
 *                   mv a/b/c/d/e a/b/c/
 * a/b/c/e/f
 *                   mv a/b/c     a/
 * a/c/e/f
 *                                          a/c/d/e/f
 *
 * XXX We'll protect against this by sampling the seq before the
 * traversal and restarting if we saw backref items whose seq was
 * greater than the start point.  It's not precise in that it doesn't
 * also capture the rename of a dir that we already traversed but it
 * lets us complete the traversal in one pass that very rarely restarts.
 *
 * XXX and worry about traversing entirely dirty backref items with
 * equal seqs that have seen crazy modification?  seems like we have to
 * sync if we see our dirty seq.
 */
int scoutfs_dir_get_backref_path(struct super_block *sb, u64 ino, u64 dir_ino,
				 char *name, u16 name_len,
				 struct list_head *list)
{
	u64 par_ino;
	int ret;

retry:
	/* get the next link name to the given inode */
	ret = add_next_linkref(sb, ino, dir_ino, name, name_len, list);
	if (ret < 0)
		goto out;

	/* then get the names of all the parent dirs */
	par_ino = first_backref_dir_ino(list);
	while (par_ino != SCOUTFS_ROOT_INO) {

		ret = add_next_linkref(sb, par_ino, 0, NULL, 0, list);
		if (ret < 0) {
			if (ret == -ENOENT) {
				/* restart if there was no parent component */
				scoutfs_dir_free_backref_path(sb, list);
				goto retry;
			}
			goto out;
		}

		par_ino = first_backref_dir_ino(list);
	}
out:
	if (ret < 0)
		scoutfs_dir_free_backref_path(sb, list);
	return ret;
}

const struct file_operations scoutfs_dir_fops = {
	.readdir	= scoutfs_readdir,
	.fsync		= scoutfs_file_fsync,
};

const struct inode_operations scoutfs_dir_iops = {
	.lookup		= scoutfs_lookup,
	.mknod		= scoutfs_mknod,
	.create		= scoutfs_create,
	.mkdir		= scoutfs_mkdir,
	.link		= scoutfs_link,
	.unlink		= scoutfs_unlink,
	.rmdir		= scoutfs_unlink,
	.setxattr	= scoutfs_setxattr,
	.getxattr	= scoutfs_getxattr,
	.listxattr	= scoutfs_listxattr,
	.removexattr	= scoutfs_removexattr,
	.symlink	= scoutfs_symlink,
};

void scoutfs_dir_exit(void)
{
	if (dentry_info_cache) {
		kmem_cache_destroy(dentry_info_cache);
		dentry_info_cache = NULL;
	}
}

int scoutfs_dir_init(void)
{
	dentry_info_cache = kmem_cache_create("scoutfs_dentry_info",
					      sizeof(struct dentry_info), 0,
					      SLAB_RECLAIM_ACCOUNT, NULL);
	if (!dentry_info_cache)
		return -ENOMEM;

	return 0;
}