diff --git a/kmod/src/compact.c b/kmod/src/compact.c index cae9d7f6..4a9792c5 100644 --- a/kmod/src/compact.c +++ b/kmod/src/compact.c @@ -24,6 +24,7 @@ #include "manifest.h" #include "trans.h" #include "counters.h" +#include "alloc.h" #include "scoutfs_trace.h" /* @@ -347,6 +348,7 @@ static int compact_segments(struct super_block *sb, struct compact_seg *lower; u32 key_bytes; u32 nr_items; + u64 segno; int ret; scoutfs_inc_counter(sb, compact_operations); @@ -444,12 +446,19 @@ static int compact_segments(struct super_block *sb, break; } - ret = scoutfs_seg_alloc(sb, &seg); + ret = scoutfs_alloc_segno(sb, &segno); if (ret) { kfree(cseg); break; } + ret = scoutfs_seg_alloc(sb, segno, &seg); + if (ret) { + scoutfs_alloc_free(sb, segno); + kfree(cseg); + break; + } + /* csegs will be claned up once they're on the list */ cseg->level = curs->lower_level; cseg->seg = seg; diff --git a/kmod/src/format.h b/kmod/src/format.h index fd361c03..5d3c184b 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -389,6 +389,8 @@ enum { SCOUTFS_NET_TRADE_TIME = 0, SCOUTFS_NET_ALLOC_INODES, SCOUTFS_NET_MANIFEST_RANGE_ENTRIES, + SCOUTFS_NET_ALLOC_SEGNO, + SCOUTFS_NET_RECORD_SEGMENT, SCOUTFS_NET_UNKNOWN, }; diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index bdd2f641..869cc60f 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -210,6 +210,45 @@ int scoutfs_manifest_add(struct super_block *sb, return 0; } +/* + * Add a manifest entry as provided by the caller instead of exploded + * out into arguments. + * + * This must be called with the manifest lock held. + */ +int scoutfs_manifest_add_ment(struct super_block *sb, + struct scoutfs_manifest_entry *add) +{ + 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; + struct scoutfs_key_buf first; + unsigned bytes; + + lockdep_assert_held(&mani->rwsem); + + init_ment_keys(add, &first, NULL); + + skey.key = &first; + skey.level = add->level; + skey.seq = le64_to_cpu(add->seq); + + bytes = scoutfs_manifest_bytes(add); + + ment = scoutfs_ring_insert(&mani->ring, &skey, bytes); + if (!ment) + return -ENOMEM; + + memcpy(ment, add, bytes); + + mani->nr_levels = max_t(u8, mani->nr_levels, add->level + 1); + add_level_count(sb, mani, super, add->level, 1); + + return 0; +} + /* * This must be called with the manifest lock held. */ @@ -268,6 +307,41 @@ int scoutfs_manifest_bytes(struct scoutfs_manifest_entry *ment) le16_to_cpu(ment->last_key_len); } +/* + * Return an allocated and filled in manifest entry. + */ +struct scoutfs_manifest_entry * +scoutfs_manifest_alloc_entry(struct super_block *sb, + struct scoutfs_key_buf *first, + struct scoutfs_key_buf *last, u64 segno, u64 seq, + u8 level) +{ + struct scoutfs_manifest_entry *ment; + struct scoutfs_key_buf ment_first; + struct scoutfs_key_buf ment_last; + unsigned key_bytes; + unsigned bytes; + + key_bytes = first->key_len + last->key_len; + bytes = offsetof(struct scoutfs_manifest_entry, keys[key_bytes]); + + ment = kmalloc(bytes, GFP_NOFS); + if (!ment) + return NULL; + + 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); + + return ment; +} + /* * XXX This feels pretty gross, but it's a simple way to give compaction * atomic updates. It'll go away once compactions go to the trouble of diff --git a/kmod/src/manifest.h b/kmod/src/manifest.h index 64936e59..e0547c26 100644 --- a/kmod/src/manifest.h +++ b/kmod/src/manifest.h @@ -8,6 +8,8 @@ int scoutfs_manifest_add(struct super_block *sb, struct scoutfs_key_buf *first, struct scoutfs_key_buf *last, u64 segno, u64 seq, u8 level); +int scoutfs_manifest_add_ment(struct super_block *sb, + struct scoutfs_manifest_entry *add); int scoutfs_manifest_dirty(struct super_block *sb, struct scoutfs_key_buf *first, u64 seq, u8 level); int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *first, @@ -19,6 +21,12 @@ void scoutfs_manifest_write_complete(struct super_block *sb); int scoutfs_manifest_bytes(struct scoutfs_manifest_entry *ment); +struct scoutfs_manifest_entry * +scoutfs_manifest_alloc_entry(struct super_block *sb, + struct scoutfs_key_buf *first, + struct scoutfs_key_buf *last, u64 segno, u64 seq, + u8 level); + int scoutfs_manifest_lock(struct super_block *sb); int scoutfs_manifest_unlock(struct super_block *sb); diff --git a/kmod/src/net.c b/kmod/src/net.c index 85143934..bbe081cf 100644 --- a/kmod/src/net.c +++ b/kmod/src/net.c @@ -24,6 +24,9 @@ #include "counters.h" #include "inode.h" #include "manifest.h" +#include "bio.h" +#include "alloc.h" +#include "seg.h" #include "scoutfs_trace.h" /* @@ -71,6 +74,11 @@ struct net_info { struct delayed_work server_work; struct sock_info *listening_sinf; + /* server commits ring changes while processing requests */ + struct rw_semaphore ring_commit_rwsem; + struct llist_head ring_commit_waiters; + struct work_struct ring_commit_work; + /* both track active sockets for destruction */ struct list_head active_socks; @@ -243,6 +251,102 @@ static void scoutfs_net_send_func(struct work_struct *work) mutex_unlock(&nti->mutex); } +struct commit_waiter { + struct completion comp; + struct llist_node node; + int ret; +}; + +/* + * This is called while still holding the rwsem that prevents commits so + * that the caller can be sure to be woken by the next commit after they + * queue and release the lock. + * + * This could queue delayed work but we're first trying to have batching + * work by having concurrent modification line up behind a commit in + * flight. Once the commit finishes it'll unlock and hopefully everyone + * will race to make their changes and they'll all be applied by the + * next commit after that. + */ +static void queue_commit_work(struct net_info *nti, struct commit_waiter *cw) +{ + lockdep_assert_held(&nti->ring_commit_rwsem); + + cw->ret = 0; + init_completion(&cw->comp); + llist_add(&cw->node, &nti->ring_commit_waiters); + queue_work(nti->proc_wq, &nti->ring_commit_work); +} + +static int wait_for_commit(struct commit_waiter *cw) +{ + wait_for_completion(&cw->comp); + return cw->ret; +} + +/* + * A core function of request processing is to modify the manifest and + * allocator. Often the processing needs to make the modifications + * persistent before replying. We'd like to batch these commits as much + * as is reasonable so that we don't degrade to a few IO round trips per + * request. + * + * Getting that batching right is bound up in the concurrency of request + * processing so a clear way to implement the batched commits is to + * implement commits with work funcs like the processing. This ring + * commit work is queued on the non-reentrant proc_wq so there will only + * ever be one commit executing at a time. + * + * Processing paths acquire the rwsem for reading while they're making + * multiple dependent changes. When they're done and want it persistent + * they add themselves to the list of waiters and queue the commit work. + * This work runs, acquires the lock to exclude other writers, and + * performs the commit. Readers can run concurrently with these + * commits. + */ +static void scoutfs_net_ring_commit_func(struct work_struct *work) +{ + struct net_info *nti = container_of(work, struct net_info, + ring_commit_work); + struct super_block *sb = nti->sb; + struct scoutfs_bio_completion comp; + struct commit_waiter *cw; + struct commit_waiter *pos; + struct llist_node *node; + int ret; + + scoutfs_bio_init_comp(&comp); + + down_write(&nti->ring_commit_rwsem); + + if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) { + ret = scoutfs_manifest_submit_write(sb, &comp) ?: + scoutfs_alloc_submit_write(sb, &comp) ?: + scoutfs_bio_wait_comp(sb, &comp) ?: + scoutfs_write_dirty_super(sb); + + /* we'd need to loop or something */ + BUG_ON(ret); + + scoutfs_manifest_write_complete(sb); + scoutfs_alloc_write_complete(sb); + + scoutfs_advance_dirty_super(sb); + } else { + ret = 0; + } + + node = llist_del_all(&nti->ring_commit_waiters); + + /* waiters always wait on completion, cw could be free after complete */ + llist_for_each_entry_safe(cw, pos, node, node) { + cw->ret = ret; + complete(&cw->comp); + } + + up_write(&nti->ring_commit_rwsem); +} + static struct send_buf *alloc_sbuf(unsigned data_len) { unsigned len = offsetof(struct send_buf, nh[0].data[data_len]); @@ -257,6 +361,98 @@ static struct send_buf *alloc_sbuf(unsigned data_len) return sbuf; } +static struct send_buf *process_record_segment(struct super_block *sb, + void *req, int req_len) +{ + DECLARE_NET_INFO(sb, nti); + struct scoutfs_manifest_entry *ment; + struct commit_waiter cw; + struct send_buf *sbuf; + int ret; + + if (req_len < sizeof(struct scoutfs_manifest_entry)) { + sbuf = ERR_PTR(-EINVAL); + goto out; + } + + ment = req; + + if (req_len != scoutfs_manifest_bytes(ment)) { + sbuf = ERR_PTR(-EINVAL); + goto out; + } + + down_read(&nti->ring_commit_rwsem); + + scoutfs_manifest_lock(sb); + ret = scoutfs_manifest_add_ment(sb, ment); + scoutfs_manifest_unlock(sb); + + if (ret == 0) + queue_commit_work(nti, &cw); + up_read(&nti->ring_commit_rwsem); + + if (ret == 0) + ret = wait_for_commit(&cw); + + sbuf = alloc_sbuf(0); + if (!sbuf) { + sbuf = ERR_PTR(-ENOMEM); + goto out; + } + + if (ret) + sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR; + else + sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS; +out: + return sbuf; +} + +static struct send_buf *process_alloc_segno(struct super_block *sb, + void *req, int req_len) +{ + DECLARE_NET_INFO(sb, nti); + __le64 * __packed lesegno; + struct commit_waiter cw; + struct send_buf *sbuf; + u64 segno; + int ret; + + if (req_len != 0) { + sbuf = ERR_PTR(-EINVAL); + goto out; + } + + down_read(&nti->ring_commit_rwsem); + + ret = scoutfs_alloc_segno(sb, &segno); + if (ret == 0) + queue_commit_work(nti, &cw); + + up_read(&nti->ring_commit_rwsem); + + if (ret == 0) + ret = wait_for_commit(&cw); + + sbuf = alloc_sbuf(sizeof(__le64)); + if (!sbuf) { + sbuf = ERR_PTR(-ENOMEM); + goto out; + } + + if (ret) { + sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR; + } else { + lesegno = (void *)sbuf->nh->data; + *lesegno = cpu_to_le64(segno); + sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS; + } + +out: + return sbuf; +} + /* * Find the manifest entries that intersect with the request's key * range. We lock the manifest and get pointers to the manifest entries @@ -330,9 +526,11 @@ out: static struct send_buf *process_alloc_inodes(struct super_block *sb, void *req, int req_len) { + DECLARE_NET_INFO(sb, nti); struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct scoutfs_net_inode_alloc *ial; + struct commit_waiter cw; struct send_buf *sbuf; int ret; u64 ino; @@ -345,22 +543,27 @@ static struct send_buf *process_alloc_inodes(struct super_block *sb, if (!sbuf) return ERR_PTR(-ENOMEM); - spin_lock(&sbi->next_ino_lock); + down_read(&nti->ring_commit_rwsem); + spin_lock(&sbi->next_ino_lock); ino = le64_to_cpu(super->next_ino); nr = min(100000ULL, ~0ULL - ino); le64_add_cpu(&super->next_ino, nr); - spin_unlock(&sbi->next_ino_lock); - /* XXX think about server ring commits */ - ret = 0; //sync_or_something(); + queue_commit_work(nti, &cw); + up_read(&nti->ring_commit_rwsem); + + ret = wait_for_commit(&cw); ial = (void *)sbuf->nh->data; ial->ino = cpu_to_le64(ino); ial->nr = cpu_to_le64(nr); - sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS; + if (ret < 0) + sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR; + else + sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS; return sbuf; } @@ -369,9 +572,9 @@ static struct send_buf *process_alloc_inodes(struct super_block *sb, * Log the time in the request and reply with our current time. */ static struct send_buf *process_trade_time(struct super_block *sb, - struct scoutfs_timespec *req, - int req_len) + void *r, int req_len) { + struct scoutfs_timespec *req = r; struct scoutfs_timespec *reply; struct send_buf *sbuf; struct timespec64 ts; @@ -397,6 +600,23 @@ static struct send_buf *process_trade_time(struct super_block *sb, return sbuf; } +typedef struct send_buf *(*proc_func_t)(struct super_block *sb, void *req, + int req_len); + +static proc_func_t type_proc_func(u8 type) +{ + static proc_func_t funcs[] = { + [SCOUTFS_NET_TRADE_TIME] = process_trade_time, + [SCOUTFS_NET_ALLOC_INODES] = process_alloc_inodes, + [SCOUTFS_NET_MANIFEST_RANGE_ENTRIES] = + process_manifest_range_entries, + [SCOUTFS_NET_ALLOC_SEGNO] = process_alloc_segno, + [SCOUTFS_NET_RECORD_SEGMENT] = process_record_segment, + }; + + return type < SCOUTFS_NET_UNKNOWN ? funcs[type] : NULL; +} + /* * Process an incoming request and queue its reply to send if the socket * is still open by the time we have the reply. @@ -405,23 +625,15 @@ static int process_request(struct net_info *nti, struct recv_buf *rbuf) { struct super_block *sb = nti->sb; struct send_buf *sbuf; + proc_func_t proc; unsigned data_len; data_len = le16_to_cpu(rbuf->nh->data_len); - - if (rbuf->nh->type == SCOUTFS_NET_TRADE_TIME) - sbuf = process_trade_time(sb, (void *)rbuf->nh->data, - data_len); - else if (rbuf->nh->type == SCOUTFS_NET_ALLOC_INODES) - sbuf = process_alloc_inodes(sb, (void *)rbuf->nh->data, - data_len); - else if (rbuf->nh->type == SCOUTFS_NET_MANIFEST_RANGE_ENTRIES) - sbuf = process_manifest_range_entries(sb, - (void *)rbuf->nh->data, - data_len); + proc = type_proc_func(rbuf->nh->type); + if (proc) + sbuf = proc(sb, (void *)rbuf->nh->data, data_len); else sbuf = ERR_PTR(-EINVAL); - if (IS_ERR(sbuf)) return PTR_ERR(sbuf); @@ -844,6 +1056,93 @@ static int add_send_buf(struct super_block *sb, int type, void *data, return 0; } +struct record_segment_args { + struct completion comp; + int ret; +}; + +static int record_segment_reply(struct super_block *sb, void *reply, int ret, + void *arg) +{ + struct record_segment_args *args = arg; + + if (ret > 0) + ret = -EINVAL; + + args->ret = ret; + complete(&args->comp); + return args->ret; +} + +int scoutfs_net_record_segment(struct super_block *sb, + struct scoutfs_segment *seg, u8 level) +{ + struct scoutfs_manifest_entry *ment; + struct record_segment_args args; + int ret; + + ment = scoutfs_seg_manifest_entry(sb, seg, level); + if (!ment) { + ret = -ENOMEM; + goto out; + } + + init_completion(&args.comp); + + ret = add_send_buf(sb, SCOUTFS_NET_RECORD_SEGMENT, ment, + scoutfs_manifest_bytes(ment), + record_segment_reply, &args); + kfree(ment); + if (ret == 0) { + wait_for_completion(&args.comp); + ret = args.ret; + } +out: + return ret; +} + +struct alloc_segno_args { + u64 segno; + struct completion comp; + int ret; +}; + +static int alloc_segno_reply(struct super_block *sb, void *reply, int ret, + void *arg) +{ + struct alloc_segno_args *args = arg; + __le64 * __packed segno = reply; + + if (ret == sizeof(__le64)) { + args->segno = le64_to_cpup(segno); + args->ret = 0; + } else { + args->ret = -EINVAL; + } + + complete(&args->comp); /* args can be freed from this point */ + return args->ret; +} + +int scoutfs_net_alloc_segno(struct super_block *sb, u64 *segno) +{ + struct alloc_segno_args args; + int ret; + + init_completion(&args.comp); + + ret = add_send_buf(sb, SCOUTFS_NET_ALLOC_SEGNO, NULL, 0, + alloc_segno_reply, &args); + if (ret == 0) { + wait_for_completion(&args.comp); + *segno = args.segno; + ret = args.ret; + if (ret == 0 && *segno == 0) + ret = -ENOSPC; + } + return ret; +} + struct manifest_range_entries_args { struct list_head *list; struct completion comp; @@ -1337,6 +1636,9 @@ int scoutfs_net_setup(struct super_block *sb) INIT_LIST_HEAD(&nti->to_send); nti->next_id = 1; INIT_DELAYED_WORK(&nti->server_work, scoutfs_net_server_func); + init_rwsem(&nti->ring_commit_rwsem); + init_llist_head(&nti->ring_commit_waiters); + INIT_WORK(&nti->ring_commit_work, scoutfs_net_ring_commit_func); INIT_LIST_HEAD(&nti->active_socks); sbi->net_info = nti; @@ -1383,8 +1685,8 @@ void scoutfs_net_destroy(struct super_block *sb) mutex_unlock(&nti->mutex); drain_workqueue(nti->sock_wq); - /* wait for processing to finish and free rbufs */ - flush_workqueue(nti->proc_wq); + /* wait for processing (and commits) to finish and free rbufs */ + drain_workqueue(nti->proc_wq); /* make sure client/server work isn't queued */ cancel_delayed_work_sync(&nti->server_work); diff --git a/kmod/src/net.h b/kmod/src/net.h index aee93bfd..973b0530 100644 --- a/kmod/src/net.h +++ b/kmod/src/net.h @@ -2,6 +2,7 @@ #define _SCOUTFS_NET_H_ struct scoutfs_key_buf; +struct scoutfs_segment; int scoutfs_net_trade_time(struct super_block *sb); int scoutfs_net_alloc_inodes(struct super_block *sb); @@ -9,6 +10,9 @@ int scoutfs_net_manifest_range_entries(struct super_block *sb, struct scoutfs_key_buf *start, struct scoutfs_key_buf *end, struct list_head *list); +int scoutfs_net_alloc_segno(struct super_block *sb, u64 *segno); +int scoutfs_net_record_segment(struct super_block *sb, + struct scoutfs_segment *seg, u8 level); int scoutfs_net_setup(struct super_block *sb); void scoutfs_net_destroy(struct super_block *sb); diff --git a/kmod/src/seg.c b/kmod/src/seg.c index b343cb5e..6e0fc04a 100644 --- a/kmod/src/seg.c +++ b/kmod/src/seg.c @@ -245,26 +245,18 @@ static u64 segno_to_blkno(u64 blkno) return blkno << (SCOUTFS_SEGMENT_SHIFT - SCOUTFS_BLOCK_SHIFT); } -int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret) +int scoutfs_seg_alloc(struct super_block *sb, u64 segno, + struct scoutfs_segment **seg_ret) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct segment_cache *cac = sbi->segment_cache; struct scoutfs_segment *existing; struct scoutfs_segment *seg; unsigned long flags; - u64 segno; int ret; - *seg_ret = NULL; - - ret = scoutfs_alloc_segno(sb, &segno); - if (ret) - goto out; - seg = alloc_seg(segno); if (!seg) { - ret = scoutfs_alloc_free(sb, segno); - BUG_ON(ret); /* XXX could make pending when allocating */ ret = -ENOMEM; goto out; } @@ -281,9 +273,9 @@ int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret) if (existing) scoutfs_seg_put(existing); - *seg_ret = seg; ret = 0; out: + *seg_ret = seg; return ret; } @@ -632,6 +624,32 @@ int scoutfs_seg_manifest_del(struct super_block *sb, return scoutfs_manifest_del(sb, &first, le64_to_cpu(sblk->seq), level); } +/* + * Return an allocated manifest entry that describes the segment, returns + * NULL if it couldn't allocate. + */ +struct scoutfs_manifest_entry * +scoutfs_seg_manifest_entry(struct super_block *sb, + struct scoutfs_segment *seg, u8 level) +{ + struct scoutfs_segment_block *sblk = off_ptr(seg, 0); + struct scoutfs_segment_item *item; + struct scoutfs_key_buf first; + struct scoutfs_key_buf last; + + item = pos_ptr(seg, 0); + scoutfs_key_init(&first, off_ptr(seg, le32_to_cpu(item->key_off)), + le16_to_cpu(item->key_len)); + + item = pos_ptr(seg, le32_to_cpu(sblk->nr_items) - 1); + scoutfs_key_init(&last, off_ptr(seg, le32_to_cpu(item->key_off)), + le16_to_cpu(item->key_len)); + + return scoutfs_manifest_alloc_entry(sb, &first, &last, + le64_to_cpu(sblk->segno), + le64_to_cpu(sblk->seq), level); +} + /* * We maintain an LRU of segments so that the shrinker can free the * oldest under memory pressure. Segments are only present in the LRU diff --git a/kmod/src/seg.h b/kmod/src/seg.h index 15c6834f..5ce0c076 100644 --- a/kmod/src/seg.h +++ b/kmod/src/seg.h @@ -19,7 +19,8 @@ int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, void scoutfs_seg_get(struct scoutfs_segment *seg); void scoutfs_seg_put(struct scoutfs_segment *seg); -int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret); +int scoutfs_seg_alloc(struct super_block *sb, u64 segno, + struct scoutfs_segment **seg_ret); int scoutfs_seg_free_segno(struct super_block *sb, struct scoutfs_segment *seg); bool scoutfs_seg_fits_single(u32 nr_items, u32 key_bytes, u32 val_bytes); @@ -41,6 +42,10 @@ int scoutfs_seg_submit_write(struct super_block *sb, struct scoutfs_segment *seg, struct scoutfs_bio_completion *comp); +struct scoutfs_manifest_entry * +scoutfs_seg_manifest_entry(struct super_block *sb, + struct scoutfs_segment *seg, u8 level); + int scoutfs_seg_setup(struct super_block *sb); void scoutfs_seg_destroy(struct super_block *sb); diff --git a/kmod/src/trans.c b/kmod/src/trans.c index aa927fb2..5661398f 100644 --- a/kmod/src/trans.c +++ b/kmod/src/trans.c @@ -24,10 +24,9 @@ #include "item.h" #include "manifest.h" #include "seg.h" -#include "alloc.h" -#include "ring.h" #include "compact.h" #include "counters.h" +#include "net.h" #include "scoutfs_trace.h" /* @@ -82,7 +81,7 @@ void scoutfs_trans_write_func(struct work_struct *work) struct super_block *sb = sbi->sb; struct scoutfs_bio_completion comp; struct scoutfs_segment *seg; - bool advance = false; + u64 segno; int ret = 0; scoutfs_bio_init_comp(&comp); @@ -91,42 +90,25 @@ void scoutfs_trans_write_func(struct work_struct *work) wait_event(sbi->trans_hold_wq, atomic_cmpxchg(&sbi->trans_holds, 0, -1) == 0); - trace_printk("items dirty %d manifest dirty %d alloc dirty %d\n", - scoutfs_item_has_dirty(sb), - scoutfs_manifest_has_dirty(sb), - scoutfs_alloc_has_dirty(sb)); + trace_printk("items dirty %d\n", scoutfs_item_has_dirty(sb)); - /* - * XXX this needs serious work to handle errors. - */ - while (scoutfs_item_has_dirty(sb)) { - seg = NULL; - ret = scoutfs_seg_alloc(sb, &seg) ?: + if (scoutfs_item_has_dirty(sb)) { + /* + * XXX only straight pass through, we're not worrying + * about leaking segnos nor duplicate manifest entries + * on crashes between us and the server. + */ + ret = scoutfs_net_alloc_segno(sb, &segno) ?: + scoutfs_seg_alloc(sb, segno, &seg) ?: scoutfs_item_dirty_seg(sb, seg) ?: - scoutfs_manifest_lock(sb) ?: - scoutfs_seg_manifest_add(sb, seg, 0) ?: - scoutfs_manifest_unlock(sb) ?: - scoutfs_seg_submit_write(sb, seg, &comp); - scoutfs_seg_put(seg); + scoutfs_seg_submit_write(sb, seg, &comp) ?: + scoutfs_bio_wait_comp(sb, &comp) ?: + scoutfs_net_record_segment(sb, seg, 0); if (ret) goto out; scoutfs_inc_counter(sb, trans_level0_seg_write); } - - if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) { - 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; - } - out: /* XXX this all needs serious work for dealing with errors */ WARN_ON_ONCE(ret); @@ -135,8 +117,6 @@ out: scoutfs_data_end_writeback(sb, ret); spin_lock(&sbi->trans_write_lock); - if (advance) - scoutfs_advance_dirty_super(sb); sbi->trans_write_count++; sbi->trans_write_ret = ret; spin_unlock(&sbi->trans_write_lock); @@ -149,7 +129,6 @@ out: } struct write_attempt { - u64 seq; u64 count; int ret; }; @@ -161,9 +140,7 @@ static int write_attempted(struct scoutfs_sb_info *sbi, int done = 1; spin_lock(&sbi->trans_write_lock); - if (le64_to_cpu(sbi->super.hdr.seq) > attempt->seq) - attempt->ret = 0; - else if (sbi->trans_write_count > attempt->count) + if (sbi->trans_write_count > attempt->count) attempt->ret = sbi->trans_write_ret; else done = 0; @@ -178,10 +155,12 @@ static void queue_trans_work(struct scoutfs_sb_info *sbi) } /* - * sync records the current dirty seq and write count and waits for - * either to change. If there's nothing to write or the write returned - * an error then only the write count advances and sets the appropriate - * return code. + * Wait for a trans commit to finish and return its error code. There + * can already be one in flight that we end up waiting for the + * completion of. This is safe because dirtying and trans commits are + * serialized. There's no way that there could have been dirty data + * before the caller got here that wouldn't be covered by a commit + * that's in flight. */ int scoutfs_sync_fs(struct super_block *sb, int wait) { @@ -197,7 +176,6 @@ int scoutfs_sync_fs(struct super_block *sb, int wait) } spin_lock(&sbi->trans_write_lock); - attempt.seq = le64_to_cpu(sbi->super.hdr.seq); attempt.count = sbi->trans_write_count; spin_unlock(&sbi->trans_write_lock);