Re: [PATCH] Relay CPU hotplug support

Mathieu Desnoyers writes:
 > Hi Tom,
 > 
 > Here is my version of the relay patch for CPU hotplug support. It uses a
 > channel list within relay, as you suggested. The patch is based on a 2.6.18
 > kernel.
 > 
 > Comments are welcome,
 > 
 > Mathieu
 > 

Hi Mathieu,

Comments below...

 > 
 > diff --git a/include/linux/relay.h b/include/linux/relay.h
 > index 24accb4..6190fa0 100644
 > --- a/include/linux/relay.h
 > +++ b/include/linux/relay.h
 > @@ -24,7 +24,7 @@ #define FIX_SIZE(x) ((((x) - 1) & PAGE_M
 >  /*
 >   * Tracks changes to rchan/rchan_buf structs
 >   */
 > -#define RELAYFS_CHANNEL_VERSION		6
 > +#define RELAYFS_CHANNEL_VERSION		7
 > 
 >  /*
 >   * Per-cpu relay channel buffer
 > @@ -41,6 +41,7 @@ struct rchan_buf
 >  	struct work_struct wake_readers; /* reader wake-up work struct */
 >  	struct dentry *dentry;		/* channel file dentry */
 >  	struct kref kref;		/* channel buffer refcount */
 > +	struct kref krefw;		/* channel buffer writers
refcount */

I still don't think this is needed (see below and previous reply)

 >  	struct page **page_array;	/* array of current buffer pages */
 >  	unsigned int page_count;	/* number of current buffer pages */
 >  	unsigned int finalized;		/* buffer has been finalized */
 > @@ -64,6 +65,10 @@ struct rchan
 >  	void *private_data;		/* for user-defined data */
 >  	size_t last_toobig;		/* tried to log event > subbuf size */
 >  	struct rchan_buf *buf[NR_CPUS]; /* per-cpu channel buffers */
 > +	int is_global;			/* One global buffer ? */
 > +	struct list_head list;		/* for channel list */
 > +	struct dentry *parent;		/* parent dentry passed to open */
 > +	char base_filename[NAME_MAX];	/* saved base filename */
 >  };
 > 
 >  /*
 > @@ -133,6 +138,8 @@ struct rchan_callbacks
 >  	 * cause relay_open() to create a single global buffer rather
 >  	 * than the default set of per-cpu buffers.
 >  	 *
 > +	 * buf->chan->buf[buf->cpu] is not set when this callback is called.
 > +	 *

If this is important, should we arrange to have it set when the
callback is called?

 >  	 * See Documentation/filesystems/relayfs.txt for more info.
 >  	 */
 >  	struct dentry *(*create_buf_file)(const char *filename,
 > @@ -162,7 +169,8 @@ struct rchan *relay_open(const char *bas
 >  			 struct dentry *parent,
 >  			 size_t subbuf_size,
 >  			 size_t n_subbufs,
 > -			 struct rchan_callbacks *cb);
 > +			 struct rchan_callbacks *cb,
 > +			 void *private_data);
 >  extern void relay_close(struct rchan *chan);
 >  extern void relay_flush(struct rchan *chan);
 >  extern void relay_subbufs_consumed(struct rchan *chan,
 > 
 > diff --git a/kernel/relay.c b/kernel/relay.c
 > index 33345e7..8dab79c 100644
 > --- a/kernel/relay.c
 > +++ b/kernel/relay.c
 > @@ -7,8 +7,14 @@
 >   * Copyright (C) 1999-2005 - Karim Yaghmour ([email protected])
 >   *
 >   * Moved to kernel/relay.c by Paul Mundt, 2006.
 > + * November 2006 - CPU hotplug support by Mathieu Desnoyers
 > + * 	([email protected])
 >   *
 >   * This file is released under the GPL.
 > + *
 > + * In case CPU_DEAD CPU hotplug is someday implemented, we protect against
 > + * buffer removal by locking cpu hotplug around each test on buf[cpu] being
 > + * NULL everywhere in the file.
 >   */
 >  #include <linux/errno.h>
 >  #include <linux/stddef.h>
 > @@ -18,6 +24,11 @@ #include <linux/string.h>
 >  #include <linux/relay.h>
 >  #include <linux/vmalloc.h>
 >  #include <linux/mm.h>
 > +#include <linux/cpu.h>
 > +
 > +/* list of open channels, for cpu hotplug */
 > +static DEFINE_MUTEX(relay_channels_mutex);
 > +static LIST_HEAD(relay_channels);
 > 
 >  /*
 >   * close() vm_op implementation for relay file mapping.
 > @@ -189,6 +200,7 @@ void relay_destroy_buf(struct rchan_buf
 >  	}
 >  	kfree(buf->padding);
 >  	kfree(buf);
 > +	chan->buf[buf->cpu] = NULL;
 >  	kref_put(&chan->kref, relay_destroy_channel);
 >  }
 > 
 > @@ -327,6 +339,7 @@ static inline void __relay_reset(struct
 >  	if (init) {
 >  		init_waitqueue_head(&buf->read_wait);
 >  		kref_init(&buf->kref);
 > +		kref_init(&buf->krefw);
 >  		INIT_WORK(&buf->wake_readers, NULL, NULL);
 >  	} else {
 >  		cancel_delayed_work(&buf->wake_readers);
 > @@ -360,70 +373,104 @@ static inline void __relay_reset(struct
 >  void relay_reset(struct rchan *chan)
 >  {
 >  	unsigned int i;
 > -	struct rchan_buf *prev = NULL;
 > 
 >  	if (!chan)
 >  		return;
 > -
 > -	for (i = 0; i < NR_CPUS; i++) {
 > -		if (!chan->buf[i] || chan->buf[i] == prev)
 > -			break;
 > -		__relay_reset(chan->buf[i], 0);
 > -		prev = chan->buf[i];
 > +	lock_cpu_hotplug();
 > +	if (chan->is_global) {
 > +		if (chan->buf[0])
 > +			__relay_reset(chan->buf[0], 0);
 > +	} else {
 > +		for_each_online_cpu(i)
 > +			if (chan->buf[i])
 > +				__relay_reset(chan->buf[i], 0);
 >  	}
 > +	unlock_cpu_hotplug();
 >  }

If the is_global case isn't affected by cpu hotplug, no need to have
those cases inside lock_cpu_hotplug().

 >  EXPORT_SYMBOL_GPL(relay_reset);
 > 
 >  /**
 >   *	relay_open_buf - create a new relay channel buffer
 >   *
 > - *	Internal - used by relay_open().
 > + *	used by relay_open() and CPU hotplug.
 >   */
 > -static struct rchan_buf *relay_open_buf(struct rchan *chan,
 > -					const char *filename,
 > -					struct dentry *parent,
 > -					int *is_global)
 > +struct rchan_buf *relay_open_buf(struct rchan *chan, unsigned int cpu)
 >  {
 > -	struct rchan_buf *buf;
 > +	struct rchan_buf *buf = NULL;
 >  	struct dentry *dentry;
 > +	char *tmpname;
 > 
 > -	if (*is_global)
 > +	if (chan->is_global) {
 > +		kref_get(&chan->buf[0]->krefw);
 >  		return chan->buf[0];
 > +	}
 > + + tmpname = kmalloc(NAME_MAX + 1, GFP_KERNEL); + if (!tmpname) +
 > goto end; + sprintf(tmpname, "%s%d", chan->base_filename, cpu);
 > 
 >  	buf = relay_create_buf(chan);
 >  	if (!buf)
 > -		return NULL;
 > +		goto free_name;
 > +
 > +	buf->cpu = cpu;
 > +	__relay_reset(buf, 1);
 > 
 >  	/* Create file in fs */
 > -	dentry = chan->cb->create_buf_file(filename, parent, S_IRUSR,
 > -					   buf, is_global);
 > -	if (!dentry) {
 > -		relay_destroy_buf(buf);
 > -		return NULL;
 > +	dentry = chan->cb->create_buf_file(tmpname, chan->parent, S_IRUSR,
 > +					   buf, &chan->is_global);
 > +	if (!dentry)
 > +		goto free_buf;
 > +
 > +	if(chan->is_global) {
 > +		chan->buf[0] = buf;

If we added a foreach_possible_cpu() chan_buf[i] = buf here, we should
then be able to ignore hotcpu events altogether for global channels.

 > +		buf->cpu = 0;
 >  	}
 > 

Also, since we're now setting chan->buf[0] here, it might make sense
to do the same for the non-global case as well and change the return
value from struct rchan_buf * to an integer return code, since it no
longer needs to return a pointer to the buf.  If it helps, we might
also be able to move it to before create_buf_file() and have access to
buf->chan->buf[], but I haven't thought about whether it would break
anything.  On second thought, it's probably cleaner to leave it as it
is.

 >  	buf->dentry = dentry;
 > -	__relay_reset(buf, 1);
 > +	goto free_name;
 > 
 > +free_buf:
 > +	relay_destroy_buf(buf);
 > +free_name:
 > +	kfree(tmpname);
 > +end:
 >  	return buf;
 >  }
 > +EXPORT_SYMBOL_GPL(relay_open_buf);
 > 
 >  /**
 > - *	relay_close_buf - close a channel buffer
 > - *	@buf: channel buffer
 > + *	relay_close_write_buf - close write to a channel buffer
 > + *	@kref: buffer writers reference
 >   *
 >   *	Marks the buffer finalized and restores the default callbacks.
 >   *	The channel buffer and channel buffer data structure are then freed
 >   *	automatically when the last reference is given up.
 >   */
 > -static inline void relay_close_buf(struct rchan_buf *buf)
 > +static void relay_close_write_buf(struct kref *kref)
 >  {
 > +	struct rchan_buf *buf = container_of(kref, struct rchan_buf, krefw);
 > +
 >  	buf->finalized = 1;
 >  	cancel_delayed_work(&buf->wake_readers);
 >  	flush_scheduled_work();
 >  	kref_put(&buf->kref, relay_remove_buf);
 >  }
 > 
 > +/**
 > + *	relay_close_buf - close a channel buffer
 > + *	@buf: channel buffer
 > + *	
 > + *	Remove a writer reference.
 > + */
 > +
 > +void relay_close_buf(struct rchan_buf *buf)
 > +{
 > +	kref_put(&buf->krefw, relay_close_write_buf);
 > +}
 > +EXPORT_SYMBOL_GPL(relay_close_buf);
 > +
 >  static inline void setup_callbacks(struct rchan *chan,
 >  				   struct rchan_callbacks *cb)
 >  {
 > @@ -446,6 +493,45 @@ static inline void setup_callbacks(struc
 >  }
 > 
 >  /**
 > + *
 > + * 	relay_hotcpu_callback - CPU hotplug callback
 > + * 	@nb: notifier block
 > + * 	@action: hotplug action to take
 > + * 	@hcpu: CPU number
 > + *
 > + * 	Returns the success/failure of the operation. (NOTIFY_OK, NOTIFY_BAD)
 > + */
 > +static int __cpuinit relay_hotcpu_callback(struct notifier_block *nb,
 > +				unsigned long action,
 > +				void *hcpu)
 > +{
 > +	unsigned int hotcpu = (unsigned long)hcpu;
 > +	struct rchan *chan;
 > +
 > +	switch(action) {
 > +	case CPU_UP_PREPARE:
 > +		mutex_lock(&relay_channels_mutex);
 > +		list_for_each_entry(chan, &relay_channels, list) {
 > +			chan->buf[hotcpu] = relay_open_buf(chan, hotcpu);
 > +			if(!chan->buf[hotcpu]) {
 > +				printk(KERN_ERR
 > +					"relay_hotcpu_callback: cpu %d buffer "
 > +					"creation failed\n", hotcpu);
 > +				mutex_unlock(&relay_channels_mutex);
 > +				return NOTIFY_BAD;
 > +			}
 > +		}
 > +		mutex_unlock(&relay_channels_mutex);
 > +		break;
 > +	case CPU_DEAD:
 > +		/* No need to flush the cpu : will be flushed upon
 > +		 * final relay_flush() call. */
 > +		break;
 > +	}
 > +	return NOTIFY_OK;
 > +}
 > +
 > +/**
 >   *	relay_open - create a new relay channel
 >   *	@base_filename: base name of files to create
 >   *	@parent: dentry of parent directory, NULL for root directory
 > @@ -464,13 +550,11 @@ struct rchan *relay_open(const char *bas
 >  			 struct dentry *parent,
 >  			 size_t subbuf_size,
 >  			 size_t n_subbufs,
 > -			 struct rchan_callbacks *cb)
 > +			 struct rchan_callbacks *cb,
 > +			 void *private_data)
 >  {
 >  	unsigned int i;
 >  	struct rchan *chan;
 > -	char *tmpname;
 > -	int is_global = 0;
 > -
 >  	if (!base_filename)
 >  		return NULL;
 > 
 > @@ -485,38 +569,40 @@ struct rchan *relay_open(const char *bas
 >  	chan->n_subbufs = n_subbufs;
 >  	chan->subbuf_size = subbuf_size;
 >  	chan->alloc_size = FIX_SIZE(subbuf_size * n_subbufs);
 > +	chan->is_global = 0;

This should already be 0 from the kcalloc.


Tom