scoutfs: allocate inode numbers per directory

Having an inode number allocation pool in the super block meant that all
allocations across the mount are interleaved.  This means that
concurrent file creation in different directories will create
overlapping inode numbers.  This leads to lock contention as reasonable
work loads will tend to distribute work by directories.

The easy fix is to have per-directory inode number allocation pools.  We
take the opportunity to clean up the network request so that the caller
gets the allocation instead of having it be fed back in via a weird
callback.

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

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;
 }
 

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);

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);
 

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;
 

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);

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,

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);