[Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Dmitry Adamushko]

[-- Attachment #1: Type: text/plain, Size: 3401 bytes --]




Hi,

I failed to find the original message from Sebastian that led to the
following change:

-----
2005-11-09  Philippe Gerum  <rpm@xenomai.org>

        * nucleus/pipe.c (xnpipe_disconnect): Flush the output queue
        upon closure. Issue spotted by Sebastian Smolorz.
        (xnpipe_release): Flush the input queue upon closure.
-----

There is one more issue as follows. The reason is that
xnpipe_open/release() are not atomic.

Briefly, a newly connected standard linux thread may get a message that
have been posted by the real-time side long time before a connection has
been opened by that thread. That message has been placed to the queue at
the time when there was another linux thread connected to the pipe.

err... I'm not sure now that we should fight against that :confused a bit:

--- details ---

THREAD #1 on CPU #1 (normal linux thread):

xnpipe_release()
{

// ok, we are locked here and noone may access the inq queue, let's free
all the messages if any
...
        if (state->output_handler != NULL)
            {
            while ((holder = getq(&state->outq)) != NULL)

state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
            }
        while ((holder = getq(&state->inq)) != NULL)
            {
            if (state->input_handler != NULL)

state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
            else if (state->alloc_handler == NULL)
                xnfree(link2mh(holder));
            }

...
    if (waitqueue_active(&state->readq))
        wake_up_interruptible_all(&state->readq);

    if (state->asyncq) /* Clear the async queue */
        {
        xnlock_get_irqsave(&nklock,s);
        removeq(&xnpipe_asyncq,&state->alink);
        clrbits(state->status,XNPIPE_USER_SIGIO);
        xnlock_put_irqrestore(&nklock,s);
        fasync_helper(-1,file,0,&state->asyncq);
        }

// here the lock is not held.

(*** EIP ***) -------> THREAD #1 is about to drop the XNPIPE_USER_CONN
flag.

    /* Free the state object. Since that time it can be open by someone
else */
    clrbits(state->status,XNPIPE_USER_CONN);
}


THREAD #2 on CPU #2 (real-time thread)

xnpipe_send()
{
...
// locked section

   xnlock_get_irqsave(&nklock,s);

// actually, the following 2 checks should be re-ordered :)

(*** EIP ***)     <----- THREAD #1 doesn't dropped the XNPIPE_USER_CONN
flag yet but the pipe is almost non-valid!

    if (!testbits(state->status,XNPIPE_USER_CONN))
        {
        xnlock_put_irqrestore(&nklock,s);
        return -EPIPE;
        }

    if (!testbits(state->status,XNPIPE_KERN_CONN))
        {
        xnlock_put_irqrestore(&nklock,s);
        return -EBADF;
        }

    inith(xnpipe_m_link(mh));
    xnpipe_m_size(mh) = size - sizeof(*mh);
    state->ionrd += xnpipe_m_size(mh);

// here the message is successfully added to the outq

    if (flags & XNPIPE_URGENT)
        prependq(&state->outq,xnpipe_m_link(mh));
    else
        appendq(&state->outq,xnpipe_m_link(mh));

}

...

In the mean time, THREAD #1 drops the XNPIPE_USER_FLAG so another standard
linux thread may open a pipe. When that happens, that thread will find a
message that have been posted when the old connection existed.

err... so is it a problem regarding desired behaviour of pipes?


---

Dmitry


[-- Attachment #2: Type: text/html, Size: 6060 bytes --]

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Dmitry Adamushko]

[-- Attachment #1: Type: text/plain, Size: 777 bytes --]


yep, it's a problem since data may be client-dependent. In such a case, for
a new client old messages are just irrelevant. And xnpipe_release() cleans
up the queus but, well, does it too earlier.

so,

1) should xnpipe_open_handler() and xnpipe_close_handler() be called
without holding a lock?

they are not used currently so I can't see.

I intend to make xnpipe_open() completely atomic.

2) the cleaning of the queues (inq, outq) must take place atomically at the
time when XNPIPE_USER_CONN is dropped.

it's about something like

lock();

__clrbits(state->status,XNPIPE_USER_CONN);

// clean up all the queues

unlock();

it looks like we can't make the whole xnpipe_release() atomic because of
PREEMPT_RT + wake_up_interruptible_all() things, right? Or no.


---

Dmitry

[-- Attachment #2: Type: text/html, Size: 1076 bytes --]

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Philippe Gerum]

Dmitry Adamushko wrote:
> yep, it's a problem since data may be client-dependent. In such a case, 
> for a new client old messages are just irrelevant. And xnpipe_release() 
> cleans up the queus but, well, does it too earlier.
> 
> so,
> 
> 1) should xnpipe_open_handler() and xnpipe_close_handler() be called 
> without holding a lock?
> 

Yes, it on purpose. I know this make things a bit trickier since this breaks the overall 
atomicity of the caller, but open/close hooks are expected to initiate/finalize 
communication sessions, and that may take an unbounded amount of time, so we definitely 
don't want to do this with the superlock being held.

> they are not used currently so I can't see.
> 
> I intend to make xnpipe_open() completely atomic.
> 
> 2) the cleaning of the queues (inq, outq) must take place atomically at 
> the time when XNPIPE_USER_CONN is dropped.
> 
> it's about something like
> 
> lock();
> 
> __clrbits(state->status,XNPIPE_USER_CONN);
> 
> // clean up all the queues
> 
> unlock();
> 
> it looks like we can't make the whole xnpipe_release() atomic because of 
> PREEMPT_RT + wake_up_interruptible_all() things, right? Or no.

You must _never_ _ever_ reschedule with the nucleus lock held; this is a major cause of 
jittery I recently stumbled upon that was induced by xnpipe_read_wait() at that time. So 
indeed, xnpipe_release() cannot be made atomic this way under a fully preemptible kernel.

> 
> 
> ---
> 
> Dmitry
> 


-- 

Philippe.

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Philippe Gerum]

Philippe Gerum wrote:
> Dmitry Adamushko wrote:
> 
>> yep, it's a problem since data may be client-dependent. In such a 
>> case, for a new client old messages are just irrelevant. And 
>> xnpipe_release() cleans up the queus but, well, does it too earlier.
>>
>> so,
>>
>> 1) should xnpipe_open_handler() and xnpipe_close_handler() be called 
>> without holding a lock?
>>
> 
> Yes, it on purpose. I know this make things a bit trickier since this 
> breaks the overall atomicity of the caller, but open/close hooks are 
> expected to initiate/finalize communication sessions, and that may take 
> an unbounded amount of time, so we definitely don't want to do this with 
> the superlock being held.
> 
>> they are not used currently so I can't see.
>>
>> I intend to make xnpipe_open() completely atomic.
>>
>> 2) the cleaning of the queues (inq, outq) must take place atomically 
>> at the time when XNPIPE_USER_CONN is dropped.
>>
>> it's about something like
>>
>> lock();
>>
>> __clrbits(state->status,XNPIPE_USER_CONN);
>>
>> // clean up all the queues
>>
>> unlock();
>>
>> it looks like we can't make the whole xnpipe_release() atomic because 
>> of PREEMPT_RT + wake_up_interruptible_all() things, right? Or no.
> 
> 
> You must _never_ _ever_ reschedule

"reschedule" in the Linux sense here; entering Xenomai's rescheduling procedure with the 
superlock held is of course perfectly valid.

  with the nucleus lock held; this is a
> major cause of jittery I recently stumbled upon that was induced by 
> xnpipe_read_wait() at that time. So indeed, xnpipe_release() cannot be 
> made atomic this way under a fully preemptible kernel.
> 
>>
>>
>> ---
>>
>> Dmitry
>>
> 
> 


-- 

Philippe.

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Dmitry Adamushko]

[-- Attachment #1: Type: text/plain, Size: 2301 bytes --]



Philippe Gerum <rpm@xenomai.org> wrote on 18.11.2005 11:14:26:

> > ...
> >
> > it looks like we can't make the whole xnpipe_release() atomic because
of
> > PREEMPT_RT + wake_up_interruptible_all() things, right? Or no.
>
> You must _never_ _ever_ reschedule with the nucleus lock held; this
> is a major cause of
> jittery I recently stumbled upon that was induced by
> xnpipe_read_wait() at that time. So
> indeed, xnpipe_release() cannot be made atomic this way under a
> fully preemptible kernel.

Yep.

Now keeping in mind the observation I have made yesterday, it looks like,
in fact, there is no need in wake_up_*(readers) call in
file_operations::release(). There is nobody to be woken up at the time when
release() is called:

1) The reference counter of "file" object is 0, i.e. there are no readers
since read() increases a counter before getting blocked.

2) noone else can use anew that "file" object since close() does the
following:

 filp = files->fd[fd];
 if (!filp)
      goto out_unlock;
 files->fd[fd] = NULL;        <--- it's invalid from now on

so it's not possible that some new readers may occur when a counter == 0.

Hm.. but we still have fasync_helper(-1,file,0,&state->asyncq); which is
about sending a signal and that's perfectly valid (a file::counter is not
involved here). And that call may lead to re-scheduling (linux
re-scheduling of course) so we can't put it in a blocked section.

So the best way I see is to have something like():

xnpipe_drop_user_conn()
{
      xnlock_get_irqsave(&nklock,s);

       while ((holder = getq(&state->outq)) != NULL)

state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
            }

        while ((holder = getq(&state->inq)) != NULL)
            {
            if (state->input_handler != NULL)

state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
            else if (state->alloc_handler == NULL)
                xnfree(link2mh(holder));
            }

      __clrbits(state->status,XNPIPE_USER_CONN);

      xnlock_put_irqrestore(&nklock,s);
}

and call it everywhere instead of clrbits(state->status,XNPIPE_USER_CONN);

This way we may be sure there are no pending messages left.


> --
>
> Philippe.

---

Dmitry

[-- Attachment #2: Type: text/html, Size: 3358 bytes --]

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Philippe Gerum]

Dmitry Adamushko wrote:
> Philippe Gerum <rpm@xenomai.org> wrote on 18.11.2005 11:14:26:
> 
>  > > ...
>  > >
>  > > it looks like we can't make the whole xnpipe_release() atomic 
> because of
>  > > PREEMPT_RT + wake_up_interruptible_all() things, right? Or no.
>  >
>  > You must _never_ _ever_ reschedule with the nucleus lock held; this
>  > is a major cause of
>  > jittery I recently stumbled upon that was induced by
>  > xnpipe_read_wait() at that time. So
>  > indeed, xnpipe_release() cannot be made atomic this way under a
>  > fully preemptible kernel.
> 
> Yep.
> 
> Now keeping in mind the observation I have made yesterday, it looks 
> like, in fact, there is no need in wake_up_*(readers) call in 
> file_operations::release(). There is nobody to be woken up at the time 
> when release() is called:
> 
> 1) The reference counter of "file" object is 0, i.e. there are no 
> readers since read() increases a counter before getting blocked.
> 
> 2) noone else can use anew that "file" object since close() does the 
> following:
> 
> filp = files->fd[fd];
> if (!filp)
> goto out_unlock;
> files->fd[fd] = NULL; <--- it's invalid from now on
> 
> so it's not possible that some new readers may occur when a counter == 0.
> 

Ack.

> Hm.. but we still have fasync_helper(-1,file,0,&state->asyncq); which is 
> about sending a signal and that's perfectly valid (a file::counter is 
> not involved here). And that call may lead to re-scheduling (linux 
> re-scheduling of course) so we can't put it in a blocked section.
> 
> So the best way I see is to have something like():
> 
> xnpipe_drop_user_conn()
> {
> xnlock_get_irqsave(&nklock,s);
> 
> while ((holder = getq(&state->outq)) != NULL)
>               
>  state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
>            }
> 
>        while ((holder = getq(&state->inq)) != NULL)
>            {
>            if (state->input_handler != NULL)
>               
>  state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
>            else if (state->alloc_handler == NULL)
>                xnfree(link2mh(holder));
>            }
> 
> __clrbits(state->status,XNPIPE_USER_CONN);
> 
> xnlock_put_irqrestore(&nklock,s);
> }
> 
> and call it everywhere instead of clrbits(state->status,XNPIPE_USER_CONN);
> 
> This way we may be sure there are no pending messages left.
> 
>

Sounds consistent, since USER_CONN flag is semantically bound to the active/inactive state 
of the associated data queues anyway.

>  > --
>  >
>  > Philippe.
> 
> ---
> 
> Dmitry
> 


-- 

Philippe.

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Sebastian Smolorz]

> Hi,
>
> I failed to find the original message from Sebastian that led to the
> following change:
>
> -----
> 2005-11-09  Philippe Gerum  <rpm@xenomai.org>
>
>         * nucleus/pipe.c (xnpipe_disconnect): Flush the output queue
>         upon closure. Issue spotted by Sebastian Smolorz.
>         (xnpipe_release): Flush the input queue upon closure.
> -----

https://mail.gna.org/public/xenomai-core/2005-11/msg00058.html

The problem was here that after a RT task and a Linux user space program had 
closed both ends of the pipe the contents remained in the pipe. So when the 
pipe was opened again the old content was read.

Sebastian

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Dmitry Adamushko]

[-- Attachment #1.1: Type: text/plain, Size: 1504 bytes --]


Philippe Gerum <rpm@xenomai.org> wrote on 18.11.2005 12:07:22:

>
> > Hm.. but we still have fasync_helper(-1,file,0,&state->asyncq); which
is
> > about sending a signal and that's perfectly valid (a file::counter is
> > not involved here). And that call may lead to re-scheduling (linux
> > re-scheduling of course) so we can't put it in a blocked section.
> >
> > So the best way I see is to have something like():
> >
> > xnpipe_drop_user_conn()
> > {
> > xnlock_get_irqsave(&nklock,s);
> >
> > while ((holder = getq(&state->outq)) != NULL)
> >
> >  state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
> >            }
> >
> >        while ((holder = getq(&state->inq)) != NULL)
> >            {
> >            if (state->input_handler != NULL)
> >
> >  state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
> >            else if (state->alloc_handler == NULL)
> >                xnfree(link2mh(holder));
> >            }
> >
> > __clrbits(state->status,XNPIPE_USER_CONN);
> >
> > xnlock_put_irqrestore(&nklock,s);
> > }
> >
> > and call it everywhere instead of
clrbits(state->status,XNPIPE_USER_CONN);
> >
> > This way we may be sure there are no pending messages left.
> >
> >
>
> Sounds consistent, since USER_CONN flag is semantically bound to the
> active/inactive state
> of the associated data queues anyway.
>

Then a patch is enclosed.


> --
>
> Philippe.

---

Dmitry


(See attached file: pipe.cleanup-user-conn.patch)(See attached file:
ChangeLog-diff.patch)

[-- Attachment #1.2: Type: text/html, Size: 2479 bytes --]

[-- Attachment #2: pipe.cleanup-user-conn.patch --]
[-- Type: application/octet-stream, Size: 4406 bytes --]

--- pipe.c-old	2005-11-18 13:31:44.000000000 +0100
+++ pipe.c	2005-11-18 14:51:16.000000000 +0100
@@ -381,16 +381,16 @@
 
     xnlock_get_irqsave(&nklock,s);
 
-    if (!testbits(state->status,XNPIPE_USER_CONN))
+    if (!testbits(state->status,XNPIPE_KERN_CONN))
 	{
 	xnlock_put_irqrestore(&nklock,s);
-	return -EPIPE;
+	return -EBADF;
 	}
 
-    if (!testbits(state->status,XNPIPE_KERN_CONN))
+    if (!testbits(state->status,XNPIPE_USER_CONN))
 	{
 	xnlock_put_irqrestore(&nklock,s);
-	return -EBADF;
+	return -EPIPE;
 	}
 
     inith(xnpipe_m_link(mh));
@@ -402,22 +402,19 @@
     else
 	appendq(&state->outq,xnpipe_m_link(mh));
 
-    if (testbits(state->status,XNPIPE_USER_CONN))
-	{
-	if (testbits(state->status,XNPIPE_USER_WREAD))
-	    {
-	    /* Wake up the userland thread waiting for input
-	       from the kernel side. */
-	    setbits(state->status,XNPIPE_USER_WREAD_READY);
-	    need_sched = 1;
-	    }
-
-	if (state->asyncq) /* Schedule asynch sig. */
-	    {
-	    setbits(state->status,XNPIPE_USER_SIGIO);
-	    need_sched = 1;
-	    }
-	}
+    if (testbits(state->status,XNPIPE_USER_WREAD))
+        {
+        /* Wake up the userland thread waiting for input
+           from the kernel side. */
+        setbits(state->status,XNPIPE_USER_WREAD_READY);
+        need_sched = 1;
+        }
+
+    if (state->asyncq) /* Schedule asynch sig. */
+        {
+        setbits(state->status,XNPIPE_USER_SIGIO);
+        need_sched = 1;
+        }
 
     xnlock_put_irqrestore(&nklock,s);
 
@@ -525,6 +522,38 @@
 }
 
 /*
+ * Clear XNPIPE_USER_CONN flag and cleanup the associated data queues
+ * in one atomic step.
+ */
+
+static void xnpipe_cleanup_user_conn(xnpipe_state_t *state)
+{
+    int minor = xnminor_from_state(state);
+    xnholder_t *holder;
+    spl_t s;
+
+    xnlock_get_irqsave(&nklock,s);
+
+    if (state->output_handler != NULL)
+        {
+        while ((holder = getq(&state->outq)) != NULL)
+    	    state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
+        }
+
+    while ((holder = getq(&state->inq)) != NULL)
+        {
+        if (state->input_handler != NULL)
+	    state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
+        else if (state->alloc_handler == NULL)
+	    xnfree(link2mh(holder));
+	}
+
+    __clrbits(state->status,XNPIPE_USER_CONN);
+
+    xnlock_put_irqrestore(&nklock,s);
+}
+
+/*
  * Open the pipe from user-space.
  */
 
@@ -569,7 +598,7 @@
 
 	    if (err != 0)
 		{
-		clrbits(state->status,XNPIPE_USER_CONN);
+		xnpipe_cleanup_user_conn(state);
 		return err;
 		}
 
@@ -581,7 +610,7 @@
 
 	if (testbits(file->f_flags,O_NONBLOCK))
 	    {
-            clrbits(state->status,XNPIPE_USER_CONN);
+	    xnpipe_cleanup_user_conn(state);
 	    xnlock_put_irqrestore(&nklock,s);
             return -EWOULDBLOCK;
 	    }
@@ -590,7 +619,7 @@
 
 	if (sigpending && !testbits(state->status,XNPIPE_KERN_CONN))
 	    {
-	    clrbits(state->status,XNPIPE_USER_CONN);
+	    xnpipe_cleanup_user_conn(state);
     	    xnlock_put_irqrestore(&nklock,s);
 	    return -ERESTARTSYS;
 	    }
@@ -606,7 +635,7 @@
 	}
 
     if(err)
-        clrbits(state->status,XNPIPE_USER_CONN);
+	xnpipe_cleanup_user_conn(state);
     
     return err;
 }
@@ -615,7 +644,6 @@
 			   struct file *file)
 {
     xnpipe_state_t *state;
-    xnholder_t *holder;
     int err = 0;
     spl_t s;
 
@@ -632,20 +660,6 @@
 	{
 	int minor = xnminor_from_state(state);
 
-	if (state->output_handler != NULL)
-	    {
-	    while ((holder = getq(&state->outq)) != NULL)
-		state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
-	    }
-
-	while ((holder = getq(&state->inq)) != NULL)
-	    {
-	    if (state->input_handler != NULL)
-		state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
-	    else if (state->alloc_handler == NULL)
-		xnfree(link2mh(holder));
-	    }
-
 	/* If a real-time kernel thread is waiting on this object,
 	   wake it up now. */
 
@@ -663,9 +677,6 @@
     else
 	xnlock_put_irqrestore(&nklock,s);
 
-    if (waitqueue_active(&state->readq))
-	wake_up_interruptible_all(&state->readq);
-
     if (state->asyncq) /* Clear the async queue */
 	{
         xnlock_get_irqsave(&nklock,s);
@@ -676,7 +687,7 @@
 	}
 
     /* Free the state object. Since that time it can be open by someone else */
-    clrbits(state->status,XNPIPE_USER_CONN);
+    xnpipe_cleanup_user_conn(state);
 
     return err;
 }

[-- Attachment #3: ChangeLog-diff.patch --]
[-- Type: application/octet-stream, Size: 452 bytes --]

--- ChangeLog-old	2005-11-18 14:42:02.000000000 +0100
+++ ChangeLog	2005-11-18 14:50:13.000000000 +0100
@@ -1,4 +1,9 @@
 
+2005-11-18  Dmitry Adamushko <dmitry.adamushko@gmail.com>
+
+	* nucleus/pipe.c (xnpipe_cleanup_user_conn): Clear XNPIPE_USER_CONN
+	flag and cleanup the associated data queues in one atomic step.
+	
 2005-11-16  Dmitry Adamushko <dmitry.adamushko@gmail.com>
 
 	* nucleus/pipe.c, skins/native/pipe.c: Support for auto-allocation

Re: [Xenomai-core] [pipe.c] hairy synchronization -> "flush the output queue upon closure"

[Philippe Gerum]

Dmitry Adamushko wrote:
> Philippe Gerum <rpm@xenomai.org> wrote on 18.11.2005 12:07:22:
> 
>  >
>  > > Hm.. but we still have fasync_helper(-1,file,0,&state->asyncq); 
> which is
>  > > about sending a signal and that's perfectly valid (a file::counter is
>  > > not involved here). And that call may lead to re-scheduling (linux
>  > > re-scheduling of course) so we can't put it in a blocked section.
>  > >
>  > > So the best way I see is to have something like():
>  > >
>  > > xnpipe_drop_user_conn()
>  > > {
>  > > xnlock_get_irqsave(&nklock,s);
>  > >
>  > > while ((holder = getq(&state->outq)) != NULL)
>  > >              
>  > >  state->output_handler(minor,link2mh(holder),-EPIPE,state->cookie);
>  > >            }
>  > >
>  > >        while ((holder = getq(&state->inq)) != NULL)
>  > >            {
>  > >            if (state->input_handler != NULL)
>  > >              
>  > >  state->input_handler(minor,link2mh(holder),-EPIPE,state->cookie);
>  > >            else if (state->alloc_handler == NULL)
>  > >                xnfree(link2mh(holder));
>  > >            }
>  > >
>  > > __clrbits(state->status,XNPIPE_USER_CONN);
>  > >
>  > > xnlock_put_irqrestore(&nklock,s);
>  > > }
>  > >
>  > > and call it everywhere instead of 
> clrbits(state->status,XNPIPE_USER_CONN);
>  > >
>  > > This way we may be sure there are no pending messages left.
>  > >
>  > >
>  >
>  > Sounds consistent, since USER_CONN flag is semantically bound to the
>  > active/inactive state
>  > of the associated data queues anyway.
>  >
> 
> Then a patch is enclosed.
> 

Applied, thanks.

> 
>  > --
>  >
>  > Philippe.
> 
> ---
> 
> Dmitry
> 
> 
> /(See attached file: pipe.cleanup-user-conn.patch)//(See attached file: 
> ChangeLog-diff.patch)/
> 


-- 

Philippe.