Commit 850bac3a authored by Chuck Lever's avatar Chuck Lever
Browse files

SUNRPC: Deduplicate thread wake-up code 



Refactor: Extract the loop that finds an idle service thread from
svc_xprt_enqueue() and svc_wake_up(). Both functions do just about
the same thing.

Note that svc_wake_up() currently does not hold the RCU read lock
while waking the target thread. It indeed should hold the lock, just
as svc_xprt_enqueue() does, to ensure the rqstp does not vanish
during the wake-up. This patch adds the RCU lock for svc_wake_up().

Note that shrinking the pool thread count is rare, and calls to
svc_wake_up() are also quite infrequent. In practice, this race is
very unlikely to be hit, so we are not marking the lock fix for
stable backport at this time.

Reviewed-by: default avatarJeff Layton <jlayton@redhat.com>
Reviewed-by: default avatarNeilBrown <neilb@suse.de>
Signed-off-by: default avatarChuck Lever <chuck.lever@oracle.com>
parent 82e5d82a

include/linux/sunrpc/svc.h

+1 −0
Original line number Diff line number Diff line
@@ -419,6 +419,7 @@ int svc_register(const struct svc_serv *, struct net *, const int, 


void		   svc_wake_up(struct svc_serv *);

void		   svc_reserve(struct svc_rqst *rqstp, int space);

bool		   svc_pool_wake_idle_thread(struct svc_pool *pool);

struct svc_pool   *svc_pool_for_cpu(struct svc_serv *serv);

char *		   svc_print_addr(struct svc_rqst *, char *, size_t);

const char *	   svc_proc_name(const struct svc_rqst *rqstp);

net/sunrpc/svc.c

+34 −0
Original line number Diff line number Diff line
@@ -687,6 +687,40 @@ svc_prepare_thread(struct svc_serv *serv, struct svc_pool *pool, int node) 
	return rqstp;

}



/**

 * svc_pool_wake_idle_thread - Awaken an idle thread in @pool

 * @pool: service thread pool

 *

 * Can be called from soft IRQ or process context. Finding an idle

 * service thread and marking it BUSY is atomic with respect to

 * other calls to svc_pool_wake_idle_thread().

 *

 * Return value:

 *   %true: An idle thread was awoken

 *   %false: No idle thread was found

 */

bool svc_pool_wake_idle_thread(struct svc_pool *pool)

{

	struct svc_rqst	*rqstp;



	rcu_read_lock();

	list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {

		if (test_and_set_bit(RQ_BUSY, &rqstp->rq_flags))

			continue;



		WRITE_ONCE(rqstp->rq_qtime, ktime_get());

		wake_up_process(rqstp->rq_task);

		rcu_read_unlock();

		percpu_counter_inc(&pool->sp_threads_woken);

		trace_svc_wake_up(rqstp->rq_task->pid);

		return true;

	}

	rcu_read_unlock();



	set_bit(SP_CONGESTED, &pool->sp_flags);

	return false;

}



/*

 * Choose a pool in which to create a new thread, for svc_set_num_threads

 */

net/sunrpc/svc_xprt.c

+8 −36
Original line number Diff line number Diff line
@@ -457,7 +457,6 @@ static bool svc_xprt_ready(struct svc_xprt *xprt) 
void svc_xprt_enqueue(struct svc_xprt *xprt)

{

	struct svc_pool *pool;

	struct svc_rqst	*rqstp = NULL;



	if (!svc_xprt_ready(xprt))

		return;
@@ -477,20 +476,7 @@ void svc_xprt_enqueue(struct svc_xprt *xprt) 
	list_add_tail(&xprt->xpt_ready, &pool->sp_sockets);

	spin_unlock_bh(&pool->sp_lock);



	/* find a thread for this xprt */

	rcu_read_lock();

	list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {

		if (test_and_set_bit(RQ_BUSY, &rqstp->rq_flags))

			continue;

		percpu_counter_inc(&pool->sp_threads_woken);

		rqstp->rq_qtime = ktime_get();

		wake_up_process(rqstp->rq_task);

		goto out_unlock;

	}

	set_bit(SP_CONGESTED, &pool->sp_flags);

	rqstp = NULL;

out_unlock:

	rcu_read_unlock();

	svc_pool_wake_idle_thread(pool);

}

EXPORT_SYMBOL_GPL(svc_xprt_enqueue);
@@ -581,7 +567,10 @@ static void svc_xprt_release(struct svc_rqst *rqstp) 
	svc_xprt_put(xprt);

}



/*

/**

 * svc_wake_up - Wake up a service thread for non-transport work

 * @serv: RPC service

 *

 * Some svc_serv's will have occasional work to do, even when a xprt is not

 * waiting to be serviced. This function is there to "kick" a task in one of

 * those services so that it can wake up and do that work. Note that we only
@@ -590,27 +579,10 @@ static void svc_xprt_release(struct svc_rqst *rqstp) 
 */

void svc_wake_up(struct svc_serv *serv)

{

	struct svc_rqst	*rqstp;

	struct svc_pool *pool;



	pool = &serv->sv_pools[0];



	rcu_read_lock();

	list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {

		/* skip any that aren't queued */

		if (test_bit(RQ_BUSY, &rqstp->rq_flags))

			continue;

		rcu_read_unlock();

		wake_up_process(rqstp->rq_task);

		trace_svc_wake_up(rqstp->rq_task->pid);

		return;

	}

	rcu_read_unlock();

	struct svc_pool *pool = &serv->sv_pools[0];



	/* No free entries available */

	if (!svc_pool_wake_idle_thread(pool))

		set_bit(SP_TASK_PENDING, &pool->sp_flags);

	smp_wmb();

	trace_svc_wake_up(0);

}

EXPORT_SYMBOL_GPL(svc_wake_up);