lkml.org 
[lkml]   [2016]   [Jan]   [27]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
/
Date
From
SubjectRe: [RFC PATCH 1/3] lib/list_batch: A simple list insertion/deletion batching facility
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

\
 
 \ /
  Last update: 2016-01-27 22:01    [W:0.296 / U:0.132 seconds]
©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site