Re: [RFC PATCH 1/3] lib/list_batch: A simple list insertion/deletion batching facility

[Waiman Long]

On 01/27/2016 11:34 AM, Peter Zijlstra wrote:
> On Tue, Jan 26, 2016 at 11:03:37AM -0500, Waiman Long wrote:
> > +static __always_inline void _list_batch_cmd(enum list_batch_cmd cmd,
> > +					    struct list_head *head,
> > +					    struct list_head *entry)
> > +{
> > +	if (cmd == lb_cmd_add)
> > +		list_add(entry, head);
> > +	else if (cmd == lb_cmd_del)
> > +		list_del(entry);
> > +	else /* cmd == lb_cmd_del_init */
> > +		list_del_init(entry);
> Maybe use switch(), GCC has fancy warns with enums and switch().

OK, I will look at the generated code to see if there is any difference.

> > +}
> > +static inline void do_list_batch(enum list_batch_cmd cmd, spinlock_t *lock,
> > +				   struct list_batch *batch,
> > +				   struct list_head *entry)
> > +{
> > +	/*
> > +	 * Fast path
> > +	 */
> > +	if (spin_trylock(lock)) {
> > +		_list_batch_cmd(cmd, batch->list, entry);
> > +		spin_unlock(lock);
> > _list_batch_cmd
> This is still quite a lot of code for an inline function

I expect the callers will call it with a constant cmd, thus optimizing 
out all the if conditional checks in _list_batch_cmd(). Taking the 
inline out will probably stop that optimization.

> > +		return;
> > +	}
> > +	do_list_batch_slowpath(cmd, lock, batch, entry);
> > +}
>
>
> > +void do_list_batch_slowpath(enum list_batch_cmd cmd, spinlock_t *lock,
> > +			    struct list_batch *batch, struct list_head *entry)
> > +{
> > +	struct list_batch_qnode node, *prev, *next, *nptr;
> > +	int loop;
> > +
> > +	/*
> > +	 * Put itself into the list_batch queue
> > +	 */
> > +	node.next  = NULL;
> > +	node.entry = entry;
> > +	node.cmd   = cmd;
> > +	node.state = lb_state_waiting;
> > +
> Here we rely on the release barrier implied by xchg() to ensure the node
> initialization is complete before the xchg() publishes the thing.
>
> But do we also need the acquire part of this barrier? From what I could
> tell, the primitive as a whole does not imply any ordering.

I think we probably won't need the acquire part, but I don't have a 
non-x86 machine that can really test out the more relaxed versions of 
the atomic ops. That is why I use the strict versions. We can always 
relax it later on with additional patches.

> > +	prev = xchg(&batch->tail,&node);
> > +
> > +	if (prev) {
> > +		WRITE_ONCE(prev->next,&node);
> > +		while (READ_ONCE(node.state) == lb_state_waiting)
> > +			cpu_relax();
> > +		if (node.state == lb_state_done)
> > +			return;
> > +		WARN_ON(node.state != lb_state_batch);
> > +	}
> > +
> > +	/*
> > +	 * We are now the queue head, we shold now acquire the lock and
> > +	 * process a batch of qnodes.
> > +	 */
> > +	loop = LB_BATCH_SIZE;
> Have you tried different sizes?

I have tried 64 and 128. Using 128 seems to give a bit better 
performance number.

> > +	next =&node;
> > +	spin_lock(lock);
> > +
> > +do_list_again:
> > +	do {
> > +		nptr = next;
> > +		_list_batch_cmd(nptr->cmd, batch->list, nptr->entry);
> > +		next = READ_ONCE(nptr->next);
> > +		/*
> > +		 * As soon as the state is marked lb_state_done, we
> > +		 * can no longer assume the content of *nptr as valid.
> > +		 * So we have to hold off marking it done until we no
> > +		 * longer need its content.
> > +		 *
> > +		 * The release barrier here is to make sure that we
> > +		 * won't access its content after marking it done.
> > +		 */
> > +		if (next)
> > +			smp_store_release(&nptr->state, lb_state_done);
> > +	} while (--loop&&  next);
> > +	if (!next) {
> > +		/*
> > +		 * The queue tail should equal to nptr, so clear it to
> > +		 * mark the queue as empty.
> > +		 */
> > +		if (cmpxchg(&batch->tail, nptr, NULL) != nptr) {
> > +			/*
> > +			 * Queue not empty, wait until the next pointer is
> > +			 * initialized.
> > +			 */
> > +			while (!(next = READ_ONCE(nptr->next)))
> > +				cpu_relax();
> > +		}
> > +		/* The above cmpxchg acts as a memory barrier */
> for what? :-)
>
> Also, if that cmpxchg() fails, it very much does _not_ act as one.
>
> I suspect you want smp_store_release() setting the state_done, just as
> above, and then use cmpxchg_relaxed().

You are right. I did forgot about there was no memory barrier guarantee 
when cmpxchg() fails. However, in that case, the READ_ONCE() and 
WRITE_ONCE() macros should still provide the necessary ordering, IMO. I 
can certainly change it to use cmpxchg_relaxed() and smp_store_release() 
instead.

> > +		WRITE_ONCE(nptr->state, lb_state_done);
> > +	}
> > +	if (next) {
> > +		if (loop)
> > +			goto do_list_again;	/* More qnodes to process */
> > +		/*
> > +		 * Mark the next qnode as head to process the next batch
> > +		 * of qnodes. The new queue head cannot proceed until we
> > +		 * release the lock.
> > +		 */
> > +		WRITE_ONCE(next->state, lb_state_batch);
> > +	}
> > +	spin_unlock(lock);
> > +}
> > +EXPORT_SYMBOL_GPL(do_list_batch_slowpath);

Cheers,
Longman