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

Move to much simpler manifest/alloc storage

Browse Source 
Using the treap to be able to incrementally read and write the manifest
and allocation storage from all nodes wasn't quite ready for prime time.
The biggest problem is that invalidating cached nodes which are the
target of native pointers, either for consistency or memory pressure, is
problematic.  This was getting in the way of adding shared support as
readers and writers try to use as much of their treap caches as they
can.  There were other serious problems that we'd run into eventually:
memory pressure from duplicate caching in native nodes and the page
cache, small IOs from reading a page at a time, the risk of
pathologically imbalanced treaps, and the ring being corrupted if the
migration balancing doesn't work (the model assumed you could always
dirty an individual node in a transaction, you have to dirty all the
parents in each new transaction).

Let's back off to a much simpler mechanism while we build the rest of
the system around it.  We can revisit aggressively optimizing this when
it's our worst problem.

We'll store the indexes that the manifest server needs in simple
preallocated rings with log entries.   The server has to read the index
in its entirety into a native rbtree before it can work on it.  We won't
access the physical ring from mounts anymore, they'll send messages to
the server.

The ring callers are now working with a pinned tree in memory so the
interface can be a bit simpler.  By storing the indexes in their own
rings the code and write path become a lot simper: we have an IO
submission path for each index instead of "dirtying" calls per index and
then a writing call.

All this is much more robust and much less likely to get in our way as
we stand up the rest of the system around it.

Signed-off-by: Zach Brown <zab@versity.com>
This commit is contained in:
Zach Brown
2017-04-18 13:51:10 -07:00
parent 86d3090982
commit 5e0e9ac12e
14 changed files with 1027 additions and 1599 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kmod/src/Makefile
+2 -2
View File
@@ -3,5 +3,5 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o
CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
scoutfs-y += alloc.o bio.o compact.o counters.o data.o dir.o kvec.o inode.o \
ioctl.o item.o key.o lock.o manifest.o msg.o net.o seg.o \
scoutfs_trace.o super.o trans.o treap.o xattr.o
ioctl.o item.o key.o lock.o manifest.o msg.o net.o ring.o seg.o \
scoutfs_trace.o super.o trans.o xattr.o
kmod/src/alloc.c
+49 -43
View File
@@ -17,13 +17,13 @@
#include "super.h"
#include "format.h"
#include "treap.h"
#include "ring.h"
#include "cmp.h"
#include "alloc.h"
#include "counters.h"
/*
* scoutfs allocates segments by storing regions of a bitmap in treap
* scoutfs allocates segments by storing regions of a bitmap in ring
* nodes.
*
* Freed segments are recorded in nodes in an rbtree. The frees can't
@@ -40,7 +40,7 @@
struct seg_alloc {
struct rw_semaphore rwsem;
struct rb_root pending_root;
struct scoutfs_treap *treap;
struct scoutfs_ring_info ring;
u64 next_segno;
};
@@ -132,7 +132,7 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno)
nr = sal->next_segno & SCOUTFS_ALLOC_REGION_MASK;
do {
reg = scoutfs_treap_lookup_next_dirty(sal->treap, &ind);
reg = scoutfs_ring_lookup_next(&sal->ring, &ind);
} while (reg == NULL && ind && (ind = 0, nr = 0, 1));
if (IS_ERR_OR_NULL(reg)) {
@@ -143,6 +143,8 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno)
goto out;
}
scoutfs_ring_dirty(&sal->ring, reg);
nr = find_next_bit_le(reg->bits, SCOUTFS_ALLOC_REGION_BITS, nr);
if (nr >= SCOUTFS_ALLOC_REGION_BITS) {
/* XXX corruption? shouldn't find empty regions */
@@ -154,12 +156,8 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno)
clear_bit_le(nr, reg->bits);
if (empty_region(reg)) {
ret = scoutfs_treap_delete(sal->treap, &ind);
/* XXX figure out what to do about this inconsistency */
if (WARN_ON_ONCE(ret))
goto out;
}
if (empty_region(reg))
scoutfs_ring_delete(&sal->ring, reg);
*segno = (ind << SCOUTFS_ALLOC_REGION_SHIFT) + nr;
sal->next_segno = *segno + 1;
@@ -178,7 +176,7 @@ out:
/*
* Record newly freed sgements in pending regions. These are applied to
* treap nodes as the transaction commits.
* ring nodes as the transaction commits.
*/
int scoutfs_alloc_free(struct super_block *sb, u64 segno)
{
@@ -234,7 +232,8 @@ int scoutfs_alloc_has_dirty(struct super_block *sb)
int ret;
down_write(&sal->rwsem);
ret = scoutfs_treap_has_dirty(sal->treap);
ret = !!(scoutfs_ring_has_dirty(&sal->ring) ||
!RB_EMPTY_ROOT(&sal->pending_root));
up_write(&sal->rwsem);
return ret;
@@ -242,13 +241,12 @@ int scoutfs_alloc_has_dirty(struct super_block *sb)
/*
* First we apply the pending frees to create the final set of dirty
* region nodes and then ask the treap to write them to ring pages.
* region nodes and then ask the ring to write them to the ring.
*/
int scoutfs_alloc_dirty_ring(struct super_block *sb)
int scoutfs_alloc_submit_write(struct super_block *sb,
struct scoutfs_bio_completion *comp)
{
DECLARE_SEG_ALLOC(sb, sal);
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
struct scoutfs_alloc_region *reg;
struct pending_region *pend;
struct rb_node *node;
@@ -262,30 +260,41 @@ int scoutfs_alloc_dirty_ring(struct super_block *sb)
ind = le64_to_cpu(pend->reg.index);
reg = scoutfs_treap_lookup_dirty(sal->treap, &ind);
if (!reg)
reg = scoutfs_treap_insert(sal->treap, &ind,
sizeof(struct scoutfs_alloc_region),
&ind);
if (IS_ERR(reg)) {
ret = PTR_ERR(reg);
goto out;
reg = scoutfs_ring_lookup(&sal->ring, &ind);
if (!reg) {
reg = scoutfs_ring_insert(&sal->ring, &ind,
sizeof(struct scoutfs_alloc_region));
if (!reg) {
ret = -ENOMEM;
goto out;
}
memset(reg, 0, sizeof(struct scoutfs_alloc_region));
reg->index = cpu_to_le64(ind);
}
reg->index = pend->reg.index;
or_region_bits(reg, &pend->reg);
scoutfs_ring_dirty(&sal->ring, reg);
rb_erase(&pend->node, &sal->pending_root);
kfree(pend);
}
scoutfs_treap_dirty_ring(sal->treap, &super->alloc_treap_root);
ret = 0;
ret = scoutfs_ring_submit_write(sb, &sal->ring, comp);
out:
up_write(&sal->rwsem);
return ret;
}
void scoutfs_alloc_write_complete(struct super_block *sb)
{
DECLARE_SEG_ALLOC(sb, sal);
down_write(&sal->rwsem);
scoutfs_ring_write_complete(&sal->ring);
up_write(&sal->rwsem);
}
/*
* Return the number of blocks free for statfs.
*/
@@ -303,7 +312,7 @@ u64 scoutfs_alloc_bfree(struct super_block *sb)
return bfree;
}
static int alloc_treap_compare(void *key, void *data)
static int alloc_ring_compare_key(void *key, void *data)
{
u64 *ind = key;
struct scoutfs_alloc_region *reg = data;
@@ -311,25 +320,20 @@ static int alloc_treap_compare(void *key, void *data)
return scoutfs_cmp_u64s(*ind, le64_to_cpu(reg->index));
}
static void alloc_treap_fill(void *data, void *fill_arg)
static int alloc_ring_compare_data(void *A, void *B)
{
struct scoutfs_alloc_region *reg = data;
u64 *ind = fill_arg;
struct scoutfs_alloc_region *a = A;
struct scoutfs_alloc_region *b = B;
memset(reg, 0, sizeof(struct scoutfs_alloc_region));
reg->index = cpu_to_le64p(ind);
return scoutfs_cmp_u64s(le64_to_cpu(a->index), le64_to_cpu(b->index));
}
static struct scoutfs_treap_ops alloc_treap_ops = {
.compare = alloc_treap_compare,
.fill = alloc_treap_fill,
};
int scoutfs_alloc_setup(struct super_block *sb)
{
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
struct seg_alloc *sal;
int ret;
/* bits need to be aligned so hosts can use native bitops */
BUILD_BUG_ON(offsetof(struct scoutfs_alloc_region, bits) &
@@ -341,11 +345,13 @@ int scoutfs_alloc_setup(struct super_block *sb)
init_rwsem(&sal->rwsem);
sal->pending_root = RB_ROOT;
sal->treap = scoutfs_treap_alloc(sb, &alloc_treap_ops,
&super->alloc_treap_root);
if (!sal->treap) {
scoutfs_ring_init(&sal->ring, &super->alloc_ring,
alloc_ring_compare_key, alloc_ring_compare_data);
ret = scoutfs_ring_load(sb, &sal->ring);
if (ret) {
kfree(sal);
return -ENOMEM;
return ret;
}
/* XXX read next_segno from super? */
@@ -362,7 +368,7 @@ void scoutfs_alloc_destroy(struct super_block *sb)
struct rb_node *node;
if (sal) {
scoutfs_treap_free(sal->treap);
scoutfs_ring_destroy(&sal->ring);
while ((node = rb_first(&sal->pending_root))) {
pend = container_of(node, struct pending_region, node);
rb_erase(&pend->node, &sal->pending_root);
kmod/src/alloc.h
+4 -1
View File
@@ -2,12 +2,15 @@
#define _SCOUTFS_ALLOC_H_
struct scoutfs_alloc_region;
struct scoutfs_bio_completion;
int scoutfs_alloc_segno(struct super_block *sb, u64 *segno);
int scoutfs_alloc_free(struct super_block *sb, u64 segno);
int scoutfs_alloc_has_dirty(struct super_block *sb);
int scoutfs_alloc_dirty_ring(struct super_block *sb);
int scoutfs_alloc_submit_write(struct super_block *sb,
struct scoutfs_bio_completion *comp);
void scoutfs_alloc_write_complete(struct super_block *sb);
u64 scoutfs_alloc_bfree(struct super_block *sb);
int scoutfs_alloc_setup(struct super_block *sb);
kmod/src/compact.c
+6 -5
View File
@@ -522,11 +522,12 @@ out:
/*
* Atomically update the manifest. We lock down the manifest so no one
* can use it while we're mucking with it. We can always delete dirty
* treap nodes without failure. So we first dirty the deletion nodes
* before modifying anything. Then we add and if any of those fail we
* can delete the dirty previous additions. Then we can delete the
* dirty existing entries without failure.
* can use it while we're mucking with it. While the current ring can
* always delete without failure we will probably have a manifest
* storage layer eventually that could return errors on deletion. We
* also also have corrupted something and try to delete an entry that
* doesn't exist. So we use an initial dirtying step to ensure that our
* later deletions succeed.
*
* XXX does locking the manifest prevent commits? I would think so?
*/
kmod/src/format.h
+21 -33
View File
@@ -50,43 +50,30 @@ struct scoutfs_block_header {
__le64 blkno;
} __packed;
struct scoutfs_treap_ref {
__le64 off;
__le64 gen;
__u8 aug_bits;
struct scoutfs_ring_entry {
__le16 data_len;
__u8 flags;
__u8 data[0];
} __packed;
/*
* The lesser and greater bits are persistent on disk so that we can migrate
* nodes from the older half of the ring.
*
* The dirty bit is only used for in-memory nodes.
*/
#define SCOUTFS_TREAP_AUG_LESSER (1 << 0)
#define SCOUTFS_TREAP_AUG_GREATER (1 << 1)
#define SCOUTFS_TREAP_AUG_HALVES (SCOUTFS_TREAP_AUG_LESSER | \
SCOUTFS_TREAP_AUG_GREATER)
#define SCOUTFS_TREAP_AUG_DIRTY (1 << 2)
#define SCOUTFS_RING_ENTRY_FLAG_DELETION (1 << 0)
/*
* Treap nodes are stored at byte offset in the ring of blocks described
* by the super block. Each reference contains the off and gen that it
* will find in the node for verification. Each node has the header
* and data payload covered by a crc.
*/
struct scoutfs_treap_node {
struct scoutfs_ring_block {
__le32 crc;
__le64 off;
__le64 gen;
__le64 prio;
struct scoutfs_treap_ref left;
struct scoutfs_treap_ref right;
__le16 bytes;
u8 data[0];
__le32 pad;
__le64 fsid;
__le64 seq;
__le64 block;
__le32 nr_entries;
struct scoutfs_ring_entry entries[0];
} __packed;
struct scoutfs_treap_root {
struct scoutfs_treap_ref ref;
struct scoutfs_ring_descriptor {
__le64 blkno;
__le64 total_blocks;
__le64 first_block;
__le64 first_seq;
__le64 nr_blocks;
} __packed;
/*
@@ -98,7 +85,7 @@ struct scoutfs_treap_root {
#define SCOUTFS_MANIFEST_FANOUT 10
struct scoutfs_manifest {
struct scoutfs_treap_root root;
struct scoutfs_ring_descriptor ring;
__le64 level_counts[SCOUTFS_MANIFEST_MAX_LEVEL];
} __packed;
@@ -246,6 +233,7 @@ struct scoutfs_symlink_key {
#define SCOUTFS_UUID_BYTES 16
/*
* The ring fields describe the statically allocated ring log. The
* head and tail indexes are logical 4k blocks offsets inside the ring.
@@ -264,7 +252,7 @@ struct scoutfs_super_block {
__le64 ring_tail_block;
__le64 ring_gen;
__le64 next_seg_seq;
struct scoutfs_treap_root alloc_treap_root;
struct scoutfs_ring_descriptor alloc_ring;
struct scoutfs_manifest manifest;
} __packed;
kmod/src/manifest.c
+78 -108
View File
@@ -20,7 +20,7 @@
#include "kvec.h"
#include "seg.h"
#include "item.h"
#include "treap.h"
#include "ring.h"
#include "cmp.h"
#include "compact.h"
#include "manifest.h"
@@ -29,24 +29,17 @@
#include "scoutfs_trace.h"
/*
* Manifest entries are stored as treap nodes in the ring.
* Manifest entries are stored in ring nodes.
*
* They're sorted first by level then by their first key. This enables
* the primary searches based on key value for looking up items in
* segments via the manifest.
*
* The treap also supports augmented searches. We get callbacks as the
* tree structure which lets us maintain data in nodes that describe
* subtrees to accelerate searches. We will record the max sequence
* numbers in subtrees for all the seq queries. We'll probably also
* have bits that direct us towards segments that contain deletion items
* for prioritized compaction.
*/
struct manifest {
struct rw_semaphore rwsem;
seqcount_t seqcount;
struct scoutfs_treap *treap;
struct scoutfs_ring_info ring;
u8 nr_levels;
/* calculated on mount, const thereafter */
@@ -81,12 +74,6 @@ struct manifest_ref {
struct scoutfs_key_buf *last;
};
struct manifest_fill_args {
struct scoutfs_manifest_entry ment;
struct scoutfs_key_buf *first;
struct scoutfs_key_buf *last;
};
/*
* Seq is only specified for operations that differentiate between
* segments with identical items by their sequence number.
@@ -174,7 +161,7 @@ static void add_level_count(struct super_block *sb, struct manifest *mani,
}
/*
* Insert a new manifest entry in the treap. The treap allocates a new
* Insert a new manifest entry in the ring. The ring allocates a new
* node for us and we fill it.
*
* This must be called with the manifest lock held.
@@ -188,40 +175,38 @@ int scoutfs_manifest_add(struct super_block *sb,
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
struct scoutfs_manifest_entry *ment;
struct manifest_fill_args args;
struct scoutfs_key_buf ment_first;
struct scoutfs_key_buf ment_last;
struct manifest_search_key skey;
unsigned key_bytes;
unsigned bytes;
int ret;
trace_scoutfs_manifest_add(sb, first, last, segno, seq, level);
key_bytes = first->key_len + last->key_len;
bytes = offsetof(struct scoutfs_manifest_entry, keys[key_bytes]);
args.ment.segno = cpu_to_le64(segno);
args.ment.seq = cpu_to_le64(seq);
args.ment.first_key_len = cpu_to_le16(first->key_len);
args.ment.last_key_len = cpu_to_le16(last->key_len);
args.ment.level = level;
args.first = first;
args.last = last;
skey.key = first;
skey.level = level;
skey.seq = seq;
ment = scoutfs_treap_insert(mani->treap, &skey, bytes, &args);
if (IS_ERR(ment)) {
ret = PTR_ERR(ment);
} else {
mani->nr_levels = max_t(u8, mani->nr_levels, level + 1);
add_level_count(sb, mani, super, level, 1);
ret = 0;
}
ment = scoutfs_ring_insert(&mani->ring, &skey, bytes);
if (!ment)
return -ENOMEM;
return ret;
ment->segno = cpu_to_le64(segno);
ment->seq = cpu_to_le64(seq);
ment->first_key_len = cpu_to_le16(first->key_len);
ment->last_key_len = cpu_to_le16(last->key_len);
ment->level = level;
init_ment_keys(ment, &ment_first, &ment_last);
scoutfs_key_copy(&ment_first, first);
scoutfs_key_copy(&ment_last, last);
mani->nr_levels = max_t(u8, mani->nr_levels, level + 1);
add_level_count(sb, mani, super, level, 1);
return 0;
}
/*
@@ -238,11 +223,11 @@ int scoutfs_manifest_dirty(struct super_block *sb,
skey.level = level;
skey.seq = seq;
ment = scoutfs_treap_lookup_dirty(mani->treap, &skey);
if (IS_ERR(ment))
return PTR_ERR(ment);
ment = scoutfs_ring_lookup(&mani->ring, &skey);
if (!ment)
return -ENOENT;
scoutfs_ring_dirty(&mani->ring, ment);
return 0;
}
@@ -255,18 +240,20 @@ int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *first,
DECLARE_MANIFEST(sb, mani);
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
struct scoutfs_manifest_entry *ment;
struct manifest_search_key skey;
int ret;
skey.key = first;
skey.level = level;
skey.seq = seq;
ret = scoutfs_treap_delete(mani->treap, &skey);
if (ret == 0)
add_level_count(sb, mani, super, level, -1ULL);
ment = scoutfs_ring_lookup(&mani->ring, &skey);
if (!ment)
return -ENOENT;
return ret;
scoutfs_ring_delete(&mani->ring, ment);
add_level_count(sb, mani, super, level, -1ULL);
return 0;
}
/*
@@ -363,19 +350,15 @@ static int get_range_refs(struct super_block *sb, struct manifest *mani,
/* get level 0 segments that overlap with the missing range */
skey.level = 0;
skey.seq = ~0ULL;
ment = scoutfs_treap_lookup_prev(mani->treap, &skey);
while (!IS_ERR_OR_NULL(ment)) {
ment = scoutfs_ring_lookup_prev(&mani->ring, &skey);
while (ment) {
if (cmp_range_ment(key, end, ment) == 0) {
ret = alloc_add_ref(sb, ref_list, ment);
if (ret)
goto out;
}
ment = scoutfs_treap_prev(mani->treap, ment);
}
if (IS_ERR(ment)) {
ret = PTR_ERR(ment);
goto out;
ment = scoutfs_ring_prev(&mani->ring, ment);
}
/* get higher level segments that overlap with the starting key */
@@ -386,12 +369,7 @@ static int get_range_refs(struct super_block *sb, struct manifest *mani,
/* XXX should use level counts to skip searches */
ment = scoutfs_treap_lookup(mani->treap, &skey);
if (IS_ERR(ment)) {
ret = PTR_ERR(ment);
goto out;
}
ment = scoutfs_ring_lookup(&mani->ring, &skey);
if (ment) {
init_ment_keys(ment, &first, &last);
ret = alloc_add_ref(sb, ref_list, ment);
@@ -625,28 +603,32 @@ int scoutfs_manifest_has_dirty(struct super_block *sb)
int ret;
down_write(&mani->rwsem);
ret = scoutfs_treap_has_dirty(mani->treap);
ret = scoutfs_ring_has_dirty(&mani->ring);
up_write(&mani->rwsem);
return ret;
}
/*
* Append the dirty manifest entries to the end of the ring.
*
* This returns 0 but can't fail.
*/
int scoutfs_manifest_dirty_ring(struct super_block *sb)
int scoutfs_manifest_submit_write(struct super_block *sb,
struct scoutfs_bio_completion *comp)
{
DECLARE_MANIFEST(sb, mani);
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
int ret;
down_write(&mani->rwsem);
scoutfs_treap_dirty_ring(mani->treap, &super->manifest.root);
ret = scoutfs_ring_submit_write(sb, &mani->ring, comp);
up_write(&mani->rwsem);
return 0;
return ret;
}
void scoutfs_manifest_write_complete(struct super_block *sb)
{
DECLARE_MANIFEST(sb, mani);
down_write(&mani->rwsem);
scoutfs_ring_write_complete(&mani->ring);
up_write(&mani->rwsem);
}
u64 scoutfs_manifest_level_count(struct super_block *sb, u8 level)
@@ -714,23 +696,19 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
/* find the oldest level 0 or the next higher order level by key */
if (level == 0) {
ment = scoutfs_treap_first(mani->treap);
if (!IS_ERR_OR_NULL(ment) && ment->level)
ment = scoutfs_ring_first(&mani->ring);
if (ment && ment->level)
ment = NULL;
} else {
skey.key = mani->compact_keys[level];
skey.level = level;
skey.seq = 0;
ment = scoutfs_treap_lookup_next(mani->treap, &skey);
ment = scoutfs_ring_lookup_next(&mani->ring, &skey);
if (ment == NULL || ment->level != level) {
scoutfs_key_set_min(skey.key);
ment = scoutfs_treap_lookup_next(mani->treap, &skey);
ment = scoutfs_ring_lookup_next(&mani->ring, &skey);
}
}
if (IS_ERR(ment)) {
ret = PTR_ERR(ment);
goto out;
}
if (ment == NULL || ment->level != level) {
/* XXX shouldn't be possible */
ret = 0;
@@ -750,14 +728,10 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
skey.key = &ment_first;
skey.level = level + 1;
skey.seq = 0;
over = scoutfs_treap_lookup_next(mani->treap, &skey);
over = scoutfs_ring_lookup_next(&mani->ring, &skey);
/* and add a fanout's worth of lower overlapping segments */
for (i = 0; i < SCOUTFS_MANIFEST_FANOUT; i++) {
if (IS_ERR(over)) {
ret = PTR_ERR(over);
goto out;
}
if (!over || over->level != (ment->level + 1))
break;
@@ -773,7 +747,7 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
if (ret)
goto out;
over = scoutfs_treap_next(mani->treap, over);
over = scoutfs_ring_next(&mani->ring, over);
}
/* record the next key to start from */
@@ -787,7 +761,7 @@ out:
}
/*
* Manifest entries for all levels are stored in a single treap.
* Manifest entries for all levels are stored in a single ring.
*
* First they're sorted by their level.
*
@@ -806,7 +780,7 @@ out:
* number. We tell the difference by the presence of a sequence number.
* A segment will never have a seq of 0.
*/
static int manifest_treap_compare(void *key, void *data)
static int manifest_ring_compare_key(void *key, void *data)
{
struct manifest_search_key *skey = key;
struct scoutfs_manifest_entry *ment = data;
@@ -842,32 +816,27 @@ out:
return cmp;
}
static void manifest_treap_fill(void *data, void *arg)
static int manifest_ring_compare_data(void *a, void *b)
{
struct scoutfs_manifest_entry *ment = data;
struct manifest_fill_args *args = arg;
struct scoutfs_key_buf ment_first;
struct scoutfs_key_buf ment_last;
struct manifest_search_key skey;
struct scoutfs_manifest_entry *ment = a;
struct scoutfs_key_buf key;
*ment = args->ment;
init_ment_keys(ment, &key, NULL);
init_ment_keys(ment, &ment_first, &ment_last);
scoutfs_key_copy(&ment_first, args->first);
scoutfs_key_copy(&ment_last, args->last);
skey.seq = le64_to_cpu(ment->seq);
skey.key = &key;
skey.level = ment->level;
return manifest_ring_compare_key(&skey, b);
}
static struct scoutfs_treap_ops manifest_treap_ops = {
.compare = manifest_treap_compare,
.fill = manifest_treap_fill,
/* update aug when we track left and right max seq */
};
int scoutfs_manifest_setup(struct super_block *sb)
{
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct scoutfs_super_block *super = &sbi->super;
struct manifest *mani;
int ret;
int i;
mani = kzalloc(sizeof(struct manifest), GFP_KERNEL);
@@ -876,12 +845,13 @@ int scoutfs_manifest_setup(struct super_block *sb)
init_rwsem(&mani->rwsem);
seqcount_init(&mani->seqcount);
mani->treap = scoutfs_treap_alloc(sb, &manifest_treap_ops,
&super->manifest.root);
if (!mani->treap) {
scoutfs_ring_init(&mani->ring, &super->manifest.ring,
manifest_ring_compare_key,
manifest_ring_compare_data);
ret = scoutfs_ring_load(sb, &mani->ring);
if (ret) {
kfree(mani);
return -ENOMEM;
return ret;
}
for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++) {
@@ -890,7 +860,7 @@ int scoutfs_manifest_setup(struct super_block *sb)
if (!mani->compact_keys[i]) {
while (--i >= 0)
scoutfs_key_free(sb, mani->compact_keys[i]);
scoutfs_treap_free(mani->treap);
scoutfs_ring_destroy(&mani->ring);
kfree(mani);
return -ENOMEM;
}
@@ -925,7 +895,7 @@ void scoutfs_manifest_destroy(struct super_block *sb)
int i;
if (mani) {
scoutfs_treap_free(mani->treap);
scoutfs_ring_destroy(&mani->ring);
for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++)
scoutfs_key_free(sb, mani->compact_keys[i]);
kfree(mani);
kmod/src/manifest.h
+4 -1
View File
@@ -2,6 +2,7 @@
#define _SCOUTFS_MANIFEST_H_
struct scoutfs_key_buf;
struct scoutfs_bio_completion;
int scoutfs_manifest_add(struct super_block *sb,
struct scoutfs_key_buf *first,
@@ -12,7 +13,9 @@ int scoutfs_manifest_dirty(struct super_block *sb,
int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *first,
u64 seq, u8 level);
int scoutfs_manifest_has_dirty(struct super_block *sb);
int scoutfs_manifest_dirty_ring(struct super_block *sb);
int scoutfs_manifest_submit_write(struct super_block *sb,
struct scoutfs_bio_completion *comp);
void scoutfs_manifest_write_complete(struct super_block *sb);
int scoutfs_manifest_lock(struct super_block *sb);
int scoutfs_manifest_unlock(struct super_block *sb);
kmod/src/ring.c
+803
View File
@@ -0,0 +1,803 @@
/*
* Copyright (C) 2017 Versity Software, Inc. All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public
* License v2 as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* General Public License for more details.
*/
#include <linux/kernel.h>
#include <linux/fs.h>
#include <linux/slab.h>
#include <linux/pagemap.h>
#include <linux/crc32c.h>
#include "super.h"
#include "format.h"
#include "bio.h"
#include "ring.h"
/*
* scoutfs stores the persistent indexes for the server in a simple log
* entries in a preallocated ring of blocks.
*
* The index is read from the log and loaded in to an rbtree in memory.
* Callers then lock around operations that work on the rbtrees. Dirty
* and deleted nodes are tracked and are eventually copied to pages that
* are written to the tail of the log.
*
* This has the great benefit of updating an index with very few (often
* one) contiguous block writes with low write amplification.
*
* This has the significant cost of requiring reading the indexes in to
* memory before doing any work and then having to hold them resident.
* This is fine for now but we'll have to address these latency and
* capacity limitations before too long.
*
* Callers are entirely responsible for locking.
*/
/*
* XXX
* - deletion entries could be smaller if we understood keys
* - shouldn't be too hard to compress
*/
/*
* @block records the logical ring index of the block that contained the
* node. As we commit a ring update we can look at the clean list to
* find the first block that we have to read out of the ring. This
* helps minimize the active region of the ring.
*
* @in_ring is used to mark nodes that were present in the ring and
* which need deletion entries written to the ring before they can be
* freed.
*/
struct ring_node {
struct rb_node rb_node;
struct list_head head;
u64 block;
u16 data_len;
u8 dirty:1,
deleted:1,
in_ring:1;
/* data is packed but callers perform native long bitops */
u8 data[0] __aligned(__alignof__(long));
};
static struct ring_node *data_rnode(void *data)
{
return data ? container_of(data, struct ring_node, data) : NULL;
}
static void *rnode_data(struct ring_node *rnode)
{
return rnode ? rnode->data : NULL;
}
static unsigned total_entry_bytes(unsigned data_len)
{
return offsetof(struct scoutfs_ring_entry, data[data_len]);
}
/*
* Each time we mark a node dirty we also dirty the oldest clean entry.
* This ensures that we never overwrite stable data.
*
* Picture a ring of blocks where the first half of the ring is full of
* existing entries. Imagine that we continuously update a set of
* entries that make up a single block. Each new update block
* invalidates the previous update block but it advances through the
* ring while the old entries are sitting idle in the first half.
* Eventually the new update blocks wrap around and clobber the old
* blocks.
*
* Now instead imagine that each time we dirty an entry in this set of
* constantly changing entries that we also go and dirty the earliest
* existing entry in the ring. Now each update is a block of the
* useless updating entries and a block of old entries that have been
* migrated. Each time we write two blocks to the ring we migrate one
* block from the start of the ring. Now by the time we fill the second
* half of the ring we've reclaimed half of the first half of the ring.
*
* So we size the ring to fit 4x the largest possible index. Now we're
* sure that we'll be able to fully migrate the index from the first
* half of the ring into the second half before it wraps around and
* starts overwriting the first.
*/
static void mark_node_dirty(struct scoutfs_ring_info *ring,
struct ring_node *rnode, bool migrate)
{
struct ring_node *pos;
long total;
if (!rnode || rnode->dirty)
return;
list_move_tail(&rnode->head, &ring->dirty_list);
rnode->dirty = 1;
ring->dirty_bytes += total_entry_bytes(rnode->data_len);
if (migrate) {
total = total_entry_bytes(rnode->data_len);
list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) {
mark_node_dirty(ring, rnode, false);
total -= total_entry_bytes(rnode->data_len);
if (total < 0)
break;
}
}
}
static void mark_node_clean(struct scoutfs_ring_info *ring,
struct ring_node *rnode)
{
if (!rnode || !rnode->dirty)
return;
list_move_tail(&rnode->head, &ring->clean_list);
rnode->dirty = 0;
ring->dirty_bytes -= total_entry_bytes(rnode->data_len);
}
static void free_node(struct scoutfs_ring_info *ring,
struct ring_node *rnode)
{
if (rnode) {
mark_node_clean(ring, rnode);
if (!list_empty(&rnode->head))
list_del_init(&rnode->head);
if (!RB_EMPTY_NODE(&rnode->rb_node))
rb_erase(&rnode->rb_node, &ring->rb_root);
kfree(rnode);
}
}
/*
* Walk the tree and return the last node traversed. cmp gives the
* caller the comparison between their key and the returned node. The
* caller can provide either their key or another nodes data to compare
* with during descent. If we're asked to insert we replace any node we
* find in the key's place.
*/
static struct ring_node *ring_rb_walk(struct scoutfs_ring_info *ring,
void *key, void *data,
struct ring_node *ins,
int *cmp)
{
struct rb_node **node = &ring->rb_root.rb_node;
struct rb_node *parent = NULL;
struct ring_node *found = NULL;
struct ring_node *rnode;
/* only provide one or the other */
BUG_ON(!!key == !!data);
while (*node) {
parent = *node;
rnode = container_of(*node, struct ring_node, rb_node);
if (key)
*cmp = ring->compare_key(key, &rnode->data);
else
*cmp = ring->compare_data(data, &rnode->data);
if (*cmp < 0) {
node = &(*node)->rb_left;
} else if (*cmp > 0) {
node = &(*node)->rb_right;
} else {
found = rnode;
break;
}
}
if (ins) {
if (found) {
rb_replace_node(&found->rb_node, &ins->rb_node,
&ring->rb_root);
RB_CLEAR_NODE(&found->rb_node);
free_node(ring, found);
} else {
rb_link_node(&ins->rb_node, parent, node);
rb_insert_color(&ins->rb_node, &ring->rb_root);
}
found = ins;
}
return found;
}
static struct ring_node *ring_rb_entry(struct rb_node *node)
{
return node ? rb_entry(node, struct ring_node, rb_node) : NULL;
}
/* return the next node, skipping deleted */
static struct ring_node *ring_rb_next(struct ring_node *rnode)
{
do {
if (rnode)
rnode = ring_rb_entry(rb_next(&rnode->rb_node));
} while (rnode && rnode->deleted);
return rnode;
}
/* return the prev node, skipping deleted */
static struct ring_node *ring_rb_prev(struct ring_node *rnode)
{
do {
if (rnode)
rnode = ring_rb_entry(rb_prev(&rnode->rb_node));
} while (rnode && rnode->deleted);
return rnode;
}
/* return the first node, skipping deleted */
static struct ring_node *ring_rb_first(struct scoutfs_ring_info *ring)
{
struct ring_node *rnode;
rnode = ring_rb_entry(rb_first(&ring->rb_root));
if (rnode && rnode->deleted)
rnode = ring_rb_next(rnode);
return rnode;
}
static struct ring_node *alloc_node(unsigned data_len)
{
struct ring_node *rnode;
rnode = kzalloc(offsetof(struct ring_node, data[data_len]), GFP_NOFS);
if (rnode) {
RB_CLEAR_NODE(&rnode->rb_node);
INIT_LIST_HEAD(&rnode->head);
rnode->data_len = data_len;
}
return rnode;
}
/*
* Insert a new node. This will replace any existing node which could
* be in any state.
*/
void *scoutfs_ring_insert(struct scoutfs_ring_info *ring, void *key,
unsigned data_len)
{
struct ring_node *rnode;
int cmp;
rnode = alloc_node(data_len);
if (!rnode)
return NULL;
ring_rb_walk(ring, key, NULL, rnode, &cmp);
/* just put it on a list, dirtying moves it to dirty */
list_add_tail(&rnode->head, &ring->dirty_list);
mark_node_dirty(ring, rnode, true);
return rnode->data;
}
void *scoutfs_ring_first(struct scoutfs_ring_info *ring)
{
return rnode_data(ring_rb_first(ring));
}
void *scoutfs_ring_lookup(struct scoutfs_ring_info *ring, void *key)
{
struct ring_node *rnode;
int cmp;
rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp);
if (rnode && (cmp || rnode->deleted))
rnode = NULL;
return rnode_data(rnode);
}
void *scoutfs_ring_lookup_next(struct scoutfs_ring_info *ring, void *key)
{
struct ring_node *rnode;
int cmp;
rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp);
if (rnode && (cmp > 1 || rnode->deleted))
rnode = ring_rb_next(rnode);
return rnode_data(rnode);
}
void *scoutfs_ring_lookup_prev(struct scoutfs_ring_info *ring, void *key)
{
struct ring_node *rnode;
int cmp;
rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp);
if (rnode && (cmp < 1 || rnode->deleted))
rnode = ring_rb_prev(rnode);
return rnode_data(rnode);
}
void *scoutfs_ring_next(struct scoutfs_ring_info *ring, void *data)
{
return rnode_data(ring_rb_next(data_rnode(data)));
}
void *scoutfs_ring_prev(struct scoutfs_ring_info *ring, void *data)
{
return rnode_data(ring_rb_prev(data_rnode(data)));
}
/*
* Calculate the most blocks we could have to use to store a given number
* of bytes of entries. At worst each block has a header and leaves one
* less than the max manifest entry unused.
*/
static unsigned most_blocks(unsigned long bytes)
{
unsigned long space;
space = SCOUTFS_BLOCK_SIZE -
sizeof(struct scoutfs_ring_block) -
(sizeof(struct scoutfs_manifest_entry) +
(2 * SCOUTFS_MAX_KEY_SIZE) - 1);
return DIV_ROUND_UP(bytes, space);
}
static u64 wrap_ring_block(struct scoutfs_ring_descriptor *rdesc, u64 block)
{
if (block >= le64_to_cpu(rdesc->total_blocks))
block -= le64_to_cpu(rdesc->total_blocks);
/* XXX callers should have verified on load */
BUG_ON(block >= le64_to_cpu(rdesc->total_blocks));
return block;
}
static u64 calc_first_dirty_block(struct scoutfs_ring_descriptor *rdesc)
{
return wrap_ring_block(rdesc, le64_to_cpu(rdesc->first_block) +
le64_to_cpu(rdesc->nr_blocks));
}
static __le32 rblk_crc(struct scoutfs_ring_block *rblk)
{
unsigned long skip = (char *)(&rblk->crc + 1) - (char *)rblk;
return cpu_to_le32(crc32c(~0, (char *)rblk + skip,
SCOUTFS_BLOCK_SIZE - skip));
}
/*
* This is called after the caller has copied all the dirty nodes into
* blocks in pages for writing. We might be able to dirty a few more
* clean nodes to fill up the end of the last dirty block to keep the
* ring blocks densely populated.
*/
static void fill_last_dirty_block(struct scoutfs_ring_info *ring,
unsigned space)
{
struct ring_node *rnode;
struct ring_node *pos;
unsigned tot;
list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) {
tot = total_entry_bytes(rnode->data_len);
if (tot > space)
break;
mark_node_dirty(ring, rnode, false);
space -= tot;
}
}
void scoutfs_ring_dirty(struct scoutfs_ring_info *ring, void *data)
{
struct ring_node *rnode;
rnode = data_rnode(data);
if (rnode)
mark_node_dirty(ring, rnode, true);
}
/*
* Delete the given node. This can free the node so the caller cannot
* use the data after calling this.
*
* If the node previously existed in the ring then we have to save it and
* write a deletion entry before freeing it.
*/
void scoutfs_ring_delete(struct scoutfs_ring_info *ring, void *data)
{
struct ring_node *rnode = data_rnode(data);
BUG_ON(rnode->deleted);
if (rnode->in_ring) {
rnode->deleted = 1;
mark_node_dirty(ring, rnode, true);
} else {
free_node(ring, rnode);
}
}
static struct scoutfs_ring_block *block_in_pages(struct page **pages,
unsigned i)
{
return page_address(pages[i / SCOUTFS_BLOCKS_PER_PAGE]) +
((i % SCOUTFS_BLOCKS_PER_PAGE) << SCOUTFS_BLOCK_SHIFT);
}
static int load_ring_block(struct scoutfs_ring_info *ring,
struct scoutfs_ring_block *rblk)
{
struct scoutfs_ring_entry *rent;
struct ring_node *rnode;
unsigned data_len;
unsigned i;
int ret = 0;
int cmp;
rent = rblk->entries;
for (i = 0; i < le32_to_cpu(rblk->nr_entries); i++) {
/* XXX verify fields? */
data_len = le16_to_cpu(rent->data_len);
if (rent->flags & SCOUTFS_RING_ENTRY_FLAG_DELETION) {
rnode = ring_rb_walk(ring, NULL, rent->data, NULL,
&cmp);
if (rnode && cmp == 0)
free_node(ring, rnode);
} else {
rnode = alloc_node(data_len);
if (!rnode) {
ret = -ENOMEM;
break;
}
rnode->block = le64_to_cpu(rblk->block);
rnode->in_ring = 1;
memcpy(rnode->data, rent->data, data_len);
ring_rb_walk(ring, NULL, rnode->data, rnode, &cmp);
list_add_tail(&rnode->head, &ring->clean_list);
}
rent = (void *)&rent->data[data_len];
}
return ret;
}
/*
* Read the ring entries into rb nodes with nice large synchronous reads.
*/
#define LOAD_BYTES (4 * 1024 * 1024)
#define LOAD_BLOCKS DIV_ROUND_UP(LOAD_BYTES, SCOUTFS_BLOCK_SIZE)
#define LOAD_PAGES DIV_ROUND_UP(LOAD_BYTES, PAGE_SIZE)
int scoutfs_ring_load(struct super_block *sb, struct scoutfs_ring_info *ring)
{
struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
struct scoutfs_ring_descriptor *rdesc = ring->rdesc;
struct scoutfs_ring_block *rblk;
struct page **pages;
unsigned read_nr;
unsigned i;
__le32 crc;
u64 block;
u64 total;
u64 seq;
u64 nr;
int ret;
pages = kcalloc(LOAD_PAGES, sizeof(struct page *), GFP_NOFS);
if (!pages)
return -ENOMEM;
for (i = 0; i < LOAD_PAGES; i++) {
pages[i] = alloc_page(GFP_NOFS);
if (!pages[i]) {
ret = -ENOMEM;
goto out;
}
}
block = le64_to_cpu(rdesc->first_block);
seq = le64_to_cpu(rdesc->first_seq);
total = le64_to_cpu(rdesc->total_blocks);
nr = le64_to_cpu(rdesc->nr_blocks);
while (nr) {
read_nr = min3(nr, (u64)LOAD_BLOCKS, total - block);
ret = scoutfs_bio_read(sb, pages, le64_to_cpu(rdesc->blkno) +
block, read_nr);
if (ret)
goto out;
for (i = 0; i < read_nr; i++) {
rblk = block_in_pages(pages, i);
crc = rblk_crc(rblk);
if (rblk->fsid != super->hdr.fsid ||
le64_to_cpu(rblk->block) != (block + i) ||
le64_to_cpu(rblk->seq) != (seq + i) ||
rblk->crc != crc) {
ret = -EIO;
goto out;
}
ret = load_ring_block(ring, rblk);
if (ret)
goto out;
}
block = wrap_ring_block(rdesc, block + read_nr);
seq += read_nr;
nr -= read_nr;
}
ret = 0;
out:
for (i = 0; pages && i < LOAD_PAGES && pages[i]; i++)
__free_page(pages[i]);
kfree(pages);
if (ret)
scoutfs_ring_destroy(ring);
return ret;
}
static struct ring_node *first_dirty_node(struct scoutfs_ring_info *ring)
{
return list_first_entry_or_null(&ring->dirty_list, struct ring_node,
head);
}
static struct ring_node *next_dirty_node(struct scoutfs_ring_info *ring,
struct ring_node *rnode)
{
if (rnode->head.next == &ring->dirty_list)
return NULL;
return list_next_entry(rnode, head);
}
static void ring_free_pages(struct scoutfs_ring_info *ring)
{
unsigned i;
if (!ring->pages)
return;
for (i = 0; i < ring->nr_pages; i++) {
if (ring->pages[i])
__free_page(ring->pages[i]);
}
kfree(ring->pages);
ring->pages = NULL;
ring->nr_pages = 0;
}
int scoutfs_ring_has_dirty(struct scoutfs_ring_info *ring)
{
return !!ring->dirty_bytes;
}
int scoutfs_ring_submit_write(struct super_block *sb,
struct scoutfs_ring_info *ring,
struct scoutfs_bio_completion *comp)
{
struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
struct scoutfs_ring_descriptor *rdesc = ring->rdesc;
struct scoutfs_ring_block *rblk;
struct scoutfs_ring_entry *rent;
struct ring_node *rnode;
struct ring_node *next;
struct page **pages;
unsigned nr_blocks;
unsigned nr_pages;
unsigned i;
u64 blkno;
u64 block;
u64 first;
u64 last;
u64 seq;
u64 nr;
u8 *end;
int ret;
if (ring->dirty_bytes == 0)
return 0;
nr_blocks = most_blocks(ring->dirty_bytes);
nr_pages = DIV_ROUND_UP(nr_blocks, SCOUTFS_BLOCKS_PER_PAGE);
ring_free_pages(ring);
pages = kcalloc(nr_pages, sizeof(struct page *), GFP_NOFS);
if (!pages)
return -ENOMEM;
ring->pages = pages;
ring->nr_pages = nr_pages;
for (i = 0; i < nr_pages; i++) {
pages[i] = alloc_page(GFP_NOFS | __GFP_ZERO);
if (!pages[i]) {
ret = -ENOMEM;
goto out;
}
}
block = ring->first_dirty_block;
seq = ring->first_dirty_seq;
rnode = first_dirty_node(ring);
for (i = 0; rnode && i < nr_blocks; i++) {
rblk = block_in_pages(pages, i);
end = (u8 *)rblk + SCOUTFS_BLOCK_SIZE;
rblk->fsid = super->hdr.fsid;
rblk->seq = cpu_to_le64(seq);
rblk->block = cpu_to_le64(block);
rent = rblk->entries;
while (rnode && &rent->data[rnode->data_len] <= end) {
rent->data_len = cpu_to_le16(rnode->data_len);
if (rnode->deleted)
rent->flags = SCOUTFS_RING_ENTRY_FLAG_DELETION;
memcpy(rent->data, rnode->data, rnode->data_len);
le32_add_cpu(&rblk->nr_entries, 1);
rnode->block = block;
rent = (void *)&rent->data[le16_to_cpu(rent->data_len)];
next = next_dirty_node(ring, rnode);
if (!next) {
fill_last_dirty_block(ring, (char *)end -
(char *)rent);
next = next_dirty_node(ring, rnode);
}
rnode = next;
}
rblk->crc = rblk_crc(rblk);
block = wrap_ring_block(rdesc, block + 1);
seq++;
}
/* update the number of blocks we actually filled */
nr_blocks = i;
/* point the descriptor at the new active region of the ring */
rnode = list_first_entry_or_null(&ring->clean_list, struct ring_node,
head);
if (rnode)
first = rnode->block;
else
first = ring->first_dirty_block;
last = wrap_ring_block(rdesc, ring->first_dirty_block + nr_blocks);
if (first < last)
nr = last - first;
else
nr = last + le64_to_cpu(rdesc->total_blocks) - first;
rdesc->first_block = cpu_to_le64(first);
rdesc->first_seq = cpu_to_le64(ring->first_dirty_seq);
rdesc->nr_blocks = cpu_to_le64(nr);
/* the contig dirty blocks in pages might wrap around ring */
blkno = le64_to_cpu(rdesc->blkno) + ring->first_dirty_block;
nr = min_t(u64, nr_blocks,
le64_to_cpu(rdesc->total_blocks) - ring->first_dirty_block);
scoutfs_bio_submit_comp(sb, WRITE, pages, blkno, nr, comp);
if (nr != nr_blocks) {
pages += nr / SCOUTFS_BLOCKS_PER_PAGE;
blkno = le64_to_cpu(rdesc->blkno);
nr = nr_blocks - nr;
scoutfs_bio_submit_comp(sb, WRITE, pages, blkno, nr, comp);
}
ret = 0;
out:
if (ret)
ring_free_pages(ring);
return ret;
}
void scoutfs_ring_write_complete(struct scoutfs_ring_info *ring)
{
struct ring_node *rnode;
struct ring_node *pos;
list_for_each_entry_safe(rnode, pos, &ring->dirty_list, head) {
if (rnode->deleted) {
free_node(ring, rnode);
} else {
mark_node_clean(ring, rnode);
rnode->in_ring = 1;
}
}
ring_free_pages(ring);
ring->dirty_bytes = 0;
ring->first_dirty_block = calc_first_dirty_block(ring->rdesc);
ring->first_dirty_seq = le64_to_cpu(ring->rdesc->first_seq) +
le64_to_cpu(ring->rdesc->nr_blocks);
}
void scoutfs_ring_init(struct scoutfs_ring_info *ring,
struct scoutfs_ring_descriptor *rdesc,
scoutfs_ring_cmp_t compare_key,
scoutfs_ring_cmp_t compare_data)
{
ring->rdesc = rdesc;
ring->compare_key = compare_key;
ring->compare_data = compare_data;
ring->rb_root = RB_ROOT;
INIT_LIST_HEAD(&ring->clean_list);
INIT_LIST_HEAD(&ring->dirty_list);
ring->dirty_bytes = 0;
ring->first_dirty_block = calc_first_dirty_block(rdesc);
ring->first_dirty_seq = le64_to_cpu(rdesc->first_seq) +
le64_to_cpu(rdesc->nr_blocks);
ring->pages = NULL;
ring->nr_pages = 0;
}
void scoutfs_ring_destroy(struct scoutfs_ring_info *ring)
{
struct ring_node *rnode;
struct ring_node *pos;
/* XXX we don't really have a coherent forced dirty unmount story */
WARN_ON_ONCE(!list_empty(&ring->dirty_list));
list_splice_init(&ring->dirty_list, &ring->clean_list);
list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) {
list_del_init(&rnode->head);
kfree(rnode);
}
ring_free_pages(ring);
scoutfs_ring_init(ring, ring->rdesc, ring->compare_key,
ring->compare_data);
}
kmod/src/ring.h
+55
View File
@@ -0,0 +1,55 @@
#ifndef _SCOUTFS_RING_H_
#define _SCOUTFS_RING_H_
struct scoutfs_bio_completion;
typedef int (*scoutfs_ring_cmp_t)(void *a, void *b);
struct scoutfs_ring_info {
struct scoutfs_ring_descriptor *rdesc;
scoutfs_ring_cmp_t compare_key;
scoutfs_ring_cmp_t compare_data;
struct rb_root rb_root;
struct list_head clean_list;
struct list_head dirty_list;
unsigned long dirty_bytes;
u64 first_dirty_block;
u64 first_dirty_seq;
struct page **pages;
unsigned long nr_pages;
};
void scoutfs_ring_init(struct scoutfs_ring_info *ring,
struct scoutfs_ring_descriptor *rdesc,
scoutfs_ring_cmp_t compare_key,
scoutfs_ring_cmp_t compare_data);
int scoutfs_ring_load(struct super_block *sb, struct scoutfs_ring_info *ring);
void *scoutfs_ring_insert(struct scoutfs_ring_info *ring, void *key,
unsigned data_len);
void *scoutfs_ring_first(struct scoutfs_ring_info *ring);
void *scoutfs_ring_lookup(struct scoutfs_ring_info *ring, void *key);
void *scoutfs_ring_lookup_next(struct scoutfs_ring_info *ring, void *key);
void *scoutfs_ring_lookup_prev(struct scoutfs_ring_info *ring, void *key);
void *scoutfs_ring_next(struct scoutfs_ring_info *ring, void *rdata);
void *scoutfs_ring_prev(struct scoutfs_ring_info *ring, void *rdata);
void scoutfs_ring_dirty(struct scoutfs_ring_info *ring, void *rdata);
void scoutfs_ring_delete(struct scoutfs_ring_info *ring, void *rdata);
int scoutfs_ring_has_dirty(struct scoutfs_ring_info *ring);
int scoutfs_ring_submit_write(struct super_block *sb,
struct scoutfs_ring_info *ring,
struct scoutfs_bio_completion *comp);
void scoutfs_ring_write_complete(struct scoutfs_ring_info *ring);
void scoutfs_ring_destroy(struct scoutfs_ring_info *ring);
#endif
kmod/src/super.c
-4
View File
@@ -32,7 +32,6 @@
#include "seg.h"
#include "bio.h"
#include "alloc.h"
#include "treap.h"
#include "compact.h"
#include "data.h"
#include "lock.h"
@@ -218,8 +217,6 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
scoutfs_item_setup(sb) ?:
scoutfs_data_setup(sb) ?:
scoutfs_alloc_setup(sb) ?:
scoutfs_treap_setup(sb) ?:
// scoutfs_buddy_setup(sb) ?:
scoutfs_compact_setup(sb) ?:
scoutfs_setup_trans(sb) ?:
scoutfs_lock_setup(sb) ?:
@@ -265,7 +262,6 @@ static void scoutfs_kill_sb(struct super_block *sb)
scoutfs_item_destroy(sb);
scoutfs_alloc_destroy(sb);
scoutfs_manifest_destroy(sb);
scoutfs_treap_destroy(sb);
scoutfs_seg_destroy(sb);
scoutfs_destroy_counters(sb);
if (sbi->kset)
kmod/src/super.h
-2
View File
@@ -10,7 +10,6 @@ struct scoutfs_counters;
struct item_cache;
struct manifest;
struct segment_cache;
struct treap_info;
struct compact_info;
struct data_info;
struct lock_info;
@@ -27,7 +26,6 @@ struct scoutfs_sb_info {
struct item_cache *item_cache;
struct segment_cache *segment_cache;
struct seg_alloc *seg_alloc;
struct treap_info *treap_info;
struct compact_info *compact_info;
struct data_info *data_info;
kmod/src/trans.c
+5 -4
View File
@@ -25,7 +25,7 @@
#include "manifest.h"
#include "seg.h"
#include "alloc.h"
#include "treap.h"
#include "ring.h"
#include "compact.h"
#include "counters.h"
#include "scoutfs_trace.h"
@@ -115,14 +115,15 @@ void scoutfs_trans_write_func(struct work_struct *work)
}
if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) {
ret = scoutfs_manifest_dirty_ring(sb) ?:
scoutfs_alloc_dirty_ring(sb) ?:
scoutfs_treap_submit_write(sb, &comp) ?:
ret = scoutfs_manifest_submit_write(sb, &comp) ?:
scoutfs_alloc_submit_write(sb, &comp) ?:
scoutfs_bio_wait_comp(sb, &comp) ?:
scoutfs_write_dirty_super(sb);
if (ret)
goto out;
scoutfs_manifest_write_complete(sb);
scoutfs_alloc_write_complete(sb);
advance = true;
}
kmod/src/treap.c
-1349
View File
File diff suppressed because it is too large Load Diff
kmod/src/treap.h
-47
View File
@@ -1,47 +0,0 @@
#ifndef _SCOUTFS_TREAP_H_
#define _SCOUTFS_TREAP_H_
struct scoutfs_bio_completion;
/*
* The runtime root that's used by operations. It's loaded and stored
* from the persistent root in the super block as transactions are written.
*/
struct scoutfs_treap;
struct scoutfs_treap_ops {
int (*compare)(void *key, void *data);
void (*fill)(void *data, void *fill_arg);
bool (*update_aug)(void *parent_data, bool left, void *node_data);
};
struct scoutfs_treap *scoutfs_treap_alloc(struct super_block *sb,
struct scoutfs_treap_ops *ops,
struct scoutfs_treap_root *root);
void scoutfs_treap_free(struct scoutfs_treap *treap);
void *scoutfs_treap_insert(struct scoutfs_treap *treap, void *key, u16 bytes,
void *fill_arg);
int scoutfs_treap_delete(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup_dirty(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup_next(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup_next_dirty(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup_prev(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_lookup_prev_dirty(struct scoutfs_treap *treap, void *key);
void *scoutfs_treap_first(struct scoutfs_treap *treap);
void *scoutfs_treap_last(struct scoutfs_treap *treap);
void *scoutfs_treap_next(struct scoutfs_treap *treap, void *data);
void *scoutfs_treap_prev(struct scoutfs_treap *treap, void *data);
int scoutfs_treap_has_dirty(struct scoutfs_treap *treap);
int scoutfs_treap_dirty_ring(struct scoutfs_treap *treap,
struct scoutfs_treap_root *root);
int scoutfs_treap_submit_write(struct super_block *sb,
struct scoutfs_bio_completion *comp);
int scoutfs_treap_setup(struct super_block *sb);
void scoutfs_treap_destroy(struct super_block *sb);
#endif