diff --git a/kmod/src/forest.c b/kmod/src/forest.c index 2915ac47..f5f259c0 100644 --- a/kmod/src/forest.c +++ b/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(<k, 0, 0); + *vers = 0; + + for (;; scoutfs_key_inc(<k)) { + ret = scoutfs_btree_next(sb, &super->logs_root, <k, &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; diff --git a/kmod/src/forest.h b/kmod/src/forest.h index e6e72a4a..b73ea7a4 100644 --- a/kmod/src/forest.h +++ b/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); diff --git a/kmod/src/format.h b/kmod/src/format.h index d418828b..7be325bc 100644 --- a/kmod/src/format.h +++ b/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; }; diff --git a/kmod/src/item.c b/kmod/src/item.c index c25e74a5..72051828 100644 --- a/kmod/src/item.c +++ b/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: diff --git a/kmod/src/lock.c b/kmod/src/lock.c index 7775b1ad..8f068ed7 100644 --- a/kmod/src/lock.c +++ b/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; diff --git a/kmod/src/lock_server.c b/kmod/src/lock_server.c index ca590635..af01f553 100644 --- a/kmod/src/lock_server.c +++ b/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); diff --git a/kmod/src/lock_server.h b/kmod/src/lock_server.h index c4fe5621..357fd5af 100644 --- a/kmod/src/lock_server.h +++ b/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 diff --git a/kmod/src/server.c b/kmod/src/server.c index 3f5eb4b5..57a7e8d5 100644 --- a/kmod/src/server.c +++ b/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;