server/named_pipe.c

   1 /*
   2  * Server-side pipe management
   3  *
   4  * Copyright (C) 1998 Alexandre Julliard
   5  * Copyright (C) 2001 Mike McCormack
   6  * Copyright 2016 Jacek Caban for CodeWeavers
   7  *
   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.
  12  *
  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.
  17  *
  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
  21  */
  22
  23 #include "config.h"
  24 #include "wine/port.h"
  25
  26 #include <assert.h>
  27 #include <string.h>
  28 #include <stdarg.h>
  29 #include <stdio.h>
  30 #include <stdlib.h>
  31
  32 #include "ntstatus.h"
  33 #define WIN32_NO_STATUS
  34 #include "windef.h"
  35 #include "winternl.h"
  36 #include "winioctl.h"
  37
  38 #include "file.h"
  39 #include "handle.h"
  40 #include "thread.h"
  41 #include "request.h"
  42 #include "security.h"
  43
  44 enum pipe_state
  45 {
  46     ps_idle_server,
  47     ps_wait_open,
  48     ps_connected_server,
  49     ps_wait_disconnect,
  50     ps_wait_connect
  51 };
  52
  53 struct named_pipe;
  54
  55 struct pipe_message
  56 {
  57     struct list          entry;      /* entry in message queue */
  58     data_size_t          read_pos;   /* already read bytes */
  59     struct iosb         *iosb;       /* message iosb */
  60     struct async        *async;      /* async of pending write */
  61 };
  62
  63 struct pipe_end
  64 {
  65     struct object        obj;        /* object header */
  66     struct fd           *fd;         /* pipe file descriptor */
  67     unsigned int         flags;      /* pipe flags */
  68     struct pipe_end     *connection; /* the other end of the pipe */
  69     data_size_t          buffer_size;/* size of buffered data that doesn't block caller */
  70     struct list          message_queue;
  71     struct async_queue   read_q;     /* read queue */
  72     struct async_queue   write_q;    /* write queue */
  73 };
  74
  75 struct pipe_server
  76 {
  77     struct pipe_end      pipe_end;   /* common header for pipe_client and pipe_server */
  78     struct list          entry;      /* entry in named pipe servers list */
  79     enum pipe_state      state;      /* server state */
  80     struct pipe_client  *client;     /* client that this server is connected to */
  81     struct named_pipe   *pipe;
  82     unsigned int         options;    /* pipe options */
  83 };
  84
  85 struct pipe_client
  86 {
  87     struct pipe_end      pipe_end;   /* common header for pipe_client and pipe_server */
  88     struct pipe_server  *server;     /* server that this client is connected to */
  89     unsigned int         flags;      /* file flags */
  90 };
  91
  92 struct named_pipe
  93 {
  94     struct object       obj;         /* object header */
  95     unsigned int        flags;
  96     unsigned int        sharing;
  97     unsigned int        maxinstances;
  98     unsigned int        outsize;
  99     unsigned int        insize;
 100     unsigned int        instances;
 101     timeout_t           timeout;
 102     struct list         servers;     /* list of servers using this pipe */
 103     struct async_queue  waiters;     /* list of clients waiting to connect */
 104 };
 105
 106 struct named_pipe_device
 107 {
 108     struct object       obj;         /* object header */
 109     struct fd          *fd;          /* pseudo-fd for ioctls */
 110     struct namespace   *pipes;       /* named pipe namespace */
 111 };
 112
 113 static void named_pipe_dump( struct object *obj, int verbose );
 114 static unsigned int named_pipe_map_access( struct object *obj, unsigned int access );
 115 static int named_pipe_link_name( struct object *obj, struct object_name *name, struct object *parent );
 116 static struct object *named_pipe_open_file( struct object *obj, unsigned int access,
 117                                             unsigned int sharing, unsigned int options );
 118 static void named_pipe_destroy( struct object *obj );
 119
 120 static const struct object_ops named_pipe_ops =
 121 {
 122     sizeof(struct named_pipe),    /* size */
 123     named_pipe_dump,              /* dump */
 124     no_get_type,                  /* get_type */
 125     no_add_queue,                 /* add_queue */
 126     NULL,                         /* remove_queue */
 127     NULL,                         /* signaled */
 128     NULL,                         /* satisfied */
 129     no_signal,                    /* signal */
 130     no_get_fd,                    /* get_fd */
 131     named_pipe_map_access,        /* map_access */
 132     default_get_sd,               /* get_sd */
 133     default_set_sd,               /* set_sd */
 134     no_lookup_name,               /* lookup_name */
 135     named_pipe_link_name,         /* link_name */
 136     default_unlink_name,          /* unlink_name */
 137     named_pipe_open_file,         /* open_file */
 138     no_close_handle,              /* close_handle */
 139     named_pipe_destroy            /* destroy */
 140 };
 141
 142 /* common server and client pipe end functions */
 143 static enum server_fd_type pipe_end_get_fd_type( struct fd *fd );
 144 static struct fd *pipe_end_get_fd( struct object *obj );
 145 static int pipe_end_read( struct fd *fd, struct async *async, file_pos_t pos );
 146 static int pipe_end_write( struct fd *fd, struct async *async_data, file_pos_t pos );
 147 static int pipe_end_flush( struct fd *fd, struct async *async );
 148 static void pipe_end_get_volume_info( struct fd *fd, unsigned int info_class );
 149 static void pipe_end_reselect_async( struct fd *fd, struct async_queue *queue );
 150
 151 /* server end functions */
 152 static void pipe_server_dump( struct object *obj, int verbose );
 153 static void pipe_server_destroy( struct object *obj);
 154 static int pipe_server_ioctl( struct fd *fd, ioctl_code_t code, struct async *async );
 155
 156 static const struct object_ops pipe_server_ops =
 157 {
 158     sizeof(struct pipe_server),   /* size */
 159     pipe_server_dump,             /* dump */
 160     no_get_type,                  /* get_type */
 161     add_queue,                    /* add_queue */
 162     remove_queue,                 /* remove_queue */
 163     default_fd_signaled,          /* signaled */
 164     no_satisfied,                 /* satisfied */
 165     no_signal,                    /* signal */
 166     pipe_end_get_fd,              /* get_fd */
 167     default_fd_map_access,        /* map_access */
 168     default_get_sd,               /* get_sd */
 169     default_set_sd,               /* set_sd */
 170     no_lookup_name,               /* lookup_name */
 171     no_link_name,                 /* link_name */
 172     NULL,                         /* unlink_name */
 173     no_open_file,                 /* open_file */
 174     fd_close_handle,              /* close_handle */
 175     pipe_server_destroy           /* destroy */
 176 };
 177
 178 static const struct fd_ops pipe_server_fd_ops =
 179 {
 180     default_fd_get_poll_events,   /* get_poll_events */
 181     default_poll_event,           /* poll_event */
 182     pipe_end_get_fd_type,         /* get_fd_type */
 183     pipe_end_read,                /* read */
 184     pipe_end_write,               /* write */
 185     pipe_end_flush,               /* flush */
 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 */
 190 };
 191
 192 /* client end functions */
 193 static void pipe_client_dump( struct object *obj, int verbose );
 194 static int pipe_client_signaled( struct object *obj, struct wait_queue_entry *entry );
 195 static void pipe_client_destroy( struct object *obj );
 196 static int pipe_client_ioctl( struct fd *fd, ioctl_code_t code, struct async *async );
 197
 198 static const struct object_ops pipe_client_ops =
 199 {
 200     sizeof(struct pipe_client),   /* size */
 201     pipe_client_dump,             /* dump */
 202     no_get_type,                  /* get_type */
 203     add_queue,                    /* add_queue */
 204     remove_queue,                 /* remove_queue */
 205     pipe_client_signaled,         /* signaled */
 206     no_satisfied,                 /* satisfied */
 207     no_signal,                    /* signal */
 208     pipe_end_get_fd,              /* get_fd */
 209     default_fd_map_access,        /* map_access */
 210     default_get_sd,               /* get_sd */
 211     default_set_sd,               /* set_sd */
 212     no_lookup_name,               /* lookup_name */
 213     no_link_name,                 /* link_name */
 214     NULL,                         /* unlink_name */
 215     no_open_file,                 /* open_file */
 216     fd_close_handle,              /* close_handle */
 217     pipe_client_destroy           /* destroy */
 218 };
 219
 220 static const struct fd_ops pipe_client_fd_ops =
 221 {
 222     default_fd_get_poll_events,   /* get_poll_events */
 223     default_poll_event,           /* poll_event */
 224     pipe_end_get_fd_type,         /* get_fd_type */
 225     pipe_end_read,                /* read */
 226     pipe_end_write,               /* write */
 227     pipe_end_flush,               /* flush */
 228     pipe_end_get_volume_info,     /* get_volume_info */
 229     pipe_client_ioctl,            /* ioctl */
 230     no_fd_queue_async,            /* queue_async */
 231     pipe_end_reselect_async       /* reselect_async */
 232 };
 233
 234 static void named_pipe_device_dump( struct object *obj, int verbose );
 235 static struct object_type *named_pipe_device_get_type( struct object *obj );
 236 static struct fd *named_pipe_device_get_fd( struct object *obj );
 237 static struct object *named_pipe_device_lookup_name( struct object *obj,
 238     struct unicode_str *name, unsigned int attr );
 239 static struct object *named_pipe_device_open_file( struct object *obj, unsigned int access,
 240                                                    unsigned int sharing, unsigned int options );
 241 static void named_pipe_device_destroy( struct object *obj );
 242 static enum server_fd_type named_pipe_device_get_fd_type( struct fd *fd );
 243 static int named_pipe_device_ioctl( struct fd *fd, ioctl_code_t code, struct async *async );
 244
 245 static const struct object_ops named_pipe_device_ops =
 246 {
 247     sizeof(struct named_pipe_device), /* size */
 248     named_pipe_device_dump,           /* dump */
 249     named_pipe_device_get_type,       /* get_type */
 250     no_add_queue,                     /* add_queue */
 251     NULL,                             /* remove_queue */
 252     NULL,                             /* signaled */
 253     no_satisfied,                     /* satisfied */
 254     no_signal,                        /* signal */
 255     named_pipe_device_get_fd,         /* get_fd */
 256     no_map_access,                    /* map_access */
 257     default_get_sd,                   /* get_sd */
 258     default_set_sd,                   /* set_sd */
 259     named_pipe_device_lookup_name,    /* lookup_name */
 260     directory_link_name,              /* link_name */
 261     default_unlink_name,              /* unlink_name */
 262     named_pipe_device_open_file,      /* open_file */
 263     fd_close_handle,                  /* close_handle */
 264     named_pipe_device_destroy         /* destroy */
 265 };
 266
 267 static const struct fd_ops named_pipe_device_fd_ops =
 268 {
 269     default_fd_get_poll_events,       /* get_poll_events */
 270     default_poll_event,               /* poll_event */
 271     named_pipe_device_get_fd_type,    /* get_fd_type */
 272     no_fd_read,                       /* read */
 273     no_fd_write,                      /* write */
 274     no_fd_flush,                      /* flush */
 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 */
 279 };
 280
 281 static void named_pipe_dump( struct object *obj, int verbose )
 282 {
 283     fputs( "Named pipe\n", stderr );
 284 }
 285
 286 static unsigned int named_pipe_map_access( struct object *obj, unsigned int access )
 287 {
 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);
 293 }
 294
 295 static void pipe_server_dump( struct object *obj, int verbose )
 296 {
 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, server->state );
 300 }
 301
 302 static void pipe_client_dump( struct object *obj, int verbose )
 303 {
 304     struct pipe_client *client = (struct pipe_client *) obj;
 305     assert( obj->ops == &pipe_client_ops );
 306     fprintf( stderr, "Named pipe client server=%p\n", client->server );
 307 }
 308
 309 static int pipe_client_signaled( struct object *obj, struct wait_queue_entry *entry )
 310 {
 311     struct pipe_client *client = (struct pipe_client *) obj;
 312
 313     return client->pipe_end.fd && is_fd_signaled(client->pipe_end.fd);
 314 }
 315
 316 static void named_pipe_destroy( struct object *obj)
 317 {
 318     struct named_pipe *pipe = (struct named_pipe *) obj;
 319
 320     assert( list_empty( &pipe->servers ) );
 321     assert( !pipe->instances );
 322     free_async_queue( &pipe->waiters );
 323 }
 324
 325 static struct fd *pipe_end_get_fd( struct object *obj )
 326 {
 327     struct pipe_end *pipe_end = (struct pipe_end *) obj;
 328     return (struct fd *) grab_object( pipe_end->fd );
 329 }
 330
 331 static void set_server_state( struct pipe_server *server, enum pipe_state state )
 332 {
 333     server->state = state;
 334
 335     switch(state)
 336     {
 337     case ps_connected_server:
 338     case ps_wait_disconnect:
 339         break;
 340     case ps_wait_open:
 341     case ps_idle_server:
 342         set_no_fd_status( server->pipe_end.fd, STATUS_PIPE_LISTENING );
 343         break;
 344     case ps_wait_connect:
 345         set_no_fd_status( server->pipe_end.fd, STATUS_PIPE_DISCONNECTED );
 346         break;
 347     }
 348 }
 349
 350
 351 static void wake_message( struct pipe_message *message )
 352 {
 353     struct async *async = message->async;
 354
 355     message->async = NULL;
 356     message->iosb->status = STATUS_SUCCESS;
 357     message->iosb->result = message->iosb->in_size;
 358     if (async)
 359     {
 360         async_terminate( async, message->iosb->result ? STATUS_ALERTED : STATUS_SUCCESS );
 361         release_object( async );
 362     }
 363 }
 364
 365 static void free_message( struct pipe_message *message )
 366 {
 367     list_remove( &message->entry );
 368     if (message->iosb) release_object( message->iosb );
 369     free( message );
 370 }
 371
 372 static void pipe_end_disconnect( struct pipe_end *pipe_end, unsigned int status )
 373 {
 374     struct pipe_end *connection = pipe_end->connection;
 375     struct pipe_message *message, *next;
 376     struct async *async;
 377
 378     pipe_end->connection = NULL;
 379
 380     if (pipe_end->fd) fd_async_wake_up( pipe_end->fd, ASYNC_TYPE_WAIT, status );
 381     async_wake_up( &pipe_end->read_q, status );
 382     LIST_FOR_EACH_ENTRY_SAFE( message, next, &pipe_end->message_queue, struct pipe_message, entry )
 383     {
 384         async = message->async;
 385         if (async || status == STATUS_PIPE_DISCONNECTED) free_message( message );
 386         if (!async) continue;
 387         async_terminate( async, status );
 388         release_object( async );
 389     }
 390     if (status == STATUS_PIPE_DISCONNECTED) set_fd_signaled( pipe_end->fd, 0 );
 391
 392     if (connection)
 393     {
 394         connection->connection = NULL;
 395         pipe_end_disconnect( connection, status );
 396     }
 397 }
 398
 399 static void pipe_end_destroy( struct pipe_end *pipe_end )
 400 {
 401     struct pipe_message *message;
 402
 403     while (!list_empty( &pipe_end->message_queue ))
 404     {
 405         message = LIST_ENTRY( list_head(&pipe_end->message_queue), struct pipe_message, entry );
 406         assert( !message->async );
 407         free_message( message );
 408     }
 409
 410     free_async_queue( &pipe_end->read_q );
 411     free_async_queue( &pipe_end->write_q );
 412     if (pipe_end->fd) release_object( pipe_end->fd );
 413 }
 414
 415 static void pipe_server_destroy( struct object *obj)
 416 {
 417     struct pipe_server *server = (struct pipe_server *)obj;
 418
 419     assert( obj->ops == &pipe_server_ops );
 420
 421     pipe_end_disconnect( &server->pipe_end, STATUS_PIPE_BROKEN );
 422
 423     pipe_end_destroy( &server->pipe_end );
 424     if (server->client)
 425     {
 426         server->client->server = NULL;
 427         server->client = NULL;
 428     }
 429
 430     assert( server->pipe->instances );
 431     server->pipe->instances--;
 432
 433     list_remove( &server->entry );
 434     release_object( server->pipe );
 435 }
 436
 437 static void pipe_client_destroy( struct object *obj)
 438 {
 439     struct pipe_client *client = (struct pipe_client *)obj;
 440     struct pipe_server *server = client->server;
 441
 442     assert( obj->ops == &pipe_client_ops );
 443
 444     pipe_end_disconnect( &client->pipe_end, STATUS_PIPE_BROKEN );
 445
 446     if (server)
 447     {
 448         switch(server->state)
 449         {
 450         case ps_connected_server:
 451             /* Don't destroy the server's fd here as we can't
 452                do a successful flush without it. */
 453             set_server_state( server, ps_wait_disconnect );
 454             break;
 455         case ps_idle_server:
 456         case ps_wait_open:
 457         case ps_wait_disconnect:
 458         case ps_wait_connect:
 459             assert( 0 );
 460         }
 461         assert( server->client );
 462         server->client = NULL;
 463         client->server = NULL;
 464     }
 465
 466     pipe_end_destroy( &client->pipe_end );
 467 }
 468
 469 static void named_pipe_device_dump( struct object *obj, int verbose )
 470 {
 471     fputs( "Named pipe device\n", stderr );
 472 }
 473
 474 static struct object_type *named_pipe_device_get_type( struct object *obj )
 475 {
 476     static const WCHAR name[] = {'D','e','v','i','c','e'};
 477     static const struct unicode_str str = { name, sizeof(name) };
 478     return get_object_type( &str );
 479 }
 480
 481 static struct fd *named_pipe_device_get_fd( struct object *obj )
 482 {
 483     struct named_pipe_device *device = (struct named_pipe_device *)obj;
 484     return (struct fd *)grab_object( device->fd );
 485 }
 486
 487 static struct object *named_pipe_device_lookup_name( struct object *obj, struct unicode_str *name,
 488                                                      unsigned int attr )
 489 {
 490     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 491     struct object *found;
 492
 493     assert( obj->ops == &named_pipe_device_ops );
 494     assert( device->pipes );
 495
 496     if (!name) return NULL;  /* open the device itself */
 497
 498     if ((found = find_object( device->pipes, name, attr | OBJ_CASE_INSENSITIVE )))
 499         name->len = 0;
 500
 501     return found;
 502 }
 503
 504 static struct object *named_pipe_device_open_file( struct object *obj, unsigned int access,
 505                                                    unsigned int sharing, unsigned int options )
 506 {
 507     return grab_object( obj );
 508 }
 509
 510 static void named_pipe_device_destroy( struct object *obj )
 511 {
 512     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 513     assert( obj->ops == &named_pipe_device_ops );
 514     if (device->fd) release_object( device->fd );
 515     free( device->pipes );
 516 }
 517
 518 static enum server_fd_type named_pipe_device_get_fd_type( struct fd *fd )
 519 {
 520     return FD_TYPE_DEVICE;
 521 }
 522
 523 struct object *create_named_pipe_device( struct object *root, const struct unicode_str *name )
 524 {
 525     struct named_pipe_device *dev;
 526
 527     if ((dev = create_named_object( root, &named_pipe_device_ops, name, 0, NULL )) &&
 528         get_error() != STATUS_OBJECT_NAME_EXISTS)
 529     {
 530         dev->pipes = NULL;
 531         if (!(dev->fd = alloc_pseudo_fd( &named_pipe_device_fd_ops, &dev->obj, 0 )) ||
 532             !(dev->pipes = create_namespace( 7 )))
 533         {
 534             release_object( dev );
 535             dev = NULL;
 536         }
 537     }
 538     return &dev->obj;
 539 }
 540
 541 static int pipe_end_flush( struct fd *fd, struct async *async )
 542 {
 543     struct pipe_end *pipe_end = get_fd_user( fd );
 544
 545     if (pipe_end->connection && !list_empty( &pipe_end->connection->message_queue ))
 546     {
 547         fd_queue_async( pipe_end->fd, async, ASYNC_TYPE_WAIT );
 548         set_error( STATUS_PENDING );
 549     }
 550     return 1;
 551 }
 552
 553 static void pipe_end_get_volume_info( struct fd *fd, unsigned int info_class )
 554 {
 555     switch (info_class)
 556     {
 557     case FileFsDeviceInformation:
 558         {
 559             static const FILE_FS_DEVICE_INFORMATION device_info =
 560             {
 561                 FILE_DEVICE_NAMED_PIPE,
 562                 FILE_DEVICE_ALLOW_APPCONTAINER_TRAVERSAL
 563             };
 564             if (get_reply_max_size() >= sizeof(device_info))
 565                 set_reply_data( &device_info, sizeof(device_info) );
 566             else
 567                 set_error( STATUS_BUFFER_TOO_SMALL );
 568             break;
 569         }
 570     default:
 571         set_error( STATUS_NOT_IMPLEMENTED );
 572     }
 573 }
 574
 575 static void message_queue_read( struct pipe_end *pipe_end, struct iosb *iosb )
 576 {
 577     struct pipe_message *message;
 578
 579     if (pipe_end->flags & NAMED_PIPE_MESSAGE_STREAM_READ)
 580     {
 581         message = LIST_ENTRY( list_head(&pipe_end->message_queue), struct pipe_message, entry );
 582         iosb->out_size = min( iosb->out_size, message->iosb->in_size - message->read_pos );
 583         iosb->status = message->read_pos + iosb->out_size < message->iosb->in_size
 584             ? STATUS_BUFFER_OVERFLOW : STATUS_SUCCESS;
 585     }
 586     else
 587     {
 588         data_size_t avail = 0;
 589         LIST_FOR_EACH_ENTRY( message, &pipe_end->message_queue, struct pipe_message, entry )
 590         {
 591             avail += message->iosb->in_size - message->read_pos;
 592             if (avail >= iosb->out_size) break;
 593         }
 594         iosb->out_size = min( iosb->out_size, avail );
 595         iosb->status = STATUS_SUCCESS;
 596     }
 597
 598     message = LIST_ENTRY( list_head(&pipe_end->message_queue), struct pipe_message, entry );
 599     if (!message->read_pos && message->iosb->in_size == iosb->out_size) /* fast path */
 600     {
 601         iosb->out_data = message->iosb->in_data;
 602         message->iosb->in_data = NULL;
 603         wake_message( message );
 604         free_message( message );
 605     }
 606     else
 607     {
 608         data_size_t write_pos = 0, writing;
 609         char *buf = NULL;
 610
 611         if (iosb->out_size && !(buf = iosb->out_data = malloc( iosb->out_size )))
 612         {
 613             iosb->out_size = 0;
 614             iosb->status = STATUS_NO_MEMORY;
 615             return;
 616         }
 617
 618         do
 619         {
 620             message = LIST_ENTRY( list_head(&pipe_end->message_queue), struct pipe_message, entry );
 621             writing = min( iosb->out_size - write_pos, message->iosb->in_size - message->read_pos );
 622             if (writing) memcpy( buf + write_pos, (const char *)message->iosb->in_data + message->read_pos, writing );
 623             write_pos += writing;
 624             message->read_pos += writing;
 625             if (message->read_pos == message->iosb->in_size)
 626             {
 627                 wake_message(message);
 628                 free_message(message);
 629             }
 630         } while (write_pos < iosb->out_size);
 631     }
 632     iosb->result = iosb->out_size;
 633 }
 634
 635 /* We call async_terminate in our reselect implementation, which causes recursive reselect.
 636  * We're not interested in such reselect calls, so we ignore them. */
 637 static int ignore_reselect;
 638
 639 static void reselect_write_queue( struct pipe_end *pipe_end );
 640
 641 static void reselect_read_queue( struct pipe_end *pipe_end )
 642 {
 643     struct async *async;
 644     struct iosb *iosb;
 645     int read_done = 0;
 646
 647     ignore_reselect = 1;
 648     while (!list_empty( &pipe_end->message_queue ) && (async = find_pending_async( &pipe_end->read_q )))
 649     {
 650         iosb = async_get_iosb( async );
 651         message_queue_read( pipe_end, iosb );
 652         async_terminate( async, iosb->result ? STATUS_ALERTED : iosb->status );
 653         release_object( async );
 654         release_object( iosb );
 655         read_done = 1;
 656     }
 657     ignore_reselect = 0;
 658
 659     if (pipe_end->connection)
 660     {
 661         if (list_empty( &pipe_end->message_queue ))
 662             fd_async_wake_up( pipe_end->connection->fd, ASYNC_TYPE_WAIT, STATUS_SUCCESS );
 663         else if (read_done)
 664             reselect_write_queue( pipe_end->connection );
 665     }
 666 }
 667
 668 static void reselect_write_queue( struct pipe_end *pipe_end )
 669 {
 670     struct pipe_message *message, *next;
 671     struct pipe_end *reader = pipe_end->connection;
 672     data_size_t avail = 0;
 673
 674     if (!reader) return;
 675
 676     ignore_reselect = 1;
 677
 678     LIST_FOR_EACH_ENTRY_SAFE( message, next, &reader->message_queue, struct pipe_message, entry )
 679     {
 680         if (message->async && message->iosb->status != STATUS_PENDING)
 681         {
 682             release_object( message->async );
 683             message->async = NULL;
 684             free_message( message );
 685         }
 686         else
 687         {
 688             avail += message->iosb->in_size - message->read_pos;
 689             if (message->iosb->status == STATUS_PENDING && (avail <= reader->buffer_size || !message->iosb->in_size))
 690                 wake_message( message );
 691         }
 692     }
 693
 694     ignore_reselect = 0;
 695     reselect_read_queue( reader );
 696 }
 697
 698 static int pipe_end_read( struct fd *fd, struct async *async, file_pos_t pos )
 699 {
 700     struct pipe_end *pipe_end = get_fd_user( fd );
 701
 702     if (!pipe_end->connection && list_empty( &pipe_end->message_queue ))
 703     {
 704         set_error( STATUS_PIPE_BROKEN );
 705         return 0;
 706     }
 707
 708     queue_async( &pipe_end->read_q, async );
 709     reselect_read_queue( pipe_end );
 710     set_error( STATUS_PENDING );
 711     return 1;
 712 }
 713
 714 static int pipe_end_write( struct fd *fd, struct async *async, file_pos_t pos )
 715 {
 716     struct pipe_end *write_end = get_fd_user( fd );
 717     struct pipe_end *read_end = write_end->connection;
 718     struct pipe_message *message;
 719
 720     if (!read_end)
 721     {
 722         set_error( STATUS_PIPE_DISCONNECTED );
 723         return 0;
 724     }
 725
 726     if (!(write_end->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE) && !get_req_data_size()) return 1;
 727
 728     if (!(message = mem_alloc( sizeof(*message) ))) return 0;
 729     message->async = (struct async *)grab_object( async );
 730     message->iosb = async_get_iosb( async );
 731     message->read_pos = 0;
 732     list_add_tail( &read_end->message_queue, &message->entry );
 733
 734     queue_async( &write_end->write_q, async );
 735     reselect_write_queue( write_end );
 736     set_error( STATUS_PENDING );
 737     return 1;
 738 }
 739
 740 static void pipe_end_reselect_async( struct fd *fd, struct async_queue *queue )
 741 {
 742     struct pipe_end *pipe_end = get_fd_user( fd );
 743
 744     if (ignore_reselect) return;
 745
 746     if (&pipe_end->write_q == queue)
 747         reselect_write_queue( pipe_end );
 748     else if (&pipe_end->read_q == queue)
 749         reselect_read_queue( pipe_end );
 750 }
 751
 752 static enum server_fd_type pipe_end_get_fd_type( struct fd *fd )
 753 {
 754     return FD_TYPE_PIPE;
 755 }
 756
 757 static int pipe_end_peek( struct pipe_end *pipe_end )
 758 {
 759     unsigned reply_size = get_reply_max_size();
 760     FILE_PIPE_PEEK_BUFFER *buffer;
 761     struct pipe_message *message;
 762     data_size_t avail = 0;
 763     data_size_t message_length = 0;
 764
 765     if (reply_size < offsetof( FILE_PIPE_PEEK_BUFFER, Data ))
 766     {
 767         set_error( STATUS_INFO_LENGTH_MISMATCH );
 768         return 0;
 769     }
 770     reply_size -= offsetof( FILE_PIPE_PEEK_BUFFER, Data );
 771
 772     if (!pipe_end->connection && list_empty( &pipe_end->message_queue ))
 773     {
 774         set_error( STATUS_PIPE_BROKEN );
 775         return 0;
 776     }
 777
 778     LIST_FOR_EACH_ENTRY( message, &pipe_end->message_queue, struct pipe_message, entry )
 779         avail += message->iosb->in_size - message->read_pos;
 780     reply_size = min( reply_size, avail );
 781
 782     if (avail && (pipe_end->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE))
 783     {
 784         message = LIST_ENTRY( list_head(&pipe_end->message_queue), struct pipe_message, entry );
 785         message_length = message->iosb->in_size - message->read_pos;
 786         reply_size = min( reply_size, message_length );
 787     }
 788
 789     if (!(buffer = set_reply_data_size( offsetof( FILE_PIPE_PEEK_BUFFER, Data[reply_size] )))) return 0;
 790     buffer->NamedPipeState    = 0;  /* FIXME */
 791     buffer->ReadDataAvailable = avail;
 792     buffer->NumberOfMessages  = 0;  /* FIXME */
 793     buffer->MessageLength     = message_length;
 794
 795     if (reply_size)
 796     {
 797         data_size_t write_pos = 0, writing;
 798         LIST_FOR_EACH_ENTRY( message, &pipe_end->message_queue, struct pipe_message, entry )
 799         {
 800             writing = min( reply_size - write_pos, message->iosb->in_size - message->read_pos );
 801             memcpy( buffer->Data + write_pos, (const char *)message->iosb->in_data + message->read_pos,
 802                     writing );
 803             write_pos += writing;
 804             if (write_pos == reply_size) break;
 805         }
 806     }
 807     return 1;
 808 }
 809
 810 static int pipe_server_ioctl( struct fd *fd, ioctl_code_t code, struct async *async )
 811 {
 812     struct pipe_server *server = get_fd_user( fd );
 813
 814     switch(code)
 815     {
 816     case FSCTL_PIPE_LISTEN:
 817         switch(server->state)
 818         {
 819         case ps_idle_server:
 820         case ps_wait_connect:
 821             fd_queue_async( server->pipe_end.fd, async, ASYNC_TYPE_WAIT );
 822             set_server_state( server, ps_wait_open );
 823             async_wake_up( &server->pipe->waiters, STATUS_SUCCESS );
 824             set_error( STATUS_PENDING );
 825             return 1;
 826         case ps_connected_server:
 827             set_error( STATUS_PIPE_CONNECTED );
 828             break;
 829         case ps_wait_disconnect:
 830             set_error( STATUS_NO_DATA_DETECTED );
 831             break;
 832         case ps_wait_open:
 833             set_error( STATUS_INVALID_HANDLE );
 834             break;
 835         }
 836         return 0;
 837
 838     case FSCTL_PIPE_DISCONNECT:
 839         switch(server->state)
 840         {
 841         case ps_connected_server:
 842             assert( server->client );
 843
 844             /* dump the client and server fds - client loses all waiting data */
 845             pipe_end_disconnect( &server->pipe_end, STATUS_PIPE_DISCONNECTED );
 846             server->client->server = NULL;
 847             server->client = NULL;
 848             set_server_state( server, ps_wait_connect );
 849             break;
 850         case ps_wait_disconnect:
 851             assert( !server->client );
 852             pipe_end_disconnect( &server->pipe_end, STATUS_PIPE_DISCONNECTED );
 853             set_server_state( server, ps_wait_connect );
 854             break;
 855         case ps_idle_server:
 856         case ps_wait_open:
 857             set_error( STATUS_PIPE_LISTENING );
 858             return 0;
 859         case ps_wait_connect:
 860             set_error( STATUS_PIPE_DISCONNECTED );
 861             return 0;
 862         }
 863         return 1;
 864
 865     case FSCTL_PIPE_PEEK:
 866         return pipe_end_peek( &server->pipe_end );
 867
 868     default:
 869         return default_fd_ioctl( fd, code, async );
 870     }
 871 }
 872
 873 static int pipe_client_ioctl( struct fd *fd, ioctl_code_t code, struct async *async )
 874 {
 875     struct pipe_client *client = get_fd_user( fd );
 876
 877     switch(code)
 878     {
 879     case FSCTL_PIPE_LISTEN:
 880         set_error( STATUS_ILLEGAL_FUNCTION );
 881         return 0;
 882
 883     case FSCTL_PIPE_PEEK:
 884         return pipe_end_peek( &client->pipe_end );
 885
 886     default:
 887         return default_fd_ioctl( fd, code, async );
 888     }
 889 }
 890
 891 static struct pipe_server *get_pipe_server_obj( struct process *process,
 892                                 obj_handle_t handle, unsigned int access )
 893 {
 894     struct object *obj;
 895     obj = get_handle_obj( process, handle, access, &pipe_server_ops );
 896     return (struct pipe_server *) obj;
 897 }
 898
 899 static void init_pipe_end( struct pipe_end *pipe_end, unsigned int pipe_flags, data_size_t buffer_size )
 900 {
 901     pipe_end->fd = NULL;
 902     pipe_end->flags = pipe_flags;
 903     pipe_end->connection = NULL;
 904     pipe_end->buffer_size = buffer_size;
 905     init_async_queue( &pipe_end->read_q );
 906     init_async_queue( &pipe_end->write_q );
 907     list_init( &pipe_end->message_queue );
 908 }
 909
 910 static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned int options,
 911                                                unsigned int pipe_flags )
 912 {
 913     struct pipe_server *server;
 914
 915     server = alloc_object( &pipe_server_ops );
 916     if (!server)
 917         return NULL;
 918
 919     server->pipe = pipe;
 920     server->client = NULL;
 921     server->options = options;
 922     init_pipe_end( &server->pipe_end, pipe_flags, pipe->insize );
 923
 924     list_add_head( &pipe->servers, &server->entry );
 925     grab_object( pipe );
 926     if (!(server->pipe_end.fd = alloc_pseudo_fd( &pipe_server_fd_ops, &server->pipe_end.obj, options )))
 927     {
 928         release_object( server );
 929         return NULL;
 930     }
 931     set_fd_signaled( server->pipe_end.fd, 1 );
 932     set_server_state( server, ps_idle_server );
 933     return server;
 934 }
 935
 936 static struct pipe_client *create_pipe_client( unsigned int flags, unsigned int pipe_flags,
 937                                                data_size_t buffer_size, unsigned int options )
 938 {
 939     struct pipe_client *client;
 940
 941     client = alloc_object( &pipe_client_ops );
 942     if (!client)
 943         return NULL;
 944
 945     client->server = NULL;
 946     client->flags = flags;
 947     init_pipe_end( &client->pipe_end, pipe_flags, buffer_size );
 948
 949     client->pipe_end.fd = alloc_pseudo_fd( &pipe_client_fd_ops, &client->pipe_end.obj, options );
 950     if (!client->pipe_end.fd)
 951     {
 952         release_object( client );
 953         return NULL;
 954     }
 955     allow_fd_caching( client->pipe_end.fd );
 956     set_fd_signaled( client->pipe_end.fd, 1 );
 957
 958     return client;
 959 }
 960
 961 static struct pipe_server *find_available_server( struct named_pipe *pipe )
 962 {
 963     struct pipe_server *server;
 964
 965     /* look for pipe servers that are listening */
 966     LIST_FOR_EACH_ENTRY( server, &pipe->servers, struct pipe_server, entry )
 967     {
 968         if (server->state == ps_wait_open)
 969             return (struct pipe_server *)grab_object( server );
 970     }
 971
 972     /* fall back to pipe servers that are idle */
 973     LIST_FOR_EACH_ENTRY( server, &pipe->servers, struct pipe_server, entry )
 974     {
 975         if (server->state == ps_idle_server)
 976             return (struct pipe_server *)grab_object( server );
 977     }
 978
 979     return NULL;
 980 }
 981
 982 static int named_pipe_link_name( struct object *obj, struct object_name *name, struct object *parent )
 983 {
 984     struct named_pipe_device *dev = (struct named_pipe_device *)parent;
 985
 986     if (parent->ops != &named_pipe_device_ops)
 987     {
 988         set_error( STATUS_OBJECT_NAME_INVALID );
 989         return 0;
 990     }
 991     namespace_add( dev->pipes, name );
 992     name->parent = grab_object( parent );
 993     return 1;
 994 }
 995
 996 static struct object *named_pipe_open_file( struct object *obj, unsigned int access,
 997                                             unsigned int sharing, unsigned int options )
 998 {
 999     struct named_pipe *pipe = (struct named_pipe *)obj;
1000     struct pipe_server *server;
1001     struct pipe_client *client;
1002     unsigned int pipe_sharing;
1003
1004     if (!(server = find_available_server( pipe )))
1005     {
1006         set_error( STATUS_PIPE_NOT_AVAILABLE );
1007         return NULL;
1008     }
1009
1010     pipe_sharing = server->pipe->sharing;
1011     if (((access & GENERIC_READ) && !(pipe_sharing & FILE_SHARE_READ)) ||
1012         ((access & GENERIC_WRITE) && !(pipe_sharing & FILE_SHARE_WRITE)))
1013     {
1014         set_error( STATUS_ACCESS_DENIED );
1015         release_object( server );
1016         return NULL;
1017     }
1018
1019     if ((client = create_pipe_client( options, pipe->flags, pipe->outsize, options )))
1020     {
1021         set_no_fd_status( server->pipe_end.fd, STATUS_BAD_DEVICE_TYPE );
1022         allow_fd_caching( server->pipe_end.fd );
1023         if (server->state == ps_wait_open)
1024             fd_async_wake_up( server->pipe_end.fd, ASYNC_TYPE_WAIT, STATUS_SUCCESS );
1025         set_server_state( server, ps_connected_server );
1026         server->client = client;
1027         client->server = server;
1028         server->pipe_end.connection = &client->pipe_end;
1029         client->pipe_end.connection = &server->pipe_end;
1030     }
1031     release_object( server );
1032     return &client->pipe_end.obj;
1033 }
1034
1035 static int named_pipe_device_ioctl( struct fd *fd, ioctl_code_t code, struct async *async )
1036 {
1037     struct named_pipe_device *device = get_fd_user( fd );
1038
1039     switch(code)
1040     {
1041     case FSCTL_PIPE_WAIT:
1042         {
1043             const FILE_PIPE_WAIT_FOR_BUFFER *buffer = get_req_data();
1044             data_size_t size = get_req_data_size();
1045             struct named_pipe *pipe;
1046             struct pipe_server *server;
1047             struct unicode_str name;
1048             timeout_t when;
1049
1050             if (size < sizeof(*buffer) ||
1051                 size < FIELD_OFFSET(FILE_PIPE_WAIT_FOR_BUFFER, Name[buffer->NameLength/sizeof(WCHAR)]))
1052             {
1053                 set_error( STATUS_INVALID_PARAMETER );
1054                 return 0;
1055             }
1056             name.str = buffer->Name;
1057             name.len = (buffer->NameLength / sizeof(WCHAR)) * sizeof(WCHAR);
1058             if (!(pipe = open_named_object( &device->obj, &named_pipe_ops, &name, 0 ))) return 0;
1059
1060             if (!(server = find_available_server( pipe )))
1061             {
1062                 queue_async( &pipe->waiters, async );
1063                 when = buffer->TimeoutSpecified ? buffer->Timeout.QuadPart : pipe->timeout;
1064                 async_set_timeout( async, when, STATUS_IO_TIMEOUT );
1065                 release_object( pipe );
1066                 set_error( STATUS_PENDING );
1067                 return 1;
1068             }
1069
1070             release_object( server );
1071             release_object( pipe );
1072             return 0;
1073         }
1074
1075     default:
1076         return default_fd_ioctl( fd, code, async );
1077     }
1078 }
1079
1080
1081 DECL_HANDLER(create_named_pipe)
1082 {
1083     struct named_pipe *pipe;
1084     struct pipe_server *server;
1085     struct unicode_str name;
1086     struct object *root;
1087     const struct security_descriptor *sd;
1088     const struct object_attributes *objattr = get_req_object_attributes( &sd, &name, &root );
1089
1090     if (!objattr) return;
1091
1092     if (!req->sharing || (req->sharing & ~(FILE_SHARE_READ | FILE_SHARE_WRITE)) ||
1093         (!(req->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE) && (req->flags & NAMED_PIPE_MESSAGE_STREAM_READ)))
1094     {
1095         if (root) release_object( root );
1096         set_error( STATUS_INVALID_PARAMETER );
1097         return;
1098     }
1099
1100     if (!name.len)  /* pipes need a root directory even without a name */
1101     {
1102         if (!objattr->rootdir)
1103         {
1104             set_error( STATUS_OBJECT_PATH_SYNTAX_BAD );
1105             return;
1106         }
1107         if (!(root = get_directory_obj( current->process, objattr->rootdir ))) return;
1108     }
1109
1110     pipe = create_named_object( root, &named_pipe_ops, &name, objattr->attributes | OBJ_OPENIF, NULL );
1111
1112     if (root) release_object( root );
1113     if (!pipe) return;
1114
1115     if (get_error() != STATUS_OBJECT_NAME_EXISTS)
1116     {
1117         /* initialize it if it didn't already exist */
1118         pipe->instances = 0;
1119         init_async_queue( &pipe->waiters );
1120         list_init( &pipe->servers );
1121         pipe->insize = req->insize;
1122         pipe->outsize = req->outsize;
1123         pipe->maxinstances = req->maxinstances;
1124         pipe->timeout = req->timeout;
1125         pipe->flags = req->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE;
1126         pipe->sharing = req->sharing;
1127     }
1128     else
1129     {
1130         if (pipe->maxinstances <= pipe->instances)
1131         {
1132             set_error( STATUS_INSTANCE_NOT_AVAILABLE );
1133             release_object( pipe );
1134             return;
1135         }
1136         if (pipe->sharing != req->sharing)
1137         {
1138             set_error( STATUS_ACCESS_DENIED );
1139             release_object( pipe );
1140             return;
1141         }
1142         clear_error(); /* clear the name collision */
1143     }
1144
1145     server = create_pipe_server( pipe, req->options, req->flags );
1146     if (server)
1147     {
1148         reply->handle = alloc_handle( current->process, server, req->access, objattr->attributes );
1149         server->pipe->instances++;
1150         if (sd) default_set_sd( &server->pipe_end.obj, sd, OWNER_SECURITY_INFORMATION |
1151                                                            GROUP_SECURITY_INFORMATION |
1152                                                            DACL_SECURITY_INFORMATION |
1153                                                            SACL_SECURITY_INFORMATION );
1154         release_object( server );
1155     }
1156
1157     release_object( pipe );
1158 }
1159
1160 DECL_HANDLER(get_named_pipe_info)
1161 {
1162     struct pipe_server *server;
1163     struct pipe_client *client = NULL;
1164
1165     server = get_pipe_server_obj( current->process, req->handle, FILE_READ_ATTRIBUTES );
1166     if (!server)
1167     {
1168         if (get_error() != STATUS_OBJECT_TYPE_MISMATCH)
1169             return;
1170
1171         clear_error();
1172         client = (struct pipe_client *)get_handle_obj( current->process, req->handle,
1173                                                        0, &pipe_client_ops );
1174         if (!client) return;
1175         server = client->server;
1176     }
1177
1178     reply->flags = client ? client->pipe_end.flags : server->pipe_end.flags;
1179     if (server)
1180     {
1181         reply->sharing      = server->pipe->sharing;
1182         reply->maxinstances = server->pipe->maxinstances;
1183         reply->instances    = server->pipe->instances;
1184         reply->insize       = server->pipe->insize;
1185         reply->outsize      = server->pipe->outsize;
1186     }
1187
1188     if (client)
1189         release_object(client);
1190     else
1191     {
1192         reply->flags |= NAMED_PIPE_SERVER_END;
1193         release_object(server);
1194     }
1195 }
1196
1197 DECL_HANDLER(set_named_pipe_info)
1198 {
1199     struct pipe_server *server;
1200     struct pipe_client *client = NULL;
1201
1202     server = get_pipe_server_obj( current->process, req->handle, FILE_WRITE_ATTRIBUTES );
1203     if (!server)
1204     {
1205         if (get_error() != STATUS_OBJECT_TYPE_MISMATCH)
1206             return;
1207
1208         clear_error();
1209         client = (struct pipe_client *)get_handle_obj( current->process, req->handle,
1210                                                        0, &pipe_client_ops );
1211         if (!client) return;
1212         if (!(server = client->server))
1213         {
1214             release_object( client );
1215             return;
1216         }
1217     }
1218
1219     if ((req->flags & ~(NAMED_PIPE_MESSAGE_STREAM_READ | NAMED_PIPE_NONBLOCKING_MODE)) ||
1220             ((req->flags & NAMED_PIPE_MESSAGE_STREAM_READ) && !(server->pipe->flags & NAMED_PIPE_MESSAGE_STREAM_WRITE)))
1221     {
1222         set_error( STATUS_INVALID_PARAMETER );
1223     }
1224     else if (client)
1225     {
1226         client->pipe_end.flags = server->pipe->flags | req->flags;
1227     }
1228     else
1229     {
1230         server->pipe_end.flags = server->pipe->flags | req->flags;
1231     }
1232
1233     if (client)
1234         release_object(client);
1235     else
1236         release_object(server);
1237 }