Messages in this thread | | | Date | Wed, 20 Jul 2016 17:16:12 -0700 | From | Davidlohr Bueso <> | Subject | Re: [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
| |