misc nits Re: [PATCH 1/2] printk: add lockless buffer

[Petr Mladek]

Hi,

there are few more small things that catched my eyes during review.
They are from the nits deparment.

On Tue 2020-01-28 17:25:47, John Ogness wrote:
> Introduce a multi-reader multi-writer lockless ringbuffer for storing
> the kernel log messages. Readers and writers may use their API from
> any context (including scheduler and NMI). This ringbuffer will make
> it possible to decouple printk() callers from any context, locking,
> or console constraints. It also makes it possible for readers to have
> full access to the ringbuffer contents at any time and context (for
> example from any panic situation).
> 
> diff --git a/kernel/printk/printk_ringbuffer.c b/kernel/printk/printk_ringbuffer.c
> new file mode 100644
> index 000000000000..796257f226ee
> --- /dev/null
> +++ b/kernel/printk/printk_ringbuffer.c
> +/**
> + * DOC: printk_ringbuffer overview

I really like the overview.

> +/* A data block: maps to the raw data within the data ring. */
> +struct prb_data_block {
> +	unsigned long	id;
> +	char		data[0];
> +};
> +
> +
> +static struct prb_data_block *to_block(struct prb_data_ring *data_ring,
> +				       unsigned long begin_lpos)
> +{
> +	char *data = &data_ring->data[DATA_INDEX(data_ring, begin_lpos)];
> +
> +	return (struct prb_data_block *)data;

Nit: Please, use "blk" instead of "data". I was slightly confused
because "data" is also one member of struct prb_data_block.


> +/* The possible responses of a descriptor state-query. */
> +enum desc_state {
> +	desc_miss,	/* ID mismatch */
> +	desc_reserved,	/* reserved, but still in use by writer */
> +	desc_committed, /* committed, writer is done */
> +	desc_reusable,	/* free, not used by any writer */

s/not used/not yet used/

> +};

[...]

> +EXPORT_SYMBOL(prb_reserve);

Please, do not export symbols if there are no plans to actually
use them from modules. It will be easier to rework the code
in the future. Nobody would need to worry about external
users.

Please, do so everywhere in the patchset.

> +/*
> + * Given @blk_lpos, return a pointer to the raw data from the data block
> + * and calculate the size of the data part. A NULL pointer is returned
> + * if @blk_lpos specifies values that could never be legal.
> + *
> + * This function (used by readers) performs strict validation on the lpos
> + * values to possibly detect bugs in the writer code. A WARN_ON_ONCE() is
> + * triggered if an internal error is detected.
> + */
> +static char *get_data(struct prb_data_ring *data_ring,
> +		      struct prb_data_blk_lpos *blk_lpos,
> +		      unsigned long *data_size)
> +{
> +	struct prb_data_block *db;
> +
> +	/* Data-less data block description. */
> +	if (blk_lpos->begin == INVALID_LPOS &&
> +	    blk_lpos->next == INVALID_LPOS) {
> +		return NULL;

Nit: There is no need for "else" after return. checkpatch.pl usually
complains about it ;-)

> +
> +	/* Regular data block: @begin less than @next and in same wrap. */
> +	} else if (DATA_WRAPS(data_ring, blk_lpos->begin) ==
> +		   DATA_WRAPS(data_ring, blk_lpos->next) &&
> +		   blk_lpos->begin < blk_lpos->next) {
> +		db = to_block(data_ring, blk_lpos->begin);
> +		*data_size = blk_lpos->next - blk_lpos->begin;
> +
> +	/* Wrapping data block: @begin is one wrap behind @next. */
> +	} else if (DATA_WRAPS(data_ring,
> +			      blk_lpos->begin + DATA_SIZE(data_ring)) ==
> +		   DATA_WRAPS(data_ring, blk_lpos->next)) {
> +		db = to_block(data_ring, 0);
> +		*data_size = DATA_INDEX(data_ring, blk_lpos->next);
> +
> +	/* Illegal block description. */
> +	} else {
> +		WARN_ON_ONCE(1);
> +		return NULL;
> +	}
> +
> +	/* A valid data block will always be aligned to the ID size. */
> +	if (WARN_ON_ONCE(blk_lpos->begin !=
> +			 ALIGN(blk_lpos->begin, sizeof(db->id))) ||
> +	    WARN_ON_ONCE(blk_lpos->next !=
> +			 ALIGN(blk_lpos->next, sizeof(db->id)))) {
> +		return NULL;
> +	}
> +
> +	/* A valid data block will always have at least an ID. */
> +	if (WARN_ON_ONCE(*data_size < sizeof(db->id)))
> +		return NULL;
> +
> +	/* Subtract descriptor ID space from size to reflect data size. */
> +	*data_size -= sizeof(db->id);
> +
> +	return &db->data[0];
> +}
> +
> +/*
> + * Read the record @id and verify that it is committed and has the sequence
> + * number @seq. On success, 0 is returned.
> + *
> + * Error return values:
> + * -EINVAL: A committed record @seq does not exist.
> + * -ENOENT: The record @seq exists, but its data is not available. This is a
> + *          valid record, so readers should continue with the next seq.
> + */
> +static int desc_read_committed(struct prb_desc_ring *desc_ring,
> +			       unsigned long id, u64 seq,
> +			       struct prb_desc *desc)
> +{

I was few times confused whether this function reads the descriptor
a safe way or not.

Please, rename it to make it clear that does only a check.
For example, check_state_commited().

> +	enum desc_state d_state;
> +
> +	d_state = desc_read(desc_ring, id, desc);
> +	if (desc->info.seq != seq)
> +		return -EINVAL;
> +	else if (d_state == desc_reusable)
> +		return -ENOENT;
> +	else if (d_state != desc_committed)
> +		return -EINVAL;
> +
> +	return 0;
> +}

Best Regards,
Petr

PS: I am sorry that the review took me so much time. I was sick, had
some other work. And I wanted to have a free mind when thinking
about this lockless stuff. I think that it actually helped me to
realize the need of more barriers discussed in the other thread.