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
49 struct list entry
; /* entry in message queue */
50 data_size_t read_pos
; /* already read bytes */
51 struct iosb
*iosb
; /* message iosb */
52 struct async
*async
; /* async of pending write */
57 struct object obj
; /* object header */
58 struct fd
*fd
; /* pipe file descriptor */
59 unsigned int flags
; /* pipe flags */
60 unsigned int state
; /* pipe state */
61 struct named_pipe
*pipe
;
62 struct pipe_end
*connection
; /* the other end of the pipe */
63 process_id_t client_pid
; /* process that created the client */
64 process_id_t server_pid
; /* process that created the server */
65 data_size_t buffer_size
;/* size of buffered data that doesn't block caller */
66 struct list message_queue
;
67 struct async_queue read_q
; /* read queue */
68 struct async_queue write_q
; /* write queue */
73 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
74 struct list entry
; /* entry in named pipe servers list */
75 unsigned int options
; /* pipe options */
76 struct async_queue listen_q
; /* listen queue */
81 struct pipe_end pipe_end
; /* common header for pipe_client and pipe_server */
82 unsigned int flags
; /* file flags */
87 struct object obj
; /* object header */
90 unsigned int maxinstances
;
93 unsigned int instances
;
95 struct list servers
; /* list of servers using this pipe */
96 struct async_queue waiters
; /* list of clients waiting to connect */
99 struct named_pipe_device
101 struct object obj
; /* object header */
102 struct fd
*fd
; /* pseudo-fd for ioctls */
103 struct namespace *pipes
; /* named pipe namespace */
106 static void named_pipe_dump( struct object
*obj
, int verbose
);
107 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
);
108 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
);
109 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
110 unsigned int sharing
, unsigned int options
);
111 static void named_pipe_destroy( struct object
*obj
);
113 static const struct object_ops named_pipe_ops
=
115 sizeof(struct named_pipe
), /* size */
116 named_pipe_dump
, /* dump */
117 no_get_type
, /* get_type */
118 no_add_queue
, /* add_queue */
119 NULL
, /* remove_queue */
121 NULL
, /* satisfied */
122 no_signal
, /* signal */
123 no_get_fd
, /* get_fd */
124 named_pipe_map_access
, /* map_access */
125 default_get_sd
, /* get_sd */
126 default_set_sd
, /* set_sd */
127 no_lookup_name
, /* lookup_name */
128 named_pipe_link_name
, /* link_name */
129 default_unlink_name
, /* unlink_name */
130 named_pipe_open_file
, /* open_file */
131 no_close_handle
, /* close_handle */
132 named_pipe_destroy
/* destroy */
135 /* common server and client pipe end functions */
136 static void pipe_end_destroy( struct object
*obj
);
137 static struct object_type
*pipe_end_get_type( struct object
*obj
);
138 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
);
139 static struct fd
*pipe_end_get_fd( struct object
*obj
);
140 static struct security_descriptor
*pipe_end_get_sd( struct object
*obj
);
141 static int pipe_end_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
142 unsigned int set_info
);
143 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
);
144 static int pipe_end_write( struct fd
*fd
, struct async
*async_data
, file_pos_t pos
);
145 static int pipe_end_flush( struct fd
*fd
, struct async
*async
);
146 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
);
147 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
);
148 static void pipe_end_get_file_info( struct fd
*fd
, unsigned int info_class
);
150 /* server end functions */
151 static void pipe_server_dump( struct object
*obj
, int verbose
);
152 static void pipe_server_destroy( struct object
*obj
);
153 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
155 static const struct object_ops pipe_server_ops
=
157 sizeof(struct pipe_server
), /* size */
158 pipe_server_dump
, /* dump */
159 pipe_end_get_type
, /* get_type */
160 add_queue
, /* add_queue */
161 remove_queue
, /* remove_queue */
162 default_fd_signaled
, /* signaled */
163 no_satisfied
, /* satisfied */
164 no_signal
, /* signal */
165 pipe_end_get_fd
, /* get_fd */
166 default_fd_map_access
, /* map_access */
167 pipe_end_get_sd
, /* get_sd */
168 pipe_end_set_sd
, /* set_sd */
169 no_lookup_name
, /* lookup_name */
170 no_link_name
, /* link_name */
171 NULL
, /* unlink_name */
172 no_open_file
, /* open_file */
173 fd_close_handle
, /* close_handle */
174 pipe_server_destroy
/* destroy */
177 static const struct fd_ops pipe_server_fd_ops
=
179 default_fd_get_poll_events
, /* get_poll_events */
180 default_poll_event
, /* poll_event */
181 pipe_end_get_fd_type
, /* get_fd_type */
182 pipe_end_read
, /* read */
183 pipe_end_write
, /* write */
184 pipe_end_flush
, /* flush */
185 pipe_end_get_file_info
, /* get_file_info */
186 pipe_end_get_volume_info
, /* get_volume_info */
187 pipe_server_ioctl
, /* ioctl */
188 no_fd_queue_async
, /* queue_async */
189 pipe_end_reselect_async
/* reselect_async */
192 /* client end functions */
193 static void pipe_client_dump( struct object
*obj
, int verbose
);
194 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
196 static const struct object_ops pipe_client_ops
=
198 sizeof(struct pipe_client
), /* size */
199 pipe_client_dump
, /* dump */
200 pipe_end_get_type
, /* get_type */
201 add_queue
, /* add_queue */
202 remove_queue
, /* remove_queue */
203 default_fd_signaled
, /* signaled */
204 no_satisfied
, /* satisfied */
205 no_signal
, /* signal */
206 pipe_end_get_fd
, /* get_fd */
207 default_fd_map_access
, /* map_access */
208 pipe_end_get_sd
, /* get_sd */
209 pipe_end_set_sd
, /* set_sd */
210 no_lookup_name
, /* lookup_name */
211 no_link_name
, /* link_name */
212 NULL
, /* unlink_name */
213 no_open_file
, /* open_file */
214 fd_close_handle
, /* close_handle */
215 pipe_end_destroy
/* destroy */
218 static const struct fd_ops pipe_client_fd_ops
=
220 default_fd_get_poll_events
, /* get_poll_events */
221 default_poll_event
, /* poll_event */
222 pipe_end_get_fd_type
, /* get_fd_type */
223 pipe_end_read
, /* read */
224 pipe_end_write
, /* write */
225 pipe_end_flush
, /* flush */
226 pipe_end_get_file_info
, /* get_file_info */
227 pipe_end_get_volume_info
, /* get_volume_info */
228 pipe_client_ioctl
, /* ioctl */
229 no_fd_queue_async
, /* queue_async */
230 pipe_end_reselect_async
/* reselect_async */
233 static void named_pipe_device_dump( struct object
*obj
, int verbose
);
234 static struct object_type
*named_pipe_device_get_type( struct object
*obj
);
235 static struct fd
*named_pipe_device_get_fd( struct object
*obj
);
236 static struct object
*named_pipe_device_lookup_name( struct object
*obj
,
237 struct unicode_str
*name
, unsigned int attr
);
238 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
239 unsigned int sharing
, unsigned int options
);
240 static void named_pipe_device_destroy( struct object
*obj
);
241 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
);
242 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
);
244 static const struct object_ops named_pipe_device_ops
=
246 sizeof(struct named_pipe_device
), /* size */
247 named_pipe_device_dump
, /* dump */
248 named_pipe_device_get_type
, /* get_type */
249 no_add_queue
, /* add_queue */
250 NULL
, /* remove_queue */
252 no_satisfied
, /* satisfied */
253 no_signal
, /* signal */
254 named_pipe_device_get_fd
, /* get_fd */
255 no_map_access
, /* map_access */
256 default_get_sd
, /* get_sd */
257 default_set_sd
, /* set_sd */
258 named_pipe_device_lookup_name
, /* lookup_name */
259 directory_link_name
, /* link_name */
260 default_unlink_name
, /* unlink_name */
261 named_pipe_device_open_file
, /* open_file */
262 fd_close_handle
, /* close_handle */
263 named_pipe_device_destroy
/* destroy */
266 static const struct fd_ops named_pipe_device_fd_ops
=
268 default_fd_get_poll_events
, /* get_poll_events */
269 default_poll_event
, /* poll_event */
270 named_pipe_device_get_fd_type
, /* get_fd_type */
271 no_fd_read
, /* read */
272 no_fd_write
, /* write */
273 no_fd_flush
, /* flush */
274 no_fd_get_file_info
, /* get_file_info */
275 no_fd_get_volume_info
, /* get_volume_info */
276 named_pipe_device_ioctl
, /* ioctl */
277 default_fd_queue_async
, /* queue_async */
278 default_fd_reselect_async
/* reselect_async */
281 static void named_pipe_dump( struct object
*obj
, int verbose
)
283 fputs( "Named pipe\n", stderr
);
286 static unsigned int named_pipe_map_access( struct object
*obj
, unsigned int access
)
288 if (access
& GENERIC_READ
) access
|= STANDARD_RIGHTS_READ
;
289 if (access
& GENERIC_WRITE
) access
|= STANDARD_RIGHTS_WRITE
| FILE_CREATE_PIPE_INSTANCE
;
290 if (access
& GENERIC_EXECUTE
) access
|= STANDARD_RIGHTS_EXECUTE
;
291 if (access
& GENERIC_ALL
) access
|= STANDARD_RIGHTS_ALL
;
292 return access
& ~(GENERIC_READ
| GENERIC_WRITE
| GENERIC_EXECUTE
| GENERIC_ALL
);
295 static void pipe_server_dump( struct object
*obj
, int verbose
)
297 struct pipe_server
*server
= (struct pipe_server
*) obj
;
298 assert( obj
->ops
== &pipe_server_ops
);
299 fprintf( stderr
, "Named pipe server pipe=%p state=%d\n", server
->pipe_end
.pipe
,
300 server
->pipe_end
.state
);
303 static void pipe_client_dump( struct object
*obj
, int verbose
)
305 struct pipe_client
*client
= (struct pipe_client
*) obj
;
306 assert( obj
->ops
== &pipe_client_ops
);
307 fprintf( stderr
, "Named pipe client server=%p\n", client
->pipe_end
.connection
);
310 static void named_pipe_destroy( struct object
*obj
)
312 struct named_pipe
*pipe
= (struct named_pipe
*) obj
;
314 assert( list_empty( &pipe
->servers
) );
315 assert( !pipe
->instances
);
316 free_async_queue( &pipe
->waiters
);
319 static struct object_type
*pipe_end_get_type( struct object
*obj
)
321 static const WCHAR name
[] = {'F','i','l','e'};
322 static const struct unicode_str str
= { name
, sizeof(name
) };
323 return get_object_type( &str
);
326 static struct fd
*pipe_end_get_fd( struct object
*obj
)
328 struct pipe_end
*pipe_end
= (struct pipe_end
*) obj
;
329 return (struct fd
*) grab_object( pipe_end
->fd
);
332 static struct pipe_message
*queue_message( struct pipe_end
*pipe_end
, struct iosb
*iosb
)
334 struct pipe_message
*message
;
336 if (!(message
= mem_alloc( sizeof(*message
) ))) return NULL
;
337 message
->iosb
= (struct iosb
*)grab_object( iosb
);
338 message
->async
= NULL
;
339 message
->read_pos
= 0;
340 list_add_tail( &pipe_end
->message_queue
, &message
->entry
);
344 static void wake_message( struct pipe_message
*message
)
346 struct async
*async
= message
->async
;
348 message
->async
= NULL
;
351 message
->iosb
->status
= STATUS_SUCCESS
;
352 message
->iosb
->result
= message
->iosb
->in_size
;
353 async_terminate( async
, message
->iosb
->result
? STATUS_ALERTED
: STATUS_SUCCESS
);
354 release_object( async
);
357 static void free_message( struct pipe_message
*message
)
359 list_remove( &message
->entry
);
360 if (message
->iosb
) release_object( message
->iosb
);
364 static void pipe_end_disconnect( struct pipe_end
*pipe_end
, unsigned int status
)
366 struct pipe_end
*connection
= pipe_end
->connection
;
367 struct pipe_message
*message
, *next
;
370 pipe_end
->connection
= NULL
;
372 pipe_end
->state
= status
== STATUS_PIPE_DISCONNECTED
373 ? FILE_PIPE_DISCONNECTED_STATE
: FILE_PIPE_CLOSING_STATE
;
374 fd_async_wake_up( pipe_end
->fd
, ASYNC_TYPE_WAIT
, status
);
375 async_wake_up( &pipe_end
->read_q
, status
);
376 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
378 async
= message
->async
;
379 if (async
|| status
== STATUS_PIPE_DISCONNECTED
) free_message( message
);
380 if (!async
) continue;
381 async_terminate( async
, status
);
382 release_object( async
);
384 if (status
== STATUS_PIPE_DISCONNECTED
) set_fd_signaled( pipe_end
->fd
, 0 );
388 connection
->connection
= NULL
;
389 pipe_end_disconnect( connection
, status
);
393 static void pipe_end_destroy( struct object
*obj
)
395 struct pipe_end
*pipe_end
= (struct pipe_end
*)obj
;
396 struct pipe_message
*message
;
398 pipe_end_disconnect( pipe_end
, STATUS_PIPE_BROKEN
);
400 while (!list_empty( &pipe_end
->message_queue
))
402 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
403 assert( !message
->async
);
404 free_message( message
);
407 free_async_queue( &pipe_end
->read_q
);
408 free_async_queue( &pipe_end
->write_q
);
409 if (pipe_end
->fd
) release_object( pipe_end
->fd
);
410 if (pipe_end
->pipe
) release_object( pipe_end
->pipe
);
413 static void pipe_server_destroy( struct object
*obj
)
415 struct pipe_server
*server
= (struct pipe_server
*)obj
;
416 struct named_pipe
*pipe
= server
->pipe_end
.pipe
;
418 assert( obj
->ops
== &pipe_server_ops
);
420 assert( pipe
->instances
);
421 if (!--pipe
->instances
) unlink_named_object( &pipe
->obj
);
422 list_remove( &server
->entry
);
424 free_async_queue( &server
->listen_q
);
425 pipe_end_destroy( obj
);
428 static void named_pipe_device_dump( struct object
*obj
, int verbose
)
430 fputs( "Named pipe device\n", stderr
);
433 static struct object_type
*named_pipe_device_get_type( struct object
*obj
)
435 static const WCHAR name
[] = {'D','e','v','i','c','e'};
436 static const struct unicode_str str
= { name
, sizeof(name
) };
437 return get_object_type( &str
);
440 static struct fd
*named_pipe_device_get_fd( struct object
*obj
)
442 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
443 return (struct fd
*)grab_object( device
->fd
);
446 static struct object
*named_pipe_device_lookup_name( struct object
*obj
, struct unicode_str
*name
,
449 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
450 struct object
*found
;
452 assert( obj
->ops
== &named_pipe_device_ops
);
453 assert( device
->pipes
);
455 if (!name
) return NULL
; /* open the device itself */
457 if ((found
= find_object( device
->pipes
, name
, attr
| OBJ_CASE_INSENSITIVE
)))
463 static struct object
*named_pipe_device_open_file( struct object
*obj
, unsigned int access
,
464 unsigned int sharing
, unsigned int options
)
466 return grab_object( obj
);
469 static void named_pipe_device_destroy( struct object
*obj
)
471 struct named_pipe_device
*device
= (struct named_pipe_device
*)obj
;
472 assert( obj
->ops
== &named_pipe_device_ops
);
473 if (device
->fd
) release_object( device
->fd
);
474 free( device
->pipes
);
477 static enum server_fd_type
named_pipe_device_get_fd_type( struct fd
*fd
)
479 return FD_TYPE_DEVICE
;
482 struct object
*create_named_pipe_device( struct object
*root
, const struct unicode_str
*name
)
484 struct named_pipe_device
*dev
;
486 if ((dev
= create_named_object( root
, &named_pipe_device_ops
, name
, 0, NULL
)) &&
487 get_error() != STATUS_OBJECT_NAME_EXISTS
)
490 if (!(dev
->fd
= alloc_pseudo_fd( &named_pipe_device_fd_ops
, &dev
->obj
, 0 )) ||
491 !(dev
->pipes
= create_namespace( 7 )))
493 release_object( dev
);
500 static int pipe_end_flush( struct fd
*fd
, struct async
*async
)
502 struct pipe_end
*pipe_end
= get_fd_user( fd
);
504 if (pipe_end
->connection
&& !list_empty( &pipe_end
->connection
->message_queue
))
506 fd_queue_async( pipe_end
->fd
, async
, ASYNC_TYPE_WAIT
);
507 set_error( STATUS_PENDING
);
512 static void pipe_end_get_file_info( struct fd
*fd
, unsigned int info_class
)
514 struct pipe_end
*pipe_end
= get_fd_user( fd
);
515 struct named_pipe
*pipe
= pipe_end
->pipe
;
519 set_error( STATUS_PIPE_DISCONNECTED
);
525 case FileNameInformation
:
527 FILE_NAME_INFORMATION
*name_info
;
528 data_size_t name_len
, reply_size
;
531 if (get_reply_max_size() < sizeof(*name_info
))
533 set_error( STATUS_INFO_LENGTH_MISMATCH
);
537 name
= get_object_name( &pipe
->obj
, &name_len
);
538 /* FIXME: We should be able to return on unlinked pipe */
541 set_error( STATUS_PIPE_DISCONNECTED
);
544 reply_size
= offsetof( FILE_NAME_INFORMATION
, FileName
[name_len
/sizeof(WCHAR
) + 1] );
545 if (reply_size
> get_reply_max_size())
547 reply_size
= get_reply_max_size();
548 set_error( STATUS_BUFFER_OVERFLOW
);
551 if (!(name_info
= set_reply_data_size( reply_size
))) return;
552 name_info
->FileNameLength
= name_len
+ sizeof(WCHAR
);
553 name_info
->FileName
[0] = '\\';
554 reply_size
-= offsetof( FILE_NAME_INFORMATION
, FileName
[1] );
555 if (reply_size
) memcpy( &name_info
->FileName
[1], name
, reply_size
);
559 no_fd_get_file_info( fd
, info_class
);
563 static struct security_descriptor
*pipe_end_get_sd( struct object
*obj
)
565 struct pipe_end
*pipe_end
= (struct pipe_end
*) obj
;
566 if (pipe_end
->pipe
) return default_get_sd( &pipe_end
->pipe
->obj
);
567 set_error( STATUS_PIPE_DISCONNECTED
);
571 static int pipe_end_set_sd( struct object
*obj
, const struct security_descriptor
*sd
,
572 unsigned int set_info
)
574 struct pipe_end
*pipe_end
= (struct pipe_end
*) obj
;
575 if (pipe_end
->pipe
) return default_set_sd( &pipe_end
->pipe
->obj
, sd
, set_info
);
576 set_error( STATUS_PIPE_DISCONNECTED
);
580 static void pipe_end_get_volume_info( struct fd
*fd
, unsigned int info_class
)
584 case FileFsDeviceInformation
:
586 static const FILE_FS_DEVICE_INFORMATION device_info
=
588 FILE_DEVICE_NAMED_PIPE
,
589 FILE_DEVICE_ALLOW_APPCONTAINER_TRAVERSAL
591 if (get_reply_max_size() >= sizeof(device_info
))
592 set_reply_data( &device_info
, sizeof(device_info
) );
594 set_error( STATUS_BUFFER_TOO_SMALL
);
598 set_error( STATUS_NOT_IMPLEMENTED
);
602 static void message_queue_read( struct pipe_end
*pipe_end
, struct iosb
*iosb
)
604 struct pipe_message
*message
;
606 if (pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)
608 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
609 iosb
->out_size
= min( iosb
->out_size
, message
->iosb
->in_size
- message
->read_pos
);
610 iosb
->status
= message
->read_pos
+ iosb
->out_size
< message
->iosb
->in_size
611 ? STATUS_BUFFER_OVERFLOW
: STATUS_SUCCESS
;
615 data_size_t avail
= 0;
616 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
618 avail
+= message
->iosb
->in_size
- message
->read_pos
;
619 if (avail
>= iosb
->out_size
) break;
621 iosb
->out_size
= min( iosb
->out_size
, avail
);
622 iosb
->status
= STATUS_SUCCESS
;
625 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
626 if (!message
->read_pos
&& message
->iosb
->in_size
== iosb
->out_size
) /* fast path */
628 iosb
->out_data
= message
->iosb
->in_data
;
629 message
->iosb
->in_data
= NULL
;
630 wake_message( message
);
631 free_message( message
);
635 data_size_t write_pos
= 0, writing
;
638 if (iosb
->out_size
&& !(buf
= iosb
->out_data
= malloc( iosb
->out_size
)))
641 iosb
->status
= STATUS_NO_MEMORY
;
647 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
648 writing
= min( iosb
->out_size
- write_pos
, message
->iosb
->in_size
- message
->read_pos
);
649 if (writing
) memcpy( buf
+ write_pos
, (const char *)message
->iosb
->in_data
+ message
->read_pos
, writing
);
650 write_pos
+= writing
;
651 message
->read_pos
+= writing
;
652 if (message
->read_pos
== message
->iosb
->in_size
)
654 wake_message(message
);
655 free_message(message
);
657 } while (write_pos
< iosb
->out_size
);
659 iosb
->result
= iosb
->out_size
;
662 /* We call async_terminate in our reselect implementation, which causes recursive reselect.
663 * We're not interested in such reselect calls, so we ignore them. */
664 static int ignore_reselect
;
666 static void reselect_write_queue( struct pipe_end
*pipe_end
);
668 static void reselect_read_queue( struct pipe_end
*pipe_end
)
675 while (!list_empty( &pipe_end
->message_queue
) && (async
= find_pending_async( &pipe_end
->read_q
)))
677 iosb
= async_get_iosb( async
);
678 message_queue_read( pipe_end
, iosb
);
679 async_terminate( async
, iosb
->result
? STATUS_ALERTED
: iosb
->status
);
680 release_object( async
);
681 release_object( iosb
);
686 if (pipe_end
->connection
)
688 if (list_empty( &pipe_end
->message_queue
))
689 fd_async_wake_up( pipe_end
->connection
->fd
, ASYNC_TYPE_WAIT
, STATUS_SUCCESS
);
691 reselect_write_queue( pipe_end
->connection
);
695 static void reselect_write_queue( struct pipe_end
*pipe_end
)
697 struct pipe_message
*message
, *next
;
698 struct pipe_end
*reader
= pipe_end
->connection
;
699 data_size_t avail
= 0;
705 LIST_FOR_EACH_ENTRY_SAFE( message
, next
, &reader
->message_queue
, struct pipe_message
, entry
)
707 if (message
->async
&& message
->iosb
->status
!= STATUS_PENDING
)
709 release_object( message
->async
);
710 message
->async
= NULL
;
711 free_message( message
);
715 avail
+= message
->iosb
->in_size
- message
->read_pos
;
716 if (message
->async
&& (avail
<= reader
->buffer_size
|| !message
->iosb
->in_size
))
717 wake_message( message
);
722 reselect_read_queue( reader
);
725 static int pipe_end_read( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
727 struct pipe_end
*pipe_end
= get_fd_user( fd
);
729 switch (pipe_end
->state
)
731 case FILE_PIPE_CONNECTED_STATE
:
733 case FILE_PIPE_DISCONNECTED_STATE
:
734 set_error( STATUS_PIPE_DISCONNECTED
);
736 case FILE_PIPE_LISTENING_STATE
:
737 set_error( STATUS_PIPE_LISTENING
);
739 case FILE_PIPE_CLOSING_STATE
:
740 if (!list_empty( &pipe_end
->message_queue
)) break;
741 set_error( STATUS_PIPE_BROKEN
);
745 queue_async( &pipe_end
->read_q
, async
);
746 reselect_read_queue( pipe_end
);
747 set_error( STATUS_PENDING
);
751 static int pipe_end_write( struct fd
*fd
, struct async
*async
, file_pos_t pos
)
753 struct pipe_end
*pipe_end
= get_fd_user( fd
);
754 struct pipe_message
*message
;
757 switch (pipe_end
->state
)
759 case FILE_PIPE_CONNECTED_STATE
:
761 case FILE_PIPE_DISCONNECTED_STATE
:
762 set_error( STATUS_PIPE_DISCONNECTED
);
764 case FILE_PIPE_LISTENING_STATE
:
765 set_error( STATUS_PIPE_LISTENING
);
767 case FILE_PIPE_CLOSING_STATE
:
768 set_error( STATUS_PIPE_CLOSING
);
772 if (!(pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
) && !get_req_data_size()) return 1;
774 iosb
= async_get_iosb( async
);
775 message
= queue_message( pipe_end
->connection
, iosb
);
776 release_object( iosb
);
777 if (!message
) return 0;
779 message
->async
= (struct async
*)grab_object( async
);
780 queue_async( &pipe_end
->write_q
, async
);
781 reselect_write_queue( pipe_end
);
782 set_error( STATUS_PENDING
);
786 static void pipe_end_reselect_async( struct fd
*fd
, struct async_queue
*queue
)
788 struct pipe_end
*pipe_end
= get_fd_user( fd
);
790 if (ignore_reselect
) return;
792 if (&pipe_end
->write_q
== queue
)
793 reselect_write_queue( pipe_end
);
794 else if (&pipe_end
->read_q
== queue
)
795 reselect_read_queue( pipe_end
);
798 static enum server_fd_type
pipe_end_get_fd_type( struct fd
*fd
)
803 static int pipe_end_peek( struct pipe_end
*pipe_end
)
805 unsigned reply_size
= get_reply_max_size();
806 FILE_PIPE_PEEK_BUFFER
*buffer
;
807 struct pipe_message
*message
;
808 data_size_t avail
= 0;
809 data_size_t message_length
= 0;
811 if (reply_size
< offsetof( FILE_PIPE_PEEK_BUFFER
, Data
))
813 set_error( STATUS_INFO_LENGTH_MISMATCH
);
816 reply_size
-= offsetof( FILE_PIPE_PEEK_BUFFER
, Data
);
818 switch (pipe_end
->state
)
820 case FILE_PIPE_CONNECTED_STATE
:
822 case FILE_PIPE_CLOSING_STATE
:
823 if (!list_empty( &pipe_end
->message_queue
)) break;
824 set_error( STATUS_PIPE_BROKEN
);
827 set_error( STATUS_INVALID_PIPE_STATE
);
831 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
832 avail
+= message
->iosb
->in_size
- message
->read_pos
;
833 reply_size
= min( reply_size
, avail
);
835 if (avail
&& (pipe_end
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
))
837 message
= LIST_ENTRY( list_head(&pipe_end
->message_queue
), struct pipe_message
, entry
);
838 message_length
= message
->iosb
->in_size
- message
->read_pos
;
839 reply_size
= min( reply_size
, message_length
);
842 if (!(buffer
= set_reply_data_size( offsetof( FILE_PIPE_PEEK_BUFFER
, Data
[reply_size
] )))) return 0;
843 buffer
->NamedPipeState
= pipe_end
->state
;
844 buffer
->ReadDataAvailable
= avail
;
845 buffer
->NumberOfMessages
= 0; /* FIXME */
846 buffer
->MessageLength
= message_length
;
850 data_size_t write_pos
= 0, writing
;
851 LIST_FOR_EACH_ENTRY( message
, &pipe_end
->message_queue
, struct pipe_message
, entry
)
853 writing
= min( reply_size
- write_pos
, message
->iosb
->in_size
- message
->read_pos
);
854 memcpy( buffer
->Data
+ write_pos
, (const char *)message
->iosb
->in_data
+ message
->read_pos
,
856 write_pos
+= writing
;
857 if (write_pos
== reply_size
) break;
863 static int pipe_end_transceive( struct pipe_end
*pipe_end
, struct async
*async
)
865 struct pipe_message
*message
;
868 if (!pipe_end
->connection
)
870 set_error( STATUS_INVALID_PIPE_STATE
);
874 if ((pipe_end
->flags
& (NAMED_PIPE_MESSAGE_STREAM_WRITE
| NAMED_PIPE_MESSAGE_STREAM_READ
))
875 != (NAMED_PIPE_MESSAGE_STREAM_WRITE
| NAMED_PIPE_MESSAGE_STREAM_READ
))
877 set_error( STATUS_INVALID_READ_MODE
);
881 /* not allowed if we already have read data buffered */
882 if (!list_empty( &pipe_end
->message_queue
))
884 set_error( STATUS_PIPE_BUSY
);
888 iosb
= async_get_iosb( async
);
889 /* ignore output buffer copy transferred because of METHOD_NEITHER */
890 iosb
->in_size
-= iosb
->out_size
;
891 /* transaction never blocks on write, so just queue a message without async */
892 message
= queue_message( pipe_end
->connection
, iosb
);
893 release_object( iosb
);
894 if (!message
) return 0;
895 reselect_read_queue( pipe_end
->connection
);
897 queue_async( &pipe_end
->read_q
, async
);
898 reselect_read_queue( pipe_end
);
899 set_error( STATUS_PENDING
);
903 static int pipe_end_get_connection_attribute( struct pipe_end
*pipe_end
)
905 const char *attr
= get_req_data();
906 data_size_t value_size
, attr_size
= get_req_data_size();
909 if (attr_size
== sizeof("ClientProcessId") && !memcmp( attr
, "ClientProcessId", attr_size
))
911 value
= &pipe_end
->client_pid
;
912 value_size
= sizeof(pipe_end
->client_pid
);
914 else if (attr_size
== sizeof("ServerProcessId") && !memcmp( attr
, "ServerProcessId", attr_size
))
916 value
= &pipe_end
->server_pid
;
917 value_size
= sizeof(pipe_end
->server_pid
);
921 set_error( STATUS_ILLEGAL_FUNCTION
);
925 if (get_reply_max_size() < value_size
)
927 set_error( STATUS_INFO_LENGTH_MISMATCH
);
931 set_reply_data( value
, value_size
);
935 static int pipe_end_ioctl( struct pipe_end
*pipe_end
, ioctl_code_t code
, struct async
*async
)
939 case FSCTL_PIPE_GET_CONNECTION_ATTRIBUTE
:
940 return pipe_end_get_connection_attribute( pipe_end
);
942 case FSCTL_PIPE_PEEK
:
943 return pipe_end_peek( pipe_end
);
945 case FSCTL_PIPE_TRANSCEIVE
:
946 return pipe_end_transceive( pipe_end
, async
);
949 return default_fd_ioctl( pipe_end
->fd
, code
, async
);
953 static int pipe_server_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
955 struct pipe_server
*server
= get_fd_user( fd
);
959 case FSCTL_PIPE_LISTEN
:
960 switch(server
->pipe_end
.state
)
962 case FILE_PIPE_LISTENING_STATE
:
963 case FILE_PIPE_DISCONNECTED_STATE
:
965 case FILE_PIPE_CONNECTED_STATE
:
966 set_error( STATUS_PIPE_CONNECTED
);
968 case FILE_PIPE_CLOSING_STATE
:
969 set_error( STATUS_PIPE_CLOSING
);
973 server
->pipe_end
.state
= FILE_PIPE_LISTENING_STATE
;
974 queue_async( &server
->listen_q
, async
);
975 async_wake_up( &server
->pipe_end
.pipe
->waiters
, STATUS_SUCCESS
);
976 set_error( STATUS_PENDING
);
979 case FSCTL_PIPE_DISCONNECT
:
980 switch(server
->pipe_end
.state
)
982 case FILE_PIPE_CONNECTED_STATE
:
983 /* dump the client connection - all data is lost */
984 assert( server
->pipe_end
.connection
);
985 release_object( server
->pipe_end
.connection
->pipe
);
986 server
->pipe_end
.connection
->pipe
= NULL
;
988 case FILE_PIPE_CLOSING_STATE
:
990 case FILE_PIPE_LISTENING_STATE
:
991 set_error( STATUS_PIPE_LISTENING
);
993 case FILE_PIPE_DISCONNECTED_STATE
:
994 set_error( STATUS_PIPE_DISCONNECTED
);
998 pipe_end_disconnect( &server
->pipe_end
, STATUS_PIPE_DISCONNECTED
);
1002 return pipe_end_ioctl( &server
->pipe_end
, code
, async
);
1006 static int pipe_client_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1008 struct pipe_client
*client
= get_fd_user( fd
);
1012 case FSCTL_PIPE_LISTEN
:
1013 set_error( STATUS_ILLEGAL_FUNCTION
);
1017 return pipe_end_ioctl( &client
->pipe_end
, code
, async
);
1021 static void init_pipe_end( struct pipe_end
*pipe_end
, struct named_pipe
*pipe
,
1022 unsigned int pipe_flags
, data_size_t buffer_size
)
1024 pipe_end
->pipe
= (struct named_pipe
*)grab_object( pipe
);
1025 pipe_end
->fd
= NULL
;
1026 pipe_end
->flags
= pipe_flags
;
1027 pipe_end
->connection
= NULL
;
1028 pipe_end
->buffer_size
= buffer_size
;
1029 init_async_queue( &pipe_end
->read_q
);
1030 init_async_queue( &pipe_end
->write_q
);
1031 list_init( &pipe_end
->message_queue
);
1034 static struct pipe_server
*create_pipe_server( struct named_pipe
*pipe
, unsigned int options
,
1035 unsigned int pipe_flags
)
1037 struct pipe_server
*server
;
1039 server
= alloc_object( &pipe_server_ops
);
1043 server
->options
= options
;
1044 init_pipe_end( &server
->pipe_end
, pipe
, pipe_flags
, pipe
->insize
);
1045 server
->pipe_end
.state
= FILE_PIPE_LISTENING_STATE
;
1046 server
->pipe_end
.server_pid
= get_process_id( current
->process
);
1048 list_add_head( &pipe
->servers
, &server
->entry
);
1049 if (!(server
->pipe_end
.fd
= alloc_pseudo_fd( &pipe_server_fd_ops
, &server
->pipe_end
.obj
, options
)))
1051 release_object( server
);
1054 allow_fd_caching( server
->pipe_end
.fd
);
1055 set_fd_signaled( server
->pipe_end
.fd
, 1 );
1056 init_async_queue( &server
->listen_q
);
1060 static struct pipe_client
*create_pipe_client( unsigned int flags
, struct named_pipe
*pipe
,
1061 data_size_t buffer_size
, unsigned int options
)
1063 struct pipe_client
*client
;
1065 client
= alloc_object( &pipe_client_ops
);
1069 client
->flags
= flags
;
1070 init_pipe_end( &client
->pipe_end
, pipe
, pipe
->flags
, buffer_size
);
1071 client
->pipe_end
.state
= FILE_PIPE_CONNECTED_STATE
;
1072 client
->pipe_end
.client_pid
= get_process_id( current
->process
);
1074 client
->pipe_end
.fd
= alloc_pseudo_fd( &pipe_client_fd_ops
, &client
->pipe_end
.obj
, options
);
1075 if (!client
->pipe_end
.fd
)
1077 release_object( client
);
1080 allow_fd_caching( client
->pipe_end
.fd
);
1081 set_fd_signaled( client
->pipe_end
.fd
, 1 );
1086 static struct pipe_server
*find_available_server( struct named_pipe
*pipe
)
1088 struct pipe_server
*server
;
1090 /* look for pipe servers that are listening */
1091 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1093 if (server
->pipe_end
.state
== FILE_PIPE_LISTENING_STATE
&& async_queued( &server
->listen_q
))
1094 return (struct pipe_server
*)grab_object( server
);
1097 /* fall back to pipe servers that are idle */
1098 LIST_FOR_EACH_ENTRY( server
, &pipe
->servers
, struct pipe_server
, entry
)
1100 if (server
->pipe_end
.state
== FILE_PIPE_LISTENING_STATE
)
1101 return (struct pipe_server
*)grab_object( server
);
1107 static int named_pipe_link_name( struct object
*obj
, struct object_name
*name
, struct object
*parent
)
1109 struct named_pipe_device
*dev
= (struct named_pipe_device
*)parent
;
1111 if (parent
->ops
!= &named_pipe_device_ops
)
1113 set_error( STATUS_OBJECT_NAME_INVALID
);
1116 namespace_add( dev
->pipes
, name
);
1117 name
->parent
= grab_object( parent
);
1121 static struct object
*named_pipe_open_file( struct object
*obj
, unsigned int access
,
1122 unsigned int sharing
, unsigned int options
)
1124 struct named_pipe
*pipe
= (struct named_pipe
*)obj
;
1125 struct pipe_server
*server
;
1126 struct pipe_client
*client
;
1127 unsigned int pipe_sharing
;
1129 if (!(server
= find_available_server( pipe
)))
1131 set_error( STATUS_PIPE_NOT_AVAILABLE
);
1135 pipe_sharing
= pipe
->sharing
;
1136 if (((access
& GENERIC_READ
) && !(pipe_sharing
& FILE_SHARE_READ
)) ||
1137 ((access
& GENERIC_WRITE
) && !(pipe_sharing
& FILE_SHARE_WRITE
)))
1139 set_error( STATUS_ACCESS_DENIED
);
1140 release_object( server
);
1144 if ((client
= create_pipe_client( options
, pipe
, pipe
->outsize
, options
)))
1146 async_wake_up( &server
->listen_q
, STATUS_SUCCESS
);
1147 server
->pipe_end
.state
= FILE_PIPE_CONNECTED_STATE
;
1148 server
->pipe_end
.connection
= &client
->pipe_end
;
1149 client
->pipe_end
.connection
= &server
->pipe_end
;
1150 server
->pipe_end
.client_pid
= client
->pipe_end
.client_pid
;
1151 client
->pipe_end
.server_pid
= server
->pipe_end
.server_pid
;
1153 release_object( server
);
1154 return &client
->pipe_end
.obj
;
1157 static int named_pipe_device_ioctl( struct fd
*fd
, ioctl_code_t code
, struct async
*async
)
1159 struct named_pipe_device
*device
= get_fd_user( fd
);
1163 case FSCTL_PIPE_WAIT
:
1165 const FILE_PIPE_WAIT_FOR_BUFFER
*buffer
= get_req_data();
1166 data_size_t size
= get_req_data_size();
1167 struct named_pipe
*pipe
;
1168 struct pipe_server
*server
;
1169 struct unicode_str name
;
1172 if (size
< sizeof(*buffer
) ||
1173 size
< FIELD_OFFSET(FILE_PIPE_WAIT_FOR_BUFFER
, Name
[buffer
->NameLength
/sizeof(WCHAR
)]))
1175 set_error( STATUS_INVALID_PARAMETER
);
1178 name
.str
= buffer
->Name
;
1179 name
.len
= (buffer
->NameLength
/ sizeof(WCHAR
)) * sizeof(WCHAR
);
1180 if (!(pipe
= open_named_object( &device
->obj
, &named_pipe_ops
, &name
, 0 ))) return 0;
1182 if (!(server
= find_available_server( pipe
)))
1184 queue_async( &pipe
->waiters
, async
);
1185 when
= buffer
->TimeoutSpecified
? buffer
->Timeout
.QuadPart
: pipe
->timeout
;
1186 async_set_timeout( async
, when
, STATUS_IO_TIMEOUT
);
1187 release_object( pipe
);
1188 set_error( STATUS_PENDING
);
1192 release_object( server
);
1193 release_object( pipe
);
1198 return default_fd_ioctl( fd
, code
, async
);
1203 DECL_HANDLER(create_named_pipe
)
1205 struct named_pipe
*pipe
;
1206 struct pipe_server
*server
;
1207 struct unicode_str name
;
1208 struct object
*root
;
1209 const struct security_descriptor
*sd
;
1210 const struct object_attributes
*objattr
= get_req_object_attributes( &sd
, &name
, &root
);
1212 if (!objattr
) return;
1214 if (!req
->sharing
|| (req
->sharing
& ~(FILE_SHARE_READ
| FILE_SHARE_WRITE
)) ||
1215 (!(req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
) && (req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
)))
1217 if (root
) release_object( root
);
1218 set_error( STATUS_INVALID_PARAMETER
);
1222 if (!name
.len
) /* pipes need a root directory even without a name */
1224 if (!objattr
->rootdir
)
1226 set_error( STATUS_OBJECT_PATH_SYNTAX_BAD
);
1229 if (!(root
= get_directory_obj( current
->process
, objattr
->rootdir
))) return;
1232 pipe
= create_named_object( root
, &named_pipe_ops
, &name
, objattr
->attributes
| OBJ_OPENIF
, NULL
);
1234 if (root
) release_object( root
);
1237 if (get_error() != STATUS_OBJECT_NAME_EXISTS
)
1239 /* initialize it if it didn't already exist */
1240 pipe
->instances
= 0;
1241 init_async_queue( &pipe
->waiters
);
1242 list_init( &pipe
->servers
);
1243 pipe
->insize
= req
->insize
;
1244 pipe
->outsize
= req
->outsize
;
1245 pipe
->maxinstances
= req
->maxinstances
;
1246 pipe
->timeout
= req
->timeout
;
1247 pipe
->flags
= req
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
;
1248 pipe
->sharing
= req
->sharing
;
1249 if (sd
) default_set_sd( &pipe
->obj
, sd
, OWNER_SECURITY_INFORMATION
|
1250 GROUP_SECURITY_INFORMATION
|
1251 DACL_SECURITY_INFORMATION
|
1252 SACL_SECURITY_INFORMATION
);
1256 if (pipe
->maxinstances
<= pipe
->instances
)
1258 set_error( STATUS_INSTANCE_NOT_AVAILABLE
);
1259 release_object( pipe
);
1262 if (pipe
->sharing
!= req
->sharing
)
1264 set_error( STATUS_ACCESS_DENIED
);
1265 release_object( pipe
);
1268 clear_error(); /* clear the name collision */
1271 server
= create_pipe_server( pipe
, req
->options
, req
->flags
);
1274 reply
->handle
= alloc_handle( current
->process
, server
, req
->access
, objattr
->attributes
);
1276 release_object( server
);
1279 release_object( pipe
);
1282 DECL_HANDLER(get_named_pipe_info
)
1284 struct pipe_end
*pipe_end
;
1286 pipe_end
= (struct pipe_end
*)get_handle_obj( current
->process
, req
->handle
,
1287 FILE_READ_ATTRIBUTES
, &pipe_server_ops
);
1290 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1294 pipe_end
= (struct pipe_end
*)get_handle_obj( current
->process
, req
->handle
,
1295 FILE_READ_ATTRIBUTES
, &pipe_client_ops
);
1296 if (!pipe_end
) return;
1301 reply
->flags
= pipe_end
->flags
;
1302 reply
->sharing
= pipe_end
->pipe
->sharing
;
1303 reply
->maxinstances
= pipe_end
->pipe
->maxinstances
;
1304 reply
->instances
= pipe_end
->pipe
->instances
;
1305 reply
->insize
= pipe_end
->pipe
->insize
;
1306 reply
->outsize
= pipe_end
->pipe
->outsize
;
1308 if (pipe_end
->obj
.ops
== &pipe_server_ops
) reply
->flags
|= NAMED_PIPE_SERVER_END
;
1310 else set_error( STATUS_PIPE_DISCONNECTED
);
1312 release_object( pipe_end
);
1315 DECL_HANDLER(set_named_pipe_info
)
1317 struct pipe_end
*pipe_end
;
1319 pipe_end
= (struct pipe_end
*)get_handle_obj( current
->process
, req
->handle
,
1320 FILE_WRITE_ATTRIBUTES
, &pipe_server_ops
);
1323 if (get_error() != STATUS_OBJECT_TYPE_MISMATCH
)
1327 pipe_end
= (struct pipe_end
*)get_handle_obj( current
->process
, req
->handle
,
1328 0, &pipe_client_ops
);
1329 if (!pipe_end
) return;
1332 if ((req
->flags
& ~(NAMED_PIPE_MESSAGE_STREAM_READ
| NAMED_PIPE_NONBLOCKING_MODE
)) ||
1333 ((req
->flags
& NAMED_PIPE_MESSAGE_STREAM_READ
) && !(pipe_end
->pipe
->flags
& NAMED_PIPE_MESSAGE_STREAM_WRITE
)))
1335 set_error( STATUS_INVALID_PARAMETER
);
1339 pipe_end
->flags
= pipe_end
->pipe
->flags
| req
->flags
;
1342 release_object( pipe_end
);