mirrors/scoutfs
1
0
Fork 0
mirror of https://github.com/versity/scoutfs.git synced 2026-07-19 14:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

scoutfs: use buffer heads

Browse Source 
Now that we have a fixed small block size we don't need our own code for
tracking contiguous memory for blocks that are larger than the page
size.  We can use buffer heads which support block sizes smaller than
the page size.

Our block API remains to enforce transactions, cheksumming, cow, and
eventually invalidating and retrying reads of stale bloks.

We set the logical blocksize of the bdev buffer cache to our fixed block
size.  We use a private bh state bit to indicate that the contents of a
block have had their checksum verified.  We use a small structure stored
at b_private to track dirty blocks so that we can control when they're
written.

The btree block traversal code uses the buffer_head lock to serialize
access to btree block contents now that the block rwsem has gone
away.  This isn't great but works for now.

Not being able to relocate blocks in the buffer cache (really fragments
of pages in the bdev page cache.. blkno determines memory location)
means that the cow path always has to copy.

Callers are easily translated: use struct buffer_head instead of
scoutfs_block and use a little helper instead of dereferencing ->data
directly.

I took the opportunity to clean up some of the inconsistent block
function names.  Now more of the functions follow the scoutfs_block_*()
pattern.

Signed-off-by: Zach Brown <zab@versity.com>
This commit is contained in:
Zach Brown
2016-08-01 11:30:36 -07:00
parent f024c70802
commit 7b18bce2e2
8 changed files with 578 additions and 715 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kmod/src/block.c
+271 -413
View File
@@ -11,10 +11,9 @@
* General Public License for more details.
*/
#include <linux/kernel.h>
#include <linux/buffer_head.h>
#include <linux/blkdev.h>
#include <linux/slab.h>
#include <linux/radix-tree.h>
#include <linux/mm.h>
#include <linux/bio.h>
#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, &copy_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);
}
kmod/src/block.h
+24 -32
View File
@@ -2,43 +2,35 @@
#define _SCOUTFS_BLOCK_H_
#include <linux/fs.h>
#include <linux/rwlock.h>
#include <linux/atomic.h>
#include <linux/buffer_head.h>
#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
kmod/src/btree.c
+168 -165
View File
@@ -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;
}
/*
kmod/src/btree.h
+1 -1
View File
@@ -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 */
kmod/src/buddy.c
+80 -80
View File
@@ -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;
}
kmod/src/super.c
+30 -19
View File
@@ -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);
}
}
kmod/src/super.h
+1 -1
View File
@@ -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;
kmod/src/trans.c
+3 -4
View File
@@ -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);