scoutfs: recover max lock write_version

Write locks are given an increasing version number as they're granted
which makes its way into items in the log btrees and is used to find the
most recent version of an item.

The initialization of the lock server's next write_version for granted
locks dates back to the initial prototype of the forest of log btrees.
It is only initialized to zero as the module is loaded.  This means that
reloading the module, perhaps by rebooting, resets all the item versions
to 0 and can lead to newly written items being ignored in favour of
older existing items with greater versions from a previous mount.

To fix this we initialize the lock server's write_version to the
greatest of all the versions in items in log btrees.  We add a field to
the log_trees struct which records the greatest version which is
maintained as we write out items in transactions.  These are read by the
server as it starts.

Then lock recovery needs to include the write_version so that the
lock_server can be sure to set the next write_version past the greatest
version in the currently granted locks.

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

kmod/src/forest.c

@@ -481,6 +481,65 @@ out:
 	return ret;
 }
 
+/*
+ * The caller is commiting items in the transaction and has found the
+ * greatest item version amongst them.  We store it in the log_trees root
+ * to send to the server.
+ */
+void scoutfs_forest_set_max_vers(struct super_block *sb, u64 max_vers)
+{
+	DECLARE_FOREST_INFO(sb, finf);
+
+	finf->our_log.max_item_vers = cpu_to_le64(max_vers);
+}
+
+/*
+ * The server is calling during setup to find the greatest item version
+ * amongst all the log tree roots.  They have the authoritative current
+ * super.
+ *
+ * Item versions are only used to compare items in log trees, not in the
+ * main fs tree.  All we have to do is find the greatest version amongst
+ * the log_trees so that new locks will have a write_version greater
+ * than all the items in the log_trees.
+ */
+int scoutfs_forest_get_max_vers(struct super_block *sb,
+				struct scoutfs_super_block *super,
+				u64 *vers)
+{
+	struct scoutfs_log_trees *lt;
+	SCOUTFS_BTREE_ITEM_REF(iref);
+	struct scoutfs_key ltk;
+	int ret;
+
+	scoutfs_key_init_log_trees(&ltk, 0, 0);
+	*vers = 0;
+
+	for (;; scoutfs_key_inc(&ltk)) {
+		ret = scoutfs_btree_next(sb, &super->logs_root, &ltk, &iref);
+		if (ret == 0) {
+			if (iref.val_len == sizeof(struct scoutfs_log_trees)) {
+				ltk = *iref.key;
+				lt = iref.val;
+				*vers = max(*vers,
+					    le64_to_cpu(lt->max_item_vers));
+			} else {
+				ret = -EIO;
+			}
+			scoutfs_btree_put_iref(&iref);
+		}
+		if (ret < 0) {
+			if (ret == -ENOENT)
+				break;
+			goto out;
+		}
+	}
+
+	ret = 0;
+out:
+	return ret;
+}
+
 int scoutfs_forest_insert_list(struct super_block *sb,
 			       struct scoutfs_btree_item_list *lst)
 {
@@ -532,9 +591,11 @@ void scoutfs_forest_init_btrees(struct super_block *sb,
 	memset(&finf->our_log, 0, sizeof(finf->our_log));
 	finf->our_log.item_root = lt->item_root;
 	finf->our_log.bloom_ref = lt->bloom_ref;
+	finf->our_log.max_item_vers = lt->max_item_vers;
 	finf->our_log.rid = lt->rid;
 	finf->our_log.nr = lt->nr;
 	finf->srch_file = lt->srch_file;
+
 	WARN_ON_ONCE(finf->srch_bl); /* commiting should have put the block */
 	finf->srch_bl = NULL;
 
@@ -560,6 +621,7 @@ void scoutfs_forest_get_btrees(struct super_block *sb,
 	lt->item_root = finf->our_log.item_root;
 	lt->bloom_ref = finf->our_log.bloom_ref;
 	lt->srch_file = finf->srch_file;
+	lt->max_item_vers = finf->our_log.max_item_vers;
 
 	scoutfs_block_put(sb, finf->srch_bl);
 	finf->srch_bl = NULL;

kmod/src/forest.h

@@ -23,6 +23,10 @@ int scoutfs_forest_read_items(struct super_block *sb,
 			      scoutfs_forest_item_cb cb, void *arg);
 int scoutfs_forest_set_bloom_bits(struct super_block *sb,
 				  struct scoutfs_lock *lock);
+void scoutfs_forest_set_max_vers(struct super_block *sb, u64 max_vers);
+int scoutfs_forest_get_max_vers(struct super_block *sb,
+				struct scoutfs_super_block *super,
+				u64 *vers);
 int scoutfs_forest_insert_list(struct super_block *sb,
 			       struct scoutfs_btree_item_list *lst);
 int scoutfs_forest_srch_add(struct super_block *sb, u64 hash, u64 ino, u64 id);

kmod/src/format.h

@@ -426,6 +426,7 @@ struct scoutfs_log_trees {
 	struct scoutfs_alloc_root data_avail;
 	struct scoutfs_alloc_root data_freed;
 	struct scoutfs_srch_file srch_file;
+	__le64 max_item_vers;
 	__le64 rid;
 	__le64 nr;
 };

kmod/src/item.c

@@ -2108,6 +2108,7 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 	struct page *page;
 	LIST_HEAD(pages);
 	LIST_HEAD(pos);
+	u64 max_vers = 0;
 	int val_len;
 	int bytes;
 	int off;
@@ -2172,6 +2173,7 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 			val_len = sizeof(item->liv) + item->val_len;
 			bytes = offsetof(struct scoutfs_btree_item_list,
 					 val[val_len]);
+			max_vers = max(max_vers, le64_to_cpu(item->liv.vers));
 
 			if (off + bytes > PAGE_SIZE) {
 				page = second;
@@ -2201,6 +2203,9 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 		read_unlock(&pg->rwlock);
 	}
 
+	/* store max item vers in forest's log_trees */
+	scoutfs_forest_set_max_vers(sb, max_vers);
+
 	/* write all the dirty items into log btree blocks */
 	ret = scoutfs_forest_insert_list(sb, first);
 out:

kmod/src/lock.c

@@ -901,6 +901,7 @@ int scoutfs_lock_recover_request(struct super_block *sb, u64 net_id,
 	for (i = 0; lock && i < SCOUTFS_NET_LOCK_MAX_RECOVER_NR; i++) {
 
 		nlr->locks[i].key = lock->start;
+		nlr->locks[i].write_version = cpu_to_le64(lock->write_version);
 		nlr->locks[i].old_mode = lock->mode;
 		nlr->locks[i].new_mode = lock->mode;
 

kmod/src/lock_server.c

@@ -88,6 +88,8 @@ struct lock_server_info {
 
 	struct scoutfs_alloc *alloc;
 	struct scoutfs_block_writer *wri;
+
+	atomic64_t write_version;
 };
 
 #define DECLARE_LOCK_SERVER_INFO(sb, name) \
@@ -494,7 +496,6 @@ static int process_waiting_requests(struct super_block *sb,
 	struct client_lock_entry *req_tmp;
 	struct client_lock_entry *gr;
 	struct client_lock_entry *gr_tmp;
-	static atomic64_t write_version = ATOMIC64_INIT(0);
 	u64 wv;
 	int ret;
 
@@ -548,7 +549,7 @@ static int process_waiting_requests(struct super_block *sb,
 
 		if (nl.new_mode == SCOUTFS_LOCK_WRITE ||
 		    nl.new_mode == SCOUTFS_LOCK_WRITE_ONLY) {
-			wv = atomic64_inc_return(&write_version);
+			wv = atomic64_inc_return(&inf->write_version);
 			nl.write_version = cpu_to_le64(wv);
 		}
 
@@ -674,6 +675,14 @@ static int finished_recovery(struct super_block *sb, u64 rid, bool cancel)
 	return ret;
 }
 
+static void set_max_write_version(struct lock_server_info *inf, u64 new)
+{
+	u64 old;
+
+	while (new > (old = atomic64_read(&inf->write_version)) &&
+	       (atomic64_cmpxchg(&inf->write_version, old, new) != old));
+}
+
 /*
  * We sent a lock recover request to the client when we received its
  * greeting while in recovery.  Here we instantiate all the locks it
@@ -737,6 +746,10 @@ int scoutfs_lock_server_recover_response(struct super_block *sb, u64 rid,
 		scoutfs_tseq_add(&inf->tseq_tree, &clent->tseq_entry);
 
 		put_server_lock(inf, snode);
+
+		/* make sure next write lock is greater than all recovered */
+		set_max_write_version(inf,
+				le64_to_cpu(nlr->locks[i].write_version));
 	}
 
 	/* send request for next batch of keys */
@@ -956,7 +969,7 @@ static void lock_server_tseq_show(struct seq_file *m,
  */
 int scoutfs_lock_server_setup(struct super_block *sb,
 			      struct scoutfs_alloc *alloc,
-			      struct scoutfs_block_writer *wri)
+			      struct scoutfs_block_writer *wri, u64 max_vers)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
@@ -981,6 +994,7 @@ int scoutfs_lock_server_setup(struct super_block *sb,
 	scoutfs_tseq_tree_init(&inf->tseq_tree, lock_server_tseq_show);
 	inf->alloc = alloc;
 	inf->wri = wri;
+	atomic64_set(&inf->write_version, max_vers); /* inc_return gives +1 */
 
 	inf->tseq_dentry = scoutfs_tseq_create("server_locks", sbi->debug_root,
 					       &inf->tseq_tree);

kmod/src/lock_server.h

@@ -13,7 +13,7 @@ int scoutfs_lock_server_farewell(struct super_block *sb, u64 rid);
 
 int scoutfs_lock_server_setup(struct super_block *sb,
 			      struct scoutfs_alloc *alloc,
-			      struct scoutfs_block_writer *wri);
+			      struct scoutfs_block_writer *wri, u64 max_vers);
 void scoutfs_lock_server_destroy(struct super_block *sb);
 
 #endif

kmod/src/server.c

@@ -37,6 +37,7 @@
 #include "trans.h"
 #include "srch.h"
 #include "alloc.h"
+#include "forest.h"
 
 /*
  * Every active mount can act as the server that listens on a net
@@ -1523,6 +1524,7 @@ static void scoutfs_server_worker(struct work_struct *work)
 	DECLARE_WAIT_QUEUE_HEAD(waitq);
 	struct sockaddr_in sin;
 	LIST_HEAD(conn_list);
+	u64 max_vers;
 	int ret;
 	int err;
 
@@ -1580,7 +1582,14 @@ static void scoutfs_server_worker(struct work_struct *work)
 	    le64_to_cpu(server->meta_avail->total_len))
 		swap(server->meta_avail, server->meta_freed);
 
-	ret = scoutfs_lock_server_setup(sb, &server->alloc, &server->wri);
+	ret = scoutfs_forest_get_max_vers(sb, super, &max_vers);
+	if (ret) {
+		scoutfs_err(sb, "server couldn't find max item vers: %d", ret);
+		goto shutdown;
+	}
+
+	ret = scoutfs_lock_server_setup(sb, &server->alloc, &server->wri,
+					max_vers);
 	if (ret)
 		goto shutdown;