diff --git a/kmod/src/client.c b/kmod/src/client.c index 16517ccb..174f7f17 100644 --- a/kmod/src/client.c +++ b/kmod/src/client.c @@ -540,30 +540,30 @@ static int client_request(struct client_info *client, int type, void *data, return ret; } -int scoutfs_client_alloc_inodes(struct super_block *sb) +/* + * Ask for a new run of allocated inode numbers. The server can return + * fewer than @count. It will success with nr == 0 if we've run out. + */ +int scoutfs_client_alloc_inodes(struct super_block *sb, u64 count, + u64 *ino, u64 *nr) { struct client_info *client = SCOUTFS_SB(sb)->client_info; struct scoutfs_net_inode_alloc ial; - u64 ino = 0; - u64 nr = 0; + __le64 lecount = cpu_to_le64(count); int ret; - ret = client_request(client, SCOUTFS_NET_ALLOC_INODES, NULL, 0, - &ial, sizeof(ial)); + ret = client_request(client, SCOUTFS_NET_ALLOC_INODES, + &lecount, sizeof(lecount), &ial, sizeof(ial)); if (ret == 0) { - ino = le64_to_cpu(ial.ino); - nr = le64_to_cpu(ial.nr); + *ino = le64_to_cpu(ial.ino); + *nr = le64_to_cpu(ial.nr); - /* catch wrapping */ - if (ino + nr < ino) + if (*nr == 0) + ret = -ENOSPC; + else if (*ino + *nr < *ino) ret = -EINVAL; } - if (ret < 0) - scoutfs_inode_fill_pool(sb, 0, 0); - else - scoutfs_inode_fill_pool(sb, ino, nr); - return ret; } diff --git a/kmod/src/client.h b/kmod/src/client.h index 71c8c4eb..c2bc3bb9 100644 --- a/kmod/src/client.h +++ b/kmod/src/client.h @@ -1,7 +1,8 @@ #ifndef _SCOUTFS_CLIENT_H_ #define _SCOUTFS_CLIENT_H_ -int scoutfs_client_alloc_inodes(struct super_block *sb); +int scoutfs_client_alloc_inodes(struct super_block *sb, u64 count, + u64 *ino, u64 *nr); int scoutfs_client_alloc_segno(struct super_block *sb, u64 *segno); int scoutfs_client_record_segment(struct super_block *sb, struct scoutfs_segment *seg, u8 level); diff --git a/kmod/src/dir.c b/kmod/src/dir.c index 5ae5cc8b..85b9143e 100644 --- a/kmod/src/dir.c +++ b/kmod/src/dir.c @@ -545,7 +545,7 @@ static struct inode *lock_hold_create(struct inode *dir, struct dentry *dentry, if (ret) return ERR_PTR(ret); - ret = scoutfs_alloc_ino(sb, &ino); + ret = scoutfs_alloc_ino(dir, &ino); if (ret) return ERR_PTR(ret); diff --git a/kmod/src/inode.c b/kmod/src/inode.c index 209bd6f9..31545f0c 100644 --- a/kmod/src/inode.c +++ b/kmod/src/inode.c @@ -47,17 +47,7 @@ * - describe data locking size problems */ -struct free_ino_pool { - wait_queue_head_t waitq; - spinlock_t lock; - u64 ino; - u64 nr; - bool in_flight; -}; - struct inode_sb_info { - struct free_ino_pool pool; - spinlock_t writeback_lock; struct rb_root writeback_inodes; }; @@ -82,6 +72,7 @@ static void scoutfs_inode_ctor(void *obj) scoutfs_per_task_init(&ci->pt_data_lock); init_rwsem(&ci->xattr_rwsem); RB_CLEAR_NODE(&ci->writeback_node); + spin_lock_init(&ci->ino_alloc.lock); inode_init_once(&ci->inode); } @@ -563,8 +554,9 @@ struct inode *scoutfs_ilookup(struct super_block *sb, u64 ino) struct inode *scoutfs_iget(struct super_block *sb, u64 ino) { - struct inode *inode; struct scoutfs_lock *lock = NULL; + struct scoutfs_inode_info *si; + struct inode *inode; int ret; ret = scoutfs_lock_ino(sb, DLM_LOCK_PR, 0, ino, &lock); @@ -580,7 +572,10 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino) if (inode->i_state & I_NEW) { /* XXX ensure refresh, instead clear in drop_inode? */ - atomic64_set(&SCOUTFS_I(inode)->last_refreshed, 0); + si = SCOUTFS_I(inode); + atomic64_set(&si->last_refreshed, 0); + si->ino_alloc.ino = 0; + si->ino_alloc.nr = 0; ret = scoutfs_inode_refresh(inode, lock, 0); if (ret) { @@ -1205,98 +1200,50 @@ u64 scoutfs_last_ino(struct super_block *sb) } /* - * Network replies refill the pool, providing ino = ~0ULL nr = 0 when - * there's no more inodes (which should never happen in practice.) + * Return an allocated and unused inode number. Returns -ENOSPC if + * we're out of inode. + * + * Each parent directory has its own pool of free inode numbers. Items + * are sorted by their inode numbers as they're stored in segments. + * This will tend to group together files that are created in a + * directory at the same time in segments. Concurrent creation across + * different directories will be stored in their own regions. + * + * Inode numbers are never reclaimed. If the inode is evicted or we're + * unmounted the pending inode numbers will be lost. Asking for a + * relatively small number from the server each time will tend to + * minimize that loss while still being large enough for typical + * directory file counts. */ -void scoutfs_inode_fill_pool(struct super_block *sb, u64 ino, u64 nr) +int scoutfs_alloc_ino(struct inode *parent, u64 *ino_ret) { - struct free_ino_pool *pool = &SCOUTFS_SB(sb)->inode_sb_info->pool; - - trace_scoutfs_inode_fill_pool(sb, ino, nr); - - spin_lock(&pool->lock); - - pool->ino = ino; - pool->nr = nr; - pool->in_flight = false; - - spin_unlock(&pool->lock); - - wake_up(&pool->waitq); -} - -static bool pool_in_flight(struct free_ino_pool *pool) -{ - bool in_flight; - - spin_lock(&pool->lock); - in_flight = pool->in_flight; - spin_unlock(&pool->lock); - - return in_flight; -} - -/* - * We have a pool of free inodes given to us by the server. If it - * empties we only ever have one request for new inodes in flight. The - * net layer calls us when it gets a reply. If there's no more inodes - * we'll get ino == ~0 and nr == 0. - */ -int scoutfs_alloc_ino(struct super_block *sb, u64 *ino) -{ - struct free_ino_pool *pool = &SCOUTFS_SB(sb)->inode_sb_info->pool; - bool request; + struct scoutfs_inode_allocator *ia = &SCOUTFS_I(parent)->ino_alloc; + struct super_block *sb = parent->i_sb; + u64 ino; + u64 nr; int ret; - *ino = 0; + spin_lock(&ia->lock); - spin_lock(&pool->lock); - - while (pool->nr == 0 && pool->ino != ~0ULL) { - if (pool->in_flight) { - request = false; - } else { - pool->in_flight = true; - request = true; - } - - spin_unlock(&pool->lock); - - if (request) { - ret = scoutfs_client_alloc_inodes(sb); - if (ret) { - spin_lock(&pool->lock); - pool->in_flight = false; - spin_unlock(&pool->lock); - wake_up(&pool->waitq); - goto out; - } - } - - ret = wait_event_interruptible(pool->waitq, - !pool_in_flight(pool)); - if (ret) + if (ia->nr == 0) { + spin_unlock(&ia->lock); + ret = scoutfs_client_alloc_inodes(sb, 10000, &ino, &nr); + if (ret < 0) goto out; - - spin_lock(&pool->lock); + spin_lock(&ia->lock); + if (ia->nr == 0) { + ia->ino = ino; + ia->nr = nr; + } } - if (pool->nr == 0) { - *ino = 0; - ret = -ENOSPC; - } else { - *ino = pool->ino++; - pool->nr--; - ret = 0; - - } - - spin_unlock(&pool->lock); + *ino_ret = ia->ino++; + ia->nr--; + spin_unlock(&ia->lock); + ret = 0; out: - - trace_scoutfs_alloc_ino(sb, ret, *ino, pool->ino, pool->nr, - pool->in_flight); + trace_scoutfs_alloc_ino(sb, ret, *ino_ret, ia->ino, ia->nr); return ret; } @@ -1327,6 +1274,8 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir, ci->have_item = false; atomic64_set(&ci->last_refreshed, scoutfs_lock_refresh_gen(lock)); ci->flags = 0; + ci->ino_alloc.ino = 0; + ci->ino_alloc.nr = 0; scoutfs_inode_set_meta_seq(inode); scoutfs_inode_set_data_seq(inode); @@ -1653,17 +1602,12 @@ out: int scoutfs_inode_setup(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct free_ino_pool *pool; struct inode_sb_info *inf; inf = kzalloc(sizeof(struct inode_sb_info), GFP_KERNEL); if (!inf) return -ENOMEM; - pool = &inf->pool; - init_waitqueue_head(&pool->waitq); - spin_lock_init(&pool->lock); - spin_lock_init(&inf->writeback_lock); inf->writeback_inodes = RB_ROOT; diff --git a/kmod/src/inode.h b/kmod/src/inode.h index 6e5309be..532bb21a 100644 --- a/kmod/src/inode.h +++ b/kmod/src/inode.h @@ -8,6 +8,12 @@ struct scoutfs_lock; +struct scoutfs_inode_allocator { + spinlock_t lock; + u64 ino; + u64 nr; +}; + struct scoutfs_inode_info { /* read or initialized for each inode instance */ u64 ino; @@ -31,6 +37,9 @@ struct scoutfs_inode_info { /* updated at on each new lock acquisition */ atomic64_t last_refreshed; + /* reset for every new inode instance */ + struct scoutfs_inode_allocator ino_alloc; + /* initialized once for slab object */ seqcount_t seqcount; bool staging; /* holder of i_mutex is staging */ @@ -80,8 +89,7 @@ int scoutfs_dirty_inode_item(struct inode *inode, struct scoutfs_lock *lock); void scoutfs_update_inode_item(struct inode *inode, struct scoutfs_lock *lock, struct list_head *ind_locks); -void scoutfs_inode_fill_pool(struct super_block *sb, u64 ino, u64 nr); -int scoutfs_alloc_ino(struct super_block *sb, u64 *ino); +int scoutfs_alloc_ino(struct inode *parent, u64 *ino); struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir, umode_t mode, dev_t rdev, u64 ino, struct scoutfs_lock *lock); diff --git a/kmod/src/scoutfs_trace.h b/kmod/src/scoutfs_trace.h index def4daf7..f8e512eb 100644 --- a/kmod/src/scoutfs_trace.h +++ b/kmod/src/scoutfs_trace.h @@ -903,32 +903,30 @@ TRACE_EVENT(scoutfs_inode_fill_pool, ); TRACE_EVENT(scoutfs_alloc_ino, - TP_PROTO(struct super_block *sb, int ret, __u64 ino, __u64 pool_ino, - __u64 nr, unsigned int in_flight), + TP_PROTO(struct super_block *sb, int ret, __u64 ino, __u64 next_ino, + __u64 next_nr), - TP_ARGS(sb, ret, ino, pool_ino, nr, in_flight), + TP_ARGS(sb, ret, ino, next_ino, next_nr), TP_STRUCT__entry( __field(__u64, fsid) __field(int, ret) __field(__u64, ino) - __field(__u64, pool_ino) - __field(__u64, nr) - __field(unsigned int, in_flight) + __field(__u64, next_ino) + __field(__u64, next_nr) ), TP_fast_assign( __entry->fsid = FSID_ARG(sb); __entry->ret = ret; __entry->ino = ino; - __entry->pool_ino = pool_ino; - __entry->nr = nr; - __entry->in_flight = in_flight; + __entry->next_ino = next_ino; + __entry->next_nr = next_nr; ), - TP_printk(FSID_FMT" ret %d ino %llu pool ino %llu nr %llu req %u " - "(racey)", __entry->fsid, __entry->ret, __entry->ino, - __entry->pool_ino, __entry->nr, __entry->in_flight) + TP_printk(FSID_FMT" ret %d ino %llu next_ino %llu next_nr %llu", + __entry->fsid, __entry->ret, __entry->ino, __entry->next_ino, + __entry->next_nr) ); TRACE_EVENT(scoutfs_evict_inode, diff --git a/kmod/src/server.c b/kmod/src/server.c index 9cba3882..1b206f42 100644 --- a/kmod/src/server.c +++ b/kmod/src/server.c @@ -304,20 +304,23 @@ static int process_alloc_inodes(struct server_connection *conn, struct scoutfs_super_block *super = &sbi->super; struct scoutfs_net_inode_alloc ial; struct commit_waiter cw; + __le64 lecount; u64 ino; u64 nr; int ret; - if (data_len != 0) { + if (data_len != sizeof(lecount)) { ret = -EINVAL; goto out; } + memcpy(&lecount, data, data_len); + down_read(&server->commit_rwsem); spin_lock(&sbi->next_ino_lock); ino = le64_to_cpu(super->next_ino); - nr = min(100000ULL, ~0ULL - ino); + nr = min(le64_to_cpu(lecount), U64_MAX - ino); le64_add_cpu(&super->next_ino, nr); spin_unlock(&sbi->next_ino_lock);