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 #include <sys/types.h>
34 #ifdef HAVE_SYS_SOCKET_H
35 #include <sys/socket.h>
44 #define WIN32_NO_STATUS
68 struct list entry
; /* entry in message queue */
69 data_size_t read_pos
; /* already read bytes */
70 struct iosb
*iosb
; /* message iosb */
71 struct async
*async
; /* async of pending write */
76 struct object obj
; /* object header */
77 struct fd
*fd
; /* pipe file descriptor */
78 unsigned int flags
; /* pipe flags */
79 struct pipe_end
*connection
; /* the other end of the pipe */
80 data_size_t buffer_size
;/* size of buffered data that doesn't block caller */
81 struct list message_queue
;
82 struct async_queue read_q
; /* read queue */
83 struct async_queue write_q
; /* write queue */
88 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
89 struct fd
*ioctl_fd
; /* file descriptor for ioctls when not connected */
90 struct list entry
; /* entry in named pipe servers list */
91 enum pipe_state state
; /* server state */
92 struct pipe_client
*client
; /* client that this server is connected to */
93 struct named_pipe
*pipe
;
94 struct timeout_user
*flush_poll
;
95 unsigned int options
; /* pipe options */
100 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
101 struct pipe_server
*server
; /* server that this client is connected to */
102 unsigned int flags
; /* file flags */
107 struct object obj
; /* object header */
109 unsigned int sharing
;
110 unsigned int maxinstances
;
111 unsigned int outsize
;
113 unsigned int instances
;
115 struct list servers
; /* list of servers using this pipe */
116 struct async_queue waiters
; /* list of clients waiting to connect */
119 struct named_pipe_device
121 struct object obj
; /* object header */
122 struct fd
*fd
; /* pseudo-fd for ioctls */
123 struct namespace *pipes
; /* named pipe namespace */
126 static void named_pipe_dump( struct object
*obj
, int verbose
);
127 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
);
128 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
);
129 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
130 unsigned int sharing
, unsigned int options
);
131 static void named_pipe_destroy( struct object
*obj
);
133 static const struct object_ops named_pipe_ops
=
135 sizeof(struct named_pipe
), /* size */
136 named_pipe_dump
, /* dump */
137 no_get_type
, /* get_type */
138 no_add_queue
, /* add_queue */
139 NULL
, /* remove_queue */
141 NULL
, /* satisfied */
142 no_signal
, /* signal */
143 no_get_fd
, /* get_fd */
144 named_pipe_map_access
, /* map_access */
145 default_get_sd
, /* get_sd */
146 default_set_sd
, /* set_sd */
147 no_lookup_name
, /* lookup_name */
148 named_pipe_link_name
, /* link_name */
149 default_unlink_name
, /* unlink_name */
150 named_pipe_open_file
, /* open_file */
151 no_close_handle
, /* close_handle */
152 named_pipe_destroy
/* destroy */
155 /* common server and client pipe end functions */
156 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
);
157 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
);
158 static int pipe_end_write( struct fd
*fd
, struct async
*async_data
, file_pos_t pos
);
159 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
);
160 static void pipe_end_queue_async( struct fd
*fd
, struct async
*async
, int type
, int count
);
161 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
);
163 /* server end functions */
164 static void pipe_server_dump( struct object
*obj
, int verbose
);
165 static struct fd
*pipe_server_get_fd( struct object
*obj
);
166 static void pipe_server_destroy( struct object
*obj
);
167 static int pipe_server_flush( struct fd
*fd
, struct async
*async
);
168 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
170 static const struct object_ops pipe_server_ops
=
172 sizeof(struct pipe_server
), /* size */
173 pipe_server_dump
, /* dump */
174 no_get_type
, /* get_type */
175 add_queue
, /* add_queue */
176 remove_queue
, /* remove_queue */
177 default_fd_signaled
, /* signaled */
178 no_satisfied
, /* satisfied */
179 no_signal
, /* signal */
180 pipe_server_get_fd
, /* get_fd */
181 default_fd_map_access
, /* map_access */
182 default_get_sd
, /* get_sd */
183 default_set_sd
, /* set_sd */
184 no_lookup_name
, /* lookup_name */
185 no_link_name
, /* link_name */
186 NULL
, /* unlink_name */
187 no_open_file
, /* open_file */
188 fd_close_handle
, /* close_handle */
189 pipe_server_destroy
/* destroy */
192 static const struct fd_ops pipe_server_fd_ops
=
194 default_fd_get_poll_events
, /* get_poll_events */
195 default_poll_event
, /* poll_event */
196 pipe_end_get_fd_type
, /* get_fd_type */
197 pipe_end_read
, /* read */
198 pipe_end_write
, /* write */
199 pipe_server_flush
, /* flush */
200 pipe_end_get_volume_info
, /* get_volume_info */
201 pipe_server_ioctl
, /* ioctl */
202 pipe_end_queue_async
, /* queue_async */
203 pipe_end_reselect_async
/* reselect_async */
206 /* client end functions */
207 static void pipe_client_dump( struct object
*obj
, int verbose
);
208 static int pipe_client_signaled( struct object
*obj
, struct wait_queue_entry
*entry
);
209 static struct fd
*pipe_client_get_fd( struct object
*obj
);
210 static void pipe_client_destroy( struct object
*obj
);
211 static int pipe_client_flush( struct fd
*fd
, struct async
*async
);
212 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
214 static const struct object_ops pipe_client_ops
=
216 sizeof(struct pipe_client
), /* size */
217 pipe_client_dump
, /* dump */
218 no_get_type
, /* get_type */
219 add_queue
, /* add_queue */
220 remove_queue
, /* remove_queue */
221 pipe_client_signaled
, /* signaled */
222 no_satisfied
, /* satisfied */
223 no_signal
, /* signal */
224 pipe_client_get_fd
, /* get_fd */
225 default_fd_map_access
, /* map_access */
226 default_get_sd
, /* get_sd */
227 default_set_sd
, /* set_sd */
228 no_lookup_name
, /* lookup_name */
229 no_link_name
, /* link_name */
230 NULL
, /* unlink_name */
231 no_open_file
, /* open_file */
232 fd_close_handle
, /* close_handle */
233 pipe_client_destroy
/* destroy */
236 static const struct fd_ops pipe_client_fd_ops
=
238 default_fd_get_poll_events
, /* get_poll_events */
239 default_poll_event
, /* poll_event */
240 pipe_end_get_fd_type
, /* get_fd_type */
241 pipe_end_read
, /* read */
242 pipe_end_write
, /* write */
243 pipe_client_flush
, /* flush */
244 pipe_end_get_volume_info
, /* get_volume_info */
245 pipe_client_ioctl
, /* ioctl */
246 pipe_end_queue_async
, /* queue_async */
247 pipe_end_reselect_async
/* reselect_async */
250 static void named_pipe_device_dump( struct object
*obj
, int verbose
);
251 static struct object_type
*named_pipe_device_get_type( struct object
*obj
);
252 static struct fd
*named_pipe_device_get_fd( struct object
*obj
);
253 static struct object
*named_pipe_device_lookup_name( struct object
*obj
,
254 struct unicode_str
*name
, unsigned int attr
);
255 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
256 unsigned int sharing
, unsigned int options
);
257 static void named_pipe_device_destroy( struct object
*obj
);
258 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
);
259 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
261 static const struct object_ops named_pipe_device_ops
=
263 sizeof(struct named_pipe_device
), /* size */
264 named_pipe_device_dump
, /* dump */
265 named_pipe_device_get_type
, /* get_type */
266 no_add_queue
, /* add_queue */
267 NULL
, /* remove_queue */
269 no_satisfied
, /* satisfied */
270 no_signal
, /* signal */
271 named_pipe_device_get_fd
, /* get_fd */
272 no_map_access
, /* map_access */
273 default_get_sd
, /* get_sd */
274 default_set_sd
, /* set_sd */
275 named_pipe_device_lookup_name
, /* lookup_name */
276 directory_link_name
, /* link_name */
277 default_unlink_name
, /* unlink_name */
278 named_pipe_device_open_file
, /* open_file */
279 fd_close_handle
, /* close_handle */
280 named_pipe_device_destroy
/* destroy */
283 static const struct fd_ops named_pipe_device_fd_ops
=
285 default_fd_get_poll_events
, /* get_poll_events */
286 default_poll_event
, /* poll_event */
287 named_pipe_device_get_fd_type
, /* get_fd_type */
288 no_fd_read
, /* read */
289 no_fd_write
, /* write */
290 no_fd_flush
, /* flush */
291 no_fd_get_volume_info
, /* get_volume_info */
292 named_pipe_device_ioctl
, /* ioctl */
293 default_fd_queue_async
, /* queue_async */
294 default_fd_reselect_async
/* reselect_async */
297 /* Returns if we handle I/O via server calls. Currently message-mode pipes are handled this way. */
298 static int use_server_io( struct pipe_end
*pipe_end
)
300 return pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
;
303 static void named_pipe_dump( struct object
*obj
, int verbose
)
305 fputs( "Named pipe\n", stderr
);
308 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
)
310 if (access
& GENERIC_READ
) access
|= STANDARD_RIGHTS_READ
;
311 if (access
& GENERIC_WRITE
) access
|= STANDARD_RIGHTS_WRITE
| FILE_CREATE_PIPE_INSTANCE
;
312 if (access
& GENERIC_EXECUTE
) access
|= STANDARD_RIGHTS_EXECUTE
;
313 if (access
& GENERIC_ALL
) access
|= STANDARD_RIGHTS_ALL
;
314 return access
& ~(GENERIC_READ
| GENERIC_WRITE
| GENERIC_EXECUTE
| GENERIC_ALL
);
317 static void pipe_server_dump( struct object
*obj
, int verbose
)
319 struct pipe_server
*server
= (struct pipe_server
*) obj
;
320 assert( obj
->ops
== &pipe_server_ops
);
321 fprintf( stderr
, "Named pipe server pipe=%p state=%d\n", server
->pipe
, server
->state
);
324 static void pipe_client_dump( struct object
*obj
, int verbose
)
326 struct pipe_client
*client
= (struct pipe_client
*) obj
;
327 assert( obj
->ops
== &pipe_client_ops
);
328 fprintf( stderr
, "Named pipe client server=%p\n", client
->server
);
331 static int pipe_client_signaled( struct object
*obj
, struct wait_queue_entry
*entry
)
333 struct pipe_client
*client
= (struct pipe_client
*) obj
;
335 return client
->pipe_end
.fd
&& is_fd_signaled(client
->pipe_end
.fd
);
338 static void named_pipe_destroy( struct object
*obj
)
340 struct named_pipe
*pipe
= (struct named_pipe
*) obj
;
342 assert( list_empty( &pipe
->servers
) );
343 assert( !pipe
->instances
);
344 free_async_queue( &pipe
->waiters
);
347 static struct fd
*pipe_client_get_fd( struct object
*obj
)
349 struct pipe_client
*client
= (struct pipe_client
*) obj
;
350 if (client
->pipe_end
.fd
)
351 return (struct fd
*) grab_object( client
->pipe_end
.fd
);
352 set_error( STATUS_PIPE_DISCONNECTED
);
356 static void set_server_state( struct pipe_server
*server
, enum pipe_state state
)
358 server
->state
= state
;
362 case ps_connected_server
:
363 case ps_wait_disconnect
:
364 assert( server
->pipe_end
.fd
);
368 assert( !server
->pipe_end
.fd
);
369 set_no_fd_status( server
->ioctl_fd
, STATUS_PIPE_LISTENING
);
371 case ps_wait_connect
:
372 assert( !server
->pipe_end
.fd
);
373 set_no_fd_status( server
->ioctl_fd
, STATUS_PIPE_DISCONNECTED
);
378 static struct fd
*pipe_server_get_fd( struct object
*obj
)
380 struct pipe_server
*server
= (struct pipe_server
*) obj
;
382 return (struct fd
*)grab_object( server
->pipe_end
.fd
? server
->pipe_end
.fd
: server
->ioctl_fd
);
386 static void notify_empty( struct pipe_server
*server
)
388 if (!server
->flush_poll
)
390 assert( server
->state
== ps_connected_server
);
391 remove_timeout_user( server
->flush_poll
);
392 server
->flush_poll
= NULL
;
393 fd_async_wake_up( server
->pipe_end
.fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
396 static void wake_message( struct pipe_message
*message
)
398 struct async
*async
= message
->async
;
400 message
->async
= NULL
;
401 message
->iosb
->status
= STATUS_SUCCESS
;
402 message
->iosb
->result
= message
->iosb
->in_size
;
405 async_terminate( async
, message
->iosb
->result
? STATUS_ALERTED
: STATUS_SUCCESS
);
406 release_object( async
);
410 static void free_message( struct pipe_message
*message
)
412 list_remove( &message
->entry
);
413 if (message
->iosb
) release_object( message
->iosb
);
417 static void pipe_end_disconnect( struct pipe_end
*pipe_end
, unsigned int status
)
419 struct pipe_end
*connection
= pipe_end
->connection
;
421 pipe_end
->connection
= NULL
;
423 if (use_server_io( pipe_end
))
425 struct pipe_message
*message
, *next
;
427 if (pipe_end
->fd
) fd_async_wake_up( pipe_end
->fd
, ASYNC_TYPE_WAIT
, status
);
428 async_wake_up( &pipe_end
->read_q
, status
);
429 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
431 async
= message
->async
;
432 if (async
|| status
== STATUS_PIPE_DISCONNECTED
) free_message( message
);
433 if (!async
) continue;
434 async_terminate( async
, status
);
435 release_object( async
);
437 if (status
== STATUS_PIPE_DISCONNECTED
) set_fd_signaled( pipe_end
->fd
, 0 );
441 connection
->connection
= NULL
;
442 pipe_end_disconnect( connection
, status
);
446 static void do_disconnect( struct pipe_server
*server
)
448 /* we may only have a server fd, if the client disconnected */
451 assert( server
->client
->server
== server
);
452 assert( server
->client
->pipe_end
.fd
);
453 if (!use_server_io( &server
->pipe_end
))
455 release_object( server
->client
->pipe_end
.fd
);
456 server
->client
->pipe_end
.fd
= NULL
;
459 assert( server
->pipe_end
.fd
);
460 if (!use_server_io( &server
->pipe_end
))
461 shutdown( get_unix_fd( server
->pipe_end
.fd
), SHUT_RDWR
);
462 release_object( server
->pipe_end
.fd
);
463 server
->pipe_end
.fd
= NULL
;
466 static void pipe_end_destroy( struct pipe_end
*pipe_end
)
468 struct pipe_message
*message
;
470 while (!list_empty( &pipe_end
->message_queue
))
472 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
473 assert( !message
->async
);
474 free_message( message
);
477 free_async_queue( &pipe_end
->read_q
);
478 free_async_queue( &pipe_end
->write_q
);
481 static void pipe_server_destroy( struct object
*obj
)
483 struct pipe_server
*server
= (struct pipe_server
*)obj
;
485 assert( obj
->ops
== &pipe_server_ops
);
487 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_BROKEN
);
489 if (server
->pipe_end
.fd
)
491 notify_empty( server
);
492 do_disconnect( server
);
495 pipe_end_destroy( &server
->pipe_end
);
498 server
->client
->server
= NULL
;
499 server
->client
= NULL
;
502 assert( server
->pipe
->instances
);
503 server
->pipe
->instances
--;
505 if (server
->ioctl_fd
) release_object( server
->ioctl_fd
);
506 list_remove( &server
->entry
);
507 release_object( server
->pipe
);
510 static void pipe_client_destroy( struct object
*obj
)
512 struct pipe_client
*client
= (struct pipe_client
*)obj
;
513 struct pipe_server
*server
= client
->server
;
515 assert( obj
->ops
== &pipe_client_ops
);
517 pipe_end_disconnect( &client
->pipe_end
, STATUS_PIPE_BROKEN
);
521 notify_empty( server
);
523 switch(server
->state
)
525 case ps_connected_server
:
526 /* Don't destroy the server's fd here as we can't
527 do a successful flush without it. */
528 set_server_state( server
, ps_wait_disconnect
);
532 case ps_wait_disconnect
:
533 case ps_wait_connect
:
536 assert( server
->client
);
537 server
->client
= NULL
;
538 client
->server
= NULL
;
541 pipe_end_destroy( &client
->pipe_end
);
542 if (client
->pipe_end
.fd
) release_object( client
->pipe_end
.fd
);
545 static void named_pipe_device_dump( struct object
*obj
, int verbose
)
547 fputs( "Named pipe device\n", stderr
);
550 static struct object_type
*named_pipe_device_get_type( struct object
*obj
)
552 static const WCHAR name
[] = {'D','e','v','i','c','e'};
553 static const struct unicode_str str
= { name
, sizeof(name
) };
554 return get_object_type( &str
);
557 static struct fd
*named_pipe_device_get_fd( struct object
*obj
)
559 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
560 return (struct fd
*)grab_object( device
->fd
);
563 static struct object
*named_pipe_device_lookup_name( struct object
*obj
, struct unicode_str
*name
,
566 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
567 struct object
*found
;
569 assert( obj
->ops
== &named_pipe_device_ops
);
570 assert( device
->pipes
);
572 if (!name
) return NULL
; /* open the device itself */
574 if ((found
= find_object( device
->pipes
, name
, attr
| OBJ_CASE_INSENSITIVE
)))
580 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
581 unsigned int sharing
, unsigned int options
)
583 return grab_object( obj
);
586 static void named_pipe_device_destroy( struct object
*obj
)
588 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
589 assert( obj
->ops
== &named_pipe_device_ops
);
590 if (device
->fd
) release_object( device
->fd
);
591 free( device
->pipes
);
594 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
)
596 return FD_TYPE_DEVICE
;
599 struct object
*create_named_pipe_device( struct object
*root
, const struct unicode_str
*name
)
601 struct named_pipe_device
*dev
;
603 if ((dev
= create_named_object( root
, &named_pipe_device_ops
, name
, 0, NULL
)) &&
604 get_error() != STATUS_OBJECT_NAME_EXISTS
)
607 if (!(dev
->fd
= alloc_pseudo_fd( &named_pipe_device_fd_ops
, &dev
->obj
, 0 )) ||
608 !(dev
->pipes
= create_namespace( 7 )))
610 release_object( dev
);
617 static int pipe_data_remaining( struct pipe_server
*server
)
622 assert( server
->client
);
624 if (use_server_io( &server
->pipe_end
))
625 return !list_empty( &server
->client
->pipe_end
.message_queue
);
627 fd
= get_unix_fd( server
->client
->pipe_end
.fd
);
634 if (0 > poll( &pfd
, 1, 0 ))
637 return pfd
.revents
&POLLIN
;
640 static void check_flushed( void *arg
)
642 struct pipe_server
*server
= (struct pipe_server
*) arg
;
644 if (pipe_data_remaining( server
))
646 server
->flush_poll
= add_timeout_user( -TICKS_PER_SEC
/ 10, check_flushed
, server
);
650 server
->flush_poll
= NULL
;
651 fd_async_wake_up( server
->pipe_end
.fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
655 static int pipe_end_flush( struct pipe_end
*pipe_end
, struct async
*async
)
657 if (use_server_io( pipe_end
) && (!pipe_end
->connection
|| list_empty( &pipe_end
->connection
->message_queue
)))
660 fd_queue_async( pipe_end
->fd
, async
, ASYNC_TYPE_WAIT
);
661 set_error( STATUS_PENDING
);
665 static int pipe_server_flush( struct fd
*fd
, struct async
*async
)
667 struct pipe_server
*server
= get_fd_user( fd
);
670 if (!server
|| server
->state
!= ps_connected_server
) return 1;
672 if (!pipe_data_remaining( server
)) return 1;
674 handle
= pipe_end_flush( &server
->pipe_end
, async
);
676 /* there's no unix way to be alerted when a pipe becomes empty, so resort to polling */
677 if (handle
&& !use_server_io( &server
->pipe_end
) && !server
->flush_poll
)
678 server
->flush_poll
= add_timeout_user( -TICKS_PER_SEC
/ 10, check_flushed
, server
);
682 static int pipe_client_flush( struct fd
*fd
, struct async
*async
)
684 struct pipe_end
*pipe_end
= get_fd_user( fd
);
685 /* FIXME: Support byte mode. */
686 return use_server_io( pipe_end
) ? pipe_end_flush( pipe_end
, async
) : 1;
689 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
)
693 case FileFsDeviceInformation
:
695 static const FILE_FS_DEVICE_INFORMATION device_info
=
697 FILE_DEVICE_NAMED_PIPE
,
698 FILE_DEVICE_ALLOW_APPCONTAINER_TRAVERSAL
700 if (get_reply_max_size() >= sizeof(device_info
))
701 set_reply_data( &device_info
, sizeof(device_info
) );
703 set_error( STATUS_BUFFER_TOO_SMALL
);
707 set_error( STATUS_NOT_IMPLEMENTED
);
711 static void message_queue_read( struct pipe_end
*pipe_end
, struct iosb
*iosb
)
713 struct pipe_message
*message
;
715 if (pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)
717 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
718 iosb
->out_size
= min( iosb
->out_size
, message
->iosb
->in_size
- message
->read_pos
);
719 iosb
->status
= message
->read_pos
+ iosb
->out_size
< message
->iosb
->in_size
720 ? STATUS_BUFFER_OVERFLOW
: STATUS_SUCCESS
;
724 data_size_t avail
= 0;
725 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
727 avail
+= message
->iosb
->in_size
- message
->read_pos
;
728 if (avail
>= iosb
->out_size
) break;
730 iosb
->out_size
= min( iosb
->out_size
, avail
);
731 iosb
->status
= STATUS_SUCCESS
;
734 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
735 if (!message
->read_pos
&& message
->iosb
->in_size
== iosb
->out_size
) /* fast path */
737 iosb
->out_data
= message
->iosb
->in_data
;
738 message
->iosb
->in_data
= NULL
;
739 wake_message( message
);
740 free_message( message
);
744 data_size_t write_pos
= 0, writing
;
747 if (iosb
->out_size
&& !(buf
= iosb
->out_data
= malloc( iosb
->out_size
)))
750 iosb
->status
= STATUS_NO_MEMORY
;
756 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
757 writing
= min( iosb
->out_size
- write_pos
, message
->iosb
->in_size
- message
->read_pos
);
758 if (writing
) memcpy( buf
+ write_pos
, (const char *)message
->iosb
->in_data
+ message
->read_pos
, writing
);
759 write_pos
+= writing
;
760 message
->read_pos
+= writing
;
761 if (message
->read_pos
== message
->iosb
->in_size
)
763 wake_message(message
);
764 free_message(message
);
766 } while (write_pos
< iosb
->out_size
);
768 iosb
->result
= iosb
->out_size
;
771 /* We call async_terminate in our reselect implementation, which causes recursive reselect.
772 * We're not interested in such reselect calls, so we ignore them. */
773 static int ignore_reselect
;
775 static void reselect_write_queue( struct pipe_end
*pipe_end
);
777 static void reselect_read_queue( struct pipe_end
*pipe_end
)
784 while (!list_empty( &pipe_end
->message_queue
) && (async
= find_pending_async( &pipe_end
->read_q
)))
786 iosb
= async_get_iosb( async
);
787 message_queue_read( pipe_end
, iosb
);
788 async_terminate( async
, iosb
->result
? STATUS_ALERTED
: iosb
->status
);
789 release_object( async
);
790 release_object( iosb
);
795 if (pipe_end
->connection
)
797 if (list_empty( &pipe_end
->message_queue
))
798 fd_async_wake_up( pipe_end
->connection
->fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
800 reselect_write_queue( pipe_end
->connection
);
804 static void reselect_write_queue( struct pipe_end
*pipe_end
)
806 struct pipe_message
*message
, *next
;
807 struct pipe_end
*reader
= pipe_end
->connection
;
808 data_size_t avail
= 0;
814 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &reader
->message_queue
, struct pipe_message
, entry
)
816 if (message
->async
&& message
->iosb
->status
!= STATUS_PENDING
)
818 release_object( message
->async
);
819 message
->async
= NULL
;
820 free_message( message
);
824 avail
+= message
->iosb
->in_size
- message
->read_pos
;
825 if (message
->iosb
->status
== STATUS_PENDING
&& (avail
<= reader
->buffer_size
|| !message
->iosb
->in_size
))
826 wake_message( message
);
831 reselect_read_queue( reader
);
834 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
836 struct pipe_end
*pipe_end
= get_fd_user( fd
);
838 if (!use_server_io( pipe_end
)) return no_fd_read( fd
, async
, pos
);
840 if (!pipe_end
->connection
&& list_empty( &pipe_end
->message_queue
))
842 set_error( STATUS_PIPE_BROKEN
);
846 queue_async( &pipe_end
->read_q
, async
);
847 reselect_read_queue( pipe_end
);
848 set_error( STATUS_PENDING
);
852 static int pipe_end_write( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
854 struct pipe_end
*write_end
= get_fd_user( fd
);
855 struct pipe_end
*read_end
= write_end
->connection
;
856 struct pipe_message
*message
;
858 if (!use_server_io( write_end
)) return no_fd_write( fd
, async
, pos
);
862 set_error( STATUS_PIPE_DISCONNECTED
);
866 if (!(message
= mem_alloc( sizeof(*message
) ))) return 0;
867 message
->async
= (struct async
*)grab_object( async
);
868 message
->iosb
= async_get_iosb( async
);
869 message
->read_pos
= 0;
870 list_add_tail( &read_end
->message_queue
, &message
->entry
);
872 queue_async( &write_end
->write_q
, async
);
873 reselect_write_queue( write_end
);
874 set_error( STATUS_PENDING
);
878 static void pipe_end_queue_async( struct fd
*fd
, struct async
*async
, int type
, int count
)
880 struct pipe_end
*pipe_end
= get_fd_user( fd
);
881 if (use_server_io( pipe_end
)) no_fd_queue_async( fd
, async
, type
, count
);
882 else default_fd_queue_async( fd
, async
, type
, count
);
885 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
)
887 struct pipe_end
*pipe_end
= get_fd_user( fd
);
889 if (ignore_reselect
) return;
891 if (!use_server_io( pipe_end
))
892 default_fd_reselect_async( fd
, queue
);
893 else if (&pipe_end
->write_q
== queue
)
894 reselect_write_queue( pipe_end
);
895 else if (&pipe_end
->read_q
== queue
)
896 reselect_read_queue( pipe_end
);
899 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
)
904 static int pipe_end_peek( struct pipe_end
*pipe_end
)
906 unsigned reply_size
= get_reply_max_size();
907 FILE_PIPE_PEEK_BUFFER
*buffer
;
908 struct pipe_message
*message
;
909 data_size_t avail
= 0;
910 data_size_t message_length
= 0;
912 if (!use_server_io( pipe_end
))
914 set_error( STATUS_NOT_SUPPORTED
);
918 if (reply_size
< offsetof( FILE_PIPE_PEEK_BUFFER
, Data
))
920 set_error( STATUS_INFO_LENGTH_MISMATCH
);
923 reply_size
-= offsetof( FILE_PIPE_PEEK_BUFFER
, Data
);
925 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
926 avail
+= message
->iosb
->in_size
- message
->read_pos
;
930 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
931 message_length
= message
->iosb
->in_size
- message
->read_pos
;
932 reply_size
= min( reply_size
, message_length
);
936 if (!(buffer
= set_reply_data_size( offsetof( FILE_PIPE_PEEK_BUFFER
, Data
[reply_size
] )))) return 0;
937 buffer
->NamedPipeState
= 0; /* FIXME */
938 buffer
->ReadDataAvailable
= avail
;
939 buffer
->NumberOfMessages
= 0; /* FIXME */
940 buffer
->MessageLength
= message_length
;
941 if (reply_size
) memcpy( buffer
->Data
, (const char *)message
->iosb
->in_data
+ message
->read_pos
, reply_size
);
945 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
947 struct pipe_server
*server
= get_fd_user( fd
);
951 case FSCTL_PIPE_LISTEN
:
952 switch(server
->state
)
955 case ps_wait_connect
:
956 fd_queue_async( server
->ioctl_fd
, async
, ASYNC_TYPE_WAIT
);
957 set_server_state( server
, ps_wait_open
);
958 async_wake_up( &server
->pipe
->waiters
, STATUS_SUCCESS
);
959 set_error( STATUS_PENDING
);
961 case ps_connected_server
:
962 set_error( STATUS_PIPE_CONNECTED
);
964 case ps_wait_disconnect
:
965 set_error( STATUS_NO_DATA_DETECTED
);
968 set_error( STATUS_INVALID_HANDLE
);
973 case FSCTL_PIPE_DISCONNECT
:
974 switch(server
->state
)
976 case ps_connected_server
:
977 assert( server
->client
);
978 assert( server
->client
->pipe_end
.fd
);
980 notify_empty( server
);
982 /* dump the client and server fds - client loses all waiting data */
983 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_DISCONNECTED
);
984 do_disconnect( server
);
985 server
->client
->server
= NULL
;
986 server
->client
= NULL
;
987 set_server_state( server
, ps_wait_connect
);
989 case ps_wait_disconnect
:
990 assert( !server
->client
);
991 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_DISCONNECTED
);
992 do_disconnect( server
);
993 set_server_state( server
, ps_wait_connect
);
997 set_error( STATUS_PIPE_LISTENING
);
999 case ps_wait_connect
:
1000 set_error( STATUS_PIPE_DISCONNECTED
);
1005 case FSCTL_PIPE_PEEK
:
1006 return pipe_end_peek( &server
->pipe_end
);
1009 return default_fd_ioctl( fd
, code
, async
);
1013 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1015 struct pipe_client
*client
= get_fd_user( fd
);
1019 case FSCTL_PIPE_PEEK
:
1020 return pipe_end_peek( &client
->pipe_end
);
1023 return default_fd_ioctl( fd
, code
, async
);
1027 static struct pipe_server
*get_pipe_server_obj( struct process
*process
,
1028 obj_handle_t handle
, unsigned int access
)
1031 obj
= get_handle_obj( process
, handle
, access
, &pipe_server_ops
);
1032 return (struct pipe_server
*) obj
;
1035 static void init_pipe_end( struct pipe_end
*pipe_end
, unsigned int pipe_flags
, data_size_t buffer_size
)
1037 pipe_end
->fd
= NULL
;
1038 pipe_end
->flags
= pipe_flags
;
1039 pipe_end
->connection
= NULL
;
1040 pipe_end
->buffer_size
= buffer_size
;
1041 init_async_queue( &pipe_end
->read_q
);
1042 init_async_queue( &pipe_end
->write_q
);
1043 list_init( &pipe_end
->message_queue
);
1046 static struct pipe_server
*create_pipe_server( struct named_pipe
*pipe
, unsigned int options
,
1047 unsigned int pipe_flags
)
1049 struct pipe_server
*server
;
1051 server
= alloc_object( &pipe_server_ops
);
1055 server
->pipe
= pipe
;
1056 server
->client
= NULL
;
1057 server
->flush_poll
= NULL
;
1058 server
->options
= options
;
1059 init_pipe_end( &server
->pipe_end
, pipe_flags
, pipe
->insize
);
1061 list_add_head( &pipe
->servers
, &server
->entry
);
1062 grab_object( pipe
);
1063 if (!(server
->ioctl_fd
= alloc_pseudo_fd( &pipe_server_fd_ops
, &server
->pipe_end
.obj
, options
)))
1065 release_object( server
);
1068 set_fd_signaled( server
->ioctl_fd
, 1 );
1069 set_server_state( server
, ps_idle_server
);
1073 static struct pipe_client
*create_pipe_client( unsigned int flags
, unsigned int pipe_flags
, data_size_t buffer_size
)
1075 struct pipe_client
*client
;
1077 client
= alloc_object( &pipe_client_ops
);
1081 client
->server
= NULL
;
1082 client
->flags
= flags
;
1083 init_pipe_end( &client
->pipe_end
, pipe_flags
, buffer_size
);
1088 static struct pipe_server
*find_available_server( struct named_pipe
*pipe
)
1090 struct pipe_server
*server
;
1092 /* look for pipe servers that are listening */
1093 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1095 if (server
->state
== ps_wait_open
)
1096 return (struct pipe_server
*)grab_object( server
);
1099 /* fall back to pipe servers that are idle */
1100 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1102 if (server
->state
== ps_idle_server
)
1103 return (struct pipe_server
*)grab_object( server
);
1109 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
)
1111 struct named_pipe_device
*dev
= (struct named_pipe_device
*)parent
;
1113 if (parent
->ops
!= &named_pipe_device_ops
)
1115 set_error( STATUS_OBJECT_NAME_INVALID
);
1118 namespace_add( dev
->pipes
, name
);
1119 name
->parent
= grab_object( parent
);
1123 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
1124 unsigned int sharing
, unsigned int options
)
1126 struct named_pipe
*pipe
= (struct named_pipe
*)obj
;
1127 struct pipe_server
*server
;
1128 struct pipe_client
*client
;
1129 unsigned int pipe_sharing
;
1132 if (!(server
= find_available_server( pipe
)))
1134 set_error( STATUS_PIPE_NOT_AVAILABLE
);
1138 pipe_sharing
= server
->pipe
->sharing
;
1139 if (((access
& GENERIC_READ
) && !(pipe_sharing
& FILE_SHARE_READ
)) ||
1140 ((access
& GENERIC_WRITE
) && !(pipe_sharing
& FILE_SHARE_WRITE
)))
1142 set_error( STATUS_ACCESS_DENIED
);
1143 release_object( server
);
1147 if ((client
= create_pipe_client( options
, pipe
->flags
, pipe
->outsize
)))
1149 if (use_server_io( &server
->pipe_end
))
1151 client
->pipe_end
.fd
= alloc_pseudo_fd( &pipe_client_fd_ops
, &client
->pipe_end
.obj
, options
);
1152 if (client
->pipe_end
.fd
)
1154 set_fd_signaled( client
->pipe_end
.fd
, 1 );
1155 server
->pipe_end
.fd
= (struct fd
*)grab_object( server
->ioctl_fd
);
1156 set_no_fd_status( server
->ioctl_fd
, STATUS_BAD_DEVICE_TYPE
);
1160 release_object( client
);
1164 else if (!socketpair( PF_UNIX
, SOCK_STREAM
, 0, fds
))
1166 assert( !server
->pipe_end
.fd
);
1168 fcntl( fds
[0], F_SETFL
, O_NONBLOCK
);
1169 fcntl( fds
[1], F_SETFL
, O_NONBLOCK
);
1173 setsockopt( fds
[0], SOL_SOCKET
, SO_RCVBUF
, &pipe
->insize
, sizeof(pipe
->insize
) );
1174 setsockopt( fds
[1], SOL_SOCKET
, SO_RCVBUF
, &pipe
->insize
, sizeof(pipe
->insize
) );
1178 setsockopt( fds
[0], SOL_SOCKET
, SO_SNDBUF
, &pipe
->outsize
, sizeof(pipe
->outsize
) );
1179 setsockopt( fds
[1], SOL_SOCKET
, SO_SNDBUF
, &pipe
->outsize
, sizeof(pipe
->outsize
) );
1182 client
->pipe_end
.fd
= create_anonymous_fd( &pipe_client_fd_ops
, fds
[1], &client
->pipe_end
.obj
, options
);
1183 server
->pipe_end
.fd
= create_anonymous_fd( &pipe_server_fd_ops
, fds
[0], &server
->pipe_end
.obj
, server
->options
);
1184 if (client
->pipe_end
.fd
&& server
->pipe_end
.fd
)
1186 fd_copy_completion( server
->ioctl_fd
, server
->pipe_end
.fd
);
1190 release_object( client
);
1197 release_object( client
);
1202 allow_fd_caching( client
->pipe_end
.fd
);
1203 allow_fd_caching( server
->pipe_end
.fd
);
1204 if (server
->state
== ps_wait_open
)
1205 fd_async_wake_up( server
->ioctl_fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
1206 set_server_state( server
, ps_connected_server
);
1207 server
->client
= client
;
1208 client
->server
= server
;
1209 server
->pipe_end
.connection
= &client
->pipe_end
;
1210 client
->pipe_end
.connection
= &server
->pipe_end
;
1213 release_object( server
);
1214 return &client
->pipe_end
.obj
;
1217 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1219 struct named_pipe_device
*device
= get_fd_user( fd
);
1223 case FSCTL_PIPE_WAIT
:
1225 const FILE_PIPE_WAIT_FOR_BUFFER
*buffer
= get_req_data();
1226 data_size_t size
= get_req_data_size();
1227 struct named_pipe
*pipe
;
1228 struct pipe_server
*server
;
1229 struct unicode_str name
;
1232 if (size
< sizeof(*buffer
) ||
1233 size
< FIELD_OFFSET(FILE_PIPE_WAIT_FOR_BUFFER
, Name
[buffer
->NameLength
/sizeof(WCHAR
)]))
1235 set_error( STATUS_INVALID_PARAMETER
);
1238 name
.str
= buffer
->Name
;
1239 name
.len
= (buffer
->NameLength
/ sizeof(WCHAR
)) * sizeof(WCHAR
);
1240 if (!(pipe
= open_named_object( &device
->obj
, &named_pipe_ops
, &name
, 0 ))) return 0;
1242 if (!(server
= find_available_server( pipe
)))
1244 queue_async( &pipe
->waiters
, async
);
1245 when
= buffer
->TimeoutSpecified
? buffer
->Timeout
.QuadPart
: pipe
->timeout
;
1246 async_set_timeout( async
, when
, STATUS_IO_TIMEOUT
);
1247 release_object( pipe
);
1248 set_error( STATUS_PENDING
);
1252 release_object( server
);
1253 release_object( pipe
);
1258 return default_fd_ioctl( fd
, code
, async
);
1263 DECL_HANDLER(create_named_pipe
)
1265 struct named_pipe
*pipe
;
1266 struct pipe_server
*server
;
1267 struct unicode_str name
;
1268 struct object
*root
;
1269 const struct security_descriptor
*sd
;
1270 const struct object_attributes
*objattr
= get_req_object_attributes( &sd
, &name
, &root
);
1272 if (!objattr
) return;
1274 if (!req
->sharing
|| (req
->sharing
& ~(FILE_SHARE_READ
| FILE_SHARE_WRITE
)) ||
1275 (!(req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
) && (req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)))
1277 if (root
) release_object( root
);
1278 set_error( STATUS_INVALID_PARAMETER
);
1282 if (!name
.len
) /* pipes need a root directory even without a name */
1284 if (!objattr
->rootdir
)
1286 set_error( STATUS_OBJECT_PATH_SYNTAX_BAD
);
1289 if (!(root
= get_directory_obj( current
->process
, objattr
->rootdir
))) return;
1292 pipe
= create_named_object( root
, &named_pipe_ops
, &name
, objattr
->attributes
| OBJ_OPENIF
, NULL
);
1294 if (root
) release_object( root
);
1297 if (get_error() != STATUS_OBJECT_NAME_EXISTS
)
1299 /* initialize it if it didn't already exist */
1300 pipe
->instances
= 0;
1301 init_async_queue( &pipe
->waiters
);
1302 list_init( &pipe
->servers
);
1303 pipe
->insize
= req
->insize
;
1304 pipe
->outsize
= req
->outsize
;
1305 pipe
->maxinstances
= req
->maxinstances
;
1306 pipe
->timeout
= req
->timeout
;
1307 pipe
->flags
= req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
;
1308 pipe
->sharing
= req
->sharing
;
1312 if (pipe
->maxinstances
<= pipe
->instances
)
1314 set_error( STATUS_INSTANCE_NOT_AVAILABLE
);
1315 release_object( pipe
);
1318 if (pipe
->sharing
!= req
->sharing
)
1320 set_error( STATUS_ACCESS_DENIED
);
1321 release_object( pipe
);
1324 clear_error(); /* clear the name collision */
1327 server
= create_pipe_server( pipe
, req
->options
, req
->flags
);
1330 reply
->handle
= alloc_handle( current
->process
, server
, req
->access
, objattr
->attributes
);
1331 server
->pipe
->instances
++;
1332 if (sd
) default_set_sd( &server
->pipe_end
.obj
, sd
, OWNER_SECURITY_INFORMATION
|
1333 GROUP_SECURITY_INFORMATION
|
1334 DACL_SECURITY_INFORMATION
|
1335 SACL_SECURITY_INFORMATION
);
1336 release_object( server
);
1339 release_object( pipe
);
1342 DECL_HANDLER(get_named_pipe_info
)
1344 struct pipe_server
*server
;
1345 struct pipe_client
*client
= NULL
;
1347 server
= get_pipe_server_obj( current
->process
, req
->handle
, FILE_READ_ATTRIBUTES
);
1350 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1354 client
= (struct pipe_client
*)get_handle_obj( current
->process
, req
->handle
,
1355 0, &pipe_client_ops
);
1356 if (!client
) return;
1357 server
= client
->server
;
1360 reply
->flags
= client
? client
->pipe_end
.flags
: server
->pipe_end
.flags
;
1363 reply
->sharing
= server
->pipe
->sharing
;
1364 reply
->maxinstances
= server
->pipe
->maxinstances
;
1365 reply
->instances
= server
->pipe
->instances
;
1366 reply
->insize
= server
->pipe
->insize
;
1367 reply
->outsize
= server
->pipe
->outsize
;
1371 release_object(client
);
1374 reply
->flags
|= NAMED_PIPE_SERVER_END
;
1375 release_object(server
);
1379 DECL_HANDLER(set_named_pipe_info
)
1381 struct pipe_server
*server
;
1382 struct pipe_client
*client
= NULL
;
1384 server
= get_pipe_server_obj( current
->process
, req
->handle
, FILE_WRITE_ATTRIBUTES
);
1387 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1391 client
= (struct pipe_client
*)get_handle_obj( current
->process
, req
->handle
,
1392 0, &pipe_client_ops
);
1393 if (!client
) return;
1394 if (!(server
= client
->server
))
1396 release_object( client
);
1401 if ((req
->flags
& ~(NAMED_PIPE_MESSAGE_STREAM_READ
| NAMED_PIPE_NONBLOCKING_MODE
)) ||
1402 ((req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
) && !(server
->pipe
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
)))
1404 set_error( STATUS_INVALID_PARAMETER
);
1408 client
->pipe_end
.flags
= server
->pipe
->flags
| req
->flags
;
1412 server
->pipe_end
.flags
= server
->pipe
->flags
| req
->flags
;
1416 release_object(client
);
1418 release_object(server
);