2 * Server-side pipe management
4 * Copyright (C) 1998 Alexandre Julliard
5 * Copyright (C) 2001 Mike McCormack
6 * Copyright 2016 Jacek Caban for CodeWeavers
8 * This library is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License as published by the Free Software Foundation; either
11 * version 2.1 of the License, or (at your option) any later version.
13 * This library is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * Lesser General Public License for more details.
18 * You should have received a copy of the GNU Lesser General Public
19 * License along with this library; if not, write to the Free Software
20 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
24 #include "wine/port.h"
33 #define WIN32_NO_STATUS
58 struct list entry
; /* entry in message queue */
59 data_size_t read_pos
; /* already read bytes */
60 struct iosb
*iosb
; /* message iosb */
61 struct async
*async
; /* async of pending write */
66 struct object obj
; /* object header */
67 struct fd
*fd
; /* pipe file descriptor */
68 unsigned int flags
; /* pipe flags */
69 unsigned int state
; /* pipe state */
70 struct pipe_end
*connection
; /* the other end of the pipe */
71 process_id_t client_pid
; /* process that created the client */
72 process_id_t server_pid
; /* process that created the server */
73 data_size_t buffer_size
;/* size of buffered data that doesn't block caller */
74 struct list message_queue
;
75 struct async_queue read_q
; /* read queue */
76 struct async_queue write_q
; /* write queue */
81 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
82 struct list entry
; /* entry in named pipe servers list */
83 enum pipe_state state
; /* server state */
84 struct pipe_client
*client
; /* client that this server is connected to */
85 struct named_pipe
*pipe
;
86 unsigned int options
; /* pipe options */
91 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
92 struct pipe_server
*server
; /* server that this client is connected to */
93 unsigned int flags
; /* file flags */
98 struct object obj
; /* object header */
100 unsigned int sharing
;
101 unsigned int maxinstances
;
102 unsigned int outsize
;
104 unsigned int instances
;
106 struct list servers
; /* list of servers using this pipe */
107 struct async_queue waiters
; /* list of clients waiting to connect */
110 struct named_pipe_device
112 struct object obj
; /* object header */
113 struct fd
*fd
; /* pseudo-fd for ioctls */
114 struct namespace *pipes
; /* named pipe namespace */
117 static void named_pipe_dump( struct object
*obj
, int verbose
);
118 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
);
119 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
);
120 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
121 unsigned int sharing
, unsigned int options
);
122 static void named_pipe_destroy( struct object
*obj
);
124 static const struct object_ops named_pipe_ops
=
126 sizeof(struct named_pipe
), /* size */
127 named_pipe_dump
, /* dump */
128 no_get_type
, /* get_type */
129 no_add_queue
, /* add_queue */
130 NULL
, /* remove_queue */
132 NULL
, /* satisfied */
133 no_signal
, /* signal */
134 no_get_fd
, /* get_fd */
135 named_pipe_map_access
, /* map_access */
136 default_get_sd
, /* get_sd */
137 default_set_sd
, /* set_sd */
138 no_lookup_name
, /* lookup_name */
139 named_pipe_link_name
, /* link_name */
140 default_unlink_name
, /* unlink_name */
141 named_pipe_open_file
, /* open_file */
142 no_close_handle
, /* close_handle */
143 named_pipe_destroy
/* destroy */
146 /* common server and client pipe end functions */
147 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
);
148 static struct fd
*pipe_end_get_fd( struct object
*obj
);
149 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
);
150 static int pipe_end_write( struct fd
*fd
, struct async
*async_data
, file_pos_t pos
);
151 static int pipe_end_flush( struct fd
*fd
, struct async
*async
);
152 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
);
153 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
);
155 /* server end functions */
156 static void pipe_server_dump( struct object
*obj
, int verbose
);
157 static struct security_descriptor
*pipe_server_get_sd( struct object
*obj
);
158 static int pipe_server_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
159 unsigned int set_info
);
160 static void pipe_server_destroy( struct object
*obj
);
161 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
162 static void pipe_server_get_file_info( struct fd
*fd
, unsigned int info_class
);
164 static const struct object_ops pipe_server_ops
=
166 sizeof(struct pipe_server
), /* size */
167 pipe_server_dump
, /* dump */
168 no_get_type
, /* get_type */
169 add_queue
, /* add_queue */
170 remove_queue
, /* remove_queue */
171 default_fd_signaled
, /* signaled */
172 no_satisfied
, /* satisfied */
173 no_signal
, /* signal */
174 pipe_end_get_fd
, /* get_fd */
175 default_fd_map_access
, /* map_access */
176 pipe_server_get_sd
, /* get_sd */
177 pipe_server_set_sd
, /* set_sd */
178 no_lookup_name
, /* lookup_name */
179 no_link_name
, /* link_name */
180 NULL
, /* unlink_name */
181 no_open_file
, /* open_file */
182 fd_close_handle
, /* close_handle */
183 pipe_server_destroy
/* destroy */
186 static const struct fd_ops pipe_server_fd_ops
=
188 default_fd_get_poll_events
, /* get_poll_events */
189 default_poll_event
, /* poll_event */
190 pipe_end_get_fd_type
, /* get_fd_type */
191 pipe_end_read
, /* read */
192 pipe_end_write
, /* write */
193 pipe_end_flush
, /* flush */
194 pipe_server_get_file_info
, /* get_file_info */
195 pipe_end_get_volume_info
, /* get_volume_info */
196 pipe_server_ioctl
, /* ioctl */
197 no_fd_queue_async
, /* queue_async */
198 pipe_end_reselect_async
/* reselect_async */
201 /* client end functions */
202 static void pipe_client_dump( struct object
*obj
, int verbose
);
203 static struct security_descriptor
*pipe_client_get_sd( struct object
*obj
);
204 static int pipe_client_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
205 unsigned int set_info
);
206 static void pipe_client_destroy( struct object
*obj
);
207 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
208 static void pipe_client_get_file_info( struct fd
*fd
, unsigned int info_class
);
210 static const struct object_ops pipe_client_ops
=
212 sizeof(struct pipe_client
), /* size */
213 pipe_client_dump
, /* dump */
214 no_get_type
, /* get_type */
215 add_queue
, /* add_queue */
216 remove_queue
, /* remove_queue */
217 default_fd_signaled
, /* signaled */
218 no_satisfied
, /* satisfied */
219 no_signal
, /* signal */
220 pipe_end_get_fd
, /* get_fd */
221 default_fd_map_access
, /* map_access */
222 pipe_client_get_sd
, /* get_sd */
223 pipe_client_set_sd
, /* set_sd */
224 no_lookup_name
, /* lookup_name */
225 no_link_name
, /* link_name */
226 NULL
, /* unlink_name */
227 no_open_file
, /* open_file */
228 fd_close_handle
, /* close_handle */
229 pipe_client_destroy
/* destroy */
232 static const struct fd_ops pipe_client_fd_ops
=
234 default_fd_get_poll_events
, /* get_poll_events */
235 default_poll_event
, /* poll_event */
236 pipe_end_get_fd_type
, /* get_fd_type */
237 pipe_end_read
, /* read */
238 pipe_end_write
, /* write */
239 pipe_end_flush
, /* flush */
240 pipe_client_get_file_info
, /* get_file_info */
241 pipe_end_get_volume_info
, /* get_volume_info */
242 pipe_client_ioctl
, /* ioctl */
243 no_fd_queue_async
, /* queue_async */
244 pipe_end_reselect_async
/* reselect_async */
247 static void named_pipe_device_dump( struct object
*obj
, int verbose
);
248 static struct object_type
*named_pipe_device_get_type( struct object
*obj
);
249 static struct fd
*named_pipe_device_get_fd( struct object
*obj
);
250 static struct object
*named_pipe_device_lookup_name( struct object
*obj
,
251 struct unicode_str
*name
, unsigned int attr
);
252 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
253 unsigned int sharing
, unsigned int options
);
254 static void named_pipe_device_destroy( struct object
*obj
);
255 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
);
256 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
258 static const struct object_ops named_pipe_device_ops
=
260 sizeof(struct named_pipe_device
), /* size */
261 named_pipe_device_dump
, /* dump */
262 named_pipe_device_get_type
, /* get_type */
263 no_add_queue
, /* add_queue */
264 NULL
, /* remove_queue */
266 no_satisfied
, /* satisfied */
267 no_signal
, /* signal */
268 named_pipe_device_get_fd
, /* get_fd */
269 no_map_access
, /* map_access */
270 default_get_sd
, /* get_sd */
271 default_set_sd
, /* set_sd */
272 named_pipe_device_lookup_name
, /* lookup_name */
273 directory_link_name
, /* link_name */
274 default_unlink_name
, /* unlink_name */
275 named_pipe_device_open_file
, /* open_file */
276 fd_close_handle
, /* close_handle */
277 named_pipe_device_destroy
/* destroy */
280 static const struct fd_ops named_pipe_device_fd_ops
=
282 default_fd_get_poll_events
, /* get_poll_events */
283 default_poll_event
, /* poll_event */
284 named_pipe_device_get_fd_type
, /* get_fd_type */
285 no_fd_read
, /* read */
286 no_fd_write
, /* write */
287 no_fd_flush
, /* flush */
288 no_fd_get_file_info
, /* get_file_info */
289 no_fd_get_volume_info
, /* get_volume_info */
290 named_pipe_device_ioctl
, /* ioctl */
291 default_fd_queue_async
, /* queue_async */
292 default_fd_reselect_async
/* reselect_async */
295 static void named_pipe_dump( struct object
*obj
, int verbose
)
297 fputs( "Named pipe\n", stderr
);
300 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
)
302 if (access
& GENERIC_READ
) access
|= STANDARD_RIGHTS_READ
;
303 if (access
& GENERIC_WRITE
) access
|= STANDARD_RIGHTS_WRITE
| FILE_CREATE_PIPE_INSTANCE
;
304 if (access
& GENERIC_EXECUTE
) access
|= STANDARD_RIGHTS_EXECUTE
;
305 if (access
& GENERIC_ALL
) access
|= STANDARD_RIGHTS_ALL
;
306 return access
& ~(GENERIC_READ
| GENERIC_WRITE
| GENERIC_EXECUTE
| GENERIC_ALL
);
309 static void pipe_server_dump( struct object
*obj
, int verbose
)
311 struct pipe_server
*server
= (struct pipe_server
*) obj
;
312 assert( obj
->ops
== &pipe_server_ops
);
313 fprintf( stderr
, "Named pipe server pipe=%p state=%d\n", server
->pipe
, server
->state
);
316 static void pipe_client_dump( struct object
*obj
, int verbose
)
318 struct pipe_client
*client
= (struct pipe_client
*) obj
;
319 assert( obj
->ops
== &pipe_client_ops
);
320 fprintf( stderr
, "Named pipe client server=%p\n", client
->server
);
323 static void named_pipe_destroy( struct object
*obj
)
325 struct named_pipe
*pipe
= (struct named_pipe
*) obj
;
327 assert( list_empty( &pipe
->servers
) );
328 assert( !pipe
->instances
);
329 free_async_queue( &pipe
->waiters
);
332 static struct fd
*pipe_end_get_fd( struct object
*obj
)
334 struct pipe_end
*pipe_end
= (struct pipe_end
*) obj
;
335 return (struct fd
*) grab_object( pipe_end
->fd
);
338 static void set_server_state( struct pipe_server
*server
, enum pipe_state state
)
340 server
->state
= state
;
344 static struct pipe_message
*queue_message( struct pipe_end
*pipe_end
, struct iosb
*iosb
)
346 struct pipe_message
*message
;
348 if (!(message
= mem_alloc( sizeof(*message
) ))) return NULL
;
349 message
->iosb
= (struct iosb
*)grab_object( iosb
);
350 message
->async
= NULL
;
351 message
->read_pos
= 0;
352 list_add_tail( &pipe_end
->message_queue
, &message
->entry
);
356 static void wake_message( struct pipe_message
*message
)
358 struct async
*async
= message
->async
;
360 message
->async
= NULL
;
363 message
->iosb
->status
= STATUS_SUCCESS
;
364 message
->iosb
->result
= message
->iosb
->in_size
;
365 async_terminate( async
, message
->iosb
->result
? STATUS_ALERTED
: STATUS_SUCCESS
);
366 release_object( async
);
369 static void free_message( struct pipe_message
*message
)
371 list_remove( &message
->entry
);
372 if (message
->iosb
) release_object( message
->iosb
);
376 static void pipe_end_disconnect( struct pipe_end
*pipe_end
, unsigned int status
)
378 struct pipe_end
*connection
= pipe_end
->connection
;
379 struct pipe_message
*message
, *next
;
382 pipe_end
->connection
= NULL
;
384 pipe_end
->state
= status
== STATUS_PIPE_DISCONNECTED
385 ? FILE_PIPE_DISCONNECTED_STATE
: FILE_PIPE_CLOSING_STATE
;
386 fd_async_wake_up( pipe_end
->fd
, ASYNC_TYPE_WAIT
, status
);
387 async_wake_up( &pipe_end
->read_q
, status
);
388 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
390 async
= message
->async
;
391 if (async
|| status
== STATUS_PIPE_DISCONNECTED
) free_message( message
);
392 if (!async
) continue;
393 async_terminate( async
, status
);
394 release_object( async
);
396 if (status
== STATUS_PIPE_DISCONNECTED
) set_fd_signaled( pipe_end
->fd
, 0 );
400 connection
->connection
= NULL
;
401 pipe_end_disconnect( connection
, status
);
405 static void pipe_end_destroy( struct pipe_end
*pipe_end
)
407 struct pipe_message
*message
;
409 while (!list_empty( &pipe_end
->message_queue
))
411 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
412 assert( !message
->async
);
413 free_message( message
);
416 free_async_queue( &pipe_end
->read_q
);
417 free_async_queue( &pipe_end
->write_q
);
418 if (pipe_end
->fd
) release_object( pipe_end
->fd
);
421 static void pipe_server_destroy( struct object
*obj
)
423 struct pipe_server
*server
= (struct pipe_server
*)obj
;
425 assert( obj
->ops
== &pipe_server_ops
);
427 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_BROKEN
);
429 pipe_end_destroy( &server
->pipe_end
);
432 server
->client
->server
= NULL
;
433 server
->client
= NULL
;
436 assert( server
->pipe
->instances
);
437 server
->pipe
->instances
--;
439 list_remove( &server
->entry
);
440 release_object( server
->pipe
);
443 static void pipe_client_destroy( struct object
*obj
)
445 struct pipe_client
*client
= (struct pipe_client
*)obj
;
446 struct pipe_server
*server
= client
->server
;
448 assert( obj
->ops
== &pipe_client_ops
);
450 pipe_end_disconnect( &client
->pipe_end
, STATUS_PIPE_BROKEN
);
454 switch(server
->state
)
456 case ps_connected_server
:
457 /* Don't destroy the server's fd here as we can't
458 do a successful flush without it. */
459 set_server_state( server
, ps_wait_disconnect
);
463 case ps_wait_disconnect
:
464 case ps_wait_connect
:
467 assert( server
->client
);
468 server
->client
= NULL
;
469 client
->server
= NULL
;
472 pipe_end_destroy( &client
->pipe_end
);
475 static void named_pipe_device_dump( struct object
*obj
, int verbose
)
477 fputs( "Named pipe device\n", stderr
);
480 static struct object_type
*named_pipe_device_get_type( struct object
*obj
)
482 static const WCHAR name
[] = {'D','e','v','i','c','e'};
483 static const struct unicode_str str
= { name
, sizeof(name
) };
484 return get_object_type( &str
);
487 static struct fd
*named_pipe_device_get_fd( struct object
*obj
)
489 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
490 return (struct fd
*)grab_object( device
->fd
);
493 static struct object
*named_pipe_device_lookup_name( struct object
*obj
, struct unicode_str
*name
,
496 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
497 struct object
*found
;
499 assert( obj
->ops
== &named_pipe_device_ops
);
500 assert( device
->pipes
);
502 if (!name
) return NULL
; /* open the device itself */
504 if ((found
= find_object( device
->pipes
, name
, attr
| OBJ_CASE_INSENSITIVE
)))
510 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
511 unsigned int sharing
, unsigned int options
)
513 return grab_object( obj
);
516 static void named_pipe_device_destroy( struct object
*obj
)
518 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
519 assert( obj
->ops
== &named_pipe_device_ops
);
520 if (device
->fd
) release_object( device
->fd
);
521 free( device
->pipes
);
524 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
)
526 return FD_TYPE_DEVICE
;
529 struct object
*create_named_pipe_device( struct object
*root
, const struct unicode_str
*name
)
531 struct named_pipe_device
*dev
;
533 if ((dev
= create_named_object( root
, &named_pipe_device_ops
, name
, 0, NULL
)) &&
534 get_error() != STATUS_OBJECT_NAME_EXISTS
)
537 if (!(dev
->fd
= alloc_pseudo_fd( &named_pipe_device_fd_ops
, &dev
->obj
, 0 )) ||
538 !(dev
->pipes
= create_namespace( 7 )))
540 release_object( dev
);
547 static int pipe_end_flush( struct fd
*fd
, struct async
*async
)
549 struct pipe_end
*pipe_end
= get_fd_user( fd
);
551 if (pipe_end
->connection
&& !list_empty( &pipe_end
->connection
->message_queue
))
553 fd_queue_async( pipe_end
->fd
, async
, ASYNC_TYPE_WAIT
);
554 set_error( STATUS_PENDING
);
559 static void pipe_end_get_file_info( struct fd
*fd
, struct named_pipe
*pipe
, unsigned int info_class
)
563 case FileNameInformation
:
565 FILE_NAME_INFORMATION
*name_info
;
566 data_size_t name_len
, reply_size
;
569 if (get_reply_max_size() < sizeof(*name_info
))
571 set_error( STATUS_INFO_LENGTH_MISMATCH
);
575 name
= get_object_name( &pipe
->obj
, &name_len
);
576 reply_size
= offsetof( FILE_NAME_INFORMATION
, FileName
[name_len
/sizeof(WCHAR
) + 1] );
577 if (reply_size
> get_reply_max_size())
579 reply_size
= get_reply_max_size();
580 set_error( STATUS_BUFFER_OVERFLOW
);
583 if (!(name_info
= set_reply_data_size( reply_size
))) return;
584 name_info
->FileNameLength
= name_len
+ sizeof(WCHAR
);
585 name_info
->FileName
[0] = '\\';
586 reply_size
-= offsetof( FILE_NAME_INFORMATION
, FileName
[1] );
587 if (reply_size
) memcpy( &name_info
->FileName
[1], name
, reply_size
);
591 no_fd_get_file_info( fd
, info_class
);
595 static struct security_descriptor
*pipe_server_get_sd( struct object
*obj
)
597 struct pipe_server
*server
= (struct pipe_server
*) obj
;
598 return default_get_sd( &server
->pipe
->obj
);
601 static struct security_descriptor
*pipe_client_get_sd( struct object
*obj
)
603 struct pipe_client
*client
= (struct pipe_client
*) obj
;
604 if (client
->server
) return default_get_sd( &client
->server
->pipe
->obj
);
605 set_error( STATUS_PIPE_DISCONNECTED
);
609 static int pipe_server_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
610 unsigned int set_info
)
612 struct pipe_server
*server
= (struct pipe_server
*) obj
;
613 return default_set_sd( &server
->pipe
->obj
, sd
, set_info
);
616 static int pipe_client_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
617 unsigned int set_info
)
619 struct pipe_client
*client
= (struct pipe_client
*) obj
;
620 if (client
->server
) return default_set_sd( &client
->server
->pipe
->obj
, sd
, set_info
);
621 set_error( STATUS_PIPE_DISCONNECTED
);
625 static void pipe_server_get_file_info( struct fd
*fd
, unsigned int info_class
)
627 struct pipe_server
*server
= get_fd_user( fd
);
628 pipe_end_get_file_info( fd
, server
->pipe
, info_class
);
631 static void pipe_client_get_file_info( struct fd
*fd
, unsigned int info_class
)
633 struct pipe_client
*client
= get_fd_user( fd
);
634 if (client
->server
) pipe_end_get_file_info( fd
, client
->server
->pipe
, info_class
);
635 else set_error( STATUS_PIPE_DISCONNECTED
);
638 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
)
642 case FileFsDeviceInformation
:
644 static const FILE_FS_DEVICE_INFORMATION device_info
=
646 FILE_DEVICE_NAMED_PIPE
,
647 FILE_DEVICE_ALLOW_APPCONTAINER_TRAVERSAL
649 if (get_reply_max_size() >= sizeof(device_info
))
650 set_reply_data( &device_info
, sizeof(device_info
) );
652 set_error( STATUS_BUFFER_TOO_SMALL
);
656 set_error( STATUS_NOT_IMPLEMENTED
);
660 static void message_queue_read( struct pipe_end
*pipe_end
, struct iosb
*iosb
)
662 struct pipe_message
*message
;
664 if (pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)
666 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
667 iosb
->out_size
= min( iosb
->out_size
, message
->iosb
->in_size
- message
->read_pos
);
668 iosb
->status
= message
->read_pos
+ iosb
->out_size
< message
->iosb
->in_size
669 ? STATUS_BUFFER_OVERFLOW
: STATUS_SUCCESS
;
673 data_size_t avail
= 0;
674 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
676 avail
+= message
->iosb
->in_size
- message
->read_pos
;
677 if (avail
>= iosb
->out_size
) break;
679 iosb
->out_size
= min( iosb
->out_size
, avail
);
680 iosb
->status
= STATUS_SUCCESS
;
683 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
684 if (!message
->read_pos
&& message
->iosb
->in_size
== iosb
->out_size
) /* fast path */
686 iosb
->out_data
= message
->iosb
->in_data
;
687 message
->iosb
->in_data
= NULL
;
688 wake_message( message
);
689 free_message( message
);
693 data_size_t write_pos
= 0, writing
;
696 if (iosb
->out_size
&& !(buf
= iosb
->out_data
= malloc( iosb
->out_size
)))
699 iosb
->status
= STATUS_NO_MEMORY
;
705 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
706 writing
= min( iosb
->out_size
- write_pos
, message
->iosb
->in_size
- message
->read_pos
);
707 if (writing
) memcpy( buf
+ write_pos
, (const char *)message
->iosb
->in_data
+ message
->read_pos
, writing
);
708 write_pos
+= writing
;
709 message
->read_pos
+= writing
;
710 if (message
->read_pos
== message
->iosb
->in_size
)
712 wake_message(message
);
713 free_message(message
);
715 } while (write_pos
< iosb
->out_size
);
717 iosb
->result
= iosb
->out_size
;
720 /* We call async_terminate in our reselect implementation, which causes recursive reselect.
721 * We're not interested in such reselect calls, so we ignore them. */
722 static int ignore_reselect
;
724 static void reselect_write_queue( struct pipe_end
*pipe_end
);
726 static void reselect_read_queue( struct pipe_end
*pipe_end
)
733 while (!list_empty( &pipe_end
->message_queue
) && (async
= find_pending_async( &pipe_end
->read_q
)))
735 iosb
= async_get_iosb( async
);
736 message_queue_read( pipe_end
, iosb
);
737 async_terminate( async
, iosb
->result
? STATUS_ALERTED
: iosb
->status
);
738 release_object( async
);
739 release_object( iosb
);
744 if (pipe_end
->connection
)
746 if (list_empty( &pipe_end
->message_queue
))
747 fd_async_wake_up( pipe_end
->connection
->fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
749 reselect_write_queue( pipe_end
->connection
);
753 static void reselect_write_queue( struct pipe_end
*pipe_end
)
755 struct pipe_message
*message
, *next
;
756 struct pipe_end
*reader
= pipe_end
->connection
;
757 data_size_t avail
= 0;
763 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &reader
->message_queue
, struct pipe_message
, entry
)
765 if (message
->async
&& message
->iosb
->status
!= STATUS_PENDING
)
767 release_object( message
->async
);
768 message
->async
= NULL
;
769 free_message( message
);
773 avail
+= message
->iosb
->in_size
- message
->read_pos
;
774 if (message
->async
&& (avail
<= reader
->buffer_size
|| !message
->iosb
->in_size
))
775 wake_message( message
);
780 reselect_read_queue( reader
);
783 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
785 struct pipe_end
*pipe_end
= get_fd_user( fd
);
787 switch (pipe_end
->state
)
789 case FILE_PIPE_CONNECTED_STATE
:
791 case FILE_PIPE_DISCONNECTED_STATE
:
792 set_error( STATUS_PIPE_DISCONNECTED
);
794 case FILE_PIPE_LISTENING_STATE
:
795 set_error( STATUS_PIPE_LISTENING
);
797 case FILE_PIPE_CLOSING_STATE
:
798 if (!list_empty( &pipe_end
->message_queue
)) break;
799 set_error( STATUS_PIPE_BROKEN
);
803 queue_async( &pipe_end
->read_q
, async
);
804 reselect_read_queue( pipe_end
);
805 set_error( STATUS_PENDING
);
809 static int pipe_end_write( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
811 struct pipe_end
*pipe_end
= get_fd_user( fd
);
812 struct pipe_message
*message
;
815 switch (pipe_end
->state
)
817 case FILE_PIPE_CONNECTED_STATE
:
819 case FILE_PIPE_DISCONNECTED_STATE
:
820 set_error( STATUS_PIPE_DISCONNECTED
);
822 case FILE_PIPE_LISTENING_STATE
:
823 set_error( STATUS_PIPE_LISTENING
);
825 case FILE_PIPE_CLOSING_STATE
:
826 set_error( STATUS_PIPE_CLOSING
);
830 if (!(pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
) && !get_req_data_size()) return 1;
832 iosb
= async_get_iosb( async
);
833 message
= queue_message( pipe_end
->connection
, iosb
);
834 release_object( iosb
);
835 if (!message
) return 0;
837 message
->async
= (struct async
*)grab_object( async
);
838 queue_async( &pipe_end
->write_q
, async
);
839 reselect_write_queue( pipe_end
);
840 set_error( STATUS_PENDING
);
844 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
)
846 struct pipe_end
*pipe_end
= get_fd_user( fd
);
848 if (ignore_reselect
) return;
850 if (&pipe_end
->write_q
== queue
)
851 reselect_write_queue( pipe_end
);
852 else if (&pipe_end
->read_q
== queue
)
853 reselect_read_queue( pipe_end
);
856 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
)
861 static int pipe_end_peek( struct pipe_end
*pipe_end
)
863 unsigned reply_size
= get_reply_max_size();
864 FILE_PIPE_PEEK_BUFFER
*buffer
;
865 struct pipe_message
*message
;
866 data_size_t avail
= 0;
867 data_size_t message_length
= 0;
869 if (reply_size
< offsetof( FILE_PIPE_PEEK_BUFFER
, Data
))
871 set_error( STATUS_INFO_LENGTH_MISMATCH
);
874 reply_size
-= offsetof( FILE_PIPE_PEEK_BUFFER
, Data
);
876 switch (pipe_end
->state
)
878 case FILE_PIPE_CONNECTED_STATE
:
880 case FILE_PIPE_CLOSING_STATE
:
881 if (!list_empty( &pipe_end
->message_queue
)) break;
882 set_error( STATUS_PIPE_BROKEN
);
885 set_error( STATUS_INVALID_PIPE_STATE
);
889 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
890 avail
+= message
->iosb
->in_size
- message
->read_pos
;
891 reply_size
= min( reply_size
, avail
);
893 if (avail
&& (pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
))
895 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
896 message_length
= message
->iosb
->in_size
- message
->read_pos
;
897 reply_size
= min( reply_size
, message_length
);
900 if (!(buffer
= set_reply_data_size( offsetof( FILE_PIPE_PEEK_BUFFER
, Data
[reply_size
] )))) return 0;
901 buffer
->NamedPipeState
= pipe_end
->state
;
902 buffer
->ReadDataAvailable
= avail
;
903 buffer
->NumberOfMessages
= 0; /* FIXME */
904 buffer
->MessageLength
= message_length
;
908 data_size_t write_pos
= 0, writing
;
909 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
911 writing
= min( reply_size
- write_pos
, message
->iosb
->in_size
- message
->read_pos
);
912 memcpy( buffer
->Data
+ write_pos
, (const char *)message
->iosb
->in_data
+ message
->read_pos
,
914 write_pos
+= writing
;
915 if (write_pos
== reply_size
) break;
921 static int pipe_end_transceive( struct pipe_end
*pipe_end
, struct async
*async
)
923 struct pipe_message
*message
;
926 if (!pipe_end
->connection
)
928 set_error( STATUS_INVALID_PIPE_STATE
);
932 if ((pipe_end
->flags
& (NAMED_PIPE_MESSAGE_STREAM_WRITE
| NAMED_PIPE_MESSAGE_STREAM_READ
))
933 != (NAMED_PIPE_MESSAGE_STREAM_WRITE
| NAMED_PIPE_MESSAGE_STREAM_READ
))
935 set_error( STATUS_INVALID_READ_MODE
);
939 /* not allowed if we already have read data buffered */
940 if (!list_empty( &pipe_end
->message_queue
))
942 set_error( STATUS_PIPE_BUSY
);
946 iosb
= async_get_iosb( async
);
947 /* ignore output buffer copy transferred because of METHOD_NEITHER */
948 iosb
->in_size
-= iosb
->out_size
;
949 /* transaction never blocks on write, so just queue a message without async */
950 message
= queue_message( pipe_end
->connection
, iosb
);
951 release_object( iosb
);
952 if (!message
) return 0;
953 reselect_read_queue( pipe_end
->connection
);
955 queue_async( &pipe_end
->read_q
, async
);
956 reselect_read_queue( pipe_end
);
957 set_error( STATUS_PENDING
);
961 static int pipe_end_get_connection_attribute( struct pipe_end
*pipe_end
)
963 const char *attr
= get_req_data();
964 data_size_t value_size
, attr_size
= get_req_data_size();
967 if (attr_size
== sizeof("ClientProcessId") && !memcmp( attr
, "ClientProcessId", attr_size
))
969 value
= &pipe_end
->client_pid
;
970 value_size
= sizeof(pipe_end
->client_pid
);
972 else if (attr_size
== sizeof("ServerProcessId") && !memcmp( attr
, "ServerProcessId", attr_size
))
974 value
= &pipe_end
->server_pid
;
975 value_size
= sizeof(pipe_end
->server_pid
);
979 set_error( STATUS_ILLEGAL_FUNCTION
);
983 if (get_reply_max_size() < value_size
)
985 set_error( STATUS_INFO_LENGTH_MISMATCH
);
989 set_reply_data( value
, value_size
);
993 static int pipe_end_ioctl( struct pipe_end
*pipe_end
, ioctl_code_t code
, struct async
*async
)
997 case FSCTL_PIPE_GET_CONNECTION_ATTRIBUTE
:
998 return pipe_end_get_connection_attribute( pipe_end
);
1000 case FSCTL_PIPE_PEEK
:
1001 return pipe_end_peek( pipe_end
);
1003 case FSCTL_PIPE_TRANSCEIVE
:
1004 return pipe_end_transceive( pipe_end
, async
);
1007 return default_fd_ioctl( pipe_end
->fd
, code
, async
);
1011 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1013 struct pipe_server
*server
= get_fd_user( fd
);
1017 case FSCTL_PIPE_LISTEN
:
1018 switch(server
->state
)
1020 case ps_idle_server
:
1021 case ps_wait_connect
:
1022 fd_queue_async( server
->pipe_end
.fd
, async
, ASYNC_TYPE_WAIT
);
1023 server
->pipe_end
.state
= FILE_PIPE_LISTENING_STATE
;
1024 set_server_state( server
, ps_wait_open
);
1025 async_wake_up( &server
->pipe
->waiters
, STATUS_SUCCESS
);
1026 set_error( STATUS_PENDING
);
1028 case ps_connected_server
:
1029 set_error( STATUS_PIPE_CONNECTED
);
1031 case ps_wait_disconnect
:
1032 set_error( STATUS_NO_DATA_DETECTED
);
1035 set_error( STATUS_INVALID_HANDLE
);
1040 case FSCTL_PIPE_DISCONNECT
:
1041 switch(server
->state
)
1043 case ps_connected_server
:
1044 assert( server
->client
);
1046 /* dump the client and server fds - client loses all waiting data */
1047 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_DISCONNECTED
);
1048 server
->client
->server
= NULL
;
1049 server
->client
= NULL
;
1050 set_server_state( server
, ps_wait_connect
);
1052 case ps_wait_disconnect
:
1053 assert( !server
->client
);
1054 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_DISCONNECTED
);
1055 set_server_state( server
, ps_wait_connect
);
1057 case ps_idle_server
:
1059 set_error( STATUS_PIPE_LISTENING
);
1061 case ps_wait_connect
:
1062 set_error( STATUS_PIPE_DISCONNECTED
);
1068 return pipe_end_ioctl( &server
->pipe_end
, code
, async
);
1072 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1074 struct pipe_client
*client
= get_fd_user( fd
);
1078 case FSCTL_PIPE_LISTEN
:
1079 set_error( STATUS_ILLEGAL_FUNCTION
);
1083 return pipe_end_ioctl( &client
->pipe_end
, code
, async
);
1087 static struct pipe_server
*get_pipe_server_obj( struct process
*process
,
1088 obj_handle_t handle
, unsigned int access
)
1091 obj
= get_handle_obj( process
, handle
, access
, &pipe_server_ops
);
1092 return (struct pipe_server
*) obj
;
1095 static void init_pipe_end( struct pipe_end
*pipe_end
, unsigned int pipe_flags
, data_size_t buffer_size
)
1097 pipe_end
->fd
= NULL
;
1098 pipe_end
->flags
= pipe_flags
;
1099 pipe_end
->connection
= NULL
;
1100 pipe_end
->buffer_size
= buffer_size
;
1101 init_async_queue( &pipe_end
->read_q
);
1102 init_async_queue( &pipe_end
->write_q
);
1103 list_init( &pipe_end
->message_queue
);
1106 static struct pipe_server
*create_pipe_server( struct named_pipe
*pipe
, unsigned int options
,
1107 unsigned int pipe_flags
)
1109 struct pipe_server
*server
;
1111 server
= alloc_object( &pipe_server_ops
);
1115 server
->pipe
= pipe
;
1116 server
->client
= NULL
;
1117 server
->options
= options
;
1118 init_pipe_end( &server
->pipe_end
, pipe_flags
, pipe
->insize
);
1119 server
->pipe_end
.state
= FILE_PIPE_LISTENING_STATE
;
1120 server
->pipe_end
.server_pid
= get_process_id( current
->process
);
1122 list_add_head( &pipe
->servers
, &server
->entry
);
1123 grab_object( pipe
);
1124 if (!(server
->pipe_end
.fd
= alloc_pseudo_fd( &pipe_server_fd_ops
, &server
->pipe_end
.obj
, options
)))
1126 release_object( server
);
1129 allow_fd_caching( server
->pipe_end
.fd
);
1130 set_fd_signaled( server
->pipe_end
.fd
, 1 );
1131 set_server_state( server
, ps_idle_server
);
1135 static struct pipe_client
*create_pipe_client( unsigned int flags
, unsigned int pipe_flags
,
1136 data_size_t buffer_size
, unsigned int options
)
1138 struct pipe_client
*client
;
1140 client
= alloc_object( &pipe_client_ops
);
1144 client
->server
= NULL
;
1145 client
->flags
= flags
;
1146 init_pipe_end( &client
->pipe_end
, pipe_flags
, buffer_size
);
1147 client
->pipe_end
.state
= FILE_PIPE_CONNECTED_STATE
;
1148 client
->pipe_end
.client_pid
= get_process_id( current
->process
);
1150 client
->pipe_end
.fd
= alloc_pseudo_fd( &pipe_client_fd_ops
, &client
->pipe_end
.obj
, options
);
1151 if (!client
->pipe_end
.fd
)
1153 release_object( client
);
1156 allow_fd_caching( client
->pipe_end
.fd
);
1157 set_fd_signaled( client
->pipe_end
.fd
, 1 );
1162 static struct pipe_server
*find_available_server( struct named_pipe
*pipe
)
1164 struct pipe_server
*server
;
1166 /* look for pipe servers that are listening */
1167 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1169 if (server
->state
== ps_wait_open
)
1170 return (struct pipe_server
*)grab_object( server
);
1173 /* fall back to pipe servers that are idle */
1174 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1176 if (server
->state
== ps_idle_server
)
1177 return (struct pipe_server
*)grab_object( server
);
1183 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
)
1185 struct named_pipe_device
*dev
= (struct named_pipe_device
*)parent
;
1187 if (parent
->ops
!= &named_pipe_device_ops
)
1189 set_error( STATUS_OBJECT_NAME_INVALID
);
1192 namespace_add( dev
->pipes
, name
);
1193 name
->parent
= grab_object( parent
);
1197 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
1198 unsigned int sharing
, unsigned int options
)
1200 struct named_pipe
*pipe
= (struct named_pipe
*)obj
;
1201 struct pipe_server
*server
;
1202 struct pipe_client
*client
;
1203 unsigned int pipe_sharing
;
1205 if (!(server
= find_available_server( pipe
)))
1207 set_error( STATUS_PIPE_NOT_AVAILABLE
);
1211 pipe_sharing
= server
->pipe
->sharing
;
1212 if (((access
& GENERIC_READ
) && !(pipe_sharing
& FILE_SHARE_READ
)) ||
1213 ((access
& GENERIC_WRITE
) && !(pipe_sharing
& FILE_SHARE_WRITE
)))
1215 set_error( STATUS_ACCESS_DENIED
);
1216 release_object( server
);
1220 if ((client
= create_pipe_client( options
, pipe
->flags
, pipe
->outsize
, options
)))
1222 if (server
->state
== ps_wait_open
)
1223 fd_async_wake_up( server
->pipe_end
.fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
1224 server
->pipe_end
.state
= FILE_PIPE_CONNECTED_STATE
;
1225 set_server_state( server
, ps_connected_server
);
1226 server
->client
= client
;
1227 client
->server
= server
;
1228 server
->pipe_end
.connection
= &client
->pipe_end
;
1229 client
->pipe_end
.connection
= &server
->pipe_end
;
1230 server
->pipe_end
.client_pid
= client
->pipe_end
.client_pid
;
1231 client
->pipe_end
.server_pid
= server
->pipe_end
.server_pid
;
1233 release_object( server
);
1234 return &client
->pipe_end
.obj
;
1237 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1239 struct named_pipe_device
*device
= get_fd_user( fd
);
1243 case FSCTL_PIPE_WAIT
:
1245 const FILE_PIPE_WAIT_FOR_BUFFER
*buffer
= get_req_data();
1246 data_size_t size
= get_req_data_size();
1247 struct named_pipe
*pipe
;
1248 struct pipe_server
*server
;
1249 struct unicode_str name
;
1252 if (size
< sizeof(*buffer
) ||
1253 size
< FIELD_OFFSET(FILE_PIPE_WAIT_FOR_BUFFER
, Name
[buffer
->NameLength
/sizeof(WCHAR
)]))
1255 set_error( STATUS_INVALID_PARAMETER
);
1258 name
.str
= buffer
->Name
;
1259 name
.len
= (buffer
->NameLength
/ sizeof(WCHAR
)) * sizeof(WCHAR
);
1260 if (!(pipe
= open_named_object( &device
->obj
, &named_pipe_ops
, &name
, 0 ))) return 0;
1262 if (!(server
= find_available_server( pipe
)))
1264 queue_async( &pipe
->waiters
, async
);
1265 when
= buffer
->TimeoutSpecified
? buffer
->Timeout
.QuadPart
: pipe
->timeout
;
1266 async_set_timeout( async
, when
, STATUS_IO_TIMEOUT
);
1267 release_object( pipe
);
1268 set_error( STATUS_PENDING
);
1272 release_object( server
);
1273 release_object( pipe
);
1278 return default_fd_ioctl( fd
, code
, async
);
1283 DECL_HANDLER(create_named_pipe
)
1285 struct named_pipe
*pipe
;
1286 struct pipe_server
*server
;
1287 struct unicode_str name
;
1288 struct object
*root
;
1289 const struct security_descriptor
*sd
;
1290 const struct object_attributes
*objattr
= get_req_object_attributes( &sd
, &name
, &root
);
1292 if (!objattr
) return;
1294 if (!req
->sharing
|| (req
->sharing
& ~(FILE_SHARE_READ
| FILE_SHARE_WRITE
)) ||
1295 (!(req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
) && (req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)))
1297 if (root
) release_object( root
);
1298 set_error( STATUS_INVALID_PARAMETER
);
1302 if (!name
.len
) /* pipes need a root directory even without a name */
1304 if (!objattr
->rootdir
)
1306 set_error( STATUS_OBJECT_PATH_SYNTAX_BAD
);
1309 if (!(root
= get_directory_obj( current
->process
, objattr
->rootdir
))) return;
1312 pipe
= create_named_object( root
, &named_pipe_ops
, &name
, objattr
->attributes
| OBJ_OPENIF
, NULL
);
1314 if (root
) release_object( root
);
1317 if (get_error() != STATUS_OBJECT_NAME_EXISTS
)
1319 /* initialize it if it didn't already exist */
1320 pipe
->instances
= 0;
1321 init_async_queue( &pipe
->waiters
);
1322 list_init( &pipe
->servers
);
1323 pipe
->insize
= req
->insize
;
1324 pipe
->outsize
= req
->outsize
;
1325 pipe
->maxinstances
= req
->maxinstances
;
1326 pipe
->timeout
= req
->timeout
;
1327 pipe
->flags
= req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
;
1328 pipe
->sharing
= req
->sharing
;
1329 if (sd
) default_set_sd( &pipe
->obj
, sd
, OWNER_SECURITY_INFORMATION
|
1330 GROUP_SECURITY_INFORMATION
|
1331 DACL_SECURITY_INFORMATION
|
1332 SACL_SECURITY_INFORMATION
);
1336 if (pipe
->maxinstances
<= pipe
->instances
)
1338 set_error( STATUS_INSTANCE_NOT_AVAILABLE
);
1339 release_object( pipe
);
1342 if (pipe
->sharing
!= req
->sharing
)
1344 set_error( STATUS_ACCESS_DENIED
);
1345 release_object( pipe
);
1348 clear_error(); /* clear the name collision */
1351 server
= create_pipe_server( pipe
, req
->options
, req
->flags
);
1354 reply
->handle
= alloc_handle( current
->process
, server
, req
->access
, objattr
->attributes
);
1355 server
->pipe
->instances
++;
1356 release_object( server
);
1359 release_object( pipe
);
1362 DECL_HANDLER(get_named_pipe_info
)
1364 struct pipe_server
*server
;
1365 struct pipe_client
*client
= NULL
;
1367 server
= get_pipe_server_obj( current
->process
, req
->handle
, FILE_READ_ATTRIBUTES
);
1370 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1374 client
= (struct pipe_client
*)get_handle_obj( current
->process
, req
->handle
,
1375 0, &pipe_client_ops
);
1376 if (!client
) return;
1377 server
= client
->server
;
1380 reply
->flags
= client
? client
->pipe_end
.flags
: server
->pipe_end
.flags
;
1383 reply
->sharing
= server
->pipe
->sharing
;
1384 reply
->maxinstances
= server
->pipe
->maxinstances
;
1385 reply
->instances
= server
->pipe
->instances
;
1386 reply
->insize
= server
->pipe
->insize
;
1387 reply
->outsize
= server
->pipe
->outsize
;
1391 release_object(client
);
1394 reply
->flags
|= NAMED_PIPE_SERVER_END
;
1395 release_object(server
);
1399 DECL_HANDLER(set_named_pipe_info
)
1401 struct pipe_server
*server
;
1402 struct pipe_client
*client
= NULL
;
1404 server
= get_pipe_server_obj( current
->process
, req
->handle
, FILE_WRITE_ATTRIBUTES
);
1407 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1411 client
= (struct pipe_client
*)get_handle_obj( current
->process
, req
->handle
,
1412 0, &pipe_client_ops
);
1413 if (!client
) return;
1414 if (!(server
= client
->server
))
1416 release_object( client
);
1421 if ((req
->flags
& ~(NAMED_PIPE_MESSAGE_STREAM_READ
| NAMED_PIPE_NONBLOCKING_MODE
)) ||
1422 ((req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
) && !(server
->pipe
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
)))
1424 set_error( STATUS_INVALID_PARAMETER
);
1428 client
->pipe_end
.flags
= server
->pipe
->flags
| req
->flags
;
1432 server
->pipe_end
.flags
= server
->pipe
->flags
| req
->flags
;
1436 release_object(client
);
1438 release_object(server
);