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