>From 1681db3268e353914faa9b87b98b21b4d213a237 Mon Sep 17 00:00:00 2001
From: Adam Martinson <adam.r.martinson@gmail.com>
Date: Tue, 18 Dec 2012 15:44:22 -0600
Subject: [PATCH 2/7] server: Either end of a named pipe is readable after the other has been closed.

(If the pipe has not been disconnected)
---
 dlls/kernel32/file.c |    2 +-
 dlls/ntdll/file.c    |   17 ++
 server/named_pipe.c  |  426 ++++++++++++++++++++++++++++++--------------------
 3 files changed, 276 insertions(+), 169 deletions(-)

diff --git a/dlls/kernel32/file.c b/dlls/kernel32/file.c
index 73b8416..33d6561 100644
--- a/dlls/kernel32/file.c
+++ b/dlls/kernel32/file.c
@@ -404,7 +404,7 @@ BOOL WINAPI ReadFile( HANDLE hFile, LPVOID buffer, DWORD bytesToRead,
           bytesRead, overlapped );
 
     if (bytesRead) *bytesRead = 0;  /* Do this before anything else */
-    if (!bytesToRead) return TRUE;
+    if (!bytesToRead && GetFileType(hFile) != FILE_TYPE_PIPE) return TRUE;
 
     if (is_console_handle(hFile))
     {
diff --git a/dlls/ntdll/file.c b/dlls/ntdll/file.c
index 1b292d4..7b36392 100644
--- a/dlls/ntdll/file.c
+++ b/dlls/ntdll/file.c
@@ -620,6 +620,23 @@ NTSTATUS WINAPI NtReadFile(HANDLE hFile, HANDLE hEvent,
         goto done;
     }
 
+    if (type == FD_TYPE_PIPE && !length)
+    {
+        struct pollfd pfd;
+        pfd.fd = unix_handle;
+        pfd.events = POLLHUP;
+        pfd.revents = 0;
+
+        status = STATUS_SUCCESS;
+        if (poll(&pfd, 1, 0) > 0 && pfd.revents == POLLHUP)
+        { /* the other end has closed, but we only return an error if the pipe is also empty */
+            char check;
+            if (recv( unix_handle, &check, sizeof(check), MSG_PEEK | MSG_DONTWAIT ) != sizeof(check))
+                status = STATUS_PIPE_BROKEN;
+        }
+        goto done;
+    }
+
     for (;;)
     {
         if ((result = read( unix_handle, (char *)buffer + total, length - total )) >= 0)
diff --git a/server/named_pipe.c b/server/named_pipe.c
index fa7ae88..fdac33e 100644
--- a/server/named_pipe.c
+++ b/server/named_pipe.c
@@ -3,6 +3,7 @@
  *
  * Copyright (C) 1998 Alexandre Julliard
  * Copyright (C) 2001 Mike McCormack
+ * Copyright (C) 2012 Adam Martinson
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -68,15 +69,11 @@ struct named_pipe;
 struct pipe_server
 {
     struct object        obj;        /* object header */
-    struct fd           *fd;         /* pipe file descriptor */
     struct fd           *ioctl_fd;   /* file descriptor for ioctls when not connected */
     struct list          entry;      /* entry in named pipe servers list */
     enum pipe_state      state;      /* server state */
-    struct pipe_end     *client;     /* client that this server is connected to */
     struct named_pipe   *pipe;
-    struct timeout_user *flush_poll;
-    struct event        *event;
-    unsigned int         options;    /* pipe options */
+    struct pipe_end     *ends[2];    /* ends that this server connects */
 };
 
 struct pipe_end
@@ -84,13 +81,15 @@ struct pipe_end
     struct object        obj;        /* object header */
     struct fd           *fd;         /* pipe file descriptor */
     struct pipe_server  *server;     /* server that this end is connected to */
+    unsigned int         options;    /* file options */
     unsigned int         flags;      /* pipe flags */
+    struct timeout_user *flush_poll;
+    struct event        *event_empty;
 };
 
 struct named_pipe
 {
     struct object       obj;         /* object header */
-    unsigned int        flags;
     unsigned int        sharing;
     unsigned int        maxinstances;
     unsigned int        outsize;
@@ -138,7 +137,6 @@ static const struct object_ops named_pipe_ops =
 static void pipe_server_dump( struct object *obj, int verbose );
 static struct fd *pipe_server_get_fd( struct object *obj );
 static void pipe_server_destroy( struct object *obj);
-static void pipe_server_flush( struct fd *fd, struct event **event );
 static enum server_fd_type pipe_server_get_fd_type( struct fd *fd );
 static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const async_data_t *async,
                                        int blocking, const void *data, data_size_t size );
@@ -167,7 +165,7 @@ static const struct fd_ops pipe_server_fd_ops =
 {
     default_fd_get_poll_events,   /* get_poll_events */
     default_poll_event,           /* poll_event */
-    pipe_server_flush,            /* flush */
+    no_flush,                     /* flush */
     pipe_server_get_fd_type,      /* get_fd_type */
     pipe_server_ioctl,            /* ioctl */
     default_fd_queue_async,       /* queue_async */
@@ -182,6 +180,8 @@ static struct fd *pipe_end_get_fd( struct object *obj );
 static void pipe_end_destroy( struct object *obj );
 static void pipe_end_flush( struct fd *fd, struct event **event );
 static enum server_fd_type pipe_end_get_fd_type( struct fd *fd );
+static obj_handle_t pipe_end_ioctl( struct fd *fd, ioctl_code_t code, const async_data_t *async,
+                                       int blocking, const void *data, data_size_t size );
 
 static const struct object_ops pipe_end_ops =
 {
@@ -209,7 +209,7 @@ static const struct fd_ops pipe_end_fd_ops =
     default_poll_event,           /* poll_event */
     pipe_end_flush,               /* flush */
     pipe_end_get_fd_type,         /* get_fd_type */
-    default_fd_ioctl,             /* ioctl */
+    pipe_end_ioctl,               /* ioctl */
     default_fd_queue_async,       /* queue_async */
     default_fd_reselect_async,    /* reselect_async */
     default_fd_cancel_async       /* cancel_async */
@@ -259,6 +259,14 @@ static const struct fd_ops named_pipe_device_fd_ops =
     default_fd_cancel_async           /* cancel_async */
 };
 
+static inline int is_server_end( struct pipe_end *end )
+{
+    int res = (end->flags & NAMED_PIPE_SERVER_END)? 1 : 0;
+    if (end->server)
+        assert( end->server->ends[!res] == end );
+    return res;
+}
+
 static void named_pipe_dump( struct object *obj, int verbose )
 {
     struct named_pipe *pipe = (struct named_pipe *) obj;
@@ -288,7 +296,10 @@ static void pipe_end_dump( struct object *obj, int verbose )
 {
     struct pipe_end *end = (struct pipe_end *) obj;
     assert( obj->ops == &pipe_end_ops );
-    fprintf( stderr, "Named pipe client server=%p\n", end->server );
+    if (is_server_end( end ))
+        pipe_server_dump( &end->server->obj, verbose );
+    else
+        fprintf( stderr, "Named pipe client server=%p\n", end->server );
 }
 
 static int pipe_end_signaled( struct object *obj, struct thread *thread )
@@ -312,6 +323,8 @@ static struct fd *pipe_end_get_fd( struct object *obj )
     struct pipe_end *end = (struct pipe_end *) obj;
     if (end->fd)
         return (struct fd *) grab_object( end->fd );
+    else if ((end->flags & NAMED_PIPE_SERVER_END) && end->server)
+        return (struct fd *) grab_object( end->server->ioctl_fd );
     set_error( STATUS_PIPE_DISCONNECTED );
     return NULL;
 }
@@ -323,17 +336,31 @@ static void set_server_state( struct pipe_server *server, enum pipe_state state
     switch(state)
     {
     case ps_connected:
-    case ps_disconnected_client:
-        assert( server->fd );
+        assert( server->ends[0] );
+        assert( server->ends[0]->fd );
+        assert( server->ends[1] );
+        assert( server->ends[1]->fd );
         break;
     case ps_wait_open:
     case ps_idle_server:
-        assert( !server->fd );
+        assert( server->ends[0] );
+        assert( !server->ends[0]->fd );
+        assert( !server->ends[1] );
         set_no_fd_status( server->ioctl_fd, STATUS_PIPE_LISTENING );
         break;
+    case ps_disconnected_client:
+        assert( !server->ends[0] );
+        assert( server->ends[1] );
+        break;
     case ps_disconnected_server:
+        assert( server->ends[0] );
+        assert( !server->ends[1] );
+        set_no_fd_status( server->ioctl_fd, STATUS_PIPE_DISCONNECTED );
+        break;
     case ps_wait_connect:
-        assert( !server->fd );
+        assert( server->ends[0] );
+        assert( !server->ends[0]->fd );
+        assert( !server->ends[1] );
         set_no_fd_status( server->ioctl_fd, STATUS_PIPE_DISCONNECTED );
         break;
     }
@@ -343,57 +370,54 @@ static struct fd *pipe_server_get_fd( struct object *obj )
 {
     struct pipe_server *server = (struct pipe_server *) obj;
 
-    return (struct fd *)grab_object( server->fd ? server->fd : server->ioctl_fd );
+    return (struct fd*)grab_object( server->ioctl_fd );
 }
 
 
-static void notify_empty( struct pipe_server *server )
+static void notify_empty( struct pipe_end *end )
 {
-    if (!server->flush_poll)
+    if (!end->flush_poll)
         return;
-    assert( server->state == ps_connected );
-    assert( server->event );
-    remove_timeout_user( server->flush_poll );
-    server->flush_poll = NULL;
-    set_event( server->event );
-    release_object( server->event );
-    server->event = NULL;
+    assert( end->server && end->server->state == ps_connected );
+    assert( end->event_empty );
+    remove_timeout_user( end->flush_poll );
+    end->flush_poll = NULL;
+    set_event( end->event_empty );
+    release_object( end->event_empty );
+    end->event_empty = NULL;
 }
 
 static void do_disconnect( struct pipe_server *server )
 {
-    /* we may only have a server fd, if the client disconnected */
-    if (server->client)
+    int is_shutdown = 0;
+    /* we may only have 1 end */
+    if (server->ends[0])
     {
-        assert( server->client->server == server );
-        assert( server->client->fd );
-        release_object( server->client->fd );
-        server->client->fd = NULL;
+        assert( server->ends[0]->server == server );
+        assert( server->ends[0]->fd );
+        shutdown( get_unix_fd( server->ends[0]->fd ), SHUT_RDWR );
+        is_shutdown = 1;
+        release_object( server->ends[0]->fd );
+        server->ends[0]->fd = NULL;
+    }
+
+    if (server->ends[1])
+    {
+        assert( server->ends[1]->server == server );
+        assert( server->ends[1]->fd );
+        if (!is_shutdown) shutdown( get_unix_fd( server->ends[1]->fd ), SHUT_RDWR );
+        release_object( server->ends[1]->fd );
+        server->ends[1]->fd = NULL;
     }
-    assert( server->fd );
-    shutdown( get_unix_fd( server->fd ), SHUT_RDWR );
-    release_object( server->fd );
-    server->fd = NULL;
 }
 
-static void pipe_server_destroy( struct object *obj)
+static void pipe_server_destroy( struct object *obj )
 {
     struct pipe_server *server = (struct pipe_server *)obj;
 
     assert( obj->ops == &pipe_server_ops );
-
-    if (server->fd)
-    {
-        notify_empty( server );
-        do_disconnect( server );
-    }
-
-    if (server->client)
-    {
-        server->client->server = NULL;
-        server->client = NULL;
-    }
-
+    assert( !server->ends[0] );
+    assert( !server->ends[1] );
     assert( server->pipe->instances );
     server->pipe->instances--;
 
@@ -402,37 +426,40 @@ static void pipe_server_destroy( struct object *obj)
     release_object( server->pipe );
 }
 
-static void pipe_end_destroy( struct object *obj)
+static void pipe_end_destroy( struct object *obj )
 {
     struct pipe_end *end = (struct pipe_end *)obj;
     struct pipe_server *server = end->server;
 
     assert( obj->ops == &pipe_end_ops );
+    if (end->event_empty)
+        notify_empty( end );
 
     if (server)
     {
-        notify_empty( server );
+        assert( server->ends[!is_server_end( end )] == end );
+        if (!server->ends[is_server_end( end )])
+            assert( server->obj.refcount == 1 );
+
+        server->ends[!is_server_end( end )] = NULL;
 
         switch(server->state)
         {
         case ps_connected:
-            /* Don't destroy the server's fd here as we can't
-               do a successful flush without it. */
-            set_server_state( server, ps_disconnected_client );
+            if (end->flags & NAMED_PIPE_SERVER_END)
+                set_server_state( server, ps_disconnected_client );
+            else
+                set_server_state( server, ps_disconnected_server );
             break;
         case ps_disconnected_server:
-            set_server_state( server, ps_wait_connect );
-            break;
+        case ps_disconnected_client:
         case ps_idle_server:
         case ps_wait_open:
-        case ps_disconnected_client:
         case ps_wait_connect:
-            assert( 0 );
             break;
         }
-        assert( server->client );
-        server->client = NULL;
-        end->server = NULL;
+
+        release_object( server );
     }
     if (end->fd) release_object( end->fd );
 }
@@ -508,14 +535,13 @@ void create_named_pipe_device( struct directory *root, const struct unicode_str
     if (dev) make_object_static( &dev->obj );
 }
 
-static int pipe_data_remaining( struct pipe_server *server )
+static int pipe_data_remaining( struct pipe_end *end )
 {
     struct pollfd pfd;
     int fd;
 
-    assert( server->client );
-
-    fd = get_unix_fd( server->client->fd );
+    assert( end->fd );
+    fd = get_unix_fd( end->fd );
     if (fd < 0)
         return 0;
     pfd.fd = fd;
@@ -530,49 +556,49 @@ static int pipe_data_remaining( struct pipe_server *server )
 
 static void check_flushed( void *arg )
 {
-    struct pipe_server *server = (struct pipe_server*) arg;
+    struct pipe_end *end = (struct pipe_end*) arg;
 
-    assert( server->event );
-    if (pipe_data_remaining( server ))
+    assert( end->event_empty );
+    if (pipe_data_remaining( end ))
     {
-        server->flush_poll = add_timeout_user( -TICKS_PER_SEC / 10, check_flushed, server );
+        end->flush_poll = add_timeout_user( -TICKS_PER_SEC / 10, check_flushed, end );
     }
     else
     {
-        /* notify_empty( server ); */
-        server->flush_poll = NULL;
-        set_event( server->event );
-        release_object( server->event );
-        server->event = NULL;
+        notify_empty( end );
+        end->flush_poll = NULL;
+        set_event( end->event_empty );
+        release_object( end->event_empty );
+        end->event_empty = NULL;
     }
 }
 
-static void pipe_server_flush( struct fd *fd, struct event **event )
+static void pipe_end_flush( struct fd *fd, struct event **event )
 {
-    struct pipe_server *server = get_fd_user( fd );
+    struct pipe_end *other_end, *end = get_fd_user( fd );
+
+    if (!end) return;
+    assert( end->server );
+    if (end->server->state != ps_connected) return;
 
-    if (!server || server->state != ps_connected) return;
+    other_end = end->server->ends[is_server_end( end )];
+    assert( other_end );
 
     /* FIXME: if multiple threads flush the same pipe,
               maybe should create a list of processes to notify */
-    if (server->flush_poll) return;
+    if (other_end->flush_poll) return;
 
-    if (pipe_data_remaining( server ))
+    if (pipe_data_remaining( other_end ))
     {
         /* this kind of sux -
            there's no unix way to be alerted when a pipe becomes empty */
-        server->event = create_event( NULL, NULL, 0, 0, 0, NULL );
-        if (!server->event) return;
-        server->flush_poll = add_timeout_user( -TICKS_PER_SEC / 10, check_flushed, server );
-        *event = server->event;
+        other_end->event_empty = create_event( NULL, NULL, 0, 0, 0, NULL );
+        if (!other_end->event_empty) return;
+        other_end->flush_poll = add_timeout_user( -TICKS_PER_SEC / 10, check_flushed, other_end );
+        *event = other_end->event_empty;
     }
 }
 
-static void pipe_end_flush( struct fd *fd, struct event **event )
-{
-    /* FIXME: what do we have to do for this? */
-}
-
 static inline int is_overlapped( unsigned int options )
 {
     return !(options & (FILE_SYNCHRONOUS_IO_ALERT | FILE_SYNCHRONOUS_IO_NONALERT));
@@ -615,6 +641,7 @@ static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const a
         {
         case ps_idle_server:
         case ps_wait_connect:
+        case ps_disconnected_server:
             if (blocking)
             {
                 async_data_t new_data = *async_data;
@@ -630,6 +657,7 @@ static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const a
 
             if (async)
             {
+                assert( server->pipe );
                 set_server_state( server, ps_wait_open );
                 if (server->pipe->waiters) async_wake_up( server->pipe->waiters, STATUS_SUCCESS );
                 release_object( async );
@@ -638,10 +666,7 @@ static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const a
             }
             break;
         case ps_connected:
-            set_error( STATUS_PIPE_CONNECTED );
-            break;
-        case ps_disconnected_server:
-            set_error( STATUS_PIPE_BUSY );
+            assert( 0 );
             break;
         case ps_disconnected_client:
             set_error( STATUS_NO_DATA_DETECTED );
@@ -656,27 +681,15 @@ static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const a
         switch(server->state)
         {
         case ps_connected:
-            assert( server->client );
-            assert( server->client->fd );
-
-            notify_empty( server );
-
-            /* dump the client and server fds, but keep the pointers
-               around - client loses all waiting data */
-            do_disconnect( server );
-            set_server_state( server, ps_disconnected_server );
-            break;
-        case ps_disconnected_client:
-            assert( !server->client );
-            do_disconnect( server );
-            set_server_state( server, ps_wait_connect );
+            assert( 0 );
             break;
         case ps_idle_server:
         case ps_wait_open:
             set_error( STATUS_PIPE_LISTENING );
             break;
-        case ps_disconnected_server:
         case ps_wait_connect:
+        case ps_disconnected_server:
+        case ps_disconnected_client:
             set_error( STATUS_PIPE_DISCONNECTED );
             break;
         }
@@ -687,6 +700,72 @@ static obj_handle_t pipe_server_ioctl( struct fd *fd, ioctl_code_t code, const a
     }
 }
 
+static obj_handle_t pipe_end_ioctl( struct fd *fd, ioctl_code_t code, const async_data_t *async_data,
+                                       int blocking, const void *data, data_size_t size )
+{
+    struct pipe_end *end = get_fd_user( fd );
+
+    assert( end->server );
+
+    switch(code)
+    {
+    case FSCTL_PIPE_DISCONNECT:
+        switch(end->server->state)
+        {
+        case ps_connected:
+            assert( end->server->ends[0] );
+            assert( end->server->ends[0]->fd );
+            assert( end->server->ends[1] );
+            assert( end->server->ends[1]->fd );
+
+            notify_empty( end->server->ends[0] );
+            notify_empty( end->server->ends[1] );
+
+            /* all waiting data lost */
+            do_disconnect( end->server );
+            if (is_server_end( end ))
+            {
+                end->server->ends[1]->server = NULL;
+                end->server->ends[1] = NULL;
+                set_server_state( end->server, ps_disconnected_server );
+                release_object( end->server );
+            }
+            else
+            {
+                /* FIXME: is this allowed?? */
+                struct pipe_server *server = end->server;
+                end->server->ends[1] = NULL;
+                end->server = NULL;
+                set_server_state( server, ps_disconnected_server );
+                release_object( server );
+            }
+            break;
+        case ps_disconnected_server:
+            assert( !end->server->ends[1] );
+            do_disconnect( end->server );
+            break;
+        case ps_disconnected_client:
+            /* FIXME: is this allowed?? */
+            assert( !end->server->ends[0] );
+            do_disconnect( end->server );
+            break;
+        case ps_idle_server:
+        case ps_wait_open:
+         case ps_wait_connect:
+            assert( 0 );
+            break;
+        }
+        return 0;
+
+    case FSCTL_PIPE_LISTEN:
+        set_error( STATUS_PIPE_CONNECTED );
+        return 0;
+
+    default:
+        return default_fd_ioctl( fd, code, async_data, blocking, data, size );
+    }
+}
+
 static struct named_pipe *create_named_pipe( struct directory *root, const struct unicode_str *name,
                                              unsigned int attr )
 {
@@ -730,15 +809,45 @@ static struct named_pipe *create_named_pipe( struct directory *root, const struc
     return pipe;
 }
 
-static struct pipe_server *get_pipe_server_obj( struct process *process,
+static struct pipe_end *get_pipe_end_obj( struct process *process,
                                 obj_handle_t handle, unsigned int access )
 {
-    struct object *obj;
-    obj = get_handle_obj( process, handle, access, &pipe_server_ops );
-    return (struct pipe_server *) obj;
+    struct pipe_end *end;
+    end = (struct pipe_end *)get_handle_obj( process, handle, 0, &pipe_end_ops );
+    if (end && is_server_end( end ))
+    {
+        release_object( end );
+        end = (struct pipe_end *)get_handle_obj( process, handle, access, &pipe_end_ops );
+    }
+    return end;
 }
 
-static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned int options )
+static struct pipe_end *create_pipe_end( struct pipe_server *server, unsigned int options, unsigned int flags )
+{
+    struct pipe_end *end;
+
+    end = alloc_object( &pipe_end_ops );
+    if (!end)
+        return NULL;
+
+    end->fd = NULL;
+    end->server = server;
+    end->options = options;
+    end->flush_poll = NULL;
+    end->event_empty = NULL;
+    end->flags = flags;
+
+    if (flags & NAMED_PIPE_SERVER_END)
+        server->ends[0] = end;
+    else
+        server->ends[1] = end;
+
+    grab_object( server );
+
+    return end;
+}
+
+static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned int options, unsigned int flags )
 {
     struct pipe_server *server;
 
@@ -746,11 +855,14 @@ static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned
     if (!server)
         return NULL;
 
-    server->fd = NULL;
     server->pipe = pipe;
-    server->client = NULL;
-    server->flush_poll = NULL;
-    server->options = options;
+    server->ends[1] = NULL;
+    server->ends[0] = create_pipe_end( server, options, flags | NAMED_PIPE_SERVER_END );
+    if (!server->ends[0])
+    {
+        release_object( server );
+        return NULL;
+    }
 
     list_add_head( &pipe->servers, &server->entry );
     grab_object( pipe );
@@ -763,21 +875,6 @@ static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned
     return server;
 }
 
-static struct pipe_end *create_pipe_end( unsigned int flags )
-{
-    struct pipe_end *end;
-
-    end = alloc_object( &pipe_end_ops );
-    if (!end)
-        return NULL;
-
-    end->fd = NULL;
-    end->server = NULL;
-    end->flags = flags;
-
-    return end;
-}
-
 static struct pipe_server *find_available_server( struct named_pipe *pipe )
 {
     struct pipe_server *server;
@@ -814,6 +911,10 @@ static struct object *named_pipe_open_file( struct object *obj, unsigned int acc
         return NULL;
     }
 
+    assert( server->ends[0] );
+    assert( !server->ends[0]->fd );
+    assert( !server->ends[1] );
+
     pipe_sharing = server->pipe->sharing;
     if (((access & GENERIC_READ) && !(pipe_sharing & FILE_SHARE_READ)) ||
         ((access & GENERIC_WRITE) && !(pipe_sharing & FILE_SHARE_WRITE)))
@@ -823,17 +924,17 @@ static struct object *named_pipe_open_file( struct object *obj, unsigned int acc
         return NULL;
     }
 
-    if ((client = create_pipe_end( options )))
+    if ((client = create_pipe_end( server, options, server->ends[0]->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE )))
     {
         if (!socketpair( PF_UNIX, SOCK_STREAM, 0, fds ))
         {
-            assert( !server->fd );
+            assert( !server->ends[0]->fd );
 
             /* for performance reasons, only set nonblocking mode when using
              * overlapped I/O. Otherwise, we will be doing too much busy
              * looping */
             if (is_overlapped( options )) fcntl( fds[1], F_SETFL, O_NONBLOCK );
-            if (is_overlapped( server->options )) fcntl( fds[0], F_SETFL, O_NONBLOCK );
+            if (is_overlapped( server->ends[0]->options )) fcntl( fds[0], F_SETFL, O_NONBLOCK );
 
             if (pipe->insize)
             {
@@ -847,17 +948,16 @@ static struct object *named_pipe_open_file( struct object *obj, unsigned int acc
             }
 
             client->fd = create_anonymous_fd( &pipe_end_fd_ops, fds[1], &client->obj, options );
-            server->fd = create_anonymous_fd( &pipe_server_fd_ops, fds[0], &server->obj, server->options );
-            if (client->fd && server->fd)
+            server->ends[0]->fd = create_anonymous_fd( &pipe_end_fd_ops, fds[0], &server->ends[0]->obj, server->ends[0]->options );
+
+            if (client->fd && server->ends[0]->fd)
             {
                 allow_fd_caching( client->fd );
-                allow_fd_caching( server->fd );
-                fd_copy_completion( server->ioctl_fd, server->fd );
+                allow_fd_caching( server->ends[0]->fd );
+                fd_copy_completion( server->ioctl_fd, server->ends[0]->fd );
                 if (server->state == ps_wait_open)
                     fd_async_wake_up( server->ioctl_fd, ASYNC_TYPE_WAIT, STATUS_SUCCESS );
                 set_server_state( server, ps_connected );
-                server->client = client;
-                client->server = server;
             }
             else
             {
@@ -978,7 +1078,6 @@ DECL_HANDLER(create_named_pipe)
         pipe->outsize = req->outsize;
         pipe->maxinstances = req->maxinstances;
         pipe->timeout = req->timeout;
-        pipe->flags = req->flags;
         pipe->sharing = req->sharing;
     }
     else
@@ -998,10 +1097,11 @@ DECL_HANDLER(create_named_pipe)
         clear_error(); /* clear the name collision */
     }
 
-    server = create_pipe_server( pipe, req->options );
+    server = create_pipe_server( pipe, req->options, req->flags );
     if (server)
     {
-        reply->handle = alloc_handle( current->process, server, req->access, req->attributes );
+        reply->handle = alloc_handle( current->process, server->ends[0], req->access, req->attributes );
+        release_object( server->ends[0] );
         server->pipe->instances++;
         release_object( server );
     }
@@ -1011,34 +1111,24 @@ DECL_HANDLER(create_named_pipe)
 
 DECL_HANDLER(get_named_pipe_info)
 {
-    struct pipe_server *server;
-    struct pipe_end *client = NULL;
+    struct pipe_end *end;
 
-    server = get_pipe_server_obj( current->process, req->handle, FILE_READ_ATTRIBUTES );
-    if (!server)
+    if ((end = get_pipe_end_obj( current->process, req->handle, FILE_READ_ATTRIBUTES )))
     {
-        if (get_error() != STATUS_OBJECT_TYPE_MISMATCH)
+        if (!end->server)
+        {
+            set_error( STATUS_PIPE_DISCONNECTED );
+            release_object( end );
             return;
-
-        clear_error();
-        client = (struct pipe_end *)get_handle_obj( current->process, req->handle,
-                                                       0, &pipe_end_ops );
-        if (!client) return;
-        server = client->server;
-    }
-
-    reply->flags        = server->pipe->flags;
-    reply->sharing      = server->pipe->sharing;
-    reply->maxinstances = server->pipe->maxinstances;
-    reply->instances    = server->pipe->instances;
-    reply->insize       = server->pipe->insize;
-    reply->outsize      = server->pipe->outsize;
-
-    if (client)
-        release_object(client);
-    else
-    {
-        reply->flags |= NAMED_PIPE_SERVER_END;
-        release_object(server);
+        }
+        assert( end->server->pipe );
+        reply->flags        = end->flags;
+        reply->sharing      = end->server->pipe->sharing;
+        reply->maxinstances = end->server->pipe->maxinstances;
+        reply->instances    = end->server->pipe->instances;
+        reply->insize       = end->server->pipe->insize;
+        reply->outsize      = end->server->pipe->outsize;
+
+        release_object(end);
     }
 }
-- 
1.7.1

