Date:   Thu, 6 Apr 2023 10:09:02 +0200
From:   Petr Mladek <pmladek@...e.com>
To:     John Ogness <john.ogness@...utronix.de>
Cc:     Sergey Senozhatsky <senozhatsky@...omium.org>,
        Steven Rostedt <rostedt@...dmis.org>,
        Thomas Gleixner <tglx@...utronix.de>,
        linux-kernel@...r.kernel.org,
        Greg Kroah-Hartman <gregkh@...uxfoundation.org>
Subject: wakeup synchronization: was: Re: [PATCH printk v1 11/18] printk:
 nobkl: Introduce printer threads

On Thu 2023-03-02 21:02:11, John Ogness wrote:
> From: Thomas Gleixner <tglx@...utronix.de>
> 
> Add the infrastructure to create a printer thread per console along
> with the required thread function, which is takeover/handover aware.
>
> --- a/kernel/printk/printk_nobkl.c
> +++ b/kernel/printk/printk_nobkl.c
> +/**
> + * cons_kthread_func - The printk thread function
> + * @__console:	Console to operate on
> + */
> +static int cons_kthread_func(void *__console)
> +{
> +	struct console *con = __console;
> +	struct cons_write_context wctxt = {
> +		.ctxt.console	= con,
> +		.ctxt.prio	= CONS_PRIO_NORMAL,
> +		.ctxt.thread	= 1,
> +	};
> +	struct cons_context *ctxt = &ACCESS_PRIVATE(&wctxt, ctxt);
> +	unsigned long flags;
> +	short con_flags;
> +	bool backlog;
> +	int cookie;
> +	int ret;
> +
> +	for (;;) {
> +		atomic_inc(&con->kthread_waiting);

Sigh, I really have hard times to scratch my head around the barriers
here. This part looks fine. The rcuwait_wait_event() provides full
barrier before checking the "condition".

But I am not sure about the counter part. It is in another patch.
IMHO, there should be a full barrier before checking
con->kthread_waiting. Something like this:

+  void cons_wake_threads(void)
+ {
+ 	struct console *con;
+ 	int cookie;
+

	/*
	 * Full barrier against rcuwait_wait_event() in	cons_kthread_func().
	 *
	 * The purpose of this barrier is to make sure that the new record is
	 * stored before checking con->kthread_waiting.
	 *
	 * It has the same purpose as the full barrier in rcuwait_wake_up().
	 * It makes sure that cons_kthread_should_wakeup() see the new record
	 * before going into sleep in rcuwait_wait_event().
	 *
	 * The extra barrier is needed here because rcuwait_wake_up() is called
	 * only when we see con->kthread_waiting set. We need to make sure
	 * that either we see con->kthread_waiting or cons_kthread_func()
	 * will see the new record when checking the condition in
	 * rcuwait_wait_event().
	 */
	smp_mb();

+ 	cookie = console_srcu_read_lock();
+ 	for_each_console_srcu(con) {
+ 		if (con->kthread && atomic_read(&con->kthread_waiting))
+ 			irq_work_queue(&con->irq_work);
+ 	}
+ 	console_srcu_read_unlock(cookie);
+ }

I think that I am right. But I am not in a good "see-barriers" mood so
I also might be wrong.

> +
> +		/*
> +		 * Provides a full memory barrier vs. cons_kthread_wake().
> +		 */
> +		ret = rcuwait_wait_event(&con->rcuwait,
> +					 cons_kthread_should_wakeup(con, ctxt),
> +					 TASK_INTERRUPTIBLE);

I am sorry but I would need some explanation for this. I am not
familiar with the rcuwait API. I looked at the code, commit messages,
and various users, and I am still not sure.

My assumption is that this allows to wait for an event on "con"
when the lifetime of this structure is synchronized using SRCU.
The counter-part calls rcuwait_wake_up() under srcu_read_lock().

I am afraid that it it is more complicated in our case.
We do not call rcuwait_wake_up() under srcu_read_lock().
We call it from an irq_work() that might be proceed later
after srcu_read_unlock().

IMHO, we need to make sure that there is no pending irq_work
and nobody could create a new one after exit from
unregister_console(). There seems to be irq_work_sync()
for this purpose.

> +
> +		atomic_dec(&con->kthread_waiting);
> +
> +		if (kthread_should_stop())
> +			break;
> +
> +		/* Wait was interrupted by a spurious signal, go back to sleep */
> +		if (ret)
> +			continue;
> +
> +		for (;;) {
[...]
> +
> +			if (console_is_usable(con, con_flags)) {
> +				/*
> +				 * If the emit fails, this context is no
> +				 * longer the owner. Abort the processing and
> +				 * wait for new records to print.
> +				 */
> +				if (!cons_emit_record(&wctxt))
> +					break;
> +
> +				backlog = ctxt->backlog;
> +			} else {
> +				backlog = false;
> +			}
[...]
> +	}
> +	return 0;
> +}
> +

Best Regards,
Petr

Powered by blists - more mailing lists