From e163f3b0990828c892a547b4596f83a6d0bf0494 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Sun, 28 Feb 2021 18:42:51 -0800 Subject: [PATCH 1/3] Use atomic holders instead of trans info lock We saw the transaction info lock showing up in profiles. We were doing quite a lot of work with that lock held. We can remove it entirely and use an atomic. Instead of a locked holders count and writer boolean we can use an atomic holders and have a high bit indicate that the write_func is pending. This turns the lock/unlock pairs in hold and release into atomic inc/cmpxchg/dec operations. Then we were checking allocators under the trans lock. Now that we have an atomic holders count we can increment it to prevent the writer from commiting and release it after the checks if we need another commit before the hold. And finally, we were freeing our allocated reservation struct under the lock. We weren't actually doing anything with the reservation struct so we can use journal_info as the nested hold counter instead of having it point to an allocated and freed struct. Signed-off-by: Zach Brown --- kmod/src/scoutfs_trace.h | 59 +++----- kmod/src/trans.c | 299 ++++++++++++++++++++++----------------- 2 files changed, 183 insertions(+), 175 deletions(-) diff --git a/kmod/src/scoutfs_trace.h b/kmod/src/scoutfs_trace.h index d440a228..387fffee 100644 --- a/kmod/src/scoutfs_trace.h +++ b/kmod/src/scoutfs_trace.h @@ -423,61 +423,34 @@ TRACE_EVENT(scoutfs_trans_write_func, TP_printk(SCSBF" dirty %lu", SCSB_TRACE_ARGS, __entry->dirty) ); -TRACE_EVENT(scoutfs_release_trans, - TP_PROTO(struct super_block *sb, void *rsv, unsigned int rsv_holders, - unsigned int tri_holders, - unsigned int tri_writing), +DECLARE_EVENT_CLASS(scoutfs_trans_hold_release_class, + TP_PROTO(struct super_block *sb, void *journal_info, int holders), - TP_ARGS(sb, rsv, rsv_holders, tri_holders, tri_writing), + TP_ARGS(sb, journal_info, holders), TP_STRUCT__entry( SCSB_TRACE_FIELDS - __field(void *, rsv) - __field(unsigned int, rsv_holders) - __field(unsigned int, tri_holders) - __field(unsigned int, tri_writing) + __field(unsigned long, journal_info) + __field(int, holders) ), TP_fast_assign( SCSB_TRACE_ASSIGN(sb); - __entry->rsv = rsv; - __entry->rsv_holders = rsv_holders; - __entry->tri_holders = tri_holders; - __entry->tri_writing = tri_writing; + __entry->journal_info = (unsigned long)journal_info; + __entry->holders = holders; ), - TP_printk(SCSBF" rsv %p holders %u trans holders %u writing %u", - SCSB_TRACE_ARGS, __entry->rsv, __entry->rsv_holders, - __entry->tri_holders, __entry->tri_writing) + TP_printk(SCSBF" journal_info 0x%0lx holders %d", + SCSB_TRACE_ARGS, __entry->journal_info, __entry->holders) ); -TRACE_EVENT(scoutfs_trans_acquired_hold, - TP_PROTO(struct super_block *sb, - void *rsv, unsigned int rsv_holders, - unsigned int tri_holders, - unsigned int tri_writing), - - TP_ARGS(sb, rsv, rsv_holders, tri_holders, tri_writing), - - TP_STRUCT__entry( - SCSB_TRACE_FIELDS - __field(void *, rsv) - __field(unsigned int, rsv_holders) - __field(unsigned int, tri_holders) - __field(unsigned int, tri_writing) - ), - - TP_fast_assign( - SCSB_TRACE_ASSIGN(sb); - __entry->rsv = rsv; - __entry->rsv_holders = rsv_holders; - __entry->tri_holders = tri_holders; - __entry->tri_writing = tri_writing; - ), - - TP_printk(SCSBF" rsv %p holders %u trans holders %u writing %u", - SCSB_TRACE_ARGS, __entry->rsv, __entry->rsv_holders, - __entry->tri_holders, __entry->tri_writing) +DEFINE_EVENT(scoutfs_trans_hold_release_class, scoutfs_trans_acquired_hold, + TP_PROTO(struct super_block *sb, void *journal_info, int holders), + TP_ARGS(sb, journal_info, holders) +); +DEFINE_EVENT(scoutfs_trans_hold_release_class, scoutfs_release_trans, + TP_PROTO(struct super_block *sb, void *journal_info, int holders), + TP_ARGS(sb, journal_info, holders) ); TRACE_EVENT(scoutfs_ioc_release, diff --git a/kmod/src/trans.c b/kmod/src/trans.c index a72257e4..742919f6 100644 --- a/kmod/src/trans.c +++ b/kmod/src/trans.c @@ -39,17 +39,15 @@ * track the relationships between dirty blocks so there's only ever one * transaction being built. * - * The copy of the on-disk super block in the fs sb info has its header - * sequence advanced so that new dirty blocks inherit this dirty - * sequence number. It's only advanced once all those dirty blocks are - * reachable after having first written them all out and then the new - * super with that seq. It's first incremented at mount. + * Committing the current dirty transaction can be triggered by sync, a + * regular background commit interval, reaching a dirty block threshold, + * or the transaction running out of its private allocator resources. + * Once all the current holders release the writing func writes out the + * dirty blocks while excluding holders until it finishes. * - * Unfortunately writers can nest. We don't bother trying to special - * case holding a transaction that you're already holding because that - * requires per-task storage. We just let anyone hold transactions - * regardless of waiters waiting to write, which risks waiters waiting a - * very long time. + * Unfortunately writing holders can nest. We track nested hold callers + * with the per-task journal_info pointer to avoid deadlocks between + * holders that might otherwise wait for a pending commit. */ /* sync dirty data at least this often */ @@ -59,9 +57,7 @@ * XXX move the rest of the super trans_ fields here. */ struct trans_info { - spinlock_t lock; - unsigned holders; - bool writing; + atomic_t holders; struct scoutfs_log_trees lt; struct scoutfs_alloc alloc; @@ -71,17 +67,9 @@ struct trans_info { #define DECLARE_TRANS_INFO(sb, name) \ struct trans_info *name = SCOUTFS_SB(sb)->trans_info -static bool drained_holders(struct trans_info *tri) -{ - bool drained; - - spin_lock(&tri->lock); - tri->writing = true; - drained = tri->holders == 0; - spin_unlock(&tri->lock); - - return drained; -} +/* avoid the high sign bit out of an abundance of caution*/ +#define TRANS_HOLDERS_WRITE_FUNC_BIT (1 << 30) +#define TRANS_HOLDERS_COUNT_MASK (TRANS_HOLDERS_WRITE_FUNC_BIT - 1) static int commit_btrees(struct super_block *sb) { @@ -126,6 +114,36 @@ bool scoutfs_trans_has_dirty(struct super_block *sb) return scoutfs_block_writer_has_dirty(sb, &tri->wri); } +/* + * This is racing with wait_event conditions, make sure our atomic + * stores and waitqueue loads are ordered. + */ +static void sub_holders_and_wake(struct super_block *sb, int val) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + DECLARE_TRANS_INFO(sb, tri); + + atomic_sub(val, &tri->holders); + smp_mb(); /* make sure sub is visible before we wake */ + if (waitqueue_active(&sbi->trans_hold_wq)) + wake_up(&sbi->trans_hold_wq); +} + +/* + * called as a wait_event condition, needs to be careful to not change + * task state and is racing with waking paths that sub_return, test, and + * wake. + */ +static bool drained_holders(struct trans_info *tri) +{ + int holders; + + smp_mb(); /* make sure task in wait_event queue before atomic read */ + holders = atomic_read(&tri->holders) & TRANS_HOLDERS_COUNT_MASK; + + return holders == 0; +} + /* * This work func is responsible for writing out all the dirty blocks * that make up the current dirty transaction. It prevents writers from @@ -162,6 +180,9 @@ void scoutfs_trans_write_func(struct work_struct *work) sbi->trans_task = current; + /* mark that we're writing so holders wait for us to finish and clear our bit */ + atomic_add(TRANS_HOLDERS_WRITE_FUNC_BIT, &tri->holders); + wait_event(sbi->trans_hold_wq, drained_holders(tri)); trace_scoutfs_trans_write_func(sb, @@ -213,11 +234,8 @@ out: spin_unlock(&sbi->trans_write_lock); wake_up(&sbi->trans_write_wq); - spin_lock(&tri->lock); - tri->writing = false; - spin_unlock(&tri->lock); - - wake_up(&sbi->trans_hold_wq); + /* we're done, wake waiting holders */ + sub_holders_and_wake(sb, TRANS_HOLDERS_WRITE_FUNC_BIT); sbi->trans_task = NULL; @@ -309,53 +327,83 @@ void scoutfs_trans_restart_sync_deadline(struct super_block *sb) } /* - * Each thread reserves space in the segment for their dirty items while - * they hold the transaction. This is calculated before the first - * transaction hold is acquired. It includes all the potential nested - * item manipulation that could happen with the transaction held. - * Including nested holds avoids having to deal with writing out partial - * transactions while a caller still holds the transaction. + * We store nested holders in the lower bits of journal_info. We use + * some higher bits as a magic value to detect if something goes + * horribly wrong and it gets clobbered. */ +#define TRANS_JI_MAGIC 0xd5700000 +#define TRANS_JI_MAGIC_MASK 0xfff00000 +#define TRANS_JI_COUNT_MASK 0x000fffff -#define SCOUTFS_RESERVATION_MAGIC 0xd57cd13b -struct scoutfs_reservation { - unsigned magic; - unsigned holders; -}; +/* returns true if a caller already had a holder counted in journal_info */ +static bool inc_journal_info_holders(void) +{ + unsigned long holders = (unsigned long)current->journal_info; + + WARN_ON_ONCE(holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) != TRANS_JI_MAGIC)); + + if (holders == 0) + holders = TRANS_JI_MAGIC; + holders++; + + current->journal_info = (void *)holders; + return (holders > (TRANS_JI_MAGIC | 1)); +} + +static void dec_journal_info_holders(void) +{ + unsigned long holders = (unsigned long)current->journal_info; + + WARN_ON_ONCE(holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) != TRANS_JI_MAGIC)); + WARN_ON_ONCE((holders & TRANS_JI_COUNT_MASK) == 0); + + holders--; + if (holders == TRANS_JI_MAGIC) + holders = 0; + + current->journal_info = (void *)holders; +} /* - * Try to hold the transaction. If a caller already holds the trans then - * we piggy back on their hold. We wait if the writer is trying to - * write out the transation. And if our items won't fit then we kick off - * a write. + * This is called as the wait_event condition for holding a transaction. + * Increment the holder count unless the writer is present. We return + * false to wait until the writer finishes and wakes us. * - * This is called as a condition for wait_event. It is very limited in - * the locking (blocking) it can do because the caller has set the task - * state before testing the condition safely race with waking after - * setting the condition. Our checking the amount of dirty metadata - * blocks and free data blocks is racy, but we don't mind the risk of - * delaying or prematurely forcing commits. + * This can be racing with itself while there's no waiters. We retry + * the cmpxchg instead of returning and waiting. */ -static bool acquired_hold(struct super_block *sb, - struct scoutfs_reservation *rsv) +static bool inc_holders_unless_writer(struct trans_info *tri) { - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - DECLARE_TRANS_INFO(sb, tri); - bool acquired = false; + int holders; - spin_lock(&tri->lock); + do { + smp_mb(); /* make sure we read after wait puts task in queue */ + holders = atomic_read(&tri->holders); + if (holders & TRANS_HOLDERS_WRITE_FUNC_BIT) + return false; - trace_scoutfs_trans_acquired_hold(sb, rsv, rsv->holders, - tri->holders, tri->writing); + } while (atomic_cmpxchg(&tri->holders, holders, holders + 1) != holders); - /* use a caller's existing reservation */ - if (rsv->holders) - goto hold; + return true; +} - /* wait until the writing thread is finished */ - if (tri->writing) - goto out; +/* + * As we drop the last trans holder we try to wake a writing thread that + * was waiting for us to finish. + */ +static void release_holders(struct super_block *sb) +{ + dec_journal_info_holders(); + sub_holders_and_wake(sb, 1); +} +/* + * The caller has incremented holders so it is blocking commits. We + * make some quick checks to see if we need to trigger and wait for + * another commit before proceeding. + */ +static bool commit_before_hold(struct super_block *sb, struct trans_info *tri) +{ /* * In theory each dirty item page could be straddling two full * blocks, requiring 4 allocations for each item cache page. @@ -365,11 +413,9 @@ static bool acquired_hold(struct super_block *sb, * that it accounts for having to dirty parent blocks and * whatever dirtying is done during the transaction hold. */ - if (scoutfs_alloc_meta_low(sb, &tri->alloc, - scoutfs_item_dirty_pages(sb) * 2)) { + if (scoutfs_alloc_meta_low(sb, &tri->alloc, scoutfs_item_dirty_pages(sb) * 2)) { scoutfs_inc_counter(sb, trans_commit_dirty_meta_full); - queue_trans_work(sbi); - goto out; + return true; } /* @@ -381,57 +427,74 @@ static bool acquired_hold(struct super_block *sb, */ if (scoutfs_alloc_meta_low(sb, &tri->alloc, 16)) { scoutfs_inc_counter(sb, trans_commit_meta_alloc_low); - queue_trans_work(sbi); - goto out; + return true; } /* Try to refill data allocator before premature enospc */ if (scoutfs_data_alloc_free_bytes(sb) <= SCOUTFS_TRANS_DATA_ALLOC_LWM) { scoutfs_inc_counter(sb, trans_commit_data_alloc_low); - queue_trans_work(sbi); + return true; + } + + return false; +} + +static bool acquired_hold(struct super_block *sb) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + DECLARE_TRANS_INFO(sb, tri); + bool acquired; + + /* if a caller already has a hold we acquire unconditionally */ + if (inc_journal_info_holders()) { + atomic_inc(&tri->holders); + acquired = true; goto out; } -hold: - rsv->holders++; - tri->holders++; + /* wait if the writer is blocking holds */ + if (!inc_holders_unless_writer(tri)) { + dec_journal_info_holders(); + acquired = false; + goto out; + } + + /* wait if we're triggering another commit */ + if (commit_before_hold(sb, tri)) { + release_holders(sb); + queue_trans_work(sbi); + acquired = false; + goto out; + } + + trace_scoutfs_trans_acquired_hold(sb, current->journal_info, atomic_read(&tri->holders)); acquired = true; - out: - - spin_unlock(&tri->lock); - return acquired; } +/* + * Try to hold the transaction. Holding the transaction prevents it + * from being committed. If a transaction is currently being written + * then we'll block until it's done and our hold can be granted. + * + * If a caller already holds the trans then we unconditionally acquire + * our hold and return to avoid deadlocks with our caller, the writing + * thread, and us. We record nested holds in a call stack with the + * journal_info pointer in the task_struct. + * + * The writing thread marks itself as a global trans_task which + * short-circuits all the hold machinery so it can call code that would + * otherwise try to hold transactions while it is writing. + */ int scoutfs_hold_trans(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_reservation *rsv; - int ret; if (current == sbi->trans_task) return 0; - rsv = current->journal_info; - if (rsv == NULL) { - rsv = kzalloc(sizeof(struct scoutfs_reservation), GFP_NOFS); - if (!rsv) - return -ENOMEM; - - rsv->magic = SCOUTFS_RESERVATION_MAGIC; - current->journal_info = rsv; - } - - BUG_ON(rsv->magic != SCOUTFS_RESERVATION_MAGIC); - - ret = wait_event_interruptible(sbi->trans_hold_wq, - acquired_hold(sb, rsv)); - if (ret && rsv->holders == 0) { - current->journal_info = NULL; - kfree(rsv); - } - return ret; + return wait_event_interruptible(sbi->trans_hold_wq, acquired_hold(sb)); } /* @@ -441,50 +504,22 @@ int scoutfs_hold_trans(struct super_block *sb) */ bool scoutfs_trans_held(void) { - struct scoutfs_reservation *rsv = current->journal_info; + unsigned long holders = (unsigned long)current->journal_info; - return rsv && rsv->magic == SCOUTFS_RESERVATION_MAGIC; + return (holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) == TRANS_JI_MAGIC)); } -/* - * As we drop the last hold in the reservation we try and wake other - * hold attempts that were waiting for space. As we drop the last trans - * holder we try to wake a writing thread that was waiting for us to - * finish. - */ void scoutfs_release_trans(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_reservation *rsv; DECLARE_TRANS_INFO(sb, tri); - bool wake = false; if (current == sbi->trans_task) return; - rsv = current->journal_info; - BUG_ON(!rsv || rsv->magic != SCOUTFS_RESERVATION_MAGIC); + release_holders(sb); - spin_lock(&tri->lock); - - trace_scoutfs_release_trans(sb, rsv, rsv->holders, tri->holders, tri->writing); - - BUG_ON(rsv->holders <= 0); - BUG_ON(tri->holders <= 0); - - if (--rsv->holders == 0) { - current->journal_info = NULL; - kfree(rsv); - wake = true; - } - - if (--tri->holders == 0) - wake = true; - - spin_unlock(&tri->lock); - - if (wake) - wake_up(&sbi->trans_hold_wq); + trace_scoutfs_release_trans(sb, current->journal_info, atomic_read(&tri->holders)); } /* @@ -513,7 +548,7 @@ int scoutfs_setup_trans(struct super_block *sb) if (!tri) return -ENOMEM; - spin_lock_init(&tri->lock); + atomic_set(&tri->holders, 0); scoutfs_block_writer_init(sb, &tri->wri); sbi->trans_write_workq = alloc_workqueue("scoutfs_trans", From c470c1c9f6f75094e7679d8cbb65ad9ca1990d07 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 2 Mar 2021 09:55:45 -0800 Subject: [PATCH 2/3] Allow read-mostly _alloc_meta_low Each transaction hold makes multiple calls to _alloc_meta_low to see if the transaction should be committed to refill allocators before the caller's hold is acquired and they can dirty blocks in the transaction. _alloc_meta_low was using a spinlock to sample the allocator list_head blocks to determine if there was space available. The lock and unlock stores were creating significant cacheline contention. The _alloc_meta_low calls are higher frequency than allocations. We can use a seqlock to have exclusive writers and allow concurrent _alloc_meta_low readers who retry if a writer intervenes. Signed-off-by: Zach Brown --- kmod/src/alloc.c | 30 ++++++++++++++++++++---------- kmod/src/alloc.h | 3 ++- 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/kmod/src/alloc.c b/kmod/src/alloc.c index 530e261f..8253a9ea 100644 --- a/kmod/src/alloc.c +++ b/kmod/src/alloc.c @@ -252,7 +252,7 @@ void scoutfs_alloc_init(struct scoutfs_alloc *alloc, { memset(alloc, 0, sizeof(struct scoutfs_alloc)); - spin_lock_init(&alloc->lock); + seqlock_init(&alloc->seqlock); mutex_init(&alloc->mutex); alloc->avail = *avail; alloc->freed = *freed; @@ -607,7 +607,8 @@ int scoutfs_alloc_meta(struct super_block *sb, struct scoutfs_alloc *alloc, if (ret < 0) goto out; - spin_lock(&alloc->lock); + write_seqlock(&alloc->seqlock); + lblk = alloc->dirty_avail_bl->data; if (WARN_ON_ONCE(lblk->nr == 0)) { /* shouldn't happen, transaction should commit first */ @@ -617,7 +618,8 @@ int scoutfs_alloc_meta(struct super_block *sb, struct scoutfs_alloc *alloc, list_block_remove(&alloc->avail, lblk, 1); ret = 0; } - spin_unlock(&alloc->lock); + + write_sequnlock(&alloc->seqlock); out: if (ret < 0) @@ -640,7 +642,8 @@ int scoutfs_free_meta(struct super_block *sb, struct scoutfs_alloc *alloc, if (ret < 0) goto out; - spin_lock(&alloc->lock); + write_seqlock(&alloc->seqlock); + lblk = alloc->dirty_freed_bl->data; if (WARN_ON_ONCE(list_block_space(lblk->nr) == 0)) { /* shouldn't happen, transaction should commit first */ @@ -649,7 +652,8 @@ int scoutfs_free_meta(struct super_block *sb, struct scoutfs_alloc *alloc, list_block_add(&alloc->freed, lblk, blkno); ret = 0; } - spin_unlock(&alloc->lock); + + write_sequnlock(&alloc->seqlock); out: scoutfs_inc_counter(sb, alloc_free_meta); @@ -1147,17 +1151,23 @@ out: /* * Returns true if meta avail and free don't have room for the given - * number of alloctions or frees. + * number of allocations or frees. This is called at a significantly + * higher frequency than allocations as writers try to enter + * transactions. This is the only reader of the seqlock which gives + * read-mostly sampling instead of bouncing a spinlock around all the + * cores. */ bool scoutfs_alloc_meta_low(struct super_block *sb, struct scoutfs_alloc *alloc, u32 nr) { + unsigned int seq; bool lo; - spin_lock(&alloc->lock); - lo = le32_to_cpu(alloc->avail.first_nr) < nr || - list_block_space(alloc->freed.first_nr) < nr; - spin_unlock(&alloc->lock); + do { + seq = read_seqbegin(&alloc->seqlock); + lo = le32_to_cpu(alloc->avail.first_nr) < nr || + list_block_space(alloc->freed.first_nr) < nr; + } while (read_seqretry(&alloc->seqlock, seq)); return lo; } diff --git a/kmod/src/alloc.h b/kmod/src/alloc.h index 8abf729c..395ef2ce 100644 --- a/kmod/src/alloc.h +++ b/kmod/src/alloc.h @@ -72,7 +72,8 @@ * transaction. */ struct scoutfs_alloc { - spinlock_t lock; + /* writers rarely modify list_head avail/freed. readers often check for _meta_alloc_low */ + seqlock_t seqlock; struct mutex mutex; struct scoutfs_block *dirty_avail_bl; struct scoutfs_block *dirty_freed_bl; From f8d39610a2f7e293451a17fe51c3e981c8ec5071 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 2 Mar 2021 11:44:24 -0800 Subject: [PATCH 3/3] Only get inode writeback_lock when adding inodes Each transaction maintains a global list of inodes to sync. It checks the inode and adds it in each write_end call per OS page. Locking and unlocking the global spinlock was showing up in profiles. At the very least, we can only get the lock once per large file that's written during a transaction. This will reduce spinlock traffic on the lock by the number of pages written per file. We'll want a better solution in the long run, but this helps for now. Signed-off-by: Zach Brown --- kmod/src/inode.c | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/kmod/src/inode.c b/kmod/src/inode.c index 36181fe5..8f16da36 100644 --- a/kmod/src/inode.c +++ b/kmod/src/inode.c @@ -1621,19 +1621,28 @@ int scoutfs_orphan_inode(struct inode *inode) } /* - * Track an inode that could have dirty pages. Used to kick off writeback - * on all dirty pages during transaction commit without tying ourselves in - * knots trying to call through the high level vfs sync methods. + * Track an inode that could have dirty pages. Used to kick off + * writeback on all dirty pages during transaction commit without tying + * ourselves in knots trying to call through the high level vfs sync + * methods. + * + * This is called by writers who hold the inode and transaction. The + * inode's presence in the rbtree is removed by destroy_inode, prevented + * by the inode hold, and by committing the transaction, which is + * prevented by holding the transaction. The inode can only go from + * empty to on the rbtree while we're here. */ void scoutfs_inode_queue_writeback(struct inode *inode) { DECLARE_INODE_SB_INFO(inode->i_sb, inf); struct scoutfs_inode_info *si = SCOUTFS_I(inode); - spin_lock(&inf->writeback_lock); - if (RB_EMPTY_NODE(&si->writeback_node)) - insert_writeback_inode(inf, si); - spin_unlock(&inf->writeback_lock); + if (RB_EMPTY_NODE(&si->writeback_node)) { + spin_lock(&inf->writeback_lock); + if (RB_EMPTY_NODE(&si->writeback_node)) + insert_writeback_inode(inf, si); + spin_unlock(&inf->writeback_lock); + } } /*