Re: [PATCH v2] ipc/msg: Implement lockless pipelined wakeups

[Sebastian Andrzej Siewior <bigeasy@linutronix.de>]

On 11/03/2015 06:30 PM, Davidlohr Bueso wrote:
> 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?

This follows the scheme we have in pipelined_send(). First wake_q_add()
then ->state (here we have the msg instead).

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

okay.

>> [...]
>>
>>         /* 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.

In pipelined_receive() (mqueue) there is the wake_q_add() with the
implicit cmpxchg barrier. The matching barrier pairing should be in
wq_sleep() but there is none. Why is it okay to have none there and I
need one here?

> Likewise,
> there's no need to remove the comment above that illustrates this.

I did not assume we need a barrier here. If we do, I keep it in the
comment / graphic but right I now, I think that it can go.

> Thanks,
> Davidlohr

Sebastian