diff --git a/kmod/src/block.c b/kmod/src/block.c index 7808ab5b..e6ddf0a8 100644 --- a/kmod/src/block.c +++ b/kmod/src/block.c @@ -11,10 +11,9 @@ * General Public License for more details. */ #include +#include +#include #include -#include -#include -#include #include "super.h" #include "format.h" @@ -23,74 +22,51 @@ #include "counters.h" #include "buddy.h" -#define DIRTY_RADIX_TAG 0 - /* + * scoutfs has a fixed 4k small block size for metadata blocks. This + * lets us consistently use buffer heads without worrying about having a + * block size greater than the page size. + * + * This block interface does the work to cow dirty blocks, track dirty + * blocks, generate checksums as they're written, only write them in + * transactions, verify checksums on read, and invalidate and retry + * reads of stale cached blocks. (That last bit only has a hint of an + * implementation.) + * * XXX - * - tie into reclaim - * - per cpu lru of refs? - * - relax locking - * - get, check, and fill slots instead of full radix walks - * - block slab - * - maybe more clever wait functions + * - tear down dirty blocks left by write errors on unmount + * - should invalidate dirty blocks if freed */ -static struct scoutfs_block *alloc_block(struct super_block *sb, u64 blkno) +struct block_bh_private { + struct super_block *sb; + struct buffer_head *bh; + struct rb_node node; +}; + +enum { + BH_ScoutfsVerified = BH_PrivateStart, +}; +BUFFER_FNS(ScoutfsVerified, scoutfs_verified) + +static int verify_block_header(struct scoutfs_sb_info *sbi, + struct buffer_head *bh) { - struct scoutfs_block *bl; - struct page *page; - - /* we'd need to be just a bit more careful */ - BUILD_BUG_ON(PAGE_SIZE > SCOUTFS_BLOCK_SIZE); - - bl = kzalloc(sizeof(struct scoutfs_block), GFP_NOFS); - if (bl) { - page = alloc_pages(GFP_NOFS, SCOUTFS_BLOCK_PAGE_ORDER); - WARN_ON_ONCE(!page); - if (page) { - init_rwsem(&bl->rwsem); - atomic_set(&bl->refcount, 1); - bl->blkno = blkno; - bl->sb = sb; - bl->page = page; - bl->data = page_address(page); - scoutfs_inc_counter(sb, block_mem_alloc); - } else { - kfree(bl); - bl = NULL; - } - } - - return bl; -} - -void scoutfs_put_block(struct scoutfs_block *bl) -{ - if (!IS_ERR_OR_NULL(bl) && atomic_dec_and_test(&bl->refcount)) { - trace_printk("freeing bl %p\n", bl); - __free_pages(bl->page, SCOUTFS_BLOCK_PAGE_ORDER); - kfree(bl); - scoutfs_inc_counter(bl->sb, block_mem_free); - } -} - -static int verify_block_header(struct super_block *sb, struct scoutfs_block *bl) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; - struct scoutfs_block_header *hdr = bl->data; + struct scoutfs_block_header *hdr = (void *)bh->b_data; u32 crc = scoutfs_crc_block(hdr); int ret = -EIO; if (le32_to_cpu(hdr->crc) != crc) { - printk("blkno %llu hdr crc %x != calculated %x\n", bl->blkno, - le32_to_cpu(hdr->crc), crc); + printk("blkno %llu hdr crc %x != calculated %x\n", + (u64)bh->b_blocknr, le32_to_cpu(hdr->crc), crc); } else if (super->hdr.fsid && hdr->fsid != super->hdr.fsid) { - printk("blkno %llu fsid %llx != super fsid %llx\n", bl->blkno, - le64_to_cpu(hdr->fsid), le64_to_cpu(super->hdr.fsid)); - } else if (le64_to_cpu(hdr->blkno) != bl->blkno) { - printk("blkno %llu invalid hdr blkno %llx\n", bl->blkno, - le64_to_cpu(hdr->blkno)); + printk("blkno %llu fsid %llx != super fsid %llx\n", + (u64)bh->b_blocknr, le64_to_cpu(hdr->fsid), + le64_to_cpu(super->hdr.fsid)); + } else if (le64_to_cpu(hdr->blkno) != bh->b_blocknr) { + printk("blkno %llu invalid hdr blkno %llx\n", + (u64)bh->b_blocknr, le64_to_cpu(hdr->blkno)); } else { ret = 0; } @@ -98,175 +74,145 @@ static int verify_block_header(struct super_block *sb, struct scoutfs_block *bl) return ret; } -static void block_read_end_io(struct bio *bio, int err) +static struct buffer_head *bh_from_bhp_node(struct rb_node *node) { - struct scoutfs_block *bl = bio->bi_private; - struct scoutfs_sb_info *sbi = SCOUTFS_SB(bl->sb); + struct block_bh_private *bhp; - if (!err && !verify_block_header(bl->sb, bl)) - set_bit(SCOUTFS_BLOCK_BIT_UPTODATE, &bl->bits); - else - set_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits); + bhp = container_of(node, struct block_bh_private, node); + return bhp->bh; +} - /* - * uncontended spin_lock in wake_up and unconditional smp_mb to - * make waitqueue_active safe are about the same cost, so we - * prefer the obviously safe choice. - */ - wake_up(&sbi->block_wq); +static struct scoutfs_sb_info *sbi_from_bh(struct buffer_head *bh) +{ + struct block_bh_private *bhp = bh->b_private; - scoutfs_put_block(bl); - bio_put(bio); + return SCOUTFS_SB(bhp->sb); +} + +static void insert_bhp_rb(struct rb_root *root, struct buffer_head *ins) +{ + struct rb_node **node = &root->rb_node; + struct rb_node *parent = NULL; + struct block_bh_private *bhp; + struct buffer_head *bh; + + while (*node) { + parent = *node; + bh = bh_from_bhp_node(*node); + + if (ins->b_blocknr < bh->b_blocknr) + node = &(*node)->rb_left; + else + node = &(*node)->rb_right; + } + + bhp = ins->b_private; + rb_link_node(&bhp->node, parent, node); + rb_insert_color(&bhp->node, root); } /* - * Once a transaction block is persistent it's fine to drop the dirty - * tag. It's been checksummed so it can be read in again. It's seq - * will be in the current transaction so it'll simply be dirtied and - * checksummed and written out again. + * Track a dirty block by allocating private data and inserting it into + * the dirty rbtree in the super block. + * + * Callers are in transactions that prevent metadata writeback so blocks + * won't be written and cleaned while we're trying to dirty them. We + * serialize racing to add dirty tracking to the same block in case the + * caller didn't. + * + * Presence in the dirty tree holds a bh ref. */ -static void block_write_end_io(struct bio *bio, int err) +static int insert_bhp(struct super_block *sb, struct buffer_head *bh) { - struct scoutfs_block *bl = bio->bi_private; - struct scoutfs_sb_info *sbi = SCOUTFS_SB(bl->sb); + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct block_bh_private *bhp; unsigned long flags; + int ret = 0; - if (!err) { - spin_lock_irqsave(&sbi->block_lock, flags); - radix_tree_tag_clear(&sbi->block_radix, - bl->blkno, DIRTY_RADIX_TAG); - spin_unlock_irqrestore(&sbi->block_lock, flags); + if (bh->b_private) + return 0; + + lock_buffer(bh); + if (bh->b_private) + goto out; + + bhp = kmalloc(sizeof(*bhp), GFP_NOFS); + if (!bhp) { + ret = -ENOMEM; + goto out; } - /* not too worried about racing ints */ - if (err && !sbi->block_write_err) - sbi->block_write_err = err; + bhp->sb = sb; + bhp->bh = bh; + get_bh(bh); + bh->b_private = bhp; - if (atomic_dec_and_test(&sbi->block_writes)) - wake_up(&sbi->block_wq); - - scoutfs_put_block(bl); - bio_put(bio); + spin_lock_irqsave(&sbi->block_lock, flags); + insert_bhp_rb(&sbi->block_dirty_tree, bh); + spin_unlock_irqrestore(&sbi->block_lock, flags); + trace_printk("blkno %llu bh %p\n", (u64)bh->b_blocknr, bh); +out: + unlock_buffer(bh); + return ret; } -static int block_submit_bio(struct scoutfs_block *bl, int rw) +static void erase_bhp(struct buffer_head *bh) { - struct super_block *sb = bl->sb; - struct bio *bio; - int ret; + struct block_bh_private *bhp = bh->b_private; + struct scoutfs_sb_info *sbi = sbi_from_bh(bh); + unsigned long flags; - if (WARN_ON_ONCE(bl->blkno >= - i_size_read(sb->s_bdev->bd_inode) >> SCOUTFS_BLOCK_SHIFT)) { - printk("trying to read bad blkno %llu\n", bl->blkno); - } + spin_lock_irqsave(&sbi->block_lock, flags); + rb_erase(&bhp->node, &sbi->block_dirty_tree); + spin_unlock_irqrestore(&sbi->block_lock, flags); + put_bh(bh); + kfree(bhp); + bh->b_private = NULL; - bio = bio_alloc(GFP_NOFS, SCOUTFS_PAGES_PER_BLOCK); - if (WARN_ON_ONCE(!bio)) - return -ENOMEM; - - bio->bi_sector = bl->blkno << (SCOUTFS_BLOCK_SHIFT - 9); - bio->bi_bdev = sb->s_bdev; - if (rw & WRITE) { - bio->bi_end_io = block_write_end_io; - } else - bio->bi_end_io = block_read_end_io; - bio->bi_private = bl; - - ret = bio_add_page(bio, bl->page, SCOUTFS_BLOCK_SIZE, 0); - if (WARN_ON_ONCE(ret != SCOUTFS_BLOCK_SIZE)) { - bio_put(bio); - return -ENOMEM; - } - - atomic_inc(&bl->refcount); - submit_bio(rw, bio); - - return 0; + trace_printk("blkno %llu bh %p\n", (u64)bh->b_blocknr, bh); } /* * Read an existing block from the device and verify its metadata header. + * The buffer head is returned unlocked and uptodate. */ -struct scoutfs_block *scoutfs_read_block(struct super_block *sb, u64 blkno) +struct buffer_head *scoutfs_block_read(struct super_block *sb, u64 blkno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_block *found; - struct scoutfs_block *bl; - unsigned long flags; + struct buffer_head *bh; int ret; - /* find an existing block, dropping if it's errored */ - spin_lock_irqsave(&sbi->block_lock, flags); + bh = sb_bread(sb, blkno); + if (!bh) { + bh = ERR_PTR(-EIO); + goto out; + } - bl = radix_tree_lookup(&sbi->block_radix, blkno); - if (bl) { - if (test_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits)) { - radix_tree_delete(&sbi->block_radix, bl->blkno); - scoutfs_put_block(bl); - bl = NULL; - } else { - atomic_inc(&bl->refcount); + if (!buffer_scoutfs_verified(bh)) { + lock_buffer(bh); + if (!buffer_scoutfs_verified(bh)) { + ret = verify_block_header(sbi, bh); + if (ret < 0) { + scoutfs_block_put(bh); + bh = ERR_PTR(ret); + } else { + set_buffer_scoutfs_verified(bh); + } } + unlock_buffer(bh); } - spin_unlock_irqrestore(&sbi->block_lock, flags); - if (bl) - goto wait; - - /* allocate a new block and try to insert it */ - bl = alloc_block(sb, blkno); - if (!bl) { - ret = -EIO; - goto out; - } - - ret = radix_tree_preload(GFP_NOFS); - if (ret) - goto out; - - spin_lock_irqsave(&sbi->block_lock, flags); - - found = radix_tree_lookup(&sbi->block_radix, blkno); - if (found) { - scoutfs_put_block(bl); - bl = found; - atomic_inc(&bl->refcount); - } else { - radix_tree_insert(&sbi->block_radix, blkno, bl); - atomic_inc(&bl->refcount); - } - - spin_unlock_irqrestore(&sbi->block_lock, flags); - radix_tree_preload_end(); - - if (!found) { - ret = block_submit_bio(bl, READ_SYNC | REQ_META); - if (ret) - goto out; - } - -wait: - ret = wait_event_interruptible(sbi->block_wq, - test_bit(SCOUTFS_BLOCK_BIT_UPTODATE, &bl->bits) || - test_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits)); - if (test_bit(SCOUTFS_BLOCK_BIT_UPTODATE, &bl->bits)) - ret = 0; - else if (test_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits)) - ret = -EIO; - out: - if (ret) { - scoutfs_put_block(bl); - bl = ERR_PTR(ret); - } - - return bl; + trace_printk("blkno %llu bh %p (ret %ld)\n", + blkno, bh, IS_ERR(bh) ? PTR_ERR(bh) : 0); + return bh; } /* - * Return the block pointed to by the caller's reference. + * Read an existing block from the device described by the caller's + * reference. * * If the reference sequence numbers don't match then we could be racing * with another writer. We back off and try again. If it happens too @@ -277,130 +223,122 @@ out: * - reads that span transactions? * - writers creating a new dirty block? */ -struct scoutfs_block *scoutfs_read_ref(struct super_block *sb, - struct scoutfs_block_ref *ref) +struct buffer_head *scoutfs_block_read_ref(struct super_block *sb, + struct scoutfs_block_ref *ref) { - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_block_header *hdr; - struct scoutfs_block *bl; - struct scoutfs_block *found; - unsigned long flags; - - bl = scoutfs_read_block(sb, le64_to_cpu(ref->blkno)); - if (!IS_ERR(bl)) { - hdr = bl->data; + struct buffer_head *bh; + bh = scoutfs_block_read(sb, le64_to_cpu(ref->blkno)); + if (!IS_ERR(bh)) { + hdr = bh_data(bh); if (WARN_ON_ONCE(hdr->seq != ref->seq)) { - /* XXX hack, make this a function */ - spin_lock_irqsave(&sbi->block_lock, flags); - found = radix_tree_lookup(&sbi->block_radix, - bl->blkno); - if (found == bl) { - radix_tree_delete(&sbi->block_radix, bl->blkno); - scoutfs_put_block(bl); - } - spin_unlock_irqrestore(&sbi->block_lock, flags); - - scoutfs_put_block(bl); - bl = ERR_PTR(-EAGAIN); + clear_buffer_uptodate(bh); + brelse(bh); + bh = ERR_PTR(-EAGAIN); } } - return bl; + return bh; } /* - * XXX This is a gross hack for writing the super. It doesn't have - * per-block write completion indication, it just knows that it's the - * only thing that will be writing. + * We stop tracking dirty metadata blocks when their IO succeeds. This + * happens in the context of transaction commit which excludes other + * metadata dirtying paths. */ -int scoutfs_write_block(struct scoutfs_block *bl) +static void block_write_end_io(struct buffer_head *bh, int uptodate) { - struct scoutfs_sb_info *sbi = SCOUTFS_SB(bl->sb); - int ret; + struct scoutfs_sb_info *sbi = sbi_from_bh(bh); - BUG_ON(atomic_read(&sbi->block_writes) != 0); + trace_printk("bh %p uptdate %d\n", bh, uptodate); - atomic_inc(&sbi->block_writes); - ret = block_submit_bio(bl, WRITE); - if (ret) - atomic_dec(&sbi->block_writes); - else - wait_event(sbi->block_wq, atomic_read(&sbi->block_writes) == 0); + /* XXX */ + unlock_buffer(bh); - return ret ?: sbi->block_write_err; + if (uptodate) { + erase_bhp(bh); + } else { + /* don't care if this is racey? */ + if (!sbi->block_write_err) + sbi->block_write_err = -EIO; + } + + if (atomic_dec_and_test(&sbi->block_writes)) + wake_up(&sbi->block_wq); } /* - * A quick cheap test so that write dirty blocks only has to return - * success or error, not also the lack of dirty blocks. - */ -int scoutfs_has_dirty_blocks(struct super_block *sb) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - - return radix_tree_tagged(&sbi->block_radix, DIRTY_RADIX_TAG); -} - -/* - * Write out all the currently dirty blocks. The caller has waited - * for all the dirty blocks to be consistent and has prevented further - * writes while we're working. + * Submit writes for all the buffer heads in the dirty block tree. The + * write transaction machinery ensures that the dirty blocks form a + * consistent image and excludes future dirtying while we're working. * - * The blocks are kept dirty so that they won't be evicted by reclaim - * while they're in flight. Reads can traverse the blocks while they're - * in flight. + * Presence in the dirty tree holds a reference. Blocks are only + * removed from the tree which drops the ref when IO completes. + * + * Blocks that see write errors remain in the dirty tree and will try to + * be written again in the next transaction commit. + * + * Reads can traverse the blocks while they're in flight. + * + * The number of blocks written is returned, or -errno on error. */ -int scoutfs_write_dirty_blocks(struct super_block *sb) +int scoutfs_block_write_dirty(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_block *blocks[16]; - struct scoutfs_block *bl; + struct buffer_head *bh; + struct rb_node *node; + struct blk_plug plug; unsigned long flags; - unsigned long blkno; - int ret; - int nr; - int i; + int count; + int err; - blkno = 0; + atomic_set(&sbi->block_writes, 1); sbi->block_write_err = 0; - ret = 0; - atomic_inc(&sbi->block_writes); + count = 0; + err = 0; - do { - /* get refs to a bunch of dirty blocks */ - spin_lock_irqsave(&sbi->block_lock, flags); - nr = radix_tree_gang_lookup_tag(&sbi->block_radix, - (void **)blocks, blkno, - ARRAY_SIZE(blocks), - DIRTY_RADIX_TAG); - if (nr > 0) - blkno = blocks[nr - 1]->blkno + 1; - for (i = 0; i < nr; i++) - atomic_inc(&blocks[i]->refcount); + blk_start_plug(&plug); + + spin_lock_irqsave(&sbi->block_lock, flags); + node = rb_first(&sbi->block_dirty_tree); + while(node) { + bh = bh_from_bhp_node(node); + node = rb_next(node); spin_unlock_irqrestore(&sbi->block_lock, flags); - /* submit them in order, being careful to put all on err */ - for (i = 0; i < nr; i++) { - bl = blocks[i]; + atomic_inc(&sbi->block_writes); + count++; + scoutfs_block_set_crc(bh); - if (ret == 0) { - /* XXX crc could be farmed out */ - scoutfs_calc_hdr_crc(bl); - atomic_inc(&sbi->block_writes); - ret = block_submit_bio(bl, WRITE); - if (ret) - atomic_dec(&sbi->block_writes); - } - scoutfs_put_block(bl); - } - } while (nr && !ret); + /* + * XXX submit_bh() forces us to lock the block while IO is + * in flight. This is unfortunate because we use the buffer + * head lock to serialize access to btree block contents. + * We should fix that and only use the buffer head lock + * when the APIs force us to. + */ + lock_buffer(bh); + + bh->b_end_io = block_write_end_io; + err = submit_bh(WRITE, bh); /* doesn't actually fail? */ + + spin_lock_irqsave(&sbi->block_lock, flags); + if (err) + break; + } + spin_unlock_irqrestore(&sbi->block_lock, flags); + + blk_finish_plug(&plug); /* wait for all io to drain */ atomic_dec(&sbi->block_writes); wait_event(sbi->block_wq, atomic_read(&sbi->block_writes) == 0); - return ret ?: sbi->block_write_err; + trace_printk("err %d sbi err %d count %d\n", + err, sbi->block_write_err, count); + + return err ?: sbi->block_write_err ?: count; } /* @@ -415,188 +353,108 @@ int scoutfs_write_dirty_blocks(struct super_block *sb) * For now we're using the dirty super block in the sb_info to track the * dirty seq. That'll be different when we have multiple btrees. * - * Callers are working in structures that have sufficient locking to - * protect references to the source block. If we've come to dirty it - * then there won't be concurrent users and we can just move it in the - * cache. - * - * The caller can ask that we either move the existing cached block to - * its new dirty blkno in the cache or copy its contents to a newly - * allocated dirty block. The caller knows if they'll ever reference - * the old clean block again (buddy does, btree doesn't.) + * Callers are responsible for serializing modification to the reference + * which is probably embedded in some other dirty persistent structure. */ -static struct scoutfs_block *dirty_ref(struct super_block *sb, - struct scoutfs_block_ref *ref, bool cow) +struct buffer_head *scoutfs_block_dirty_ref(struct super_block *sb, + struct scoutfs_block_ref *ref) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_block_header *hdr; - struct scoutfs_block *copy_bl = NULL; - struct scoutfs_block *found; - struct scoutfs_block *bl; - unsigned long flags; - u64 clean_blkno; + struct buffer_head *copy_bh = NULL; + struct buffer_head *bh; u64 blkno = 0; int ret; int err; - bl = scoutfs_read_block(sb, le64_to_cpu(ref->blkno)); - if (IS_ERR(bl) || ref->seq == sbi->super.hdr.seq) - return bl; - - clean_blkno = bl->blkno; - - ret = scoutfs_buddy_dirty(sb, clean_blkno, 0); - if (ret < 0) - goto out; + bh = scoutfs_block_read(sb, le64_to_cpu(ref->blkno)); + if (IS_ERR(bh) || ref->seq == sbi->super.hdr.seq) + return bh; ret = scoutfs_buddy_alloc_same(sb, &blkno, 0, le64_to_cpu(ref->blkno)); if (ret < 0) goto out; - if (cow) { - copy_bl = alloc_block(sb, blkno); - if (IS_ERR(copy_bl)) { - ret = PTR_ERR(copy_bl); - goto out; - } - set_bit(SCOUTFS_BLOCK_BIT_UPTODATE, ©_bl->bits); + copy_bh = scoutfs_block_dirty(sb, blkno); + if (IS_ERR(copy_bh)) { + ret = PTR_ERR(copy_bh); + goto out; } - ret = radix_tree_preload(GFP_NOFS); + ret = scoutfs_buddy_free(sb, bh->b_blocknr, 0); if (ret) goto out; - spin_lock_irqsave(&sbi->block_lock, flags); + memcpy(copy_bh->b_data, bh->b_data, SCOUTFS_BLOCK_SIZE); - /* delete anything at the new blkno */ - found = radix_tree_lookup(&sbi->block_radix, blkno); - if (found) { - radix_tree_delete(&sbi->block_radix, blkno); - scoutfs_put_block(found); - } - - if (cow) { - /* copy contents to the new block, hdr updated below */ - memcpy(copy_bl->data, bl->data, SCOUTFS_BLOCK_SIZE); - scoutfs_put_block(bl); - bl = copy_bl; - copy_bl = NULL; - } else { - /* move the existing block to its new dirty blkno */ - found = radix_tree_lookup(&sbi->block_radix, bl->blkno); - if (found == bl) { - radix_tree_delete(&sbi->block_radix, bl->blkno); - atomic_dec(&bl->refcount); - } - } - - bl->blkno = blkno; - hdr = bl->data; + hdr = bh_data(copy_bh); hdr->blkno = cpu_to_le64(blkno); hdr->seq = sbi->super.hdr.seq; ref->blkno = hdr->blkno; ref->seq = hdr->seq; - /* insert the dirty block at its new blkno */ - radix_tree_insert(&sbi->block_radix, blkno, bl); - radix_tree_tag_set(&sbi->block_radix, blkno, DIRTY_RADIX_TAG); - atomic_inc(&bl->refcount); - - spin_unlock_irqrestore(&sbi->block_lock, flags); - radix_tree_preload_end(); - - /* free clean blkno after preload end enables preemption */ - err = scoutfs_buddy_free(sb, clean_blkno, 0); - WARN_ON(err); /* XXX corruption (dirtying should prevent) */ ret = 0; out: - scoutfs_put_block(copy_bl); + scoutfs_block_put(bh); if (ret) { - if (blkno) { - err = scoutfs_buddy_free(sb, blkno, 0); - WARN_ON_ONCE(err); /* XXX hmm */ + if (!IS_ERR_OR_NULL(copy_bh)) { + err = scoutfs_buddy_free(sb, copy_bh->b_blocknr, 0); + WARN_ON_ONCE(err); /* freeing dirty must work */ } - scoutfs_put_block(bl); - bl = ERR_PTR(ret); + scoutfs_block_put(copy_bh); + copy_bh = ERR_PTR(ret); } - return bl; -} - -struct scoutfs_block *scoutfs_block_cow_ref(struct super_block *sb, - struct scoutfs_block_ref *ref) -{ - return dirty_ref(sb, ref, true); -} -struct scoutfs_block *scoutfs_block_dirty_ref(struct super_block *sb, - struct scoutfs_block_ref *ref) -{ - return dirty_ref(sb, ref, false); + return copy_bh; } /* - * Return a newly allocated metadata block with an updated block header - * to match the current dirty seq. Callers are responsible for - * serializing access to the block and for zeroing unwritten block - * contents. + * Return a dirty metadata block with an updated block header to match + * the current dirty seq. Callers are responsible for serializing + * access to the block and for zeroing unwritten block contents. */ -struct scoutfs_block *scoutfs_new_block(struct super_block *sb, u64 blkno) +struct buffer_head *scoutfs_block_dirty(struct super_block *sb, u64 blkno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_block_header *hdr; - struct scoutfs_block *found; - struct scoutfs_block *bl; - unsigned long flags; + struct buffer_head *bh; int ret; /* allocate a new block and try to insert it */ - bl = alloc_block(sb, blkno); - if (!bl) { - ret = -EIO; + bh = sb_getblk(sb, blkno); + if (!bh) { + bh = ERR_PTR(-ENOMEM); goto out; } - set_bit(SCOUTFS_BLOCK_BIT_UPTODATE, &bl->bits); - - ret = radix_tree_preload(GFP_NOFS); - if (ret) + ret = insert_bhp(sb, bh); + if (ret < 0) { + scoutfs_block_put(bh); + bh = ERR_PTR(ret); goto out; + } - hdr = bl->data; + hdr = bh_data(bh); *hdr = sbi->super.hdr; hdr->blkno = cpu_to_le64(blkno); hdr->seq = sbi->super.hdr.seq; - spin_lock_irqsave(&sbi->block_lock, flags); - found = radix_tree_lookup(&sbi->block_radix, blkno); - if (found) { - radix_tree_delete(&sbi->block_radix, blkno); - scoutfs_put_block(found); - } - - radix_tree_insert(&sbi->block_radix, blkno, bl); - radix_tree_tag_set(&sbi->block_radix, blkno, DIRTY_RADIX_TAG); - atomic_inc(&bl->refcount); - spin_unlock_irqrestore(&sbi->block_lock, flags); - - radix_tree_preload_end(); - ret = 0; + set_buffer_uptodate(bh); + set_buffer_scoutfs_verified(bh); out: - if (ret) { - scoutfs_put_block(bl); - bl = ERR_PTR(ret); - } + trace_printk("blkno %llu bh %p (ret %ld)\n", + blkno, bh, IS_ERR(bh) ? PTR_ERR(bh) : 0); - return bl; + return bh; } /* * Allocate a new dirty writable block. The caller must be in a * transaction so that we can assign the dirty seq. */ -struct scoutfs_block *scoutfs_alloc_block(struct super_block *sb) +struct buffer_head *scoutfs_block_dirty_alloc(struct super_block *sb) { - struct scoutfs_block *bl; + struct buffer_head *bh; u64 blkno; int ret; int err; @@ -605,26 +463,26 @@ struct scoutfs_block *scoutfs_alloc_block(struct super_block *sb) if (ret < 0) return ERR_PTR(ret); - bl = scoutfs_new_block(sb, blkno); - if (IS_ERR(bl)) { + bh = scoutfs_block_dirty(sb, blkno); + if (IS_ERR(bh)) { err = scoutfs_buddy_free(sb, blkno, 0); - WARN_ON_ONCE(err); /* XXX hmm */ + WARN_ON_ONCE(err); /* freeing dirty must work */ } - return bl; + return bh; } -void scoutfs_calc_hdr_crc(struct scoutfs_block *bl) +void scoutfs_block_set_crc(struct buffer_head *bh) { - struct scoutfs_block_header *hdr = bl->data; + struct scoutfs_block_header *hdr = bh_data(bh); hdr->crc = cpu_to_le32(scoutfs_crc_block(hdr)); } -void scoutfs_zero_block_tail(struct scoutfs_block *bl, size_t off) +void scoutfs_block_zero(struct buffer_head *bh, size_t off) { if (WARN_ON_ONCE(off > SCOUTFS_BLOCK_SIZE)) return; if (off < SCOUTFS_BLOCK_SIZE) - memset(bl->data + off, 0, SCOUTFS_BLOCK_SIZE - off); + memset((char *)bh->b_data + off, 0, SCOUTFS_BLOCK_SIZE - off); } diff --git a/kmod/src/block.h b/kmod/src/block.h index b464f0c3..8bb2a136 100644 --- a/kmod/src/block.h +++ b/kmod/src/block.h @@ -2,43 +2,35 @@ #define _SCOUTFS_BLOCK_H_ #include -#include -#include +#include -#define SCOUTFS_BLOCK_BIT_UPTODATE (1 << 0) -#define SCOUTFS_BLOCK_BIT_ERROR (1 << 1) +struct buffer_head *scoutfs_block_read(struct super_block *sb, u64 blkno); +struct buffer_head *scoutfs_block_read_ref(struct super_block *sb, + struct scoutfs_block_ref *ref); -struct scoutfs_block { - struct rw_semaphore rwsem; - atomic_t refcount; - u64 blkno; - - unsigned long bits; - - struct super_block *sb; - /* only high order page alloc for now */ - struct page *page; - void *data; -}; - -struct scoutfs_block *scoutfs_read_block(struct super_block *sb, u64 blkno); -struct scoutfs_block *scoutfs_new_block(struct super_block *sb, u64 blkno); -struct scoutfs_block *scoutfs_alloc_block(struct super_block *sb); - -struct scoutfs_block *scoutfs_read_ref(struct super_block *sb, - struct scoutfs_block_ref *ref); -struct scoutfs_block *scoutfs_block_cow_ref(struct super_block *sb, +struct buffer_head *scoutfs_block_dirty(struct super_block *sb, u64 blkno); +struct buffer_head *scoutfs_block_dirty_alloc(struct super_block *sb); +struct buffer_head *scoutfs_block_dirty_ref(struct super_block *sb, struct scoutfs_block_ref *ref); -struct scoutfs_block *scoutfs_block_dirty_ref(struct super_block *sb, - struct scoutfs_block_ref *ref); -int scoutfs_has_dirty_blocks(struct super_block *sb); -int scoutfs_write_block(struct scoutfs_block *bl); -int scoutfs_write_dirty_blocks(struct super_block *sb); +int scoutfs_block_write_dirty(struct super_block *sb); -void scoutfs_put_block(struct scoutfs_block *bl); +void scoutfs_block_set_crc(struct buffer_head *bh); +void scoutfs_block_zero(struct buffer_head *bh, size_t off); -void scoutfs_calc_hdr_crc(struct scoutfs_block *bl); -void scoutfs_zero_block_tail(struct scoutfs_block *bl, size_t off); +/* XXX seems like this should be upstream :) */ +static inline void *bh_data(struct buffer_head *bh) +{ + return (void *)bh->b_data; +} + +static inline void scoutfs_block_put(struct buffer_head *bh) +{ + if (!IS_ERR_OR_NULL(bh)) { + trace_printk("putting bh %p count %d\n", + bh, atomic_read(&bh->b_count)); + brelse(bh); + } +} #endif diff --git a/kmod/src/btree.c b/kmod/src/btree.c index ceecbee1..df0b75e3 100644 --- a/kmod/src/btree.c +++ b/kmod/src/btree.c @@ -294,14 +294,14 @@ static void compact_items(struct scoutfs_btree_block *bt) * Allocate and initialize a new tree block. The caller adds references * to it. */ -static struct scoutfs_block *alloc_tree_block(struct super_block *sb) +static struct buffer_head *alloc_tree_block(struct super_block *sb) { struct scoutfs_btree_block *bt; - struct scoutfs_block *bl; + struct buffer_head *bh; - bl = scoutfs_alloc_block(sb); - if (!IS_ERR(bl)) { - bt = bl->data; + bh = scoutfs_block_dirty_alloc(sb); + if (!IS_ERR(bh)) { + bt = bh_data(bh); bt->treap.off = 0; bt->total_free = cpu_to_le16(SCOUTFS_BLOCK_SIZE - @@ -310,7 +310,7 @@ static struct scoutfs_block *alloc_tree_block(struct super_block *sb) bt->nr_items = 0; } - return bl; + return bh; } /* the caller has ensured that the free must succeed */ @@ -324,22 +324,22 @@ static void free_tree_block(struct super_block *sb, __le64 blkno) * Allocate a new tree block and point the root at it. The caller * is responsible for the items in the new root block. */ -static struct scoutfs_block *grow_tree(struct super_block *sb, +static struct buffer_head *grow_tree(struct super_block *sb, struct scoutfs_btree_root *root) { struct scoutfs_block_header *hdr; - struct scoutfs_block *bl; + struct buffer_head *bh; - bl = alloc_tree_block(sb); - if (!IS_ERR(bl)) { - hdr = bl->data; + bh = alloc_tree_block(sb); + if (!IS_ERR(bh)) { + hdr = bh_data(bh); root->height++; root->ref.blkno = hdr->blkno; root->ref.seq = hdr->seq; } - return bl; + return bh; } /* @@ -380,18 +380,18 @@ static void create_parent_item(struct scoutfs_btree_block *parent, * the child that we return. It's skipping locking the new parent as it * descends but that's fine. */ -static struct scoutfs_block *try_split(struct super_block *sb, +static struct buffer_head *try_split(struct super_block *sb, struct scoutfs_btree_root *root, int level, struct scoutfs_key *key, unsigned int val_len, struct scoutfs_btree_block *parent, struct scoutfs_btree_item *par_item, - struct scoutfs_block *right_bl) + struct buffer_head *right_bh) { - struct scoutfs_btree_block *right = right_bl->data; + struct scoutfs_btree_block *right = bh_data(right_bh); struct scoutfs_btree_block *left; - struct scoutfs_block *left_bl; - struct scoutfs_block *par_bl = NULL; + struct buffer_head *left_bh; + struct buffer_head *par_bh = NULL; unsigned int bytes; if (level) @@ -399,35 +399,35 @@ static struct scoutfs_block *try_split(struct super_block *sb, bytes = val_bytes(val_len); if (le16_to_cpu(right->tail_free) >= bytes) - return right_bl; + return right_bh; if (le16_to_cpu(right->total_free) >= bytes) { compact_items(right); - return right_bl; + return right_bh; } /* alloc split neighbour first to avoid unwinding tree growth */ - left_bl = alloc_tree_block(sb); - if (IS_ERR(left_bl)) { - scoutfs_put_block(right_bl); - return left_bl; + left_bh = alloc_tree_block(sb); + if (IS_ERR(left_bh)) { + scoutfs_block_put(right_bh); + return left_bh; } - left = left_bl->data; + left = bh_data(left_bh); if (!parent) { - par_bl = grow_tree(sb, root); - if (IS_ERR(par_bl)) { + par_bh = grow_tree(sb, root); + if (IS_ERR(par_bh)) { free_tree_block(sb, left->hdr.blkno); - scoutfs_put_block(left_bl); - scoutfs_put_block(right_bl); - return par_bl; + scoutfs_block_put(left_bh); + scoutfs_block_put(right_bh); + return par_bh; } - parent = par_bl->data; + parent = bh_data(par_bh); } /* only grow the tree once we have the split neighbour */ - if (par_bl) { + if (par_bh) { struct scoutfs_key maximal; scoutfs_set_max_key(&maximal); create_parent_item(parent, right, &maximal); @@ -438,19 +438,19 @@ static struct scoutfs_block *try_split(struct super_block *sb, if (scoutfs_key_cmp(key, greatest_key(left)) <= 0) { /* insertion will go to the new left block */ - scoutfs_put_block(right_bl); - right_bl = left_bl; + scoutfs_block_put(right_bh); + right_bh = left_bh; } else { /* insertion will still go through us, might need to compact */ - scoutfs_put_block(left_bl); + scoutfs_block_put(left_bh); if (le16_to_cpu(right->tail_free) < bytes) compact_items(right); } - scoutfs_put_block(par_bl); + scoutfs_block_put(par_bh); - return right_bl; + return right_bh; } /* @@ -474,21 +474,21 @@ static struct scoutfs_block *try_split(struct super_block *sb, * * XXX this could more cleverly chose a merge candidate sibling */ -static struct scoutfs_block *try_merge(struct super_block *sb, +static struct buffer_head *try_merge(struct super_block *sb, struct scoutfs_btree_root *root, struct scoutfs_btree_block *parent, struct scoutfs_btree_item *par_item, - struct scoutfs_block *bl) + struct buffer_head *bh) { - struct scoutfs_btree_block *bt = bl->data; + struct scoutfs_btree_block *bt = bh_data(bh); struct scoutfs_btree_block *sib_bt; - struct scoutfs_block *sib_bl; + struct buffer_head *sib_bh; struct scoutfs_btree_item *sib_item; int to_move; bool move_right; if (le16_to_cpu(bt->total_free) <= SCOUTFS_BTREE_FREE_LIMIT) - return bl; + return bh; /* move items right into our block if we have a left sibling */ sib_item = bt_prev(parent, par_item); @@ -499,13 +499,13 @@ static struct scoutfs_block *try_merge(struct super_block *sb, move_right = true; } - sib_bl = scoutfs_block_dirty_ref(sb, (void *)sib_item->val); - if (IS_ERR(sib_bl)) { + sib_bh = scoutfs_block_dirty_ref(sb, (void *)sib_item->val); + if (IS_ERR(sib_bh)) { /* XXX do we need to unlock this? don't think so */ - scoutfs_put_block(bl); - return sib_bl; + scoutfs_block_put(bh); + return sib_bh; } - sib_bt = sib_bl->data; + sib_bt = bh_data(sib_bh); if (used_total(sib_bt) <= le16_to_cpu(bt->total_free)) to_move = used_total(sib_bt); @@ -538,7 +538,7 @@ static struct scoutfs_block *try_merge(struct super_block *sb, free_tree_block(sb, parent->hdr.blkno); } - return bl; + return bh; } enum { @@ -549,40 +549,42 @@ enum { WALK_DIRTY, }; +static inline void lock_root(struct scoutfs_sb_info *sbi, bool dirty) +{ + if (dirty) + down_write(&sbi->btree_rwsem); + else + down_read(&sbi->btree_rwsem); +} + +static inline void unlock_root(struct scoutfs_sb_info *sbi, bool dirty) +{ + if (dirty) + up_write(&sbi->btree_rwsem); + else + up_read(&sbi->btree_rwsem); +} + /* * As we descend we lock parent blocks (or the root), then lock the child, * then unlock the parent. */ -static void lock_block(struct scoutfs_sb_info *sbi, struct scoutfs_block *bl, - bool dirty) +static inline void lock_block(struct scoutfs_sb_info *sbi, + struct buffer_head *bh, bool dirty) { - struct rw_semaphore *rwsem; - - if (bl == NULL) - rwsem = &sbi->btree_rwsem; + if (bh == NULL) + lock_root(sbi, dirty); else - rwsem = &bl->rwsem; - - if (dirty) - down_write(rwsem); - else - down_read(rwsem); + lock_buffer(bh); } -static void unlock_block(struct scoutfs_sb_info *sbi, struct scoutfs_block *bl, - bool dirty) +static inline void unlock_block(struct scoutfs_sb_info *sbi, + struct buffer_head *bh, bool dirty) { - struct rw_semaphore *rwsem; - - if (bl == NULL) - rwsem = &sbi->btree_rwsem; + if (bh == NULL) + unlock_root(sbi, dirty); else - rwsem = &bl->rwsem; - - if (dirty) - up_write(rwsem); - else - up_read(rwsem); + unlock_buffer(bh); } static u64 item_block_ref_seq(struct scoutfs_btree_item *item) @@ -647,7 +649,7 @@ item_after_seq(struct scoutfs_btree_block *bt, struct scoutfs_key *key, * in the next sibling's block. This is used by iteration to advance to * the next block when they're done with the block this returns. */ -static struct scoutfs_block *btree_walk(struct super_block *sb, +static struct buffer_head *btree_walk(struct super_block *sb, struct scoutfs_key *key, struct scoutfs_key *next_key, unsigned int val_len, u64 seq, int op) @@ -655,8 +657,8 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_btree_block *parent = NULL; struct scoutfs_btree_root *root; - struct scoutfs_block *par_bl = NULL; - struct scoutfs_block *bl = NULL; + struct buffer_head *par_bh = NULL; + struct buffer_head *bh = NULL; struct scoutfs_btree_item *item = NULL; struct scoutfs_block_ref *ref; unsigned int level; @@ -667,7 +669,7 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, if (next_key) scoutfs_set_max_key(next_key); - lock_block(sbi, par_bl, dirty); + lock_block(sbi, par_bh, dirty); /* XXX one for now */ root = &sbi->super.btree_root; @@ -676,31 +678,31 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, if (!root->height) { if (op == WALK_INSERT) { - bl = ERR_PTR(-ENOENT); + bh = ERR_PTR(-ENOENT); } else { - bl = grow_tree(sb, root); - if (!IS_ERR(bl)) - lock_block(sbi, bl, dirty); + bh = grow_tree(sb, root); + if (!IS_ERR(bh)) + lock_block(sbi, bh, dirty); } - unlock_block(sbi, par_bl, dirty); - return bl; + unlock_block(sbi, par_bh, dirty); + return bh; } /* skip the whole tree if the root ref's seq is old */ if (op == WALK_NEXT_SEQ && le64_to_cpu(ref->seq) < seq) { - unlock_block(sbi, par_bl, dirty); + unlock_block(sbi, par_bh, dirty); return ERR_PTR(-ENOENT); } while (level--) { /* XXX hmm, need to think about retry */ if (dirty) { - bl = scoutfs_block_dirty_ref(sb, ref); + bh = scoutfs_block_dirty_ref(sb, ref); } else { - bl = scoutfs_read_ref(sb, ref); + bh = scoutfs_block_read_ref(sb, ref); } - if (IS_ERR(bl)) + if (IS_ERR(bh)) break; /* @@ -714,23 +716,23 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, } if (op == WALK_INSERT) - bl = try_split(sb, root, level, key, val_len, parent, - item, bl); + bh = try_split(sb, root, level, key, val_len, parent, + item, bh); if ((op == WALK_DELETE) && parent) - bl = try_merge(sb, root, parent, item, bl); - if (IS_ERR(bl)) + bh = try_merge(sb, root, parent, item, bh); + if (IS_ERR(bh)) break; - lock_block(sbi, bl, dirty); + lock_block(sbi, bh, dirty); if (!level) break; /* unlock parent before searching so others can use it */ - unlock_block(sbi, par_bl, dirty); - scoutfs_put_block(par_bl); - par_bl = bl; - parent = par_bl->data; + unlock_block(sbi, par_bh, dirty); + scoutfs_block_put(par_bh); + par_bh = bh; + parent = bh_data(par_bh); /* * Find the parent item that references the next child @@ -742,9 +744,9 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, if (!item) { /* current block dropped as parent below */ if (op == WALK_NEXT_SEQ) { - bl = ERR_PTR(-ENOENT); + bh = ERR_PTR(-ENOENT); } else { - bl = ERR_PTR(-EIO); + bh = ERR_PTR(-EIO); } break; } @@ -752,17 +754,17 @@ static struct scoutfs_block *btree_walk(struct super_block *sb, ref = (void *)item->val; } - unlock_block(sbi, par_bl, dirty); - scoutfs_put_block(par_bl); + unlock_block(sbi, par_bh, dirty); + scoutfs_block_put(par_bh); - return bl; + return bh; } static void set_cursor(struct scoutfs_btree_cursor *curs, - struct scoutfs_block *bl, + struct buffer_head *bh, struct scoutfs_btree_item *item, bool write) { - curs->bl = bl; + curs->bh = bh; curs->item = item; curs->key = &item->key; curs->seq = le64_to_cpu(item->seq); @@ -779,22 +781,24 @@ int scoutfs_btree_lookup(struct super_block *sb, struct scoutfs_key *key, struct scoutfs_btree_cursor *curs) { struct scoutfs_btree_item *item; - struct scoutfs_block *bl; + struct scoutfs_btree_block *bt; + struct buffer_head *bh; int ret; - BUG_ON(curs->bl); + BUG_ON(curs->bh); - bl = btree_walk(sb, key, NULL, 0, 0, 0); - if (IS_ERR(bl)) - return PTR_ERR(bl); + bh = btree_walk(sb, key, NULL, 0, 0, 0); + if (IS_ERR(bh)) + return PTR_ERR(bh); - item = bt_lookup(bl->data, key); + bt = bh_data(bh); + item = bt_lookup(bt, key); if (item) { - set_cursor(curs, bl, item, false); + set_cursor(curs, bh, item, false); ret = 0; } else { - up_read(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, false); + scoutfs_block_put(bh); ret = -ENOENT; } @@ -815,25 +819,25 @@ int scoutfs_btree_insert(struct super_block *sb, struct scoutfs_key *key, { struct scoutfs_btree_item *item; struct scoutfs_btree_block *bt; - struct scoutfs_block *bl; + struct buffer_head *bh; int ret; - BUG_ON(curs->bl); + BUG_ON(curs->bh); - bl = btree_walk(sb, key, NULL, val_len, 0, WALK_INSERT); - if (IS_ERR(bl)) - return PTR_ERR(bl); - bt = bl->data; + bh = btree_walk(sb, key, NULL, val_len, 0, WALK_INSERT); + if (IS_ERR(bh)) + return PTR_ERR(bh); + bt = bh_data(bh); /* XXX should this return -eexist? */ item = bt_lookup(bt, key); if (!item) { item = create_item(bt, key, val_len); - set_cursor(curs, bl, item, true); + set_cursor(curs, bh, item, true); ret = 0; } else { - up_write(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, true); + scoutfs_block_put(bh); ret = -ENOENT; } @@ -850,13 +854,13 @@ int scoutfs_btree_delete(struct super_block *sb, struct scoutfs_key *key) struct scoutfs_btree_root *root; struct scoutfs_btree_item *item; struct scoutfs_btree_block *bt; - struct scoutfs_block *bl; + struct buffer_head *bh; int ret; - bl = btree_walk(sb, key, NULL, 0, 0, WALK_DELETE); - if (IS_ERR(bl)) - return PTR_ERR(bl); - bt = bl->data; + bh = btree_walk(sb, key, NULL, 0, 0, WALK_DELETE); + if (IS_ERR(bh)) + return PTR_ERR(bh); + bt = bh_data(bh); item = bt_lookup(bt, key); if (item) { @@ -879,8 +883,8 @@ int scoutfs_btree_delete(struct super_block *sb, struct scoutfs_key *key) ret = -ENOENT; } - up_write(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, true); + scoutfs_block_put(bh); return ret; } @@ -905,7 +909,7 @@ static int btree_next(struct super_block *sb, struct scoutfs_key *first, struct scoutfs_btree_cursor *curs) { struct scoutfs_btree_block *bt; - struct scoutfs_block *bl; + struct buffer_head *bh; struct scoutfs_key key = *first; struct scoutfs_key next_key; int ret; @@ -914,50 +918,50 @@ static int btree_next(struct super_block *sb, struct scoutfs_key *first, return 0; /* find the next item after the cursor, releasing if we're done */ - if (curs->bl) { + if (curs->bh) { + bt = bh_data(curs->bh); key = curs->item->key; scoutfs_inc_key(&key); - curs->item = item_next_seq(curs->bl->data, curs->item, - 0, seq, op); + curs->item = item_next_seq(bt, curs->item, 0, seq, op); if (curs->item) - set_cursor(curs, curs->bl, curs->item, curs->write); + set_cursor(curs, curs->bh, curs->item, curs->write); else scoutfs_btree_release(curs); } /* find the leaf that contains the next item after the key */ - while (!curs->bl && scoutfs_key_cmp(&key, last) <= 0) { + while (!curs->bh && scoutfs_key_cmp(&key, last) <= 0) { - bl = btree_walk(sb, &key, &next_key, 0, seq, op); + bh = btree_walk(sb, &key, &next_key, 0, seq, op); /* next seq walks can terminate in parents with old seqs */ - if (op == WALK_NEXT_SEQ && bl == ERR_PTR(-ENOENT)) { + if (op == WALK_NEXT_SEQ && bh == ERR_PTR(-ENOENT)) { key = next_key; continue; } - if (IS_ERR(bl)) { - if (bl == ERR_PTR(-ENOENT)) + if (IS_ERR(bh)) { + if (bh == ERR_PTR(-ENOENT)) break; - return PTR_ERR(bl); + return PTR_ERR(bh); } - bt = bl->data; + bt = bh_data(bh); /* keep trying leaves until next_key passes last */ - curs->item = item_after_seq(bl->data, &key, 0, seq, op); + curs->item = item_after_seq(bt, &key, 0, seq, op); if (!curs->item) { key = next_key; - up_read(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, false); + scoutfs_block_put(bh); continue; } if (curs->item) { - set_cursor(curs, bl, curs->item, false); + set_cursor(curs, bh, curs->item, false); } else { - up_read(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, false); + scoutfs_block_put(bh); } break; } @@ -997,22 +1001,24 @@ int scoutfs_btree_since(struct super_block *sb, struct scoutfs_key *first, int scoutfs_btree_dirty(struct super_block *sb, struct scoutfs_key *key) { struct scoutfs_btree_item *item; - struct scoutfs_block *bl; + struct scoutfs_btree_block *bt; + struct buffer_head *bh; int ret; - bl = btree_walk(sb, key, NULL, 0, 0, WALK_DIRTY); - if (IS_ERR(bl)) - return PTR_ERR(bl); + bh = btree_walk(sb, key, NULL, 0, 0, WALK_DIRTY); + if (IS_ERR(bh)) + return PTR_ERR(bh); - item = bt_lookup(bl->data, key); + bt = bh_data(bh); + item = bt_lookup(bt, key); if (item) { ret = 0; } else { ret = -ENOENT; } - up_write(&bl->rwsem); - scoutfs_put_block(bl); + unlock_block(NULL, bh, true); + scoutfs_block_put(bh); return ret; } @@ -1026,31 +1032,28 @@ void scoutfs_btree_update(struct super_block *sb, struct scoutfs_key *key, { struct scoutfs_btree_item *item; struct scoutfs_btree_block *bt; - struct scoutfs_block *bl; + struct buffer_head *bh; - BUG_ON(curs->bl); + BUG_ON(curs->bh); - bl = btree_walk(sb, key, NULL, 0, 0, WALK_DIRTY); - BUG_ON(IS_ERR(bl)); + bh = btree_walk(sb, key, NULL, 0, 0, WALK_DIRTY); + BUG_ON(IS_ERR(bh)); + bt = bh_data(bh); - item = bt_lookup(bl->data, key); + item = bt_lookup(bt, key); BUG_ON(!item); - bt = bl->data; item->seq = bt->hdr.seq; - set_cursor(curs, bl, item, true); + set_cursor(curs, bh, item, true); } void scoutfs_btree_release(struct scoutfs_btree_cursor *curs) { - if (curs->bl) { - if (curs->write) - up_write(&curs->bl->rwsem); - else - up_read(&curs->bl->rwsem); - scoutfs_put_block(curs->bl); + if (curs->bh) { + unlock_block(NULL, curs->bh, curs->write); + scoutfs_block_put(curs->bh); } - curs->bl = NULL; + curs->bh = NULL; } /* diff --git a/kmod/src/btree.h b/kmod/src/btree.h index fb9b1716..2dc05bb5 100644 --- a/kmod/src/btree.h +++ b/kmod/src/btree.h @@ -3,7 +3,7 @@ struct scoutfs_btree_cursor { /* for btree.c */ - struct scoutfs_block *bl; + struct buffer_head *bh; struct scoutfs_btree_item *item; /* for callers */ diff --git a/kmod/src/buddy.c b/kmod/src/buddy.c index c6e5974b..52ff738f 100644 --- a/kmod/src/buddy.c +++ b/kmod/src/buddy.c @@ -212,8 +212,8 @@ static int bitmap_alloc(struct super_block *sb, u64 *blkno) struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_bitmap_block *st_bm; struct scoutfs_bitmap_block *bm; - struct scoutfs_block *st_bl; - struct scoutfs_block *bm_bl; + struct buffer_head *st_bh; + struct buffer_head *bm_bh; int size; int ret; int d; @@ -226,18 +226,18 @@ static int bitmap_alloc(struct super_block *sb, u64 *blkno) return -EIO; /* dirty the bitmap block */ - bm_bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_bm_ref); - if (IS_ERR(bm_bl)) - return PTR_ERR(bm_bl); - bm = bm_bl->data; + bm_bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_bm_ref); + if (IS_ERR(bm_bh)) + return PTR_ERR(bm_bh); + bm = bh_data(bm_bh); /* read the stable bitmap block */ - st_bl = scoutfs_read_ref(sb, &sbi->stable_super.buddy_bm_ref); - if (IS_ERR(st_bl)) { - ret = PTR_ERR(st_bl); + st_bh = scoutfs_block_read_ref(sb, &sbi->stable_super.buddy_bm_ref); + if (IS_ERR(st_bh)) { + ret = PTR_ERR(st_bh); goto out; } - st_bm = st_bl->data; + st_bm = bh_data(st_bh); /* find the first bit that is set in both dirty and stable bitmaps */ size = le32_to_cpu(sbi->super.buddy_blocks); @@ -255,8 +255,8 @@ static int bitmap_alloc(struct super_block *sb, u64 *blkno) clear_bit_le(d, &bm->bits); ret = 0; out: - scoutfs_put_block(st_bl); - scoutfs_put_block(bm_bl); + scoutfs_block_put(st_bh); + scoutfs_block_put(bm_bh); return ret; } @@ -265,7 +265,7 @@ static int bitmap_free(struct super_block *sb, u64 blkno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_bitmap_block *bm; - struct scoutfs_block *bl; + struct buffer_head *bh; int nr; /* mkfs should have ensured that there's bitmap blocks */ @@ -273,14 +273,14 @@ static int bitmap_free(struct super_block *sb, u64 blkno) if (sbi->super.buddy_bm_ref.blkno == 0) return -EIO; - bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_bm_ref); - if (IS_ERR(bl)) - return PTR_ERR(bl); - bm = bl->data; + bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_bm_ref); + if (IS_ERR(bh)) + return PTR_ERR(bh); + bm = bh_data(bh); nr = blkno - (SCOUTFS_BUDDY_BM_BLKNO + SCOUTFS_BUDDY_BM_NR); set_bit_le(nr, bm->bits); - scoutfs_put_block(bl); + scoutfs_block_put(bh); return 0; } @@ -289,13 +289,13 @@ static int bitmap_free(struct super_block *sb, u64 blkno) * Give the caller a dirty buddy block. If the slot hasn't been used * yet then we need to allocate and initialize a new block. */ -static struct scoutfs_block *dirty_buddy_block(struct super_block *sb, int sl, +static struct buffer_head *dirty_buddy_block(struct super_block *sb, int sl, struct scoutfs_buddy_slot *slot) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct scoutfs_buddy_block *bud; - struct scoutfs_block *bl; + struct buffer_head *bh; u64 blkno; int count; int order; @@ -305,19 +305,19 @@ static struct scoutfs_block *dirty_buddy_block(struct super_block *sb, int sl, /* the fast path is to dirty an existing block */ if (slot->ref.blkno) - return scoutfs_block_cow_ref(sb, &slot->ref); + return scoutfs_block_dirty_ref(sb, &slot->ref); ret = bitmap_alloc(sb, &blkno); if (ret) return ERR_PTR(ret); - bl = scoutfs_new_block(sb, blkno); - if (IS_ERR(bl)) { + bh = scoutfs_block_dirty(sb, blkno); + if (IS_ERR(bh)) { bitmap_free(sb, blkno); - return bl; + return bh; } - bud = bl->data; - scoutfs_zero_block_tail(bl, sizeof(bud->hdr)); + bud = bh_data(bh); + scoutfs_block_zero(bh, sizeof(bud->hdr)); /* mark the initial run of highest orders free */ count = slot_count(super, sl); @@ -345,7 +345,7 @@ static struct scoutfs_block *dirty_buddy_block(struct super_block *sb, int sl, update_free_orders(slot, bud); - return bl; + return bh; } /* @@ -414,29 +414,29 @@ static int alloc_slot(struct super_block *sb, int sl, struct scoutfs_super_block *super = &sbi->super; struct scoutfs_buddy_block *bud; struct scoutfs_buddy_block *st_bud; - struct scoutfs_block *st_bl; - struct scoutfs_block *bl; + struct buffer_head *st_bh; + struct buffer_head *bh; int found; int ret; int nr; int i; /* initialize or dirty the slot's buddy block */ - bl = dirty_buddy_block(sb, sl, slot); - if (IS_ERR(bl)) - return PTR_ERR(bl); - bud = bl->data; + bh = dirty_buddy_block(sb, sl, slot); + if (IS_ERR(bh)) + return PTR_ERR(bh); + bud = bh_data(bh); /* read stable slots's buddy block if there is one */ if (stable_ref->blkno) { - st_bl = scoutfs_read_ref(sb, stable_ref); - if (IS_ERR(st_bl)) { - ret = PTR_ERR(st_bl); + st_bh = scoutfs_block_read_ref(sb, stable_ref); + if (IS_ERR(st_bh)) { + ret = PTR_ERR(st_bh); goto out; } - st_bud = st_bl->data; + st_bud = bh_data(st_bh); } else { - st_bl = NULL; + st_bh = NULL; st_bud = NULL; } @@ -459,8 +459,8 @@ static int alloc_slot(struct super_block *sb, int sl, update_free_orders(slot, bud); ret = 0; out: - scoutfs_put_block(st_bl); - scoutfs_put_block(bl); + scoutfs_block_put(st_bh); + scoutfs_block_put(bh); return ret; } @@ -482,8 +482,8 @@ static int alloc_order(struct super_block *sb, u64 *blkno, int order) struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_buddy_indirect *st_ind; struct scoutfs_buddy_indirect *ind; - struct scoutfs_block *st_bl = NULL; - struct scoutfs_block *bl = NULL; + struct buffer_head *st_bh = NULL; + struct buffer_head *bh = NULL; u8 mask; int ret; int i; @@ -496,20 +496,20 @@ static int alloc_order(struct super_block *sb, u64 *blkno, int order) } /* get the dirty indirect block */ - bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_ind_ref); - if (IS_ERR(bl)) { - ret = PTR_ERR(bl); + bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_ind_ref); + if (IS_ERR(bh)) { + ret = PTR_ERR(bh); goto out; } - ind = bl->data; + ind = bh_data(bh); /* get the stable indirect block */ - st_bl = scoutfs_read_ref(sb, &sbi->stable_super.buddy_ind_ref); - if (IS_ERR(st_bl)) { - ret = PTR_ERR(st_bl); + st_bh = scoutfs_block_read_ref(sb, &sbi->stable_super.buddy_ind_ref); + if (IS_ERR(st_bh)) { + ret = PTR_ERR(st_bh); goto out; } - st_ind = st_bl->data; + st_ind = bh_data(st_bh); mask = ~0U << order; @@ -531,8 +531,8 @@ static int alloc_order(struct super_block *sb, u64 *blkno, int order) } out: - scoutfs_put_block(st_bl); - scoutfs_put_block(bl); + scoutfs_block_put(st_bh); + scoutfs_block_put(bh); return ret; } @@ -597,7 +597,7 @@ int scoutfs_buddy_alloc(struct super_block *sb, u64 *blkno, int order) static int bitmap_dirty(struct super_block *sb, u64 blkno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_block *bl; + struct buffer_head *bh; /* mkfs should have ensured that there's bitmap blocks */ /* XXX corruption */ @@ -605,11 +605,11 @@ static int bitmap_dirty(struct super_block *sb, u64 blkno) return -EIO; /* dirty the bitmap block */ - bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_bm_ref); - if (IS_ERR(bl)) - return PTR_ERR(bl); + bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_bm_ref); + if (IS_ERR(bh)) + return PTR_ERR(bh); - scoutfs_put_block(bl); + scoutfs_block_put(bh); return 0; } @@ -618,8 +618,8 @@ static int buddy_dirty(struct super_block *sb, u64 blkno, int order) struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct scoutfs_buddy_indirect *ind; - struct scoutfs_block *ind_bl = NULL; - struct scoutfs_block *bl = NULL; + struct buffer_head *ind_bh = NULL; + struct buffer_head *bh = NULL; int ret; int sl; @@ -632,23 +632,23 @@ static int buddy_dirty(struct super_block *sb, u64 blkno, int order) } /* get the dirty indirect block */ - ind_bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_ind_ref); - if (IS_ERR(ind_bl)) { - ret = PTR_ERR(ind_bl); + ind_bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_ind_ref); + if (IS_ERR(ind_bh)) { + ret = PTR_ERR(ind_bh); goto out; } - ind = ind_bl->data; + ind = bh_data(ind_bh); sl = indirect_slot(super, blkno); - bl = dirty_buddy_block(sb, sl, &ind->slots[sl]); - if (IS_ERR(bl)) - ret = PTR_ERR(bl); + bh = dirty_buddy_block(sb, sl, &ind->slots[sl]); + if (IS_ERR(bh)) + ret = PTR_ERR(bh); else ret = 0; out: mutex_unlock(&sbi->buddy_mutex); - scoutfs_put_block(ind_bl); - scoutfs_put_block(bl); + scoutfs_block_put(ind_bh); + scoutfs_block_put(bh); return ret; } @@ -702,8 +702,8 @@ static int buddy_free(struct super_block *sb, u64 blkno, int order) struct scoutfs_super_block *super = &sbi->super; struct scoutfs_buddy_indirect *ind; struct scoutfs_buddy_block *bud; - struct scoutfs_block *ind_bl = NULL; - struct scoutfs_block *bl = NULL; + struct buffer_head *ind_bh = NULL; + struct buffer_head *bh = NULL; int ret; int sl; int nr; @@ -721,20 +721,20 @@ static int buddy_free(struct super_block *sb, u64 blkno, int order) goto out; } - ind_bl = scoutfs_block_cow_ref(sb, &sbi->super.buddy_ind_ref); - if (IS_ERR(ind_bl)) { - ret = PTR_ERR(ind_bl); + ind_bh = scoutfs_block_dirty_ref(sb, &sbi->super.buddy_ind_ref); + if (IS_ERR(ind_bh)) { + ret = PTR_ERR(ind_bh); goto out; } - ind = ind_bl->data; + ind = bh_data(ind_bh); sl = indirect_slot(super, blkno); - bl = scoutfs_block_cow_ref(sb, &ind->slots[sl].ref); - if (IS_ERR(bl)) { - ret = PTR_ERR(bl); + bh = scoutfs_block_dirty_ref(sb, &ind->slots[sl].ref); + if (IS_ERR(bh)) { + ret = PTR_ERR(bh); goto out; } - bud = bl->data; + bud = bh_data(bh); /* * Merge our region with its free buddy and then try to merge @@ -754,11 +754,11 @@ static int buddy_free(struct super_block *sb, u64 blkno, int order) set_buddy_bit(bud, i, nr); update_free_orders(&ind->slots[sl], bud); - scoutfs_put_block(bl); + scoutfs_block_put(bh); ret = 0; out: mutex_unlock(&sbi->buddy_mutex); - scoutfs_put_block(ind_bl); + scoutfs_block_put(ind_bh); return ret; } diff --git a/kmod/src/super.c b/kmod/src/super.c index 8d7137b4..baa384d0 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -55,6 +55,8 @@ void scoutfs_advance_dirty_super(struct super_block *sb) super->hdr.blkno = cpu_to_le64(SCOUTFS_SUPER_BLKNO); le64_add_cpu(&super->hdr.seq, 1); + + trace_printk("super seq now %llu\n", le64_to_cpu(super->hdr.seq)); } /* @@ -64,22 +66,24 @@ void scoutfs_advance_dirty_super(struct super_block *sb) int scoutfs_write_dirty_super(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - size_t sz = sizeof(struct scoutfs_super_block); - u64 blkno = le64_to_cpu(sbi->super.hdr.blkno); - struct scoutfs_block *bl; + struct scoutfs_super_block *super; + struct buffer_head *bh; int ret; /* XXX prealloc? */ - bl = scoutfs_new_block(sb, blkno); - if (WARN_ON_ONCE(IS_ERR(bl))) - return PTR_ERR(bl); + bh = sb_getblk(sb, le64_to_cpu(sbi->super.hdr.blkno)); + if (!bh) + return -ENOMEM; + super = bh_data(bh); - memcpy(bl->data, &sbi->super, sz); - memset(bl->data + sz, 0, SCOUTFS_BLOCK_SIZE - sz); - scoutfs_calc_hdr_crc(bl); - ret = scoutfs_write_block(bl); + *super = sbi->super; + scoutfs_block_zero(bh, sizeof(struct scoutfs_super_block)); + scoutfs_block_set_crc(bh); - scoutfs_put_block(bl); + mark_buffer_dirty(bh); + ret = sync_dirty_buffer(bh); + + scoutfs_block_put(bh); return ret; } @@ -87,19 +91,18 @@ static int read_supers(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super; - struct scoutfs_block *bl = NULL; + struct buffer_head *bh = NULL; int found = -1; int i; for (i = 0; i < SCOUTFS_SUPER_NR; i++) { - scoutfs_put_block(bl); - bl = scoutfs_read_block(sb, SCOUTFS_SUPER_BLKNO + i); - if (IS_ERR(bl)) { + scoutfs_block_put(bh); + bh = scoutfs_block_read(sb, SCOUTFS_SUPER_BLKNO + i); + if (IS_ERR(bh)) { scoutfs_warn(sb, "couldn't read super block %u", i); continue; } - - super = bl->data; + super = bh_data(bh); if (super->id != cpu_to_le64(SCOUTFS_SUPER_ID)) { scoutfs_warn(sb, "super block %u has invalid id %llx", @@ -114,7 +117,7 @@ static int read_supers(struct super_block *sb) } } - scoutfs_put_block(bl); + scoutfs_block_put(bh); if (found < 0) { scoutfs_err(sb, "unable to read valid super block"); @@ -147,7 +150,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) spin_lock_init(&sbi->next_ino_lock); spin_lock_init(&sbi->block_lock); - INIT_RADIX_TREE(&sbi->block_radix, GFP_NOFS); + sbi->block_dirty_tree = RB_ROOT; init_waitqueue_head(&sbi->block_wq); atomic_set(&sbi->block_writes, 0); mutex_init(&sbi->buddy_mutex); @@ -158,6 +161,11 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) INIT_WORK(&sbi->trans_write_work, scoutfs_trans_write_func); init_waitqueue_head(&sbi->trans_write_wq); + if (!sb_set_blocksize(sb, SCOUTFS_BLOCK_SIZE)) { + printk(KERN_ERR "couldn't set blocksize\n"); + return -EINVAL; + } + /* XXX can have multiple mounts of a device, need mount id */ sbi->kset = kset_create_and_add(sb->s_id, NULL, &scoutfs_kset->kobj); if (!sbi->kset) @@ -198,6 +206,9 @@ static void scoutfs_kill_sb(struct super_block *sb) scoutfs_destroy_counters(sb); if (sbi->kset) kset_unregister(sbi->kset); + + /* XXX write errors can leave dirty blocks */ + WARN_ON_ONCE(!RB_EMPTY_ROOT(&sbi->block_dirty_tree)); kfree(sbi); } } diff --git a/kmod/src/super.h b/kmod/src/super.h index f0a2e8cf..7c3617b4 100644 --- a/kmod/src/super.h +++ b/kmod/src/super.h @@ -19,7 +19,7 @@ struct scoutfs_sb_info { spinlock_t next_ino_lock; spinlock_t block_lock; - struct radix_tree_root block_radix; + struct rb_root block_dirty_tree; wait_queue_head_t block_wq; atomic_t block_writes; int block_write_err; diff --git a/kmod/src/trans.c b/kmod/src/trans.c index e1376546..3291b83c 100644 --- a/kmod/src/trans.c +++ b/kmod/src/trans.c @@ -63,14 +63,13 @@ void scoutfs_trans_write_func(struct work_struct *work) /* XXX probably want to write out dirty pages in inodes */ - if (scoutfs_has_dirty_blocks(sb)) { - ret = scoutfs_write_dirty_blocks(sb) ?: - scoutfs_write_dirty_super(sb); + ret = scoutfs_block_write_dirty(sb); + if (ret > 0) { + ret = scoutfs_write_dirty_super(sb); if (!ret) advance = 1; } - spin_lock(&sbi->trans_write_lock); if (advance) scoutfs_advance_dirty_super(sb);