Remove almost all interruptible waits

As subsystems were built I tended to use interruptible waits in the hope
that we'd let users break out of most waits.

The reality is that we have significant code paths that have trouble
unwinding.  Final inode deletion during iput->evict in a task is a good
example.  It's madness to have a pending signal turn an inode deletion
from an efficient inline operation to a deferred background orphan inode
scan deletion.

It also happens that golang built pre-emptive thread scheduling around
signals.  Under load we see a surprising amount of signal spam and it
has created surprising error cases which would have otherwise been fine.

This changes waits to expect that IOs (including network commands) will
complete reasonably promptly.  We remove all interruptible waits with
the notable exception of breaking out of a pending mount.  That requires
shuffling setup around a little bit so that the first network message we
wait for is the lock for getting the root inode.

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

kmod/src/block.c

@@ -645,9 +645,11 @@ static struct block_private *block_read(struct super_block *sb, u64 blkno)
 			goto out;
 	}
 
-	ret = wait_event_interruptible(binf->waitq, uptodate_or_error(bp));
-	if (ret == 0 && test_bit(BLOCK_BIT_ERROR, &bp->bits))
+	wait_event(binf->waitq, uptodate_or_error(bp));
+	if (test_bit(BLOCK_BIT_ERROR, &bp->bits))
 		ret = -EIO;
+	else
+		ret = 0;
 
 out:
 	if (ret < 0) {

kmod/src/client.c

@@ -623,10 +623,8 @@ void scoutfs_client_destroy(struct super_block *sb)
 						 client_farewell_response,
 						 NULL, NULL);
 		if (ret == 0) {
-			ret = wait_for_completion_interruptible(
-							&client->farewell_comp);
-			if (ret == 0)
-				ret = client->farewell_error;
+			wait_for_completion(&client->farewell_comp);
+			ret = client->farewell_error;
 		}
 		if (ret) {
 			scoutfs_inc_counter(sb, client_farewell_error);

kmod/src/counters.h

@@ -88,7 +88,6 @@
 	EXPAND_COUNTER(forest_read_items)			\
 	EXPAND_COUNTER(forest_roots_next_hint)			\
 	EXPAND_COUNTER(forest_set_bloom_bits)			\
-	EXPAND_COUNTER(inode_evict_intr)			\
 	EXPAND_COUNTER(item_clear_dirty)			\
 	EXPAND_COUNTER(item_create)				\
 	EXPAND_COUNTER(item_delete)				\

kmod/src/dir.c

@@ -462,7 +462,7 @@ out:
 	else if (ino == 0)
 		inode = NULL;
 	else
-		inode = scoutfs_iget(sb, ino);
+		inode = scoutfs_iget(sb, ino, 0);
 
 	/*
 	 * We can't splice dir aliases into the dcache.  dir entries

kmod/src/export.c

@@ -81,7 +81,7 @@ static struct dentry *scoutfs_fh_to_dentry(struct super_block *sb,
 	trace_scoutfs_fh_to_dentry(sb, fh_type, sfid);
 
 	if (scoutfs_valid_fileid(fh_type))
-		inode = scoutfs_iget(sb, le64_to_cpu(sfid->ino));
+		inode = scoutfs_iget(sb, le64_to_cpu(sfid->ino), 0);
 
 	return d_obtain_alias(inode);
 }
@@ -100,7 +100,7 @@ static struct dentry *scoutfs_fh_to_parent(struct super_block *sb,
 
 	if (scoutfs_valid_fileid(fh_type) &&
 	    fh_type == FILEID_SCOUTFS_WITH_PARENT)
-		inode = scoutfs_iget(sb, le64_to_cpu(sfid->parent_ino));
+		inode = scoutfs_iget(sb, le64_to_cpu(sfid->parent_ino), 0);
 
 	return d_obtain_alias(inode);
 }
@@ -123,7 +123,7 @@ static struct dentry *scoutfs_get_parent(struct dentry *child)
 	scoutfs_dir_free_backref_path(sb, &list);
 	trace_scoutfs_get_parent(sb, inode, ino);
 
-	inode = scoutfs_iget(sb, ino);
+	inode = scoutfs_iget(sb, ino, 0);
 
 	return d_obtain_alias(inode);
 }

kmod/src/fence.c

@@ -376,7 +376,7 @@ int scoutfs_fence_wait_fenced(struct super_block *sb, long timeout_jiffies)
 	bool error;
 	long ret;
 
-	ret = wait_event_interruptible_timeout(fi->waitq, all_fenced(fi, &error), timeout_jiffies);
+	ret = wait_event_timeout(fi->waitq, all_fenced(fi, &error), timeout_jiffies);
 	if (ret == 0)
 		ret = -ETIMEDOUT;
 	else if (ret > 0)

kmod/src/forest.c

@@ -747,9 +747,6 @@ int scoutfs_forest_setup(struct super_block *sb)
 		goto out;
 	}
 
-	queue_delayed_work(finf->workq, &finf->log_merge_dwork,
-			   msecs_to_jiffies(LOG_MERGE_DELAY_MS));
-
 	ret = 0;
 out:
 	if (ret)
@@ -758,6 +755,14 @@ out:
 	return 0;
 }
 
+void scoutfs_forest_start(struct super_block *sb)
+{
+	DECLARE_FOREST_INFO(sb, finf);
+
+	queue_delayed_work(finf->workq, &finf->log_merge_dwork,
+			   msecs_to_jiffies(LOG_MERGE_DELAY_MS));
+}
+
 void scoutfs_forest_stop(struct super_block *sb)
 {
 	DECLARE_FOREST_INFO(sb, finf);

kmod/src/forest.h

@@ -39,6 +39,7 @@ void scoutfs_forest_get_btrees(struct super_block *sb,
 			       struct scoutfs_log_trees *lt);
 
 int scoutfs_forest_setup(struct super_block *sb);
+void scoutfs_forest_start(struct super_block *sb);
 void scoutfs_forest_stop(struct super_block *sb);
 void scoutfs_forest_destroy(struct super_block *sb);
 

kmod/src/inode.c

@@ -662,14 +662,14 @@ struct inode *scoutfs_ilookup(struct super_block *sb, u64 ino)
 	return ilookup5(sb, ino, scoutfs_iget_test, &ino);
 }
 
-struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
+struct inode *scoutfs_iget(struct super_block *sb, u64 ino, int lkf)
 {
 	struct scoutfs_lock *lock = NULL;
 	struct scoutfs_inode_info *si;
 	struct inode *inode;
 	int ret;
 
-	ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_READ, 0, ino, &lock);
+	ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_READ, lkf, ino, &lock);
 	if (ret)
 		return ERR_PTR(ret);
 
@@ -1627,11 +1627,6 @@ void scoutfs_evict_inode(struct inode *inode)
 		scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
 		scoutfs_unlock(sb, orph_lock, SCOUTFS_LOCK_WRITE_ONLY);
 	}
-	if (ret == -ERESTARTSYS) {
-		/* can be in task with pending, could be found as orphan */
-		scoutfs_inc_counter(sb, inode_evict_intr);
-		ret = 0;
-	}
 	if (ret < 0) {
 		scoutfs_err(sb, "error %d while checking to delete inode nr %llu, it might linger.",
 			    ret, ino);
@@ -1827,7 +1822,7 @@ static void inode_orphan_scan_worker(struct work_struct *work)
 		}
 
 		/* try to cached and evict unused inode to delete, can be racing */
-		inode = scoutfs_iget(sb, ino);
+		inode = scoutfs_iget(sb, ino, 0);
 		if (IS_ERR(inode)) {
 			ret = PTR_ERR(inode);
 			if (ret == -ENOENT)
@@ -1970,12 +1965,11 @@ int scoutfs_inode_setup(struct super_block *sb)
  * many other subsystems like networking and the server.  We only kick
  * it off once everything is ready.
  */
-int scoutfs_inode_start(struct super_block *sb)
+void scoutfs_inode_start(struct super_block *sb)
 {
 	DECLARE_INODE_SB_INFO(sb, inf);
 
 	schedule_orphan_dwork(inf);
-	return 0;
 }
 
 void scoutfs_inode_stop(struct super_block *sb)

kmod/src/inode.h

@@ -77,7 +77,7 @@ int scoutfs_drop_inode(struct inode *inode);
 void scoutfs_evict_inode(struct inode *inode);
 void scoutfs_inode_queue_iput(struct inode *inode);
 
-struct inode *scoutfs_iget(struct super_block *sb, u64 ino);
+struct inode *scoutfs_iget(struct super_block *sb, u64 ino, int lkf);
 struct inode *scoutfs_ilookup(struct super_block *sb, u64 ino);
 
 void scoutfs_inode_init_index_key(struct scoutfs_key *key, u8 type, u64 major,
@@ -132,7 +132,7 @@ void scoutfs_inode_exit(void);
 int scoutfs_inode_init(void);
 
 int scoutfs_inode_setup(struct super_block *sb);
-int scoutfs_inode_start(struct super_block *sb);
+void scoutfs_inode_start(struct super_block *sb);
 void scoutfs_inode_stop(struct super_block *sb);
 void scoutfs_inode_destroy(struct super_block *sb);
 

kmod/src/lock.c

@@ -1095,8 +1095,14 @@ static int lock_key_range(struct super_block *sb, enum scoutfs_lock_mode mode, i
 
 		trace_scoutfs_lock_wait(sb, lock);
 
-		ret = wait_event_interruptible(lock->waitq,
-					       lock_wait_cond(sb, lock, mode));
+		if (flags & SCOUTFS_LKF_INTERRUPTIBLE) {
+			ret = wait_event_interruptible(lock->waitq,
+						       lock_wait_cond(sb, lock, mode));
+		} else {
+			wait_event(lock->waitq, lock_wait_cond(sb, lock, mode));
+			ret = 0;
+		}
+
 		spin_lock(&linfo->lock);
 		if (ret)
 			break;

kmod/src/lock.h

@@ -6,7 +6,8 @@
 
 #define SCOUTFS_LKF_REFRESH_INODE	0x01 /* update stale inode from item */
 #define SCOUTFS_LKF_NONBLOCK		0x02 /* only use already held locks */
-#define SCOUTFS_LKF_INVALID		(~((SCOUTFS_LKF_NONBLOCK << 1) - 1))
+#define SCOUTFS_LKF_INTERRUPTIBLE	0x04 /* pending signals return -ERESTARTSYS */
+#define SCOUTFS_LKF_INVALID		(~((SCOUTFS_LKF_INTERRUPTIBLE << 1) - 1))
 
 #define SCOUTFS_LOCK_NR_MODES		SCOUTFS_LOCK_INVALID
 

kmod/src/net.c

@@ -1486,8 +1486,7 @@ int scoutfs_net_connect(struct super_block *sb,
 			struct scoutfs_net_connection *conn,
 			struct sockaddr_in *sin, unsigned long timeout_ms)
 {
-	int error = 0;
-	int ret;
+	int ret = 0;
 
 	spin_lock(&conn->lock);
 	conn->connect_sin = *sin;
@@ -1495,10 +1494,8 @@ int scoutfs_net_connect(struct super_block *sb,
 	spin_unlock(&conn->lock);
 
 	queue_work(conn->workq, &conn->connect_work);
-
-	ret = wait_event_interruptible(conn->waitq,
-				       connect_result(conn, &error));
-	return ret ?: error;
+	wait_event(conn->waitq, connect_result(conn, &ret));
+	return ret;
 }
 
 static void set_valid_greeting(struct scoutfs_net_connection *conn)
@@ -1802,11 +1799,8 @@ int scoutfs_net_sync_request(struct super_block *sb,
 					 sync_response, &sreq, &id);
 
 	if (ret == 0) {
-		ret = wait_for_completion_interruptible(&sreq.comp);
-		if (ret == -ERESTARTSYS)
-			scoutfs_net_cancel_request(sb, conn, cmd, id);
-		else
-			ret = sreq.error;
+		wait_for_completion(&sreq.comp);
+		ret = sreq.error;
 	}
 
 	return ret;

kmod/src/server.c

@@ -3564,7 +3564,7 @@ static void scoutfs_server_worker(struct work_struct *work)
 
 	queue_reclaim_work(server, 0);
 
-	/* wait_event/wake_up provide barriers */
+	/* interruptible mostly to avoid stuck messages */
 	wait_event_interruptible(server->waitq, test_shutting_down(server));
 
 shutdown:

kmod/src/super.c

@@ -630,15 +630,16 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	      scoutfs_quorum_setup(sb) ?:
 	      scoutfs_client_setup(sb) ?:
 	      scoutfs_volopt_setup(sb) ?:
-	      scoutfs_trans_get_log_trees(sb) ?:
-	      scoutfs_srch_setup(sb) ?:
-	      scoutfs_inode_start(sb);
+	      scoutfs_srch_setup(sb);
 	if (ret)
 		goto out;
 
-	inode = scoutfs_iget(sb, SCOUTFS_ROOT_INO);
+	/* this interruptible iget lets hung mount be aborted with ctl-c */
+	inode = scoutfs_iget(sb, SCOUTFS_ROOT_INO, SCOUTFS_LKF_INTERRUPTIBLE);
 	if (IS_ERR(inode)) {
 		ret = PTR_ERR(inode);
+		if (ret == -ERESTARTSYS)
+			ret = -EINTR;
 		goto out;
 	}
 
@@ -648,10 +649,15 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 		goto out;
 	}
 
-	ret = scoutfs_client_advance_seq(sb, &sbi->trans_seq);
+	/* send requests once iget progress shows we had a server */
+	ret = scoutfs_trans_get_log_trees(sb) ?:
+	      scoutfs_client_advance_seq(sb, &sbi->trans_seq);
 	if (ret)
 		goto out;
 
+	/* start up background services that use everything else */
+	scoutfs_inode_start(sb);
+	scoutfs_forest_start(sb);
 	scoutfs_trans_restart_sync_deadline(sb);
 	ret = 0;
 out:

kmod/src/trans.c

@@ -291,7 +291,7 @@ static void queue_trans_work(struct scoutfs_sb_info *sbi)
 int scoutfs_trans_sync(struct super_block *sb, int wait)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	struct write_attempt attempt;
+	struct write_attempt attempt = { .ret = 0 };
 	int ret;
 
 
@@ -306,10 +306,8 @@ int scoutfs_trans_sync(struct super_block *sb, int wait)
 
 	queue_trans_work(sbi);
 
-	ret = wait_event_interruptible(sbi->trans_write_wq,
-				       write_attempted(sbi, &attempt));
-	if (ret == 0)
-		ret = attempt.ret;
+	wait_event(sbi->trans_write_wq, write_attempted(sbi, &attempt));
+	ret = attempt.ret;
 
 	return ret;
 }
@@ -496,9 +494,7 @@ int scoutfs_hold_trans(struct super_block *sb, bool allocing)
 		/* wait until the writer work is finished */
 		if (!inc_holders_unless_writer(tri)) {
 			dec_journal_info_holders();
-			ret = wait_event_interruptible(sbi->trans_hold_wq, holders_no_writer(tri));
-			if (ret < 0)
-				break;
+			wait_event(sbi->trans_hold_wq, holders_no_writer(tri));
 			continue;
 		}
 
@@ -514,10 +510,7 @@ int scoutfs_hold_trans(struct super_block *sb, bool allocing)
 			seq = scoutfs_trans_sample_seq(sb);
 			release_holders(sb);
 			queue_trans_work(sbi);
-			ret = wait_event_interruptible(sbi->trans_hold_wq,
-						       scoutfs_trans_sample_seq(sb) != seq);
-			if (ret < 0)
-				break;
+			wait_event(sbi->trans_hold_wq, scoutfs_trans_sample_seq(sb) != seq);
 			continue;
 		}