scoutfs: add basic segment reading

Add the most basic ability to read items from log segment blocks.  If
an item isn't in the cache then we walk segments in the manifest and
check for the item in each one.

This is just the core fundamental code.  There's still a lot to do:
basic corruption validation, multi-block segments, bloom filters and
arrays to optimize segment misses, and some day the ability to read file
data items directly into page cache pages.  The manifest locking is also
super broken.

But this is enough to let us mount and stat the root inode!

Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/Makefile

@@ -1,4 +1,4 @@
 obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o
  
 scoutfs-y += block.o chunk.o crc.o dir.o inode.o item.o manifest.o msg.o \
-	     ring.o super.o
+	     ring.o segment.o super.o

kmod/src/item.c

@@ -19,6 +19,7 @@
 #include "super.h"
 #include "key.h"
 #include "item.h"
+#include "segment.h"
 
 /*
  * describe:
@@ -159,15 +160,9 @@ static struct scoutfs_item *alloc_item(struct scoutfs_key *key,
 	return item;
 }
 
-/*
- * Create a new item stored at the given key.  Return it with a reference.
- * return an ERR_PTR with ENOMEM or EEXIST.
- *
- * The caller is responsible for initializing the item's value.
- */
-struct scoutfs_item *scoutfs_item_create(struct super_block *sb,
-					 struct scoutfs_key *key,
-					 unsigned int val_len)
+static struct scoutfs_item *create_item(struct super_block *sb,
+					struct scoutfs_key *key,
+					unsigned int val_len, bool dirty)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_item *existing;
@@ -183,8 +178,12 @@ struct scoutfs_item *scoutfs_item_create(struct super_block *sb,
 	existing = find_item(sb, &sbi->item_root, key, 0);
 	if (!existing) {
 		insert_item(sb, &sbi->item_root, item);
-		insert_item(sb, &sbi->dirty_item_root, item);
-		atomic_add(2, &item->refcount);
+		atomic_inc(&item->refcount);
+		if (dirty) {
+			insert_item(sb, &sbi->dirty_item_root, item);
+			atomic_inc(&item->refcount);
+		}
+
 	}
 	spin_unlock_irqrestore(&sbi->item_lock, flags);
 
@@ -198,6 +197,30 @@ struct scoutfs_item *scoutfs_item_create(struct super_block *sb,
 	return item;
 }
 
+/*
+ * Create a new item stored at the given key.  Return it with a reference.
+ * return an ERR_PTR with ENOMEM or EEXIST.
+ *
+ * The caller is responsible for initializing the item's value.
+ */
+struct scoutfs_item *scoutfs_item_create(struct super_block *sb,
+					 struct scoutfs_key *key,
+					 unsigned int val_len)
+{
+	return create_item(sb, key, val_len, true);
+}
+
+/*
+ * Allocate a new clean item in the cache for the caller to fill.  If the
+ * item already exists then -EEXIST is returned.
+ */
+struct scoutfs_item *scoutfs_clean_item(struct super_block *sb,
+				        struct scoutfs_key *key,
+				        unsigned int val_len)
+{
+	return create_item(sb, key, val_len, false);
+}
+
 /*
  * The caller is still responsible for unlocking and putting the item.
  *
@@ -230,6 +253,10 @@ void scoutfs_item_delete(struct super_block *sb, struct scoutfs_item *item)
 	spin_unlock_irqrestore(&sbi->item_lock, flags);
 }
 
+/*
+ * Find an item in the cache.  If it isn't present then we try to read
+ * it from log segements.
+ */
 static struct scoutfs_item *item_lookup(struct super_block *sb,
 					struct scoutfs_key *key, int np)
 {
@@ -237,15 +264,17 @@ static struct scoutfs_item *item_lookup(struct super_block *sb,
 	struct scoutfs_item *item;
 	unsigned long flags;
 
-	spin_lock_irqsave(&sbi->item_lock, flags);
+	do {
+		spin_lock_irqsave(&sbi->item_lock, flags);
 
-	item = find_item(sb, &sbi->item_root, key, np);
-	if (item)
-		atomic_inc(&item->refcount);
-	else
-		item = ERR_PTR(-ENOENT);
+		item = find_item(sb, &sbi->item_root, key, np);
+		if (item)
+			atomic_inc(&item->refcount);
 
-	spin_unlock_irqrestore(&sbi->item_lock, flags);
+		spin_unlock_irqrestore(&sbi->item_lock, flags);
+		if (!item)
+			item = scoutfs_read_segment_item(sb, key);
+	} while (item == ERR_PTR(-EEXIST));
 
 	return item;
 }

kmod/src/item.h

@@ -19,6 +19,9 @@ struct scoutfs_item {
 struct scoutfs_item *scoutfs_item_create(struct super_block *sb,
 					 struct scoutfs_key *key,
 					 unsigned int val_len);
+struct scoutfs_item *scoutfs_clean_item(struct super_block *sb,
+				        struct scoutfs_key *key,
+				        unsigned int val_len);
 struct scoutfs_item *scoutfs_item_lookup(struct super_block *sb,
 					 struct scoutfs_key *key);
 struct scoutfs_item *scoutfs_item_next(struct super_block *sb,

kmod/src/key.h

@@ -31,6 +31,27 @@ static inline int scoutfs_key_cmp(struct scoutfs_key *a, struct scoutfs_key *b)
 	       le64_cmp(a->offset, b->offset);
 }
 
+/*
+ * return -ve, 0, +ve if the key is less than, contained within, or greater
+ * than the given range of keys.
+ */
+static inline int scoutfs_key_cmp_range(struct scoutfs_key *key,
+					struct scoutfs_key *first,
+					struct scoutfs_key *last)
+{
+	int cmp;
+
+	WARN_ON_ONCE(scoutfs_key_cmp(first, last) > 0);
+
+	cmp = scoutfs_key_cmp(key, first);
+	if (cmp > 0) {
+		cmp = scoutfs_key_cmp(key, last);
+		if (cmp < 0)
+			cmp = 0;
+	}
+	return cmp;
+}
+
 	
 static inline void scoutfs_set_key(struct scoutfs_key *key, u64 inode, u8 type,
 				   u64 offset)

kmod/src/manifest.c

@@ -74,6 +74,29 @@ static void insert_mnode(struct rb_root *root,
 	rb_insert_color(&ins->node, root);
 }
 
+static struct scoutfs_manifest_node *find_mnode(struct rb_root *root,
+						struct scoutfs_key *key)
+{
+	struct rb_node *node = root->rb_node;
+	struct scoutfs_manifest_node *mnode;
+	int cmp;
+
+	while (node) {
+		mnode = rb_entry(node, struct scoutfs_manifest_node, node);
+
+		cmp = scoutfs_key_cmp_range(key, &mnode->ment.first,
+					    &mnode->ment.last);
+		if (cmp < 0)
+			node = node->rb_left;
+		else if (cmp > 0)
+			node = node->rb_right;
+		else
+			return mnode;
+	}
+
+	return NULL;
+}
+
 static struct scoutfs_manifest_node *delete_mnode(struct scoutfs_manifest *mani,
 						  u64 blkno)
 
@@ -151,6 +174,69 @@ int scoutfs_add_manifest(struct super_block *sb,
 	return 0;
 }
 
+/*
+ * Fill the caller's ment with the next log segment in the manifest that
+ * might contain the given key.  The ment is initialized to 0 to return
+ * the first entry.
+ *
+ * This can return multiple log segments from level 0 in decreasing age.
+ * Then it can return at most one log segment in each level that
+ * intersects with the given key.
+ */
+bool scoutfs_next_manifest_segment(struct super_block *sb,
+				   struct scoutfs_key *key,
+				   struct scoutfs_ring_manifest_entry *ment)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_manifest *mani = sbi->mani;
+	struct scoutfs_manifest_node *mnode;
+	bool found = false;
+	int i;
+
+	if (ment->level >= SCOUTFS_MAX_LEVEL)
+		return false;
+
+	spin_lock(&mani->lock);
+
+	if (ment->level == 0) {
+		if (ment->blkno) {
+			mnode = radix_tree_lookup(&mani->blkno_radix,
+						  le64_to_cpu(ment->blkno));
+			mnode = list_next_entry(mnode, head);
+		} else {
+			mnode = list_first_entry(&mani->level_zero,
+						 struct scoutfs_manifest_node,
+						 head);
+		}
+
+		list_for_each_entry_from(mnode, &mani->level_zero, head) {
+			if (scoutfs_key_cmp_range(key, &mnode->ment.first,
+						  &mnode->ment.last) == 0) {
+				*ment = mnode->ment;
+				found = true;
+				break;
+			}
+		}
+	}
+
+	if (!found) {
+		for (i = ment->level + 1; i <= SCOUTFS_MAX_LEVEL; i++) {
+			mnode = find_mnode(&mani->levels[i].root, key);
+			if (mnode) {
+				*ment = mnode->ment;
+				found = true;
+				break;
+			}
+		}
+		if (!found)
+			ment->level = SCOUTFS_MAX_LEVEL;
+	}
+
+	spin_unlock(&mani->lock);
+
+	return found;
+}
+
 int scoutfs_setup_manifest(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);

kmod/src/manifest.h

@@ -8,4 +8,8 @@ int scoutfs_add_manifest(struct super_block *sb,
 		         struct scoutfs_ring_manifest_entry *ment);
 void scoutfs_delete_manifest(struct super_block *sb, u64 blkno);
 
+bool scoutfs_next_manifest_segment(struct super_block *sb,
+				   struct scoutfs_key *key,
+				   struct scoutfs_ring_manifest_entry *ment);
+
 #endif

kmod/src/segment.c

@@ -0,0 +1,103 @@
+/*
+ * Copyright (C) 2016 Versity Software, Inc.  All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License v2 as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+#include <linux/kernel.h>
+#include <linux/buffer_head.h>
+#include <linux/fs.h>
+#include <linux/spinlock.h>
+#include <linux/rbtree.h>
+#include <linux/slab.h>
+
+#include "super.h"
+#include "key.h"
+#include "item.h"
+#include "segment.h"
+#include "manifest.h"
+#include "block.h"
+
+static struct scoutfs_item_header *next_ihdr(struct scoutfs_item_header *ihdr)
+{
+	return (void *)(ihdr + 1) + le16_to_cpu(ihdr->len);
+}
+
+/*
+ * Use the manifest to search log segments for the most recent version
+ * of the item with the given key.  Return a reference to the item after
+ * it's been added to the item cache.
+ */
+struct scoutfs_item *scoutfs_read_segment_item(struct super_block *sb,
+					       struct scoutfs_key *key)
+{
+	struct scoutfs_ring_manifest_entry ment;
+	struct scoutfs_item_header *ihdr;
+	struct scoutfs_item_block *iblk;
+	struct scoutfs_item *item;
+	struct buffer_head *bh;
+	int cmp;
+	int err;
+	int i;
+
+	/* XXX hold manifest */
+
+	memset(&ment, 0, sizeof(struct scoutfs_ring_manifest_entry));
+
+	item = NULL;
+	err = -ENOENT;
+	while (scoutfs_next_manifest_segment(sb, key, &ment)) {
+
+		bh = scoutfs_read_block(sb, le64_to_cpu(ment.blkno));
+		if (!bh) {
+			err = -EIO;
+			break;
+		}
+
+		iblk = (void *)bh->b_data;
+		/* XXX seq corruption */
+
+		ihdr = (void *)(iblk + 1);
+
+		/* XXX test bloom filter blocks */
+		/* XXX binary search of key array */
+		/* XXX could populate more from granted range */
+
+		for (i = 0; i < le32_to_cpu(iblk->nr_items);
+		     i++, ihdr = next_ihdr(ihdr)) {
+			cmp = scoutfs_key_cmp(key, &ihdr->key);
+			if (cmp > 0)
+				continue;
+			if (cmp < 0)
+				break;
+
+			item = scoutfs_clean_item(sb, key,
+						  le16_to_cpu(ihdr->len));
+			if (IS_ERR(item)) {
+				err = PTR_ERR(item);
+			} else {
+				memcpy(item->val, (void *)(ihdr + 1),
+				       item->val_len);
+				err = 0;
+			}
+			break;
+		}
+
+		brelse(bh);
+		if (item) /* also breaks for IS_ERR */
+			break;
+	}
+
+	/* XXX release manifest */
+
+	if (err)
+		item = ERR_PTR(err);
+
+	return item;
+}

kmod/src/segment.h

@@ -0,0 +1,7 @@
+#ifndef _SCOUTFS_SEGMENT_H_
+#define _SCOUTFS_SEGMENT_H_
+
+struct scoutfs_item *scoutfs_read_segment_item(struct super_block *sb,
+					       struct scoutfs_key *key);
+
+#endif