lkml.org 
[lkml]   [2016]   [Jul]   [20]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
/
Date
From
SubjectRe: [PATCH v4] ipc/msg: Implement lockless pipelined wakeups
On Wed, 20 Jul 2016, Sebastian Andrzej Siewior wrote:

>This patch moves the wakeup_process() invocation so it is not done under
>the perm->lock by making use of a lockless wake_q. With this change, the
>waiter is woken up once the message has been assigned and it does not
>need to loop on SMP if the message points to NULL. In the signal case we
>still need to check the pointer under the lock to verify the state.
>
>This change should also avoid the introduction of preempt_disable() in
>-RT which avoids a busy-loop which pools for the NULL -> !NULL
>change if the waiter has a higher priority compared to the waker.
>
>This has been tested with Manred's pmsg-shared tool on a "AMD A10-7800
>Radeon R7, 12 Compute Cores 4C+8G":
>
>test | before | after | diff
>-----------------|------------|------------|----------
>pmsg-shared 8 60 | 19,347,422 | 30,442,191 | + ~57.34 %
>pmsg-shared 4 60 | 21,367,197 | 35,743,458 | + ~67.28 %
>pmsg-shared 2 60 | 22,884,224 | 24,278,200 | + ~6.09 %
>
>v3???v4: - drop smp_wmb in the error case as per Davidlohr
>v2???v3: - add smp_[rw]mb back including the usage graphic of them
> - use READ_ONCE / WRITE_ONCE after the removal of the volatile
> attribute.
>v1???v2:
> - msg_receiver.r_msg is no longer volatile. After all we no
> longer have that busy loop.
> - added a comment while we do wake_q_add() followed by the
> assignment of ->r_msg and not the other way around.
>
>Cc: Davidlohr Bueso <dave@stgolabs.net>
>Cc: Manfred Spraul <manfred@colorfullife.com>
>Cc: Andrew Morton <akpm@linux-foundation.org>
>Signed-off-by: Sebastian Andrzej Siewior <bigeasy@linutronix.de>

This looks good to me, and is now very similar to the posix flavor of
msg queues in this regard. fwiw I threw it under ltp msgqueue specific
tests without things breaking. Just a small comments below, otherwise
feel free to add my:

Reviewed-by: Davidlohr Bueso <dave@stgolabs.net>

>-static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
>+static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg,
>+ struct wake_q_head *wake_q)
> {
> struct msg_receiver *msr, *t;
>
>@@ -577,27 +571,23 @@ static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
>
> list_del(&msr->r_list);
> if (msr->r_maxsize < msg->m_ts) {
>- /* initialize pipelined send ordering */
>- msr->r_msg = NULL;
>- wake_up_process(msr->r_tsk);
>- /* barrier (B) see barrier comment below */
>- smp_wmb();
>- msr->r_msg = ERR_PTR(-E2BIG);
>+ WRITE_ONCE(msr->r_msg, ERR_PTR(-E2BIG));
>+ /*
>+ * rely on wake_q_add() barrier instead of
>+ * explicit smp_wmb
>+ */
>+ wake_q_add(wake_q, msr->r_tsk);
> } else {
>- msr->r_msg = NULL;
> msq->q_lrpid = task_pid_vnr(msr->r_tsk);
> msq->q_rtime = get_seconds();
>- wake_up_process(msr->r_tsk);
> /*
>- * Ensure that the wakeup is visible before
>- * setting r_msg, as the receiving can otherwise
>- * exit - once r_msg is set, the receiver can
>- * continue. See lockless receive part 1 and 2
>- * in do_msgrcv(). Barrier (B).
>+ * Ensure that we see the new r_msg after the
>+ * wake up or the old value forcing to take the
>+ * queue lock.
> */
>- smp_wmb();
>- msr->r_msg = msg;
>-
>+ WRITE_ONCE(msr->r_msg, msg);
>+ smp_wmb(); /* barrier (B) */
>+ wake_q_add(wake_q, msr->r_tsk);

Just as with expunge_all and the E2BIG case, could you remove that explicit
barrier (B) and just rely on wake_q_add?

Thanks,
Davidlohr

\
 
 \ /
  Last update: 2016-07-21 02:41    [W:0.259 / U:0.132 seconds]
©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site