diff --git a/kmod/src/Makefile b/kmod/src/Makefile index e5712be8..dae6c279 100644 --- a/kmod/src/Makefile +++ b/kmod/src/Makefile @@ -1,4 +1,4 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o -scoutfs-y += block.o chunk.o crc.o dir.o inode.o item.o manifest.o msg.o \ - ring.o segment.o super.o +scoutfs-y += block.o bloom.o chunk.o crc.o dir.o inode.o manifest.o msg.o \ + ring.o segment.o skip.o super.o diff --git a/kmod/src/block.c b/kmod/src/block.c index 9102aaaa..8326382f 100644 --- a/kmod/src/block.c +++ b/kmod/src/block.c @@ -21,25 +21,13 @@ BUFFER_FNS(Private_Verified, private_verified) - -/* - * A quick metadata read wrapper which knows how to validate the - * block header. - */ -struct buffer_head *scoutfs_read_block(struct super_block *sb, u64 blkno) +static void verify_block_header(struct super_block *sb, struct buffer_head *bh) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; - struct scoutfs_block_header *hdr; - struct buffer_head *bh; - u32 crc; - - bh = sb_bread(sb, blkno); - if (!bh || buffer_private_verified(bh)) - return bh; - - hdr = (void *)bh->b_data; - crc = scoutfs_crc_block(hdr); + struct scoutfs_block_header *hdr = (void *)bh->b_data; + u32 crc = scoutfs_crc_block(hdr); + u64 blkno = bh->b_blocknr; if (le32_to_cpu(hdr->crc) != crc) { printk("blkno %llu hdr crc %x != calculated %x\n", blkno, @@ -52,49 +40,67 @@ struct buffer_head *scoutfs_read_block(struct super_block *sb, u64 blkno) le64_to_cpu(hdr->blkno)); } else { set_buffer_private_verified(bh); - return bh; } - - brelse(bh); - return NULL; } /* - * Return a locked dirty buffer with undefined contents. The caller is - * responsible for initializing the entire block. Callers can try and - * read from these dirty blocks so we mark them verified so that they - * don't try to check uninitialized crcs. + * Read an existing block from the device and verify its metadata header. */ -struct buffer_head *scoutfs_dirty_bh(struct super_block *sb, u64 blkno) +struct buffer_head *scoutfs_read_block(struct super_block *sb, u64 blkno) { struct buffer_head *bh; - bh = sb_getblk(sb, blkno); - if (bh) { - lock_buffer(bh); - set_buffer_uptodate(bh); - mark_buffer_dirty(bh); - set_buffer_private_verified(bh); + bh = sb_bread(sb, blkno); + if (!bh || buffer_private_verified(bh)) + return bh; + + lock_buffer(bh); + if (!buffer_private_verified(bh)) + verify_block_header(sb, bh); + unlock_buffer(bh); + + if (!buffer_private_verified(bh)) { + brelse(bh); + bh = NULL; } return bh; } /* - * Return a locked dirty buffer with a partially initialized block - * header. The caller has to calculate the header crc before unlocking - * the block. The header will have the sequence number of the dirty super - * by default. + * Read the block that contains the given byte offset in the given chunk. */ -struct buffer_head *scoutfs_dirty_block(struct super_block *sb, u64 blkno) +struct buffer_head *scoutfs_read_block_off(struct super_block *sb, u64 blkno, + u32 off) +{ + if (WARN_ON_ONCE(off >= SCOUTFS_CHUNK_SIZE)) + return ERR_PTR(-EINVAL); + + return scoutfs_read_block(sb, blkno + (off >> SCOUTFS_BLOCK_SHIFT)); +} + +/* + * Return a newly allocated metadata block with an updated block header + * to match the current dirty super block. Callers are responsible for + * serializing access to the block and for zeroing unwritten block + * contents. + */ +struct buffer_head *scoutfs_new_block(struct super_block *sb, u64 blkno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct scoutfs_block_header *hdr; struct buffer_head *bh; - bh = scoutfs_dirty_bh(sb, blkno); + bh = sb_getblk(sb, blkno); if (bh) { + if (!buffer_uptodate(bh) || buffer_private_verified(bh)) { + lock_buffer(bh); + set_buffer_uptodate(bh); + set_buffer_private_verified(bh); + unlock_buffer(bh); + } + hdr = (void *)bh->b_data; *hdr = super->hdr; hdr->blkno = cpu_to_le64(blkno); diff --git a/kmod/src/block.h b/kmod/src/block.h index 30d79864..7be8ed6d 100644 --- a/kmod/src/block.h +++ b/kmod/src/block.h @@ -2,8 +2,9 @@ #define _SCOUTFS_BLOCK_H_ struct buffer_head *scoutfs_read_block(struct super_block *sb, u64 blkno); -struct buffer_head *scoutfs_dirty_bh(struct super_block *sb, u64 blkno); -struct buffer_head *scoutfs_dirty_block(struct super_block *sb, u64 blkno); +struct buffer_head *scoutfs_read_block_off(struct super_block *sb, u64 blkno, + u32 off); +struct buffer_head *scoutfs_new_block(struct super_block *sb, u64 blkno); void scoutfs_calc_hdr_crc(struct buffer_head *bh); #endif diff --git a/kmod/src/bloom.c b/kmod/src/bloom.c new file mode 100644 index 00000000..df528afd --- /dev/null +++ b/kmod/src/bloom.c @@ -0,0 +1,125 @@ +/* + * 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 +#include +#include +#include +#include + +#include "super.h" +#include "format.h" +#include "block.h" +#include "bloom.h" + +/* + * Each log segment starts with a bloom filters that spans multiple + * blocks. It's used to test for the presence of key in the log segment + * without having to read and search the much larger array of items and + * their keys. + */ + +/* XXX garbage hack until we have siphash */ +static u32 bloom_hash(struct scoutfs_key *key, __le32 salt) +{ + return crc32c(le32_to_cpu(salt), key, sizeof(struct scoutfs_key)); +} + +/* + * Find the bits in the bloom filter for the given key. The caller calculates + * these once and uses them to test all the blocks. + */ +void scoutfs_calc_bloom_bits(struct scoutfs_bloom_bits *bits, + struct scoutfs_key *key, __le32 *salts) +{ + unsigned h_bits = 0; + unsigned int b; + unsigned s = 0; + u64 h = 0; + int i; + + BUILD_BUG_ON(SCOUTFS_BLOOM_BIT_WIDTH > 32); + + for (i = 0; i < SCOUTFS_BLOOM_BITS; i++) { + if (h_bits < SCOUTFS_BLOOM_BIT_WIDTH) { + h = (h << 32) | bloom_hash(key, salts[s++]); + h_bits += 32; + } + + b = h & SCOUTFS_BLOOM_BIT_MASK; + h >>= SCOUTFS_BLOOM_BIT_WIDTH; + h_bits -= SCOUTFS_BLOOM_BIT_WIDTH; + + bits->block[i] = (b / SCOUTFS_BLOOM_BITS_PER_BLOCK) % + SCOUTFS_BLOOM_BLOCKS; + bits->bit_off[i] = b % SCOUTFS_BLOOM_BITS_PER_BLOCK; + } +} + +/* + * Set the caller's bit numbers in the bloom filter contained in bloom + * blocks starting at the given block number. The caller has + * initialized the blocks and is responsible for locking and dirtying + * and writeout. + */ +int scoutfs_set_bloom_bits(struct super_block *sb, u64 blkno, + struct scoutfs_bloom_bits *bits) +{ + struct scoutfs_bloom_block *blm; + struct buffer_head *bh; + int ret = 0; + int i; + + for (i = 0; i < SCOUTFS_BLOOM_BITS; i++) { + bh = scoutfs_read_block(sb, blkno + bits->block[i]); + if (!bh) { + ret = -EIO; + break; + } + + blm = (void *)bh->b_data; + set_bit_le(bits->bit_off[i], blm->bits); + + brelse(bh); + } + + return ret; +} + +/* + * Returns zero if the bits' key can't be found in the block, true if it + * might, and -errno if IO fails. + */ +int scoutfs_test_bloom_bits(struct super_block *sb, u64 blkno, + struct scoutfs_bloom_bits *bits) +{ + struct scoutfs_bloom_block *blm; + struct buffer_head *bh; + int ret; + int i; + + for (i = 0; i < SCOUTFS_BLOOM_BITS; i++) { + bh = scoutfs_read_block(sb, blkno + bits->block[i]); + if (!bh) { + ret = -EIO; + break; + } + + blm = (void *)bh->b_data; + ret = !!test_bit_le(bits->bit_off[i], blm->bits); + brelse(bh); + if (!ret) + break; + } + + return ret; +} diff --git a/kmod/src/bloom.h b/kmod/src/bloom.h new file mode 100644 index 00000000..4e843fbe --- /dev/null +++ b/kmod/src/bloom.h @@ -0,0 +1,16 @@ +#ifndef _SCOUTFS_BLOOM_H_ +#define _SCOUTFS_BLOOM_H_ + +struct scoutfs_bloom_bits { + u16 bit_off[SCOUTFS_BLOOM_BITS]; + u8 block[SCOUTFS_BLOOM_BITS]; +}; + +void scoutfs_calc_bloom_bits(struct scoutfs_bloom_bits *bits, + struct scoutfs_key *key, __le32 *salts); +int scoutfs_test_bloom_bits(struct super_block *sb, u64 blkno, + struct scoutfs_bloom_bits *bits); +int scoutfs_set_bloom_bits(struct super_block *sb, u64 blkno, + struct scoutfs_bloom_bits *bits); + +#endif diff --git a/kmod/src/crc.c b/kmod/src/crc.c index 9869cbd1..cde9a1ae 100644 --- a/kmod/src/crc.c +++ b/kmod/src/crc.c @@ -10,6 +10,7 @@ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * General Public License for more details. */ +#include #include #include "format.h" diff --git a/kmod/src/dir.c b/kmod/src/dir.c index 921fbdef..e0aac04d 100644 --- a/kmod/src/dir.c +++ b/kmod/src/dir.c @@ -20,7 +20,7 @@ #include "dir.h" #include "inode.h" #include "key.h" -#include "item.h" +#include "segment.h" #include "super.h" /* @@ -110,26 +110,26 @@ static unsigned int dent_bytes(unsigned int name_len) return sizeof(struct scoutfs_dirent) + name_len; } -static unsigned int dent_val_off(struct scoutfs_item *item, +static unsigned int dent_val_off(struct scoutfs_item_ref *ref, struct scoutfs_dirent *dent) { - return (char *)dent - (char *)item->val; + return (char *)dent - (char *)ref->val; } -static inline struct scoutfs_dirent *next_dent(struct scoutfs_item *item, +static inline struct scoutfs_dirent *next_dent(struct scoutfs_item_ref *ref, struct scoutfs_dirent *dent) { unsigned int next_off; - next_off = dent_val_off(item, dent) + dent_bytes(dent->name_len); - if (next_off == item->val_len) + next_off = dent_val_off(ref, dent) + dent_bytes(dent->name_len); + if (next_off == ref->val_len) return NULL; - return item->val + next_off; + return ref->val + next_off; } -#define for_each_item_dent(item, dent) \ - for (dent = item->val; dent; dent = next_dent(item, dent)) +#define for_each_item_dent(ref, dent) \ + for (dent = (ref)->val; dent; dent = next_dent(ref, dent)) struct dentry_info { /* @@ -175,10 +175,10 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, { struct super_block *sb = dir->i_sb; struct scoutfs_dirent *dent; - struct scoutfs_item *item; struct dentry_info *di; struct scoutfs_key key; struct inode *inode; + DECLARE_SCOUTFS_ITEM_REF(ref); u64 ino = 0; u32 h = 0; u32 nr = 0; @@ -198,14 +198,12 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, h = name_hash(dir, dentry->d_name.name, dentry->d_name.len); scoutfs_set_key(&key, scoutfs_ino(dir), SCOUTFS_DIRENT_KEY, h); - item = scoutfs_item_lookup(sb, &key); - if (IS_ERR(item)) { - ret = PTR_ERR(item); + ret = scoutfs_read_item(sb, &key, &ref); + if (ret) goto out; - } ret = -ENOENT; - for_each_item_dent(item, dent) { + for_each_item_dent(&ref, dent) { if (names_equal(dentry->d_name.name, dentry->d_name.len, dent->name, dent->name_len)) { ino = le64_to_cpu(dent->ino); @@ -215,7 +213,7 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, } } - scoutfs_item_put(item); + scoutfs_put_ref(&ref); out: if (ret == -ENOENT) { inode = NULL; @@ -254,7 +252,7 @@ static int dir_emit_dots(struct file *file, void *dirent, filldir_t filldir) /* * readdir finds the next entry at or past the hash|coll_nr stored in - * the ctx->pos (f_pos). + * the current file position. * * It will need to be careful not to read past the region of the dirent * hash offset keys that it has access to. @@ -263,65 +261,63 @@ 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; + DECLARE_SCOUTFS_ITEM_REF(ref); struct scoutfs_dirent *dent; - struct scoutfs_key last_key; - struct scoutfs_item *item; - struct scoutfs_key key; - u32 nr; - u32 off; - u64 pos; + struct scoutfs_key first; + struct scoutfs_key last; + LIST_HEAD(iter_list); int ret = 0; + u32 off; + u32 pos; + u32 nr; if (!dir_emit_dots(file, dirent, filldir)) return 0; - scoutfs_set_key(&last_key, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, + scoutfs_set_key(&first, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, + file->f_pos >> SCOUTFS_DIRENT_COLL_BITS); + scoutfs_set_key(&last, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, SCOUTFS_DIRENT_OFF_MASK); - do { - off = file->f_pos >> SCOUTFS_DIRENT_COLL_BITS; - nr = file->f_pos & SCOUTFS_DIRENT_COLL_MASK; - - scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, - off); - item = scoutfs_item_next(sb, &key); - if (IS_ERR(item)) { - ret = PTR_ERR(item); - if (ret == -ENOENT) - ret = 0; + for(;;) { + scoutfs_put_ref(&ref); + ret = scoutfs_next_item(sb, &first, &last, &iter_list, &ref); + if (ret) break; - } - if (scoutfs_key_cmp(&item->key, &last_key) > 0) { - scoutfs_item_put(item); - break; - } - - /* reset nr to 0 if we found the next item */ - if (scoutfs_key_offset(&item->key) != off) + /* start from first collision if we're in a new item */ + if (scoutfs_key_offset(&first) == scoutfs_key_offset(ref.key)) + nr = file->f_pos & SCOUTFS_DIRENT_COLL_MASK; + else nr = 0; - pos = scoutfs_key_offset(&item->key) - << SCOUTFS_DIRENT_COLL_BITS; - for_each_item_dent(item, dent) { + off = scoutfs_key_offset(ref.key) << SCOUTFS_DIRENT_COLL_BITS; + for_each_item_dent(&ref, dent) { if (dent->coll_nr < nr) continue; + pos = off | dent->coll_nr; + if (filldir(dirent, dent->name, dent->name_len, pos, le64_to_cpu(dent->ino), dentry_type(dent->type))) break; - file->f_pos = (pos | dent->coll_nr) + 1; + file->f_pos = pos + 1; } + /* done if filldir broke the loop */ + if (dent) + break; - scoutfs_item_put(item); + first = *ref.key; + scoutfs_inc_key(&first); + } - /* advance to the next hash value if we finished item */ - if (dent == NULL) - file->f_pos = pos + (1 << SCOUTFS_DIRENT_COLL_BITS); + scoutfs_put_ref(&ref); + scoutfs_put_iter_list(&iter_list); - } while (dent == NULL); + if (ret == -ENOENT) + ret = 0; return ret; } @@ -332,12 +328,11 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, struct super_block *sb = dir->i_sb; struct inode *inode = NULL; struct scoutfs_dirent *dent; - struct scoutfs_item *item; + DECLARE_SCOUTFS_ITEM_REF(ref); struct dentry_info *di; struct scoutfs_key key; int bytes; int ret; - int off; u64 nr; u64 h; @@ -356,60 +351,32 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, scoutfs_set_key(&key, scoutfs_ino(dir), SCOUTFS_DIRENT_KEY, h); bytes = dent_bytes(dentry->d_name.len); - item = scoutfs_item_lookup(sb, &key); - if (item == ERR_PTR(-ENOENT)) { - item = scoutfs_item_create(sb, &key, bytes); - if (!IS_ERR(item)) { - /* mark a newly created item */ - dent = item->val; - dent->name_len = 0; + ret = scoutfs_read_item(sb, &key, &ref); + if (ret != -ENOENT) { + /* XXX implement many hashes, not coll nr */ + if (WARN_ON_ONCE(!ret)) { + scoutfs_put_ref(&ref); + ret = -ENOSPC; } - } - if (IS_ERR(item)) { - ret = PTR_ERR(item); goto out; } - ret = 0; - nr = 0; - for_each_item_dent(item, dent) { - /* the common case of a newly created item */ - if (!dent->name_len) - break; - - /* XXX check for eexist? can't happen? */ - - /* found a free coll nr, insert here */ - if (nr < dent->coll_nr) { - off = dent_val_off(item, dent); - ret = scoutfs_item_expand(item, off, bytes); - if (!ret) - dent = item->val + off; - break; - } - - /* the item's full */ - if (nr++ == SCOUTFS_DIRENT_COLL_MASK) { - ret = -ENOSPC; - break; - } - } - - if (!ret) { - dent->ino = cpu_to_le64(scoutfs_ino(inode)); - dent->type = mode_to_type(inode->i_mode); - dent->coll_nr = nr; - dent->name_len = dentry->d_name.len; - memcpy(dent->name, dentry->d_name.name, dent->name_len); - di->key_offset = h; - di->coll_nr = nr; - } - - scoutfs_item_put(item); - + ret = scoutfs_create_item(sb, &key, bytes, &ref); if (ret) goto out; + dent = ref.val; + nr = 0; + dent->ino = cpu_to_le64(scoutfs_ino(inode)); + dent->type = mode_to_type(inode->i_mode); + dent->coll_nr = nr; + dent->name_len = dentry->d_name.len; + memcpy(dent->name, dentry->d_name.name, dent->name_len); + di->key_offset = h; + di->coll_nr = nr; + + scoutfs_put_ref(&ref); + 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; @@ -452,8 +419,7 @@ 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_dirent *dent; - struct scoutfs_item *item; + DECLARE_SCOUTFS_ITEM_REF(ref); struct dentry_info *di; struct scoutfs_key key; int ret = 0; @@ -471,33 +437,12 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry) scoutfs_set_key(&key, scoutfs_ino(dir), SCOUTFS_DIRENT_KEY, di->key_offset); - item = scoutfs_item_lookup(sb, &key); - if (IS_ERR(item)) { - ret = PTR_ERR(item); + ret = scoutfs_read_item(sb, &key, &ref); + if (ret) goto out; - } - - /* XXX error to not find the coll nr we were looking for? */ - for_each_item_dent(item, dent) { - if (dent->coll_nr != di->coll_nr) - continue; - - /* XXX compare names and eio? */ - - if (item->val_len == dent_bytes(dent->name_len)) { - scoutfs_item_delete(sb, item); - ret = 0; - } else { - ret = scoutfs_item_shrink(item, - dent_val_off(item, dent), - dent_bytes(dent->name_len)); - } - dent = NULL; - break; - } - - scoutfs_item_put(item); + ret = scoutfs_delete_item(sb, &ref); + scoutfs_put_ref(&ref); if (ret) goto out; diff --git a/kmod/src/format.h b/kmod/src/format.h index bafaef80..1310f8d7 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -13,6 +13,7 @@ */ #define SCOUTFS_BLOCK_SHIFT 12 #define SCOUTFS_BLOCK_SIZE (1 << SCOUTFS_BLOCK_SHIFT) +#define SCOUTFS_BLOCK_MASK (SCOUTFS_BLOCK_SIZE - 1) /* * The allocator works on larger chunks. Smaller metadata structures @@ -34,6 +35,19 @@ #define SCOUTFS_SUPER_BLKNO ((64 * 1024) >> SCOUTFS_BLOCK_SHIFT) #define SCOUTFS_SUPER_NR 2 +/* + * 7 bits in a ~76k bloom filter gives ~1% false positive for our max + * of 64k items. + * + * n = 65,536, p = 0.01 (1 in 100) → m = 628,167 (76.68KB), k = 7 + */ +#define SCOUTFS_BLOOM_BITS 7 +#define SCOUTFS_BLOOM_BIT_WIDTH 20 /* 2^20 > m */ +#define SCOUTFS_BLOOM_BIT_MASK ((1 << SCOUTFS_BLOOM_BIT_WIDTH) - 1) +#define SCOUTFS_BLOOM_BLOCKS ((76 * 1024) / SCOUTFS_BLOCK_SIZE) +#define SCOUTFS_BLOOM_SALTS \ + DIV_ROUND_UP(SCOUTFS_BLOOM_BITS * SCOUTFS_BLOOM_BIT_WIDTH, 32) + /* * This header is found at the start of every block so that we can * verify that it's what we were looking for. The crc and padding @@ -64,6 +78,7 @@ struct scoutfs_super_block { struct scoutfs_block_header hdr; __le64 id; __u8 uuid[SCOUTFS_UUID_BYTES]; + __le32 bloom_salts[SCOUTFS_BLOOM_SALTS]; __le64 total_chunks; __le64 ring_map_blkno; __le64 ring_map_seq; @@ -149,22 +164,43 @@ struct scoutfs_ring_bitmap { __le64 bits[2]; } __packed; + +struct scoutfs_bloom_block { + struct scoutfs_block_header hdr; + __le64 bits[0]; +} __packed; + +#define SCOUTFS_BLOOM_BITS_PER_BLOCK \ + (((SCOUTFS_BLOCK_SIZE - sizeof(struct scoutfs_block_header)) / 8) * 64) + /* - * To start the log segments are a trivial single item block. We'll - * flesh this out into larger blocks once the rest of the architecture - * is in place. + * Items in log segments are sorted in a skip list by their key. We + * have a rough limit of 64k items. + */ +#define SCOUTFS_SKIP_HEIGHT 16 +struct scoutfs_skip_root { + __le32 next[SCOUTFS_SKIP_HEIGHT]; +} __packed; + +/* + * An item block follows the bloom filter blocks at the start of a log + * segment. Its skip root references the item structs which then + * reference the item values in the rest of the block. The references + * are byte offsets from the start of the chunk. */ struct scoutfs_item_block { struct scoutfs_block_header hdr; struct scoutfs_key first; struct scoutfs_key last; - __le32 nr_items; - /* struct scoutfs_item_header items[0] .. */ + struct scoutfs_skip_root skip_root; } __packed; -struct scoutfs_item_header { +struct scoutfs_item { struct scoutfs_key key; + __le32 offset; __le16 len; + u8 skip_height; + __le32 skip_next[0]; } __packed; struct scoutfs_timespec { diff --git a/kmod/src/inode.c b/kmod/src/inode.c index 521a1671..99187b2c 100644 --- a/kmod/src/inode.c +++ b/kmod/src/inode.c @@ -19,7 +19,7 @@ #include "super.h" #include "key.h" #include "inode.h" -#include "item.h" +#include "segment.h" #include "dir.h" /* @@ -110,17 +110,17 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode) static int scoutfs_read_locked_inode(struct inode *inode) { struct super_block *sb = inode->i_sb; - struct scoutfs_item *item; + DECLARE_SCOUTFS_ITEM_REF(ref); struct scoutfs_key key; + int ret; scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_INODE_KEY, 0); - item = scoutfs_item_lookup(sb, &key); - if (IS_ERR(item)) - return PTR_ERR(item); - - load_inode(inode, item->val); - scoutfs_item_put(item); + ret = scoutfs_read_item(sb, &key, &ref); + if (!ret) { + load_inode(inode, ref.val); + scoutfs_put_ref(&ref); + } return 0; } @@ -200,16 +200,17 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode) void scoutfs_update_inode_item(struct inode *inode) { struct super_block *sb = inode->i_sb; - struct scoutfs_item *item; + DECLARE_SCOUTFS_ITEM_REF(ref); struct scoutfs_key key; + int ret; scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_INODE_KEY, 0); - item = scoutfs_item_lookup(sb, &key); - BUG_ON(IS_ERR(item)); + ret = scoutfs_read_item(sb, &key, &ref); + BUG_ON(ret); - store_inode(item->val, inode); - scoutfs_item_put(item); + store_inode(ref.val, inode); + scoutfs_put_ref(&ref); } /* @@ -221,9 +222,10 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir, { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_inode_info *ci; - struct scoutfs_item *item; + DECLARE_SCOUTFS_ITEM_REF(ref); struct scoutfs_key key; struct inode *inode; + int ret; inode = new_inode(sb); if (!inode) @@ -242,12 +244,14 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir, scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_INODE_KEY, 0); - item = scoutfs_item_create(inode->i_sb, &key, - sizeof(struct scoutfs_inode)); - if (IS_ERR(item)) { + ret = scoutfs_create_item(inode->i_sb, &key, + sizeof(struct scoutfs_inode), &ref); + if (ret) { iput(inode); - inode = ERR_CAST(item); + return ERR_PTR(ret); } + + scoutfs_put_ref(&ref); return inode; } diff --git a/kmod/src/item.c b/kmod/src/item.c deleted file mode 100644 index 1f253826..00000000 --- a/kmod/src/item.c +++ /dev/null @@ -1,463 +0,0 @@ -/* - * Copyright (C) 2015 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 -#include -#include -#include -#include - -#include "super.h" -#include "key.h" -#include "item.h" -#include "segment.h" - -/* - * describe: - * - tracks per-item dirty state for writing - * - decouples vfs cache lifetimes from item lifetimes - * - item-granular cache for things vfs doesn't cache (readdir, xattr) - * - * XXX: - * - warnings for invalid keys/lens - * - memory pressure - */ - -enum { - ITW_NEXT = 1, - ITW_PREV, -}; - -static inline struct scoutfs_item *node_item(struct super_block *sb, - struct rb_root *root, - struct rb_node *node) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - unsigned long off; - - if (root == &sbi->item_root) - off = offsetof(struct scoutfs_item, node); - else - off = offsetof(struct scoutfs_item, dirty_node); - - return (void *)((char *)node - off); -} - -static inline struct rb_node *item_node(struct super_block *sb, - struct rb_root *root, - struct scoutfs_item *item) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - unsigned long off; - - if (root == &sbi->item_root) - off = offsetof(struct scoutfs_item, node); - else - off = offsetof(struct scoutfs_item, dirty_node); - - return (void *)((char *)item + off); -} - -/* - * Insert a new item in the tree. The caller must have done a lookup to - * ensure that the key is not already present. - */ -static void insert_item(struct super_block *sb, struct rb_root *root, - struct scoutfs_item *ins) -{ - struct rb_node **node = &root->rb_node; - struct rb_node *parent = NULL; - struct scoutfs_item *item; - int cmp; - - while (*node) { - parent = *node; - item = node_item(sb, root, *node); - - cmp = scoutfs_key_cmp(&ins->key, &item->key); - BUG_ON(cmp == 0); - if (cmp < 0) - node = &(*node)->rb_left; - else - node = &(*node)->rb_right; - } - - rb_link_node(item_node(sb, root, ins), parent, node); - rb_insert_color(item_node(sb, root, ins), root); -} - -enum { - FI_NEXT = 1, - FI_PREV, -}; - -/* - * Walk the tree looking for an item. - * - * If NEXT or PREV are specified then those will be returned - * if the specific item isn't found. - */ -static struct scoutfs_item *find_item(struct super_block *sb, - struct rb_root *root, - struct scoutfs_key *key, int np) -{ - struct rb_node *node = root->rb_node; - struct scoutfs_item *found = NULL; - struct scoutfs_item *item; - int cmp; - - while (node) { - item = node_item(sb, root, node); - - cmp = scoutfs_key_cmp(key, &item->key); - if (cmp < 0) { - if (np == FI_NEXT) - found = item; - node = node->rb_left; - } else if (cmp > 0) { - if (np == FI_PREV) - found = item; - node = node->rb_right; - } else { - found = item; - break; - } - } - - return found; -} - -static struct scoutfs_item *alloc_item(struct scoutfs_key *key, - unsigned int val_len) -{ - struct scoutfs_item *item; - void *val; - - item = kmalloc(sizeof(struct scoutfs_item), GFP_NOFS); - val = kmalloc(val_len, GFP_NOFS); - if (!item || !val) { - kfree(item); - kfree(val); - return ERR_PTR(-ENOMEM); - } - - RB_CLEAR_NODE(&item->node); - RB_CLEAR_NODE(&item->dirty_node); - atomic_set(&item->refcount, 1); - item->key = *key; - item->val_len = val_len; - item->val = val; - - return item; -} - -static struct scoutfs_item *create_item(struct super_block *sb, - struct scoutfs_key *key, - unsigned int val_len, bool dirty) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_item *existing; - struct scoutfs_item *item; - unsigned long flags; - - item = alloc_item(key, val_len); - if (IS_ERR(item)) - return item; - - spin_lock_irqsave(&sbi->item_lock, flags); - - existing = find_item(sb, &sbi->item_root, key, 0); - if (!existing) { - insert_item(sb, &sbi->item_root, item); - atomic_inc(&item->refcount); - if (dirty) { - insert_item(sb, &sbi->dirty_item_root, item); - atomic_inc(&item->refcount); - } - - } - spin_unlock_irqrestore(&sbi->item_lock, flags); - - if (existing) { - scoutfs_item_put(item); - item = ERR_PTR(-EEXIST); - } - - trace_printk("item %p key "CKF" val_len %d\n", item, CKA(key), val_len); - - return item; -} - -/* - * Create a new item stored at the given key. Return it with a reference. - * return an ERR_PTR with ENOMEM or EEXIST. - * - * The caller is responsible for initializing the item's value. - */ -struct scoutfs_item *scoutfs_item_create(struct super_block *sb, - struct scoutfs_key *key, - unsigned int val_len) -{ - return create_item(sb, key, val_len, true); -} - -/* - * Allocate a new clean item in the cache for the caller to fill. If the - * item already exists then -EEXIST is returned. - */ -struct scoutfs_item *scoutfs_clean_item(struct super_block *sb, - struct scoutfs_key *key, - unsigned int val_len) -{ - return create_item(sb, key, val_len, false); -} - -/* - * The caller is still responsible for unlocking and putting the item. - * - * We don't try and optimize away the lock for items that are already - * removed from the tree. The caller's locking and item behaviour means - * that racing to remove an item is extremely rare. - * - * XXX for now we're just removing it from the rbtree. We'd need to leave - * behind a deletion record for lsm. - */ -void scoutfs_item_delete(struct super_block *sb, struct scoutfs_item *item) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - unsigned long flags; - - spin_lock_irqsave(&sbi->item_lock, flags); - - if (!RB_EMPTY_NODE(&item->dirty_node)) { - rb_erase(&item->dirty_node, &sbi->dirty_item_root); - RB_CLEAR_NODE(&item->dirty_node); - scoutfs_item_put(item); - } - - if (!RB_EMPTY_NODE(&item->node)) { - rb_erase(&item->node, &sbi->item_root); - RB_CLEAR_NODE(&item->node); - scoutfs_item_put(item); - } - - spin_unlock_irqrestore(&sbi->item_lock, flags); -} - -/* - * Find an item in the cache. If it isn't present then we try to read - * it from log segements. - */ -static struct scoutfs_item *item_lookup(struct super_block *sb, - struct scoutfs_key *key, int np) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_item *item; - unsigned long flags; - unsigned retried = 0; - int ret; - - do { - spin_lock_irqsave(&sbi->item_lock, flags); - - item = find_item(sb, &sbi->item_root, key, np); - if (item) - atomic_inc(&item->refcount); - - spin_unlock_irqrestore(&sbi->item_lock, flags); - if (!item) { - if (np == FI_NEXT) - ret = scoutfs_read_next_item(sb, key); - else - ret = scoutfs_read_item(sb, key); - if (ret) - item = ERR_PTR(ret); - } - } while (!item && !retried++); - - if (!item) - item = ERR_PTR(-ENOENT); - - return item; -} - -struct scoutfs_item *scoutfs_item_lookup(struct super_block *sb, - struct scoutfs_key *key) -{ - return item_lookup(sb, key, 0); -} - -struct scoutfs_item *scoutfs_item_next(struct super_block *sb, - struct scoutfs_key *key) -{ - return item_lookup(sb, key, FI_NEXT); -} - -struct scoutfs_item *scoutfs_item_prev(struct super_block *sb, - struct scoutfs_key *key) -{ - return item_lookup(sb, key, FI_PREV); -} - -/* - * Expand the item's value by inserting bytes at the given offset. The - * new bytes are not initialized. - */ -int scoutfs_item_expand(struct scoutfs_item *item, int off, int bytes) -{ - void *val; - - /* XXX bytes too big */ - if (WARN_ON_ONCE(off < 0 || off > item->val_len)) - return -EINVAL; - - val = kmalloc(item->val_len + bytes, GFP_NOFS); - if (!val) - return -ENOMEM; - - memcpy(val, item->val, off); - memcpy(val + off + bytes, item->val + off, item->val_len - off); - - kfree(item->val); - item->val = val; - item->val_len += bytes; - - return 0; -} - -/* - * Shrink the item's value by remove bytes at the given offset. - */ -int scoutfs_item_shrink(struct scoutfs_item *item, int off, int bytes) -{ - void *val; - - if (WARN_ON_ONCE(off < 0 || off >= item->val_len || - bytes <= 0 || (off + bytes) > item->val_len || - bytes == item->val_len)) - return -EINVAL; - - val = kmalloc(item->val_len - bytes, GFP_NOFS); - if (!val) - return -ENOMEM; - - memcpy(val, item->val, off); - memcpy(val + off, item->val + off + bytes, - item->val_len - (off + bytes)); - - kfree(item->val); - item->val = val; - item->val_len -= bytes; - - return 0; -} - -void scoutfs_item_mark_dirty(struct super_block *sb, struct scoutfs_item *item) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - unsigned long flags; - - spin_lock_irqsave(&sbi->item_lock, flags); - - if (RB_EMPTY_NODE(&item->dirty_node)) { - insert_item(sb, &sbi->dirty_item_root, item); - atomic_inc(&item->refcount); - } - - spin_unlock_irqrestore(&sbi->item_lock, flags); -} - -/* - * Mark all the dirty items clean by emptying the dirty rbtree. The - * caller should be preventing writes from dirtying new items. - * - * We erase leaf nodes with no children to minimize rotation - * overhead during erase. Dirty items must be in the main rbtree if - * they're in the dirty rbtree so the puts here shouldn't free the - * items. - */ -void scoutfs_item_all_clean(struct super_block *sb) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct rb_root *root = &sbi->dirty_item_root; - struct scoutfs_item *item; - struct rb_node *node; - unsigned long flags; - - spin_lock_irqsave(&sbi->item_lock, flags); - - node = sbi->dirty_item_root.rb_node; - while (node) { - if (node->rb_left) - node = node->rb_left; - else if (node->rb_right) - node = node->rb_right; - else { - item = node_item(sb, root, node); - node = rb_parent(node); - - trace_printk("item %p key "CKF"\n", - item, CKA(&item->key)); - rb_erase(&item->dirty_node, root); - RB_CLEAR_NODE(&item->dirty_node); - scoutfs_item_put(item); - } - } - - spin_unlock_irqrestore(&sbi->item_lock, flags); -} - -/* - * If the item is null then the first dirty item is returned. If an - * item is given then the next dirty item is returned. NULL is returned - * if there are no more dirty items. - * - * The caller is given a reference that it has to put. The given item - * will always have its item dropped including if it returns NULL. - */ -struct scoutfs_item *scoutfs_item_next_dirty(struct super_block *sb, - struct scoutfs_item *item) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_item *next_item; - struct rb_node *node; - unsigned long flags; - - spin_lock_irqsave(&sbi->item_lock, flags); - - if (item) - node = rb_next(&item->dirty_node); - else - node = rb_first(&sbi->dirty_item_root); - - if (node) { - next_item = node_item(sb, &sbi->dirty_item_root, node); - atomic_inc(&next_item->refcount); - } else { - next_item = NULL; - } - - spin_unlock_irqrestore(&sbi->item_lock, flags); - - scoutfs_item_put(item); - - return next_item; -} - -void scoutfs_item_put(struct scoutfs_item *item) -{ - if (!IS_ERR_OR_NULL(item) && atomic_dec_and_test(&item->refcount)) { - WARN_ON_ONCE(!RB_EMPTY_NODE(&item->node)); - WARN_ON_ONCE(!RB_EMPTY_NODE(&item->dirty_node)); - kfree(item); - } -} diff --git a/kmod/src/item.h b/kmod/src/item.h deleted file mode 100644 index b8225a20..00000000 --- a/kmod/src/item.h +++ /dev/null @@ -1,40 +0,0 @@ -#ifndef _SCOUTFS_ITEM_H_ -#define _SCOUTFS_ITEM_H_ - -#include "format.h" - -struct scoutfs_item { - struct rb_node node; - struct rb_node dirty_node; - atomic_t refcount; - - /* the key is constant for the life of the item */ - struct scoutfs_key key; - - /* the value can be changed by expansion or shrinking */ - unsigned int val_len; - void *val; -}; - -struct scoutfs_item *scoutfs_item_create(struct super_block *sb, - struct scoutfs_key *key, - unsigned int val_len); -struct scoutfs_item *scoutfs_clean_item(struct super_block *sb, - struct scoutfs_key *key, - unsigned int val_len); -struct scoutfs_item *scoutfs_item_lookup(struct super_block *sb, - struct scoutfs_key *key); -struct scoutfs_item *scoutfs_item_next(struct super_block *sb, - struct scoutfs_key *key); -struct scoutfs_item *scoutfs_item_prev(struct super_block *sb, - struct scoutfs_key *key); -int scoutfs_item_expand(struct scoutfs_item *item, int off, int bytes); -int scoutfs_item_shrink(struct scoutfs_item *item, int off, int bytes); -void scoutfs_item_delete(struct super_block *sb, struct scoutfs_item *item); -void scoutfs_item_mark_dirty(struct super_block *sb, struct scoutfs_item *item); -struct scoutfs_item *scoutfs_item_next_dirty(struct super_block *sb, - struct scoutfs_item *item); -void scoutfs_item_all_clean(struct super_block *sb); -void scoutfs_item_put(struct scoutfs_item *item); - -#endif diff --git a/kmod/src/key.h b/kmod/src/key.h index 80e60668..c06f898c 100644 --- a/kmod/src/key.h +++ b/kmod/src/key.h @@ -32,27 +32,28 @@ static inline int scoutfs_key_cmp(struct scoutfs_key *a, struct scoutfs_key *b) } /* - * return -ve, 0, +ve if the key is less than, contained within, or greater - * than the given range of keys. + * return -ve if the first range is completely before the second, +ve for + * completely after, and 0 if they intersect. */ -static inline int scoutfs_key_cmp_range(struct scoutfs_key *key, +static inline int scoutfs_cmp_key_ranges(struct scoutfs_key *a_first, + struct scoutfs_key *a_last, + struct scoutfs_key *b_first, + struct scoutfs_key *b_last) +{ + if (scoutfs_key_cmp(a_last, b_first) < 0) + return -1; + if (scoutfs_key_cmp(a_first, b_last) > 0) + return 1; + return 0; +} + +static inline int scoutfs_cmp_key_range(struct scoutfs_key *key, struct scoutfs_key *first, struct scoutfs_key *last) { - int cmp; - - WARN_ON_ONCE(scoutfs_key_cmp(first, last) > 0); - - cmp = scoutfs_key_cmp(key, first); - if (cmp > 0) { - cmp = scoutfs_key_cmp(key, last); - if (cmp < 0) - cmp = 0; - } - return cmp; + return scoutfs_cmp_key_ranges(key, key, first, last); } - static inline void scoutfs_set_key(struct scoutfs_key *key, u64 inode, u8 type, u64 offset) { diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index 8666ac49..e1d06e12 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -85,7 +85,7 @@ static struct scoutfs_manifest_node *find_mnode(struct rb_root *root, while (node) { mnode = rb_entry(node, struct scoutfs_manifest_node, node); - cmp = scoutfs_key_cmp_range(key, &mnode->ment.first, + cmp = scoutfs_cmp_key_range(key, &mnode->ment.first, &mnode->ment.last); if (cmp < 0) node = node->rb_left; @@ -213,15 +213,24 @@ int scoutfs_new_manifest(struct super_block *sb, /* * Fill the caller's ment with the next log segment in the manifest that - * might contain the given key. The ment is initialized to 0 to return - * the first entry. + * might contain the given range. The caller initializes the ment to + * zeros to find the first log segment. * * This can return multiple log segments from level 0 in decreasing age. * Then it can return at most one log segment in each level that - * intersects with the given key. + * intersects the given range. + * + * Returns true if an entry was found and is now described in ment, + * false when there are no more segments that contain the range. + * + * XXX could use the l0 seq to walk the list and skipb locks we've + * already seen. I'm not sure that we'll be able to keep manifest + * entries pinned while we're away blocking. We might fail to find the + * last entry's block in the radix when we return. */ -bool scoutfs_next_manifest_segment(struct super_block *sb, - struct scoutfs_key *key, +bool scoutfs_foreach_range_segment(struct super_block *sb, + struct scoutfs_key *first, + struct scoutfs_key *last, struct scoutfs_ring_manifest_entry *ment) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); @@ -247,8 +256,9 @@ bool scoutfs_next_manifest_segment(struct super_block *sb, } list_for_each_entry_from(mnode, &mani->level_zero, head) { - if (scoutfs_key_cmp_range(key, &mnode->ment.first, - &mnode->ment.last) == 0) { + if (scoutfs_cmp_key_ranges(first, last, + &mnode->ment.first, + &mnode->ment.last) == 0) { *ment = mnode->ment; found = true; break; @@ -257,8 +267,15 @@ bool scoutfs_next_manifest_segment(struct super_block *sb, } if (!found) { + /* + * The log segments in the each level fully cover the + * key range and don't overlap. So we will always find + * a segment that matches whatever key we look for. We + * look for the start of the range because iterators are + * walk the keyspace sequentially. + */ for (i = ment->level + 1; i <= SCOUTFS_MAX_LEVEL; i++) { - mnode = find_mnode(&mani->levels[i].root, key); + mnode = find_mnode(&mani->levels[i].root, first); if (mnode) { *ment = mnode->ment; found = true; diff --git a/kmod/src/manifest.h b/kmod/src/manifest.h index 407bfa28..bab32764 100644 --- a/kmod/src/manifest.h +++ b/kmod/src/manifest.h @@ -10,8 +10,9 @@ int scoutfs_new_manifest(struct super_block *sb, struct scoutfs_ring_manifest_entry *ment); void scoutfs_delete_manifest(struct super_block *sb, u64 blkno); -bool scoutfs_next_manifest_segment(struct super_block *sb, - struct scoutfs_key *key, +bool scoutfs_foreach_range_segment(struct super_block *sb, + struct scoutfs_key *first, + struct scoutfs_key *last, struct scoutfs_ring_manifest_entry *ment); #endif diff --git a/kmod/src/ring.c b/kmod/src/ring.c index 095c30b2..19642e21 100644 --- a/kmod/src/ring.c +++ b/kmod/src/ring.c @@ -18,7 +18,6 @@ #include "dir.h" #include "inode.h" #include "key.h" -#include "item.h" #include "super.h" #include "manifest.h" #include "chunk.h" @@ -113,14 +112,14 @@ static struct buffer_head *read_ring_block(struct super_block *sb, u64 block) /* * Return a dirty locked logical ring block. */ -static struct buffer_head *dirty_ring_block(struct super_block *sb, u64 block) +static struct buffer_head *new_ring_block(struct super_block *sb, u64 block) { u64 blkno = map_ring_block(sb, block); if (!blkno) return NULL; - return scoutfs_dirty_block(sb, blkno); + return scoutfs_new_block(sb, blkno); } int scoutfs_replay_ring(struct super_block *sb) @@ -186,7 +185,7 @@ int scoutfs_dirty_ring_entry(struct super_block *sb, u8 type, void *data, if (block >= le64_to_cpu(super->ring_total_blocks)) block -= le64_to_cpu(super->ring_total_blocks); - bh = dirty_ring_block(sb, block); + bh = new_ring_block(sb, block); if (!bh) { ret = -ENOMEM; goto out; @@ -242,6 +241,7 @@ int scoutfs_finish_dirty_ring(struct super_block *sb) * the block without walking all the items. */ scoutfs_calc_hdr_crc(bh); + mark_buffer_dirty(bh); unlock_buffer(bh); brelse(bh); diff --git a/kmod/src/segment.c b/kmod/src/segment.c index 0c3d1adb..9ea41977 100644 --- a/kmod/src/segment.c +++ b/kmod/src/segment.c @@ -19,321 +19,611 @@ #include "super.h" #include "key.h" -#include "item.h" #include "segment.h" #include "manifest.h" #include "block.h" #include "chunk.h" #include "ring.h" +#include "bloom.h" +#include "skip.h" -static struct scoutfs_item_header *next_ihdr(struct scoutfs_item_header *ihdr) + +/* + * scoutfs log segments are large multi-block structures that contain + * key/value items. This file implements manipulations of the items. + * + * Each log segment starts with a bloom filter to supports quickly + * testing for key values without having to search the whole block for a + * key. + * + * After the bloom filter come the packed structures that describe the + * items that are present in the block. They're sorted in a skip list + * to support reasonably efficient insertion, sorted iteration, and + * deletion. + * + * Finally the item values are stored at the end of the block. This + * supports finding that an item's key isn't present by only reading the + * item structs, not the values. + * + * All told, should we chose to, we can have three large portions of the + * blocks resident for searching. It's likely that we'll keep the bloom + * filters hot but that the items and especially the values may age out + * of the cache. + */ + +void scoutfs_put_ref(struct scoutfs_item_ref *ref) { - return (void *)(ihdr + 1) + le16_to_cpu(ihdr->len); + if (ref->item_bh) + brelse(ref->item_bh); + if (ref->val_bh) + brelse(ref->val_bh); + + memset(ref, 0, sizeof(struct scoutfs_item_ref)); +} + +/* private to here */ +struct scoutfs_item_iter { + struct list_head list; + struct buffer_head *bh; + struct scoutfs_item *item; + u64 blkno; + bool restart_after; +}; + +void scoutfs_put_iter_list(struct list_head *list) +{ + struct scoutfs_item_iter *iter; + struct scoutfs_item_iter *pos; + + list_for_each_entry_safe(iter, pos, list, list) { + list_del_init(&iter->list); + brelse(iter->bh); + kfree(iter); + } } /* - * Use the manifest to search log segments for the most recent version - * of the item with the given key. This only returns an error if it - * fails to determine if the item exists or not. It's up to the caller - * to retry the lookup after success. + * The caller has a pointer to an item and a reference to its block. We + * read the value block and populate the reference. + * + * The item references get their own buffer head references so that the + * caller doesn't have to play funny games. They always have to drop + * their release bh. If this succeeds then they also need to put the + * ref. */ -int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key) +static int populate_ref(struct super_block *sb, u64 blkno, + struct buffer_head *item_bh, struct scoutfs_item *item, + struct scoutfs_item_ref *ref) { - struct scoutfs_ring_manifest_entry ment; - struct scoutfs_item_header *ihdr; - struct scoutfs_item_block *iblk; - struct scoutfs_item *item = NULL; struct buffer_head *bh; - int ret = 0; - int cmp; - int i; + + bh = scoutfs_read_block_off(sb, blkno, le32_to_cpu(item->offset)); + if (!bh) + return -EIO; + + ref->key = &item->key; + ref->val_len = le16_to_cpu(item->len); + ref->val = bh->b_data + (le32_to_cpu(item->offset) & + SCOUTFS_BLOCK_MASK); + get_bh(item_bh); + ref->item_bh = item_bh; + ref->val_bh = bh; + + return 0; +} + +/* + * Return a reference to the item at the given key. We walk the manifest + * to find blocks that might contain the key from most recent to oldest. + * To find the key in each log segment we test it's bloom filter and + * then search through the item keys. The first matching item we find + * is returned. + * + * XXX lock the dirty log segment? + * + * -ENOENT is returned if the item isn't present. The caller needs to put + * the ref if we return success. + */ +int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_item_ref *ref) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_ring_manifest_entry ment; + struct scoutfs_item *item = NULL; + struct scoutfs_bloom_bits bits; + struct buffer_head *bh; + int ret; /* XXX hold manifest */ + scoutfs_calc_bloom_bits(&bits, key, sbi->super.bloom_salts); + + item = NULL; + ret = -ENOENT; memset(&ment, 0, sizeof(struct scoutfs_ring_manifest_entry)); + while (scoutfs_foreach_range_segment(sb, key, key, &ment)) { - while (scoutfs_next_manifest_segment(sb, key, &ment)) { + /* XXX read-ahead all bloom blocks */ - bh = scoutfs_read_block(sb, le64_to_cpu(ment.blkno)); - if (!bh) { - ret = -EIO; + ret = scoutfs_test_bloom_bits(sb, le64_to_cpu(ment.blkno), + &bits); + if (ret < 0) break; + if (!ret) { + ret = -ENOENT; + continue; } - iblk = (void *)bh->b_data; - /* XXX seq corruption */ + /* XXX read-ahead all item header blocks */ - ihdr = (void *)(iblk + 1); - - /* XXX test bloom filter blocks */ - /* XXX binary search of key array */ - /* XXX could populate more from granted range */ - - for (i = 0; i < le32_to_cpu(iblk->nr_items); - i++, ihdr = next_ihdr(ihdr)) { - cmp = scoutfs_key_cmp(key, &ihdr->key); - if (cmp > 0) + ret = scoutfs_skip_lookup(sb, le64_to_cpu(ment.blkno), key, + &bh, &item); + if (ret) { + if (ret == -ENOENT) continue; - if (cmp < 0) - break; - - item = scoutfs_clean_item(sb, key, - le16_to_cpu(ihdr->len)); - if (IS_ERR(item)) { - ret = PTR_ERR(item); - } else { - memcpy(item->val, (void *)(ihdr + 1), - item->val_len); - } break; } - - brelse(bh); - if (item) /* also breaks for IS_ERR */ - break; + break; } /* XXX release manifest */ - scoutfs_item_put(item); + /* XXX read-ahead all value blocks? */ + + if (!ret) { + ret = populate_ref(sb, le64_to_cpu(ment.blkno), bh, item, ref); + brelse(bh); + } + return ret; } /* - * Reading the next item is more expensive than looking up a specific - * item. We can't use the bloom filters because we don't know what key - * is next. We have to search blocks at all levels because the next - * item could be in any of them. - * - * After having gone to the trouble to establish next item positions in - * all the blocks we take the opportunity to amortize that cost and - * insert multiple items. - * - * This only returns an error if it was unsure if there's a next item - * or not. It will return success if there were no next items. The caller - * is responsible for retrying the lookup after reading. + * The dirty_item_off points to the byte offset after the last item. + * Advance it past block tails and initial block headers until there's + * room for an item with the given skip list elements height. Then set + * the dirty_item_off past the item offset item we return. */ -struct item_block_cursor { - struct list_head list; - - struct buffer_head *bh; - struct scoutfs_item_header *ihdr; - unsigned int i; -}; -int scoutfs_read_next_item(struct super_block *sb, - struct scoutfs_key *first_key) +static int add_item_off(struct scoutfs_sb_info *sbi, int height) { - struct scoutfs_ring_manifest_entry ment; - struct scoutfs_item_header *least; - struct scoutfs_item_header *ihdr; + int len = offsetof(struct scoutfs_item, skip_next[height]); + int off = sbi->dirty_item_off; + int tail_free; + + /* item's can't cross a block boundary */ + tail_free = SCOUTFS_BLOCK_SIZE - (off & SCOUTFS_BLOCK_MASK); + if (tail_free < len) + off += tail_free + sizeof(struct scoutfs_block_header); + + sbi->dirty_item_off = off + len; + return off; +} + +/* + * The dirty_val_off points to the first byte of the last value that + * was allocated. Subtract the offset to make room for a new item + * of the given length. If that crosses a block boundary or wanders + * into the block header then pull it back into the tail of the previous + * block. + */ +static int sub_val_off(struct scoutfs_sb_info *sbi, int len) +{ + int off = sbi->dirty_val_off - len; + int block_off; + int tail_free; + + /* values can't start in a block header */ + block_off = off & SCOUTFS_BLOCK_MASK; + if (block_off < sizeof(struct scoutfs_block_header)) + off -= (block_off + 1); + + /* values can't cross a block boundary */ + tail_free = SCOUTFS_BLOCK_SIZE - (off & SCOUTFS_BLOCK_MASK); + if (tail_free < len) + off -= len - tail_free; + + sbi->dirty_val_off = off; + return off; +} + +/* + * Initialize the buffers for the next dirty segment. We have to initialize + * the bloom filter bits and the item block header. + * + * XXX we need to really pin the blocks somehow + */ +static int start_dirty_segment(struct super_block *sb, u64 blkno) +{ + struct scoutfs_bloom_block *blm; struct scoutfs_item_block *iblk; - struct item_block_cursor *curs; - struct item_block_cursor *tmp; - struct scoutfs_item *item; - struct scoutfs_key key; struct buffer_head *bh; - LIST_HEAD(cursors); int ret = 0; - int pass; int i; - /* XXX hold manifest */ - - memset(&ment, 0, sizeof(struct scoutfs_ring_manifest_entry)); - - /* find all the log segments that contain our key */ - key = *first_key; - while (scoutfs_next_manifest_segment(sb, &key, &ment)) { - - curs = kmalloc(sizeof(struct item_block_cursor), GFP_NOFS); - if (!curs) { - ret = -ENOMEM; - goto out; - } - - bh = scoutfs_read_block(sb, le64_to_cpu(ment.blkno)); + for (i = 0; i < SCOUTFS_BLOCKS_PER_CHUNK; i++) { + bh = scoutfs_new_block(sb, blkno + i); if (!bh) { ret = -EIO; - goto out; + break; } - /* XXX verify */ - iblk = (void *)bh->b_data; + if (i < SCOUTFS_BLOOM_BLOCKS) { + blm = (void *)bh->b_data; + memset(blm->bits, 0, SCOUTFS_BLOCK_SIZE - + offsetof(struct scoutfs_bloom_block, bits)); + } - curs->bh = bh; - curs->i = 0; - curs->ihdr = (void *)(iblk + 1); - list_add_tail(&curs->list, &cursors); + if (i == SCOUTFS_BLOOM_BLOCKS) { + iblk = (void *)bh->b_data; + /* also zero first unused item slot */ + memset(&iblk->skip_root, 0, sizeof(iblk->skip_root) + + sizeof(struct scoutfs_item)); + } + + /* bh is pinned by sbi->dirty_blkno */ } - /* there can be no segments that contain the item */ - if (list_empty(&cursors)) { - ret = 0; + while (ret && i--) { + /* unwind pinned blocks on failure */ + bh = sb_getblk(sb, blkno + i); + if (bh) { + brelse(bh); + brelse(bh); + } + } + + return ret; +} + +/* + * Zero the portion of this block that intersects with the free space in + * the middle of the segment. @start and @end are chunk-relative byte + * offsets of the inclusive start and exclusive end of the free region. + */ +static void zero_unused_block(struct super_block *sb, struct buffer_head *bh, + u32 start, u32 end) +{ + u32 off = bh->b_blocknr << SCOUTFS_BLOCK_SHIFT; + + /* see if the segment range falls outside our block */ + if (start >= off + SCOUTFS_BLOCK_SIZE || end <= off) + return; + + /* convert the chunk offsets to our block offsets */ + start = max(start, off) - off; + end = min(off + SCOUTFS_BLOCK_SIZE, end) - off; + + /* don't zero block headers */ + start = max_t(u32, start, sizeof(struct scoutfs_block_header)); + end = max_t(u32, start, sizeof(struct scoutfs_block_header)); + + if (start < end) + memset(bh->b_data + start, 0, end - start); +} + +/* + * Finish off a dirty segment if we have one. Calculate the checksums of + * all the blocks, mark them dirty, and drop their pinned reference. + */ +int scoutfs_finish_dirty_segment(struct super_block *sb) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping; + struct buffer_head *bh; + u64 blkno; + int ret = 0; + u64 i; + + /* XXX sync doesn't lock this test? */ + blkno = sbi->dirty_blkno; + if (!blkno) + return 0; + + for (i = 0; i < SCOUTFS_BLOCKS_PER_CHUNK; i++) { + bh = scoutfs_read_block(sb, blkno + i); + /* should have been pinned */ + if (WARN_ON_ONCE(!bh)) { + ret = -EIO; + break; + } + + zero_unused_block(sb, bh, sbi->dirty_item_off, + sbi->dirty_val_off); + + scoutfs_calc_hdr_crc(bh); + mark_buffer_dirty(bh); + brelse(bh); + /* extra release to unpin */ + brelse(bh); + } + + /* + * XXX the manifest entry for this log segment has a key range + * that is much too large. We should shrink it here to reflect + * the real keys. That would reduce the number of blocks involved + * in merging it into level 1. + */ + + /* + * Try to kick off a background write of the finished segment. Callers + * can wait for the buffers in writeback if they need to. + */ + if (!ret) { + filemap_fdatawrite_range(mapping, blkno << SCOUTFS_CHUNK_SHIFT, + ((blkno + 1) << SCOUTFS_CHUNK_SHIFT) - 1); + sbi->dirty_blkno = 0; + } + + return ret; +} + +/* + * Return a reference to a newly allocated and initialized item in a + * block in the currently dirty log segment. + * + * Item creation is purposely kept very simple. Item and value offset + * allocation proceed from either end of the log segment. Once they + * intersect the log segment is full and written out. Deleted dirty + * items don't reclaim their space. The free space will be reclaimed by + * the level 0 -> level 1 merge that happens anyway. Not reclaiming + * free space makes item location more rigid and lets us relax the + * locking requirements of item references. An item reference doesn't + * have to worry about unrelated item modification moving their item + * around to, say, defragment free space. + */ +int scoutfs_create_item(struct super_block *sb, struct scoutfs_key *key, + unsigned bytes, struct scoutfs_item_ref *ref) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_ring_manifest_entry ment; + struct scoutfs_bloom_bits bits; + struct scoutfs_item *item; + struct buffer_head *bh; + int item_off; + int val_off; + int height; + u64 blkno; + int ret = 0; + + /* XXX how big should items really get? */ + if (WARN_ON_ONCE(bytes == 0 || bytes > 4096)) + return -EINVAL; + + height = scoutfs_skip_random_height(); + + mutex_lock(&sbi->dirty_mutex); + +next_chunk: + if (!sbi->dirty_blkno) { + ret = scoutfs_alloc_chunk(sb, &blkno); + if (ret) + goto out; + + /* XXX free blkno on error? */ + ret = start_dirty_segment(sb, blkno); + if (ret) + goto out; + + /* + * We need a local manifest in memory to find items as + * we insert them in the dirty segment. We don't know + * what keys are going to be used so we cover the whole + * thing. + * + * XXX But we're also adding it to the ring here. We should + * add it as its finalized and its item range is collapsed. + */ + ment.blkno = cpu_to_le64(blkno); + ment.seq = sbi->super.hdr.seq; + ment.level = 0; + memset(&ment.first, 0, sizeof(ment.first)); + memset(&ment.last, ~0, sizeof(ment.last)); + ret = scoutfs_new_manifest(sb, &ment); + if (ret) + goto out; + + sbi->dirty_blkno = blkno; + sbi->dirty_item_off = + (SCOUTFS_BLOCK_SIZE * SCOUTFS_BLOOM_BLOCKS) + + sizeof(struct scoutfs_item_block); + sbi->dirty_val_off = SCOUTFS_CHUNK_SIZE; + } + + item_off = add_item_off(sbi, height); + val_off = sub_val_off(sbi, bytes); + + if (item_off > val_off) { + ret = scoutfs_finish_dirty_segment(sb); + if (ret) + goto out; + goto next_chunk; + } + + /* XXX fix up this error handling in general */ + + bh = scoutfs_read_block_off(sb, sbi->dirty_blkno, item_off); + if (!bh) { + ret = -EIO; goto out; } - /* XXX arbitrary number of next items to insert */ - for (pass = 0; pass < 16; pass++) { + /* populate iblk first and last? better than in manifest? */ - least = NULL; - list_for_each_entry(curs, &cursors, list) { - iblk = (void *)curs->bh->b_data; - ihdr = curs->ihdr; - i = curs->i; + item = (void *)bh->b_data + (item_off & SCOUTFS_BLOCK_MASK); + item->key = *key; + item->offset = cpu_to_le32(val_off); + item->len = cpu_to_le16(bytes); + item->skip_height = height; - /* Find the next item past the search key. */ - for (; i < le32_to_cpu(iblk->nr_items); i++) { - if (scoutfs_key_cmp(&key, &ihdr->key) <= 0) - break; + ret = scoutfs_skip_insert(sb, sbi->dirty_blkno, item, item_off); + if (ret) + goto out; - ihdr = next_ihdr(ihdr); - } + ret = populate_ref(sb, sbi->dirty_blkno, bh, item, ref); + brelse(bh); + if (ret) + goto out; - /* - * If we fall off a block then we can't know if - * we have the least key without checking the - * next block at that level. It could have an - * item less than the least in our other blocks. - */ - if (WARN_ON_ONCE(i == le32_to_cpu(iblk->nr_items))) { - ret = -EIO; + /* XXX delete skip on failure? */ + + /* set the bloom bits last because we can't unset them */ + scoutfs_calc_bloom_bits(&bits, key, sbi->super.bloom_salts); + ret = scoutfs_set_bloom_bits(sb, sbi->dirty_blkno, &bits); +out: + WARN_ON_ONCE(ret); /* XXX error paths are not robust */ + mutex_unlock(&sbi->dirty_mutex); + return ret; +} + +/* + * This is a really cheesy temporary delete method. It only works on items + * that are stored in dirty blocks. The caller is responsible for dropping + * the ref. XXX be less bad. + */ +int scoutfs_delete_item(struct super_block *sb, struct scoutfs_item_ref *ref) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + u64 blkno; + int ret; + + blkno = round_down(ref->item_bh->b_blocknr, SCOUTFS_BLOCKS_PER_CHUNK); + if (WARN_ON_ONCE(blkno != sbi->dirty_blkno)) + return -EINVAL; + + ret = scoutfs_skip_delete(sb, blkno, ref->key); + WARN_ON_ONCE(ret); + return ret; +} + +/* + * Return a reference to the next item in the inclusive search range. + * The caller should have access to the search key range. + * + * We walk the manifest to find all the log segments that could contain + * the start of the range. We hold cursors on the blocks in the + * segments. Each next item iteration comes from finding the least of + * the next item at all these cursors. + * + * If we exhaust a segment at a given level we may need to search the + * next segment in that level to find the next item. The manifest may + * have changed under us while we walked our old set of segments. So we + * restart the entire search to get another consistent collection of + * segments to search. + * + * We put the segment references and iteration cursors in a list in the + * caller so that they can find many next items by advancing the cursors + * without having to walk the manifest and perform initial binary + * searches in each segment. + * + * The caller is responsible for putting the item ref if we return + * success. -ENOENT is returned if there are no more items in the + * search range. + * + * XXX this is wonky. We don't want to search the manifest for the + * range, just the initial value. Then we record the last key in + * segments we finish and only restart if least is > that or there are + * no least. We have to advance the first key when restarting the + * search. + */ +int scoutfs_next_item(struct super_block *sb, struct scoutfs_key *first, + struct scoutfs_key *last, struct list_head *iter_list, + struct scoutfs_item_ref *ref) +{ + struct scoutfs_ring_manifest_entry ment; + struct scoutfs_item_iter *least; + struct scoutfs_item_iter *iter; + struct scoutfs_item_iter *pos; + int ret; + +restart: + if (list_empty(iter_list)) { + + /* + * Find all the segments that intersect the search range + * and find the next item in the block from the start + * of the range. + */ + memset(&ment, 0, sizeof(struct scoutfs_ring_manifest_entry)); + while (scoutfs_foreach_range_segment(sb, first, last, &ment)) { + iter = kzalloc(sizeof(struct scoutfs_item_iter), + GFP_NOFS); + if (!iter) { + ret = -ENOMEM; goto out; } /* - * Remember the newest least key in the blocks that's - * past the search key. + * We will restart the walk of the manifest blocks if + * we iterate over all the items in this block without + * exhausting the search range. */ - if (!least || - scoutfs_key_cmp(&ihdr->key, &least->key) < 0) - least = ihdr; + if (ment.level > 0 && + scoutfs_key_cmp(&ment.last, last) < 0) + iter->restart_after = true; - curs->ihdr = ihdr; - curs->i = i; + iter->blkno = le64_to_cpu(ment.blkno); + list_add_tail(&iter->list, iter_list); + } + if (list_empty(iter_list)) { + ret = -ENOENT; + goto out; + } + } + + least = NULL; + ret = 0; + list_for_each_entry_safe(iter, pos, iter_list, list) { + + /* search towards the first key if we haven't yet */ + if (!iter->item) { + ret = scoutfs_skip_search(sb, iter->blkno, first, + &iter->bh, &iter->item); } - /* start the next search past the next key */ - key = least->key; - scoutfs_inc_key(&key); + /* then iterate until we find or pass the first key */ + while (!ret && scoutfs_key_cmp(&iter->item->key, first) < 0) { + ret = scoutfs_skip_next(sb, iter->blkno, + &iter->bh, &iter->item); + } - /* insert the next item (XXX if it's not deleted) */ - item = scoutfs_clean_item(sb, &least->key, - le16_to_cpu(least->len)); - if (IS_ERR(item)) { - ret = PTR_ERR(item); - if (ret == -EEXIST) + /* we're done with this block if we past the last key */ + while (!ret && scoutfs_key_cmp(&iter->item->key, last) > 0) { + brelse(iter->bh); + iter->bh = NULL; + iter->item = NULL; + ret = -ENOENT; + } + + if (ret == -ENOENT) { + if (iter->restart_after) { + /* need next block at this level */ + scoutfs_put_iter_list(iter_list); + goto restart; + } else { + /* this level is done */ + list_del_init(&iter->list); + brelse(iter->bh); + kfree(iter); continue; - break; - } - - memcpy(item->val, (void *)(least + 1), item->val_len); - scoutfs_item_put(item); - } -out: - list_for_each_entry_safe(curs, tmp, &cursors, list) { - brelse(curs->bh); - list_del_init(&curs->list); - kfree(curs); - } - return ret; -} - -static int finish_item_block(struct super_block *sb, struct buffer_head *bh, - void *until) -{ - struct scoutfs_item_block *iblk = (void *)bh->b_data; - struct scoutfs_ring_manifest_entry ment; - - memset(until, 0, (void *)bh->b_data + SCOUTFS_BLOCK_SIZE - until); - scoutfs_calc_hdr_crc(bh); - unlock_buffer(bh); - brelse(bh); - - ment.blkno = cpu_to_le64(bh->b_blocknr); - ment.seq = iblk->hdr.seq; - ment.level = 0; - ment.first = iblk->first; - ment.last = iblk->last; - - return scoutfs_new_manifest(sb, &ment); -} - -/* - * Write all the currently dirty items in newly allocated log segments. - * New ring entries are added as the alloc bitmap is modified and as the - * manifest is updated. If we write out all the item and ring blocks then - * we write a new super that references those new blocks. - */ -int scoutfs_write_dirty_items(struct super_block *sb) -{ - struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping; - struct scoutfs_item_header *ihdr; - struct scoutfs_item_block *iblk; - struct scoutfs_item *item; - struct buffer_head *bh; - int val_space; - u64 blkno; - int ret; - - /* XXX wait until transactions are complete */ - - item = NULL; - iblk = NULL; - while ((item = scoutfs_item_next_dirty(sb, item))) { - - if (iblk && (item->val_len > val_space)) { - iblk = NULL; - ret = finish_item_block(sb, bh, ihdr); - if (ret) - break; - } - - if (!iblk) { - /* get the next item block */ - ret = scoutfs_alloc_chunk(sb, &blkno); - if (ret) - break; - - bh = scoutfs_dirty_block(sb, blkno); - if (!bh) { - ret = -ENOMEM; - break; } - - iblk = (void *)bh->b_data; - iblk->first = item->key; - iblk->nr_items = 0; - ihdr = (void *)(iblk + 1); - /* XXX assuming that val_space is big enough */ } + if (ret) + goto out; - iblk->last = item->key; - ihdr->key = item->key; - ihdr->len = cpu_to_le16(item->val_len); - memcpy((void *)(ihdr + 1), item->val, item->val_len); - le32_add_cpu(&iblk->nr_items, 1); - - /* XXX assuming that the next ihdr fits */ - ihdr = (void *)(ihdr + 1) + le16_to_cpu(ihdr->len); - val_space = (char *)iblk + SCOUTFS_BLOCK_SIZE - - (char *)(ihdr + 1); + /* remember the most recent smallest key from the first */ + if (!least || + scoutfs_key_cmp(&iter->item->key, &least->item->key) < 0) + least = iter; } - scoutfs_item_put(item); /* only if the loop aborted */ - - /* finish writing if we did work and haven't failed */ - if (iblk && !ret) { - ret = finish_item_block(sb, bh, ihdr) ?: - scoutfs_finish_dirty_ring(sb) ?: - filemap_write_and_wait(mapping) ?: - scoutfs_write_dirty_super(sb); - if (!ret) { - scoutfs_advance_dirty_super(sb); - scoutfs_item_all_clean(sb); - } - } - - /* XXX better tear down down in the error case */ - + if (least) + ret = populate_ref(sb, least->blkno, least->bh, least->item, + ref); + else + ret = -ENOENT; +out: + if (ret) + scoutfs_put_iter_list(iter_list); return ret; + } diff --git a/kmod/src/segment.h b/kmod/src/segment.h index fd0fda69..6b41579b 100644 --- a/kmod/src/segment.h +++ b/kmod/src/segment.h @@ -1,9 +1,33 @@ #ifndef _SCOUTFS_SEGMENT_H_ #define _SCOUTFS_SEGMENT_H_ -int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key); -int scoutfs_read_next_item(struct super_block *sb, - struct scoutfs_key *first_key); -int scoutfs_write_dirty_items(struct super_block *sb); +struct scoutfs_item_ref { + /* usable by callers */ + struct scoutfs_key *key; + unsigned int val_len; + void *val; + + /* private buffer head refs */ + struct buffer_head *item_bh; + struct buffer_head *val_bh; +}; + +#define DECLARE_SCOUTFS_ITEM_REF(name) \ + struct scoutfs_item_ref name = {NULL ,} + +void scoutfs_put_ref(struct scoutfs_item_ref *ref); +void scoutfs_put_iter_list(struct list_head *list); + +int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_item_ref *ref); +int scoutfs_create_item(struct super_block *sb, struct scoutfs_key *key, + unsigned bytes, struct scoutfs_item_ref *ref); +int scoutfs_delete_item(struct super_block *sb, struct scoutfs_item_ref *ref); +int scoutfs_next_item(struct super_block *sb, struct scoutfs_key *first, + struct scoutfs_key *last, struct list_head *iter_list, + struct scoutfs_item_ref *ref); + +int scoutfs_finish_dirty_segment(struct super_block *sb); + #endif diff --git a/kmod/src/skip.c b/kmod/src/skip.c new file mode 100644 index 00000000..69bae2b8 --- /dev/null +++ b/kmod/src/skip.c @@ -0,0 +1,325 @@ +/* + * 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 +#include +#include + +#include "format.h" +#include "key.h" +#include "block.h" +#include "skip.h" + +/* + * The items in a log segment block are sorted by their keys in a skip + * list. The skip list was chosen because it is so easy to implement + * and could, maybe some day, offer solid concurrent updates and reads. + * It also adds surprisingly little per-item overhead because half of + * the items only have one link. + * + * The list is rooted in the item block which follows the last bloom + * block in the segment. The links in the skip list elements are byte + * offsets of the start of items relative to the start of the log + * segment. + * + * We chose a limit on the height of 16 links. That gives around 64k + * items without going too crazy. That's around the higher end of the + * number of items we expect in log segments. + * + * This isn't quite a generic implementation. It knows that the items + * are rooted in the item block at a given offset in the log segment. + * It knows that the pointers are items and where the skip links are in + * its struct. It knows to compare the items by their key. + * + * The caller is completely responsible for serialization. + * + * The buffer_head reads here won't be as expensive as they might seem. + * The caller holds the blocks pinned so the worst case are block device + * page radix rcu lookups. Repeated reads of the recent blocks will hit + * the per-cpu lru bh reference caches. + */ + +struct skip_path { + struct buffer_head *root_bh; + + /* + * Pointers to the buffer heads which contain the blocks which are + * referenced by the next pointers in the path. + */ + struct buffer_head *bh[SCOUTFS_SKIP_HEIGHT]; + + /* + * Store the location of the index that references the item that + * we found. Insertion will modify the referenced index to add + * an entry before the item and deletion will modify the referenced + * index to remove the item. + */ + __le32 *next[SCOUTFS_SKIP_HEIGHT]; +}; + +#define DECLARE_SKIP_PATH(name) \ + struct skip_path name = {NULL, } + +/* + * Not all byte offsets are possible locations of items. Items have to + * be after the bloom blocks and item block header, can't be in + * the block headers for the rest of the blocks, and can't be a partial + * struct at the end of a block. + * + * This is just a rough check. It doesn't catch items offsets that overlap + * with other items or values. + */ +static int invalid_item_off(u32 off) +{ + return off < ((SCOUTFS_BLOCK_SIZE * SCOUTFS_BLOOM_BLOCKS) + + sizeof(struct scoutfs_item_block)) || + (off & SCOUTFS_BLOCK_MASK) < + sizeof(struct scoutfs_block_header) || + (off & SCOUTFS_BLOCK_MASK) > + (SCOUTFS_BLOCK_SIZE - sizeof(struct scoutfs_item)); +} + +/* + * Set the caller's item to the item in the segment at the given byte + * offset and set their bh to the block that contains it. + */ +static int skip_read_item(struct super_block *sb, u64 blkno, __le32 off, + struct buffer_head **bh, struct scoutfs_item **item) +{ + if (WARN_ON_ONCE(invalid_item_off(le32_to_cpu(off)))) + return -EINVAL; + + *bh = scoutfs_read_block_off(sb, blkno, le32_to_cpu(off)); + if (!(*bh)) { + *bh = NULL; + *item = NULL; + return -EIO; + } + + *item = (void *)(*bh)->b_data + (le32_to_cpu(off) & SCOUTFS_BLOCK_MASK); + return 0; +} + +/* + * Find the next item in the skiplist with a key greater than or equal + * to the given key. Set the path pointers to the hops before this item + * so that we can modify those pointers to insert an item before it in + * the list or delete it. + * + * The caller is responsible for initializing the path and cleaning it up. + */ +static int skip_search(struct super_block *sb, u64 blkno, + struct skip_path *path, struct scoutfs_key *key, + int *cmp) +{ + struct scoutfs_item_block *iblk; + struct scoutfs_item *item; + struct buffer_head *bh; + __le32 *next; + int ret = 0; + int i; + + /* fake lesser comparison for insertion into an empty list */ + *cmp = -1; + + bh = scoutfs_read_block(sb, blkno + SCOUTFS_BLOOM_BLOCKS); + if (!bh) + return -EIO; + + /* XXX verify */ + iblk = (void *)bh->b_data; + next = iblk->skip_root.next; + path->root_bh = bh; + + for (i = SCOUTFS_SKIP_HEIGHT - 1; i >= 0; i--) { + while (next[i]) { + ret = skip_read_item(sb, blkno, next[i], &bh, &item); + if (ret) + goto out; + + *cmp = scoutfs_key_cmp(key, &item->key); + if (*cmp <= 0) { + brelse(bh); + break; + } + + next = item->skip_next; + if (path->bh[i]) + brelse(path->bh[i]); + path->bh[i] = bh; + } + + path->next[i] = &next[i]; + } +out: + return ret; +} + +static void skip_release_path(struct skip_path *path) +{ + int i; + + if (path->root_bh) + brelse(path->root_bh); + + for (i = 0; i < SCOUTFS_SKIP_HEIGHT; i++) { + if (path->bh[i]) { + brelse(path->bh[i]); + path->bh[i] = NULL; + } + } +} + +/* + * We want heights with a distribution of 1 / (2^h). Half the items + * have a height of 1, a quarter have 2, an eighth have 3, etc. + * + * Finding the first low set bit in a random number achieves this + * nicely. ffs() even counts the bits from 1 so it matches our height. + * + * But ffs() returns 0 if no bits are set. We prevent a 0 height and + * limit the max height returned by oring in our max height. + */ +u8 scoutfs_skip_random_height(void) +{ + return ffs(get_random_int() | (1 << (SCOUTFS_SKIP_HEIGHT - 1))); +} + +/* + * Insert a new item in the item block's skip list. The caller provides + * an initialized item, particularly it's skip height and key, and + * the byte offset in the log segment of the item struct. + */ +int scoutfs_skip_insert(struct super_block *sb, u64 blkno, + struct scoutfs_item *item, u32 off) +{ + DECLARE_SKIP_PATH(path); + int cmp; + int ret; + int i; + + if (WARN_ON_ONCE(invalid_item_off(off)) || + WARN_ON_ONCE(item->skip_height > SCOUTFS_SKIP_HEIGHT)) + return -EINVAL; + + ret = skip_search(sb, blkno, &path, &item->key, &cmp); + if (ret == 0) { + if (cmp == 0) { + ret = -EEXIST; + } else { + for (i = 0; i < item->skip_height; i++) { + item->skip_next[i] = *path.next[i]; + *path.next[i] = cpu_to_le32(off); + } + } + } + + skip_release_path(&path); + return ret; +} + +static int skip_lookup(struct super_block *sb, u64 blkno, + struct scoutfs_key *key, struct buffer_head **bh, + struct scoutfs_item **item, bool exact) +{ + DECLARE_SKIP_PATH(path); + int cmp; + int ret; + + ret = skip_search(sb, blkno, &path, key, &cmp); + if (ret == 0) { + if ((exact && cmp) || *path.next[0] == 0) { + ret = -ENOENT; + } else { + ret = skip_read_item(sb, blkno, *path.next[0], + bh, item); + } + } + + skip_release_path(&path); + return ret; +} + +/* + * Find the item at the given key in the skip list. + */ +int scoutfs_skip_lookup(struct super_block *sb, u64 blkno, + struct scoutfs_key *key, struct buffer_head **bh, + struct scoutfs_item **item) +{ + return skip_lookup(sb, blkno, key, bh, item, true); +} + +/* + * Find the next item after the given key in the skip list. + */ +int scoutfs_skip_search(struct super_block *sb, u64 blkno, + struct scoutfs_key *key, struct buffer_head **bh, + struct scoutfs_item **item) +{ + return skip_lookup(sb, blkno, key, bh, item, false); +} + +int scoutfs_skip_delete(struct super_block *sb, u64 blkno, + struct scoutfs_key *key) +{ + struct scoutfs_item *item; + DECLARE_SKIP_PATH(path); + struct buffer_head *bh; + int cmp; + int ret; + int i; + + ret = skip_search(sb, blkno, &path, key, &cmp); + if (ret == 0) { + if (*path.next[0] && cmp) { + ret = -ENOENT; + } else { + ret = skip_read_item(sb, blkno, *path.next[0], + &bh, &item); + if (!ret) { + for (i = 0; i < item->skip_height; i++) + *path.next[i] = item->skip_next[i]; + brelse(bh); + } + } + } + + skip_release_path(&path); + return ret; +} + +/* + * The caller has found a valid item with search or lookup. We can use + * the lowest level links to advance through the rest of the items. The + * caller has made sure that this is safe. + */ +int scoutfs_skip_next(struct super_block *sb, u64 blkno, + struct buffer_head **bh, struct scoutfs_item **item) +{ + __le32 next; + + if (!(*bh)) + return -ENOENT; + + next = (*item)->skip_next[0]; + brelse(*bh); + + if (!next) { + *bh = NULL; + *item = NULL; + return -ENOENT; + } + + return skip_read_item(sb, blkno, next, bh, item); +} diff --git a/kmod/src/skip.h b/kmod/src/skip.h new file mode 100644 index 00000000..979719cc --- /dev/null +++ b/kmod/src/skip.h @@ -0,0 +1,18 @@ +#ifndef _SCOUTFS_SKIP_H_ +#define _SCOUTFS_SKIP_H_ + +u8 scoutfs_skip_random_height(void); +int scoutfs_skip_insert(struct super_block *sb, u64 blkno, + struct scoutfs_item *item, u32 off); +int scoutfs_skip_lookup(struct super_block *sb, u64 blkno, + struct scoutfs_key *key, struct buffer_head **bh, + struct scoutfs_item **item); +int scoutfs_skip_search(struct super_block *sb, u64 blkno, + struct scoutfs_key *key, struct buffer_head **bh, + struct scoutfs_item **item); +int scoutfs_skip_delete(struct super_block *sb, u64 blkno, + struct scoutfs_key *key); +int scoutfs_skip_next(struct super_block *sb, u64 blkno, + struct buffer_head **bh, struct scoutfs_item **item); + +#endif diff --git a/kmod/src/super.c b/kmod/src/super.c index 3bdfc743..ba200876 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -28,10 +28,23 @@ #include "ring.h" #include "segment.h" +/* + * We've been dirtying log segment blocks and ring blocks as items were + * modified. sync makes sure that they're all persistent and updates + * the super. + * + * XXX need to synchronize with transactions + * XXX is state clean after errors? + */ static int scoutfs_sync_fs(struct super_block *sb, int wait) { - /* XXX always waiting */ - return scoutfs_write_dirty_items(sb); + struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping; + + return scoutfs_finish_dirty_segment(sb) ?: + scoutfs_finish_dirty_ring(sb) ?: + filemap_write_and_wait(mapping) ?: + scoutfs_write_dirty_super(sb) ?: + scoutfs_advance_dirty_super(sb); } static const struct super_operations scoutfs_super_ops = { @@ -45,7 +58,7 @@ static const struct super_operations scoutfs_super_ops = { * every time it wants to dirty it and eventually write it to reference * dirty data that's been written. */ -void scoutfs_advance_dirty_super(struct super_block *sb) +int scoutfs_advance_dirty_super(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; @@ -57,6 +70,8 @@ void scoutfs_advance_dirty_super(struct super_block *sb) super->hdr.blkno = cpu_to_le64(SCOUTFS_SUPER_BLKNO + blkno); le64_add_cpu(&super->hdr.seq, 1); + + return 0; } /* @@ -71,16 +86,16 @@ int scoutfs_write_dirty_super(struct super_block *sb) size_t sz; int ret; - bh = scoutfs_dirty_block(sb, le64_to_cpu(super->hdr.blkno)); + bh = scoutfs_new_block(sb, le64_to_cpu(super->hdr.blkno)); if (!bh) return -ENOMEM; sz = sizeof(struct scoutfs_super_block); memcpy(bh->b_data, super, sz); memset(bh->b_data + sz, 0, SCOUTFS_BLOCK_SIZE - sz); - scoutfs_calc_hdr_crc(bh); - unlock_buffer(bh); + scoutfs_calc_hdr_crc(bh); + mark_buffer_dirty(bh); ret = sync_dirty_buffer(bh); brelse(bh); @@ -170,6 +185,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) sbi->item_root = RB_ROOT; sbi->dirty_item_root = RB_ROOT; spin_lock_init(&sbi->chunk_alloc_lock); + mutex_init(&sbi->dirty_mutex); if (!sb_set_blocksize(sb, SCOUTFS_BLOCK_SIZE)) { printk(KERN_ERR "couldn't set blocksize\n"); @@ -209,9 +225,15 @@ static struct dentry *scoutfs_mount(struct file_system_type *fs_type, int flags, static void scoutfs_kill_sb(struct super_block *sb) { + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + kill_block_super(sb); - scoutfs_destroy_manifest(sb); - kfree(sb->s_fs_info); + if (sbi) { + /* kill block super should have synced */ + WARN_ON_ONCE(sbi->dirty_blkno); + scoutfs_destroy_manifest(sb); + kfree(sbi); + } } static struct file_system_type scoutfs_fs_type = { diff --git a/kmod/src/super.h b/kmod/src/super.h index 604448ca..1663034d 100644 --- a/kmod/src/super.h +++ b/kmod/src/super.h @@ -26,6 +26,11 @@ struct scoutfs_sb_info { struct scoutfs_ring_entry *dirty_ring_ent; unsigned int dirty_ring_ent_avail; + /* pinned log segment during fs modifications */ + struct mutex dirty_mutex; + u64 dirty_blkno; + int dirty_item_off; + int dirty_val_off; }; static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb) @@ -33,7 +38,7 @@ static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb) return sb->s_fs_info; } -void scoutfs_advance_dirty_super(struct super_block *sb); +int scoutfs_advance_dirty_super(struct super_block *sb); int scoutfs_write_dirty_super(struct super_block *sb); #endif