scoutfs: more dirty segment locking

The segment code wasn't always locking around concurrent accesses to the
dirty segment.  This is mostly a problem for updating all the next
elements in skip list modification.  But we also want to serialize dirty
block writing.

Add a little helper function to acquire the dirty mutex when we're
reading from the current dirty segment.

Bring sync in to segment.c so it's clear that it's intimately related to
the dirty segment.

The item deletion hack was totally unlocked.

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

kmod/src/segment.c

@@ -27,7 +27,6 @@
 #include "bloom.h"
 #include "skip.h"
 
-
 /*
  * scoutfs log segments are large multi-block structures that contain
  * key/value items.  This file implements manipulations of the items.
@@ -112,6 +111,26 @@ static int populate_ref(struct super_block *sb, u64 blkno,
 	return 0;
 }
 
+/*
+ * Segments are immutable once they're written.  As they're being
+ * dirtied we need to lock concurrent access.  XXX the dirty blkno test
+ * is probably racey.  We could use reader/writer locks here.  And we
+ * could probably make the skip lists support concurrent access.
+ */
+static bool try_lock_dirty_mutex(struct super_block *sb, u64 blkno)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+
+	if (blkno == sbi->dirty_blkno) {
+		mutex_lock(&sbi->dirty_mutex);
+		if (blkno == sbi->dirty_blkno)
+			return true;
+		mutex_unlock(&sbi->dirty_mutex);
+	}
+
+	return false;
+}
+
 /*
  * Return a reference to the item at the given key.  We walk the manifest
  * to find blocks that might contain the key from most recent to oldest.
@@ -132,6 +151,7 @@ int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key,
 	struct scoutfs_item *item = NULL;
 	struct scoutfs_bloom_bits bits;
 	struct buffer_head *bh;
+	bool locked;
 	int ret;
 
 	/* XXX hold manifest */
@@ -156,8 +176,11 @@ int scoutfs_read_item(struct super_block *sb, struct scoutfs_key *key,
 
 		/* XXX read-ahead all item header blocks */
 
+		locked = try_lock_dirty_mutex(sb, le64_to_cpu(ment.blkno));
 		ret = scoutfs_skip_lookup(sb, le64_to_cpu(ment.blkno), key,
 					  &bh, &item);
+		if (locked)
+			mutex_unlock(&sbi->dirty_mutex);
 		if (ret) {
 			if (ret == -ENOENT)
 				continue;
@@ -311,19 +334,16 @@ static void zero_unused_block(struct super_block *sb, struct buffer_head *bh,
  * Finish off a dirty segment if we have one.  Calculate the checksums of
  * all the blocks, mark them dirty, and drop their pinned reference.
  */
-int scoutfs_finish_dirty_segment(struct super_block *sb)
+static int finish_dirty_segment(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping;
 	struct buffer_head *bh;
-	u64 blkno;
+	u64 blkno = sbi->dirty_blkno;
 	int ret = 0;
 	u64 i;
 
-	/* XXX sync doesn't lock this test? */
-	blkno = sbi->dirty_blkno;
-	if (!blkno)
-		return 0;
+	WARN_ON_ONCE(!blkno);
 
 	for (i = 0; i < SCOUTFS_BLOCKS_PER_CHUNK; i++) {
 		bh = scoutfs_read_block(sb, blkno + i);
@@ -363,6 +383,33 @@ int scoutfs_finish_dirty_segment(struct super_block *sb)
 	return ret;
 }
 
+/*
+ * We've been dirtying log segment blocks and ring blocks as items were
+ * modified.  sync makes sure that they're all persistent and updates
+ * the super.
+ *
+ * XXX need to synchronize with transactions
+ * XXX is state clean after errors?
+ */
+int scoutfs_sync_fs(struct super_block *sb, int wait)
+{
+	struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping;
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	int ret = 0;
+
+	mutex_unlock(&sbi->dirty_mutex);
+	if (sbi->dirty_blkno) {
+		ret = finish_dirty_segment(sb) ?:
+		      scoutfs_finish_dirty_ring(sb) ?:
+		      filemap_write_and_wait(mapping) ?:
+		      scoutfs_write_dirty_super(sb) ?:
+		      scoutfs_advance_dirty_super(sb);
+	}
+	mutex_unlock(&sbi->dirty_mutex);
+	return ret;
+}
+
+
 /*
  * Return a reference to a newly allocated and initialized item in a
  * block in the currently dirty log segment.
@@ -441,7 +488,7 @@ next_chunk:
 	trace_printk("item_off %u val_off %u\n", item_off, val_off);
 
 	if (item_off > val_off) {
-		ret = scoutfs_finish_dirty_segment(sb);
+		ret = finish_dirty_segment(sb);
 		if (ret)
 			goto out;
 		goto next_chunk;
@@ -537,12 +584,18 @@ int scoutfs_delete_item(struct super_block *sb, struct scoutfs_item_ref *ref)
 	u64 blkno;
 	int ret;
 
+	mutex_lock(&sbi->dirty_mutex);
+
 	blkno = round_down(ref->item_bh->b_blocknr, SCOUTFS_BLOCKS_PER_CHUNK);
-	if (WARN_ON_ONCE(blkno != sbi->dirty_blkno))
-		return -EINVAL;
+	if (WARN_ON_ONCE(blkno != sbi->dirty_blkno)) {
+		ret = -EINVAL;
+	} else {
+		ret = scoutfs_skip_delete(sb, blkno, ref->key);
+		WARN_ON_ONCE(ret);
+	}
+
+	mutex_unlock(&sbi->dirty_mutex);
 
-	ret = scoutfs_skip_delete(sb, blkno, ref->key);
-	WARN_ON_ONCE(ret);
 	return ret;
 }
 
@@ -580,10 +633,12 @@ int scoutfs_next_item(struct super_block *sb, struct scoutfs_key *first,
 		      struct scoutfs_key *last, struct list_head *iter_list,
 		      struct scoutfs_item_ref *ref)
 {
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_ring_manifest_entry ment;
 	struct scoutfs_item_iter *least;
 	struct scoutfs_item_iter *iter;
 	struct scoutfs_item_iter *pos;
+	bool locked;
 	int ret;
 
 restart:
@@ -625,6 +680,8 @@ restart:
 	ret = 0;
 	list_for_each_entry_safe(iter, pos, iter_list, list) {
 
+		locked = try_lock_dirty_mutex(sb, iter->blkno);
+
 		/* search towards the first key if we haven't yet */
 		if (!iter->item) {
 			ret = scoutfs_skip_search(sb, iter->blkno, first,
@@ -637,6 +694,9 @@ restart:
 						&iter->bh, &iter->item);
 		}
 
+		if (locked)
+			mutex_unlock(&sbi->dirty_mutex);
+
 		/* we're done with this block if we past the last key */
 		while (!ret && scoutfs_key_cmp(&iter->item->key, last) > 0) {
 			brelse(iter->bh);

kmod/src/segment.h

@@ -29,7 +29,7 @@ int scoutfs_next_item(struct super_block *sb, struct scoutfs_key *first,
 		      struct scoutfs_key *last, struct list_head *iter_list,
 		      struct scoutfs_item_ref *ref);
 
-int scoutfs_finish_dirty_segment(struct super_block *sb);
+int scoutfs_sync_fs(struct super_block *sb, int wait);
 
 
 #endif

kmod/src/super.c

@@ -29,25 +29,6 @@
 #include "segment.h"
 #include "scoutfs_trace.h"
 
-/*
- * We've been dirtying log segment blocks and ring blocks as items were
- * modified.  sync makes sure that they're all persistent and updates
- * the super.
- *
- * XXX need to synchronize with transactions
- * XXX is state clean after errors?
- */
-static int scoutfs_sync_fs(struct super_block *sb, int wait)
-{
-	struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping;
-
-	return scoutfs_finish_dirty_segment(sb) ?:
-	       scoutfs_finish_dirty_ring(sb) ?:
-	       filemap_write_and_wait(mapping) ?:
-	       scoutfs_write_dirty_super(sb) ?:
-	       scoutfs_advance_dirty_super(sb);
-}
-
 static const struct super_operations scoutfs_super_ops = {
 	.alloc_inode = scoutfs_alloc_inode,
 	.destroy_inode = scoutfs_destroy_inode,