Re: [PATCH 1/3] relay - clean up subbuf switch

[Andrew Morton <akpm@linux-foundation.org>]

On Tue, 23 Sep 2008 00:27:02 -0500
Tom Zanussi <zanussi@comcast.net> wrote:

> Clean up relay_switch_subbuf() and make waking up consumers optional.
>     
> Over time, relay_switch_subbuf() has accumulated some cruft - this
> patch cleans it up and at the same time makes available some of it
> available as common functions that any subbuf-switch implementor would
> need (this is partially in preparation for the next patch, which makes
> the subbuf-switch function completely replaceable).  It also removes
> the hard-coded reader wakeup and moves it into a replaceable callback
> called notify_consumers(); this allows any given tracer to implement
> consumer notification as it sees fit.
> 
> Signed-off-by: Tom Zanussi <tzanussi@gmail.com>
> 
> diff --git a/include/linux/relay.h b/include/linux/relay.h
> index 953fc05..17f0515 100644
> --- a/include/linux/relay.h
> +++ b/include/linux/relay.h
> @@ -159,6 +159,15 @@ struct rchan_callbacks
>  	 * The callback should return 0 if successful, negative if not.
>  	 */
>  	int (*remove_buf_file)(struct dentry *dentry);
> +
> +	/*
> +	 * notify_consumers - new sub-buffer available, let consumers know
> +	 * @buf: the channel buffer
> +	 *
> +	 * Called during sub-buffer switch.  Applications which don't
> +	 * want to notify anyone should implement an empty version.
> +	 */
> +        void (*notify_consumers)(struct rchan_buf *buf);
>  };

Does this comment format and placement get properly processed by the
kerneldoc tools?

>  /*
> @@ -186,6 +195,48 @@ extern size_t relay_switch_subbuf(struct rchan_buf *buf,
>  				  size_t length);
>  
>  /**
> + *	relay_event_toobig - is event too big to fit in a sub-buffer?
> + *	@buf: relay channel buffer
> + *	@length: length of event
> + *
> + *	Returns 1 if too big, 0 otherwise.
> + *
> + *	switch_subbuf() helper function.
> + */
> +static inline int relay_event_toobig(struct rchan_buf *buf, size_t length)
> +{
> +	return length > buf->chan->subbuf_size;
> +}
> +
> +/**
> + *	relay_update_filesize - increase relay file i_size by length
> + *	@buf: relay channel buffer
> + *	@length: length to add
> + *
> + *	switch_subbuf() helper function.
> + */
> +static inline void relay_update_filesize(struct rchan_buf *buf, size_t length)
> +{
> +	if (buf->dentry)
> +		buf->dentry->d_inode->i_size +=	length;
> +	else
> +		buf->early_bytes += length;
> +
> +	smp_mb();
> +}

What locking protects the non-atomic modification of the 64-bit i_size
here?

> +/**
> + *	relay_inc_produced - increase number of sub-buffers produced by 1
> + *	@buf: relay channel buffer
> + *
> + *	switch_subbuf() helper function.
> + */
> +static inline void relay_inc_produced(struct rchan_buf *buf)
> +{
> +	buf->subbufs_produced++;
> +}

This also needs caller-provided locking.  That's part of the function's
interface and should be documented,

> +/**
>   *	relay_write - write data into the channel
>   *	@chan: relay channel
>   *	@data: data to be written
> diff --git a/kernel/relay.c b/kernel/relay.c
> index 8d13a78..53652f1 100644
> --- a/kernel/relay.c
> +++ b/kernel/relay.c
> @@ -324,6 +324,21 @@ static int remove_buf_file_default_callback(struct dentry *dentry)
>  	return -EINVAL;
>  }
>  
> +/*
> + * notify_consumers() default callback.
> + */
> +static void notify_consumers_default_callback(struct rchan_buf *buf)
> +{
> +	if (waitqueue_active(&buf->read_wait))
> +		/*
> +		 * Calling wake_up_interruptible() from here
> +		 * will deadlock if we happen to be logging
> +		 * from the scheduler (trying to re-grab
> +		 * rq->lock), so defer it.
> +		 */
> +		__mod_timer(&buf->timer, jiffies + 1);
> +}
> +
>  /* relay channel default callbacks */
>  static struct rchan_callbacks default_channel_callbacks = {
>  	.subbuf_start = subbuf_start_default_callback,
> @@ -331,6 +346,7 @@ static struct rchan_callbacks default_channel_callbacks = {
>  	.buf_unmapped = buf_unmapped_default_callback,
>  	.create_buf_file = create_buf_file_default_callback,
>  	.remove_buf_file = remove_buf_file_default_callback,
> +	.notify_consumers = notify_consumers_default_callback,
>  };
>  
>  /**
> @@ -508,6 +524,8 @@ static void setup_callbacks(struct rchan *chan,
>  		cb->create_buf_file = create_buf_file_default_callback;
>  	if (!cb->remove_buf_file)
>  		cb->remove_buf_file = remove_buf_file_default_callback;
> +	if (!cb->notify_consumers)
> +		cb->notify_consumers = notify_consumers_default_callback;
>  	chan->cb = cb;
>  }
>  
> @@ -726,30 +744,17 @@ size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
>  	void *old, *new;
>  	size_t old_subbuf, new_subbuf;
>  
> -	if (unlikely(length > buf->chan->subbuf_size))
> +	if (unlikely(relay_event_toobig(buf, length)))
>  		goto toobig;
>  
>  	if (buf->offset != buf->chan->subbuf_size + 1) {
>  		buf->prev_padding = buf->chan->subbuf_size - buf->offset;
>  		old_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
>  		buf->padding[old_subbuf] = buf->prev_padding;
> -		buf->subbufs_produced++;
> -		if (buf->dentry)
> -			buf->dentry->d_inode->i_size +=
> -				buf->chan->subbuf_size -
> -				buf->padding[old_subbuf];
> -		else
> -			buf->early_bytes += buf->chan->subbuf_size -
> -					    buf->padding[old_subbuf];
> -		smp_mb();
> -		if (waitqueue_active(&buf->read_wait))
> -			/*
> -			 * Calling wake_up_interruptible() from here
> -			 * will deadlock if we happen to be logging
> -			 * from the scheduler (trying to re-grab
> -			 * rq->lock), so defer it.
> -			 */
> -			__mod_timer(&buf->timer, jiffies + 1);
> +		relay_inc_produced(buf);
> +		relay_update_filesize(buf, buf->chan->subbuf_size -
> +				      buf->padding[old_subbuf]);
> +		buf->chan->cb->notify_consumers(buf);
>  	}
>  
>  	old = buf->data;
> @@ -763,7 +768,7 @@ size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
>  	buf->data = new;
>  	buf->padding[new_subbuf] = 0;
>  
> -	if (unlikely(length + buf->offset > buf->chan->subbuf_size))
> +	if (unlikely(relay_event_toobig(buf, length + buf->offset)))
>  		goto toobig;

I think you can put the unlikely() into relay_event_toobig() and gcc
will dtrt.  If that has any value.

>  	return length;
>