Re: [PATCH 0/5] stop_machine: cleanups and fix

[Oleg Nesterov]

OK, let me send this patch as 6/5 in reply to 0/5. Added a couple
of helpers and the comment.

Please let me know what do you think.

On 06/30, Oleg Nesterov wrote:
>
> On 06/30, Oleg Nesterov wrote:
> >
> > On 06/30, Oleg Nesterov wrote:
> > >
> > > But let me send some cleanups first. Plus I believe I found another
> > > stop_machine bug, see the last patch. So I hope these changes make
> > > sense in any case.
> >
> > The last patch fixes the bug, I think. Say, stop_one_cpu(X) can race
> > with _cpu_down(X)->stop_machine() so that the kernel will crash if this
> > CPU X becomes online again.
> >
> > The window after cpu_stopper_thread() returns and before smpboot_thread()
> > calls ->park() is tiny, but still this is possible afaics.
> 
> Now lets try to remove lglocks from stop_machine.c. See the patch at
> the end. It lacks the comments and the changelog, just for review.
> 
> On top of this series, so let me show the code with this patch applied,
> 
> 	static int cpu_stop_queue_two_works(int cpu1, struct cpu_stop_work *work1,
> 					    int cpu2, struct cpu_stop_work *work2)
> 	{
> 		struct cpu_stopper *stopper1 = per_cpu_ptr(&cpu_stopper, cpu1);
> 		struct cpu_stopper *stopper2 = per_cpu_ptr(&cpu_stopper, cpu2);
> 		int err;
> 	retry:
> 		spin_lock_irq(&stopper1->lock);
> 		spin_lock_nested(&stopper2->lock, SINGLE_DEPTH_NESTING);
> 		err = -ENOENT;
> 		if (!cpu_active(cpu1) || !cpu_active(cpu2))
> 			goto unlock;
> 
> 		BUG_ON(!stopper1->enabled || !stopper2->enabled);
> 
> 		err = -EDEADLK;
> 		if (list_empty(&stopper1->stop_work.list) !=
> 		    list_empty(&stopper2->stop_work.list))
> 			goto unlock;
> 
> 		err = 0;
> 		list_add_tail(&work1->list, &stopper1->works);
> 		list_add_tail(&work2->list, &stopper2->works);
> 		wake_up_process(stopper1->thread);
> 		wake_up_process(stopper2->thread);
> 	unlock:
> 		spin_unlock(&stopper2->lock);
> 		spin_unlock_irq(&stopper1->lock);
> 
> 		if (unlikely(err == -EDEADLK)) {
> 			cond_resched();
> 			goto retry;
> 		}
> 		return err;
> 	}
> 
> 	int stop_two_cpus(unsigned int cpu1, unsigned int cpu2, cpu_stop_fn_t fn, void *arg)
> 	{
> 		struct cpu_stop_done done;
> 		struct cpu_stop_work work1, work2;
> 		struct multi_stop_data msdata = {
> 			.fn = fn,
> 			.data = arg,
> 			.num_threads = 2,
> 			.active_cpus = cpumask_of(cpu1),
> 		};
> 
> 		set_state(&msdata, MULTI_STOP_PREPARE);
> 		cpu_stop_init_done(&done, 2);
> 
> 		work1.fn   = work2.fn   = multi_cpu_stop;
> 		work1.arg  = work2.arg  = &msdata;
> 		work1.done = work2.done = &done;
> 
> 		if (cpu1 > cpu2)
> 			swap(cpu1, cpu2);
> 		if (cpu_stop_queue_two_works(cpu1, &work1, cpu2, &work2))
> 			return -ENOENT;
> 
> 		wait_for_completion(&done.completion);
> 		BUG_ON(!done.executed);
> 		return done.ret;
> 	}
> 
> Note the -EDEADLK check in cpu_stop_queue_two_works(). This avoids the
> race with stop_cpus(). We need to ensure that if we race with stop_cpus()
> then either stop_cpus() wins and queues both works on these CPU's, or
> we win this race and queue both works.
> 
> The "false positive" -EDEADLK can happen if we race with, say,
> stop_cpus(cpumask_of(cpu1)). But this is very unlikely (in fact it is
> always called with cpumask == cpu_online_mask), and in this case we
> need to wait anyway, so I think the "busy wait" loop can work.
> 
> As for cpu_active() checks... This was copied from the current code,
> but I think they should die later. This needs another cleanup, imo
> the stopper->enabled logic should be improved.
> 
> What do you think?
> 
> Oleg.
> 
> ------------------------------------------------------------------------------
> Subject: [PATCH] stop_machine: kill stop_cpus_lock and lg_double_lock/unlock()
> 
> ---
>  include/linux/lglock.h  |    5 ---
>  kernel/locking/lglock.c |   22 -----------
>  kernel/stop_machine.c   |   92 +++++++++++++++++++++++++---------------------
>  3 files changed, 50 insertions(+), 69 deletions(-)
> 
> diff --git a/include/linux/lglock.h b/include/linux/lglock.h
> index c92ebd1..0081f00 100644
> --- a/include/linux/lglock.h
> +++ b/include/linux/lglock.h
> @@ -52,15 +52,10 @@ struct lglock {
>  	static struct lglock name = { .lock = &name ## _lock }
>  
>  void lg_lock_init(struct lglock *lg, char *name);
> -
>  void lg_local_lock(struct lglock *lg);
>  void lg_local_unlock(struct lglock *lg);
>  void lg_local_lock_cpu(struct lglock *lg, int cpu);
>  void lg_local_unlock_cpu(struct lglock *lg, int cpu);
> -
> -void lg_double_lock(struct lglock *lg, int cpu1, int cpu2);
> -void lg_double_unlock(struct lglock *lg, int cpu1, int cpu2);
> -
>  void lg_global_lock(struct lglock *lg);
>  void lg_global_unlock(struct lglock *lg);
>  
> diff --git a/kernel/locking/lglock.c b/kernel/locking/lglock.c
> index 951cfcd..86ae2ae 100644
> --- a/kernel/locking/lglock.c
> +++ b/kernel/locking/lglock.c
> @@ -60,28 +60,6 @@ void lg_local_unlock_cpu(struct lglock *lg, int cpu)
>  }
>  EXPORT_SYMBOL(lg_local_unlock_cpu);
>  
> -void lg_double_lock(struct lglock *lg, int cpu1, int cpu2)
> -{
> -	BUG_ON(cpu1 == cpu2);
> -
> -	/* lock in cpu order, just like lg_global_lock */
> -	if (cpu2 < cpu1)
> -		swap(cpu1, cpu2);
> -
> -	preempt_disable();
> -	lock_acquire_shared(&lg->lock_dep_map, 0, 0, NULL, _RET_IP_);
> -	arch_spin_lock(per_cpu_ptr(lg->lock, cpu1));
> -	arch_spin_lock(per_cpu_ptr(lg->lock, cpu2));
> -}
> -
> -void lg_double_unlock(struct lglock *lg, int cpu1, int cpu2)
> -{
> -	lock_release(&lg->lock_dep_map, 1, _RET_IP_);
> -	arch_spin_unlock(per_cpu_ptr(lg->lock, cpu1));
> -	arch_spin_unlock(per_cpu_ptr(lg->lock, cpu2));
> -	preempt_enable();
> -}
> -
>  void lg_global_lock(struct lglock *lg)
>  {
>  	int i;
> diff --git a/kernel/stop_machine.c b/kernel/stop_machine.c
> index 12484e5..d53c86c 100644
> --- a/kernel/stop_machine.c
> +++ b/kernel/stop_machine.c
> @@ -20,7 +20,6 @@
>  #include <linux/kallsyms.h>
>  #include <linux/smpboot.h>
>  #include <linux/atomic.h>
> -#include <linux/lglock.h>
>  
>  /*
>   * Structure to determine completion condition and record errors.  May
> @@ -47,14 +46,6 @@ struct cpu_stopper {
>  static DEFINE_PER_CPU(struct cpu_stopper, cpu_stopper);
>  static bool stop_machine_initialized = false;
>  
> -/*
> - * Avoids a race between stop_two_cpus and global stop_cpus, where
> - * the stoppers could get queued up in reverse order, leading to
> - * system deadlock. Using an lglock means stop_two_cpus remains
> - * relatively cheap.
> - */
> -DEFINE_STATIC_LGLOCK(stop_cpus_lock);
> -
>  static void cpu_stop_init_done(struct cpu_stop_done *done, unsigned int nr_todo)
>  {
>  	memset(done, 0, sizeof(*done));
> @@ -213,6 +204,42 @@ static int multi_cpu_stop(void *data)
>  	return err;
>  }
>  
> +static int cpu_stop_queue_two_works(int cpu1, struct cpu_stop_work *work1,
> +				    int cpu2, struct cpu_stop_work *work2)
> +{
> +	struct cpu_stopper *stopper1 = per_cpu_ptr(&cpu_stopper, cpu1);
> +	struct cpu_stopper *stopper2 = per_cpu_ptr(&cpu_stopper, cpu2);
> +	int err;
> +retry:
> +	spin_lock_irq(&stopper1->lock);
> +	spin_lock_nested(&stopper2->lock, SINGLE_DEPTH_NESTING);
> +	err = -ENOENT;
> +	if (!cpu_active(cpu1) || !cpu_active(cpu2))
> +		goto unlock;
> +
> +	BUG_ON(!stopper1->enabled || !stopper2->enabled);
> +
> +	err = -EDEADLK;
> +	if (list_empty(&stopper1->stop_work.list) !=
> +	    list_empty(&stopper2->stop_work.list))
> +		goto unlock;
> +
> +	err = 0;
> +	list_add_tail(&work1->list, &stopper1->works);
> +	list_add_tail(&work2->list, &stopper2->works);
> +	wake_up_process(stopper1->thread);
> +	wake_up_process(stopper2->thread);
> +unlock:
> +	spin_unlock(&stopper2->lock);
> +	spin_unlock_irq(&stopper1->lock);
> +
> +	if (unlikely(err == -EDEADLK)) {
> +		cond_resched();
> +		goto retry;
> +	}
> +	return err;
> +}
> +
>  /**
>   * stop_two_cpus - stops two cpus
>   * @cpu1: the cpu to stop
> @@ -228,48 +255,28 @@ int stop_two_cpus(unsigned int cpu1, unsigned int cpu2, cpu_stop_fn_t fn, void *
>  {
>  	struct cpu_stop_done done;
>  	struct cpu_stop_work work1, work2;
> -	struct multi_stop_data msdata;
> -
> -	preempt_disable();
> -	msdata = (struct multi_stop_data){
> +	struct multi_stop_data msdata = {
>  		.fn = fn,
>  		.data = arg,
>  		.num_threads = 2,
>  		.active_cpus = cpumask_of(cpu1),
>  	};
>  
> -	work1 = work2 = (struct cpu_stop_work){
> -		.fn = multi_cpu_stop,
> -		.arg = &msdata,
> -		.done = &done
> -	};
> -
> -	cpu_stop_init_done(&done, 2);
>  	set_state(&msdata, MULTI_STOP_PREPARE);
> +	cpu_stop_init_done(&done, 2);
>  
> -	/*
> -	 * If we observe both CPUs active we know _cpu_down() cannot yet have
> -	 * queued its stop_machine works and therefore ours will get executed
> -	 * first. Or its not either one of our CPUs that's getting unplugged,
> -	 * in which case we don't care.
> -	 *
> -	 * This relies on the stopper workqueues to be FIFO.
> -	 */
> -	if (!cpu_active(cpu1) || !cpu_active(cpu2)) {
> -		preempt_enable();
> -		return -ENOENT;
> -	}
> -
> -	lg_double_lock(&stop_cpus_lock, cpu1, cpu2);
> -	cpu_stop_queue_work(cpu1, &work1);
> -	cpu_stop_queue_work(cpu2, &work2);
> -	lg_double_unlock(&stop_cpus_lock, cpu1, cpu2);
> +	work1.fn   = work2.fn   = multi_cpu_stop;
> +	work1.arg  = work2.arg  = &msdata;
> +	work1.done = work2.done = &done;
>  
> -	preempt_enable();
> +	if (cpu1 > cpu2)
> +		swap(cpu1, cpu2);
> +	if (cpu_stop_queue_two_works(cpu1, &work1, cpu2, &work2))
> +		return -ENOENT;
>  
>  	wait_for_completion(&done.completion);
> -
> -	return done.executed ? done.ret : -ENOENT;
> +	BUG_ON(!done.executed);
> +	return done.ret;
>  }
>  
>  /**
> @@ -308,7 +315,7 @@ static void queue_stop_cpus_work(const struct cpumask *cpumask,
>  	 * preempted by a stopper which might wait for other stoppers
>  	 * to enter @fn which can lead to deadlock.
>  	 */
> -	lg_global_lock(&stop_cpus_lock);
> +	preempt_disable();
>  	for_each_cpu(cpu, cpumask) {
>  		work = &per_cpu(cpu_stopper.stop_work, cpu);
>  		work->fn = fn;
> @@ -316,7 +323,7 @@ static void queue_stop_cpus_work(const struct cpumask *cpumask,
>  		work->done = done;
>  		cpu_stop_queue_work(cpu, work);
>  	}
> -	lg_global_unlock(&stop_cpus_lock);
> +	preempt_enable();
>  }
>  
>  static int __stop_cpus(const struct cpumask *cpumask,
> @@ -505,6 +512,7 @@ static int __init cpu_stop_init(void)
>  
>  		spin_lock_init(&stopper->lock);
>  		INIT_LIST_HEAD(&stopper->works);
> +		INIT_LIST_HEAD(&stopper->stop_work.list);
>  	}
>  
>  	BUG_ON(smpboot_register_percpu_thread(&cpu_stop_threads));
> -- 
> 1.5.5.1
> 

--
To unsubscribe from this list: send the line "unsubscribe linux-kernel" in
the body of a message to majordomo@xxxxxxxxxxxxxxx
More majordomo info at  http://vger.kernel.org/majordomo-info.html
Please read the FAQ at  http://www.tux.org/lkml/