dropped handling: was: Re: [PATCH printk v1 10/18] printk: nobkl: Add emit function and callback functions for atomic printing

[Petr Mladek]

On Thu 2023-03-02 21:02:10, John Ogness wrote:
> From: Thomas Gleixner <tglx@xxxxxxxxxxxxx>
> 
> Implement an emit function for non-BKL consoles to output printk
> messages. It utilizes the lockless printk_get_next_message() and
> console_prepend_dropped() functions to retrieve/build the output
> message. The emit function includes the required safety points to
> check for handover/takeover and calls a new write_atomic callback
> of the console driver to output the message. It also includes proper
> handling for updating the non-BKL console sequence number.
> 
> --- a/kernel/printk/printk_nobkl.c
> +++ b/kernel/printk/printk_nobkl.c
> @@ -1086,6 +1086,123 @@ bool console_exit_unsafe(struct cons_write_context *wctxt)
>  	return __console_update_unsafe(wctxt, false);
>  }
>  
> +/**
> + * cons_get_record - Fill the buffer with the next pending ringbuffer record
> + * @wctxt:	The write context which will be handed to the write function
> + *
> + * Returns:	True if there are records available. If the next record should
> + *		be printed, the output buffer is filled and @wctxt->outbuf
> + *		points to the text to print. If @wctxt->outbuf is NULL after
> + *		the call, the record should not be printed but the caller must
> + *		still update the console sequence number.
> + *
> + *		False means that there are no pending records anymore and the
> + *		printing can stop.
> + */
> +static bool cons_get_record(struct cons_write_context *wctxt)
> +{
> +	struct cons_context *ctxt = &ACCESS_PRIVATE(wctxt, ctxt);
> +	struct console *con = ctxt->console;
> +	bool is_extended = console_srcu_read_flags(con) & CON_EXTENDED;
> +	struct printk_message pmsg = {
> +		.pbufs = ctxt->pbufs,
> +	};
> +
> +	if (!printk_get_next_message(&pmsg, ctxt->newseq, is_extended, true))
> +		return false;
> +
> +	ctxt->newseq = pmsg.seq;
> +	ctxt->dropped += pmsg.dropped;
> +
> +	if (pmsg.outbuf_len == 0) {
> +		wctxt->outbuf = NULL;
> +	} else {
> +		if (ctxt->dropped && !is_extended)
> +			console_prepend_dropped(&pmsg, ctxt->dropped);
> +		wctxt->outbuf = &pmsg.pbufs->outbuf[0];
> +	}
> +
> +	wctxt->len = pmsg.outbuf_len;

This function seems to be needed only because we duplicate the information
in both struct printk_message and struct cons_write_context.

I think that we will not need this function at all if we bundle
struct printk_message into struct cons_context. I mean to replace:

struct cons_context {
	[...]
	struct printk_buffers	*pbufs;
	u64			newseq;
	unsigned long		dropped;
	[...]
}

with

struct cons_context {
	[...]
	struct printk_message pmsg;
	[...]
}

> +
> +	return true;
> +}
> +
> +/**
> + * cons_emit_record - Emit record in the acquired context
> + * @wctxt:	The write context that will be handed to the write function
> + *
> + * Returns:	False if the operation was aborted (takeover or handover).
> + *		True otherwise
> + *
> + * When false is returned, the caller is not allowed to touch console state.
> + * The console is owned by someone else. If the caller wants to print more
> + * it has to reacquire the console first.
> + *
> + * When true is returned, @wctxt->ctxt.backlog indicates whether there are
> + * still records pending in the ringbuffer,
> + */
> +static int __maybe_unused cons_emit_record(struct cons_write_context *wctxt)
> +{
> +	struct cons_context *ctxt = &ACCESS_PRIVATE(wctxt, ctxt);
> +	struct console *con = ctxt->console;
> +	bool done = false;
> +
> +	/*
> +	 * @con->dropped is not protected in case of hostile takeovers so
> +	 * the update below is racy. Annotate it accordingly.
> +	 */
> +	ctxt->dropped = data_race(READ_ONCE(con->dropped));
> +
> +	/* Fill the output buffer with the next record */
> +	ctxt->backlog = cons_get_record(wctxt);
> +	if (!ctxt->backlog)
> +		return true;
> +
> +	/* Safety point. Don't touch state in case of takeover */
> +	if (!console_can_proceed(wctxt))
> +		return false;
> +
> +	/* Counterpart to the read above */
> +	WRITE_ONCE(con->dropped, ctxt->dropped);

These racy hacks with ctxt-> dropped won't be needed if we bundle
strcut printk_message into struct cons_context.

> +
> +	/*
> +	 * In case of skipped records, Update sequence state in @con.
> +	 */
> +	if (!wctxt->outbuf)
> +		goto update;
> +
> +	/* Tell the driver about potential unsafe state */
> +	wctxt->unsafe = ctxt->state.unsafe;
> +
> +	if (!ctxt->thread && con->write_atomic) {
> +		done = con->write_atomic(con, wctxt);
> +	} else {
> +		cons_release(ctxt);
> +		WARN_ON_ONCE(1);
> +		return false;
> +	}
> +
> +	/* If not done, the write was aborted due to takeover */
> +	if (!done)
> +		return false;
> +
> +	/* If there was a dropped message, it has now been output. */
> +	if (ctxt->dropped) {
> +		ctxt->dropped = 0;
> +		/* Counterpart to the read above */
> +		WRITE_ONCE(con->dropped, ctxt->dropped);

I suggest to use atomic_t for con->dropped and use

		atomic_sub(ctxt->dropped, &con->dropped);

> +	}
> +update:
> +	ctxt->newseq++;
> +	/*
> +	 * The sequence update attempt is not part of console_release()
> +	 * because in panic situations the console is not released by
> +	 * the panic CPU until all records are written. On 32bit the
> +	 * sequence is separate from state anyway.
> +	 */
> +	return cons_seq_try_update(ctxt);
> +}

Best Regards,
Petr