lkml.org 
[lkml]   [2015]   [Nov]   [3]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
/
Date
From
SubjectRe: [PATCH v2] ipc/msg: Implement lockless pipelined wakeups
On Tue, 03 Nov 2015, Sebastian Andrzej Siewior wrote:

>@@ -577,26 +570,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();
>+ wake_q_add(wake_q, msr->r_tsk);
> msr->r_msg = ERR_PTR(-E2BIG);
> } 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).
>- */
>- smp_wmb();
>+ wake_q_add(wake_q, msr->r_tsk);
> msr->r_msg = msg;
>+ /*
>+ * Rely on the implicit cmpxchg barrier from
>+ * wake_q_add such that we can ensure that
>+ * updating msr->r_msg is the last write
>+ * operation: As once set, the receiver can
>+ * continue, and if we don't have the reference
>+ * count from the wake_q, yet, at that point we
>+ * can later have a use-after-free condition and
>+ * bogus wakeup.
>+ */

Not sure why you placed the comment here. Why not between smp_wmb() and the r_msg
write as we have it?

You might also want to add a reference to this comment in expunge_all(), which
does the same thing.

> [...]
>
> /* Lockless receive, part 2:
>- * Wait until pipelined_send or expunge_all are outside of
>- * wake_up_process(). There is a race with exit(), see
>- * ipc/mqueue.c for the details. The correct serialization
>- * ensures that a receiver cannot continue without the wakeup
>- * being visibible _before_ setting r_msg:
>+ * The work in pipelined_send() and expunge_all():
>+ * - Set pointer to message
>+ * - Queue the receiver task for later wakeup
>+ * - Wake up the process after the lock is dropped.
> *
>- * CPU 0 CPU 1
>- * <loop receiver>
>- * smp_rmb(); (A) <-- pair -. <waker thread>
>- * <load ->r_msg> | msr->r_msg = NULL;
>- * | wake_up_process();
>- * <continue> `------> smp_wmb(); (B)
>- * msr->r_msg = msg;
>- *
>- * Where (A) orders the message value read and where (B) orders
>- * the write to the r_msg -- done in both pipelined_send and
>- * expunge_all.
>+ * Should the process wake up before this wakeup (due to a
>+ * signal) it will either see the message and continue ...
> */
>- for (;;) {
>- /*
>- * Pairs with writer barrier in pipelined_send
>- * or expunge_all.
>- */
>- smp_rmb(); /* barrier (A) */
>- msg = (struct msg_msg *)msr_d.r_msg;
>- if (msg)
>- break;
>
>- /*
>- * The cpu_relax() call is a compiler barrier
>- * which forces everything in this loop to be
>- * re-loaded.
>- */
>- cpu_relax();
>- }
>-
>- /* Lockless receive, part 3:
>- * If there is a message or an error then accept it without
>- * locking.
>- */
>+ msg = msr_d.r_msg;

But you're getting rid of the barrier pairing (smp_rmb) we have in pipelined sends
and expunge_all, which is necesary even if we don't busy wait on nil. Likewise,
there's no need to remove the comment above that illustrates this.

Thanks,
Davidlohr


\
 
 \ /
  Last update: 2015-11-03 18:41    [W:0.057 / U:0.364 seconds]
©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site