server/named_pipe.c

   1 /*
   2  * Server-side pipe management
   3  *
   4  * Copyright (C) 1998 Alexandre Julliard
   5  * Copyright (C) 2001 Mike McCormack
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2.1 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  20  *
  21  * TODO:
  22  *   message mode
  23  */
  24
  25 #include "config.h"
  26 #include "wine/port.h"
  27
  28 #include <assert.h>
  29 #include <fcntl.h>
  30 #include <string.h>
  31 #include <stdarg.h>
  32 #include <stdio.h>
  33 #include <stdlib.h>
  34 #include <sys/time.h>
  35 #include <sys/types.h>
  36 #ifdef HAVE_SYS_SOCKET_H
  37 #include <sys/socket.h>
  38 #endif
  39 #include <time.h>
  40 #include <unistd.h>
  41 #ifdef HAVE_POLL_H
  42 #include <poll.h>
  43 #endif
  44
  45 #include "ntstatus.h"
  46 #define WIN32_NO_STATUS
  47 #include "windef.h"
  48 #include "winternl.h"
  49
  50 #include "file.h"
  51 #include "handle.h"
  52 #include "thread.h"
  53 #include "request.h"
  54
  55 enum pipe_state
  56 {
  57     ps_idle_server,
  58     ps_wait_open,
  59     ps_connected_server,
  60     ps_wait_disconnect,
  61     ps_disconnected_server,
  62     ps_wait_connect
  63 };
  64
  65 struct named_pipe;
  66
  67 struct pipe_server
  68 {
  69     struct object        obj;        /* object header */
  70     struct fd           *fd;         /* pipe file descriptor */
  71     struct list          entry;      /* entry in named pipe servers list */
  72     enum pipe_state      state;      /* server state */
  73     struct pipe_client  *client;     /* client that this server is connected to */
  74     struct named_pipe   *pipe;
  75     struct timeout_user *flush_poll;
  76     struct event        *event;
  77     struct async_queue  *wait_q;     /* only a single one can be queued */
  78     unsigned int         options;    /* pipe options */
  79 };
  80
  81 struct pipe_client
  82 {
  83     struct object        obj;        /* object header */
  84     struct fd           *fd;         /* pipe file descriptor */
  85     struct pipe_server  *server;     /* server that this client is connected to */
  86     unsigned int         flags;      /* file flags */
  87 };
  88
  89 struct named_pipe
  90 {
  91     struct object       obj;         /* object header */
  92     unsigned int        flags;
  93     unsigned int        maxinstances;
  94     unsigned int        outsize;
  95     unsigned int        insize;
  96     unsigned int        timeout;
  97     unsigned int        instances;
  98     struct list         servers;     /* list of servers using this pipe */
  99     struct async_queue *waiters;     /* list of clients waiting to connect */
 100 };
 101
 102 struct named_pipe_device
 103 {
 104     struct object       obj;         /* object header */
 105     struct fd          *fd;          /* pseudo-fd for ioctls */
 106     struct namespace   *pipes;       /* named pipe namespace */
 107 };
 108
 109 static void named_pipe_dump( struct object *obj, int verbose );
 110 static unsigned int named_pipe_map_access( struct object *obj, unsigned int access );
 111 static struct object *named_pipe_open_file( struct object *obj, unsigned int access,
 112                                             unsigned int sharing, unsigned int options );
 113 static void named_pipe_destroy( struct object *obj );
 114
 115 static const struct object_ops named_pipe_ops =
 116 {
 117     sizeof(struct named_pipe),    /* size */
 118     named_pipe_dump,              /* dump */
 119     no_add_queue,                 /* add_queue */
 120     NULL,                         /* remove_queue */
 121     NULL,                         /* signaled */
 122     NULL,                         /* satisfied */
 123     no_signal,                    /* signal */
 124     no_get_fd,                    /* get_fd */
 125     named_pipe_map_access,        /* map_access */
 126     no_lookup_name,               /* lookup_name */
 127     named_pipe_open_file,         /* open_file */
 128     no_close_handle,              /* close_handle */
 129     named_pipe_destroy            /* destroy */
 130 };
 131
 132 /* functions common to server and client */
 133 static unsigned int pipe_map_access( struct object *obj, unsigned int access );
 134
 135 /* server end functions */
 136 static void pipe_server_dump( struct object *obj, int verbose );
 137 static struct fd *pipe_server_get_fd( struct object *obj );
 138 static void pipe_server_destroy( struct object *obj);
 139 static void pipe_server_flush( struct fd *fd, struct event **event );
 140 static enum server_fd_type pipe_server_get_fd_type( struct fd *fd );
 141
 142 static const struct object_ops pipe_server_ops =
 143 {
 144     sizeof(struct pipe_server),   /* size */
 145     pipe_server_dump,             /* dump */
 146     add_queue,                    /* add_queue */
 147     remove_queue,                 /* remove_queue */
 148     default_fd_signaled,          /* signaled */
 149     no_satisfied,                 /* satisfied */
 150     no_signal,                    /* signal */
 151     pipe_server_get_fd,           /* get_fd */
 152     pipe_map_access,              /* map_access */
 153     no_lookup_name,               /* lookup_name */
 154     no_open_file,                 /* open_file */
 155     fd_close_handle,              /* close_handle */
 156     pipe_server_destroy           /* destroy */
 157 };
 158
 159 static const struct fd_ops pipe_server_fd_ops =
 160 {
 161     default_fd_get_poll_events,   /* get_poll_events */
 162     default_poll_event,           /* poll_event */
 163     pipe_server_flush,            /* flush */
 164     pipe_server_get_fd_type,      /* get_fd_type */
 165     default_fd_queue_async,       /* queue_async */
 166     default_fd_reselect_async,    /* reselect_async */
 167     default_fd_cancel_async,      /* cancel_async */
 168 };
 169
 170 /* client end functions */
 171 static void pipe_client_dump( struct object *obj, int verbose );
 172 static struct fd *pipe_client_get_fd( struct object *obj );
 173 static void pipe_client_destroy( struct object *obj );
 174 static void pipe_client_flush( struct fd *fd, struct event **event );
 175 static enum server_fd_type pipe_client_get_fd_type( struct fd *fd );
 176
 177 static const struct object_ops pipe_client_ops =
 178 {
 179     sizeof(struct pipe_client),   /* size */
 180     pipe_client_dump,             /* dump */
 181     add_queue,                    /* add_queue */
 182     remove_queue,                 /* remove_queue */
 183     default_fd_signaled,          /* signaled */
 184     no_satisfied,                 /* satisfied */
 185     no_signal,                    /* signal */
 186     pipe_client_get_fd,           /* get_fd */
 187     pipe_map_access,              /* map_access */
 188     no_lookup_name,               /* lookup_name */
 189     no_open_file,                 /* open_file */
 190     fd_close_handle,              /* close_handle */
 191     pipe_client_destroy           /* destroy */
 192 };
 193
 194 static const struct fd_ops pipe_client_fd_ops =
 195 {
 196     default_fd_get_poll_events,   /* get_poll_events */
 197     default_poll_event,           /* poll_event */
 198     pipe_client_flush,            /* flush */
 199     pipe_client_get_fd_type,      /* get_fd_type */
 200     default_fd_queue_async,       /* queue_async */
 201     default_fd_reselect_async,    /* reselect_async */
 202     default_fd_cancel_async       /* cancel_async */
 203 };
 204
 205 static void named_pipe_device_dump( struct object *obj, int verbose );
 206 static struct fd *named_pipe_device_get_fd( struct object *obj );
 207 static struct object *named_pipe_device_lookup_name( struct object *obj,
 208     struct unicode_str *name, unsigned int attr );
 209 static struct object *named_pipe_device_open_file( struct object *obj, unsigned int access,
 210                                                    unsigned int sharing, unsigned int options );
 211 static void named_pipe_device_destroy( struct object *obj );
 212 static enum server_fd_type named_pipe_device_get_fd_type( struct fd *fd );
 213
 214 static const struct object_ops named_pipe_device_ops =
 215 {
 216     sizeof(struct named_pipe_device), /* size */
 217     named_pipe_device_dump,           /* dump */
 218     no_add_queue,                     /* add_queue */
 219     NULL,                             /* remove_queue */
 220     NULL,                             /* signaled */
 221     no_satisfied,                     /* satisfied */
 222     no_signal,                        /* signal */
 223     named_pipe_device_get_fd,         /* get_fd */
 224     pipe_map_access,                  /* map_access */
 225     named_pipe_device_lookup_name,    /* lookup_name */
 226     named_pipe_device_open_file,      /* open_file */
 227     fd_close_handle,                  /* close_handle */
 228     named_pipe_device_destroy         /* destroy */
 229 };
 230
 231 static const struct fd_ops named_pipe_device_fd_ops =
 232 {
 233     default_fd_get_poll_events,       /* get_poll_events */
 234     default_poll_event,               /* poll_event */
 235     no_flush,                         /* flush */
 236     named_pipe_device_get_fd_type,    /* get_fd_type */
 237     default_fd_queue_async,           /* queue_async */
 238     default_fd_reselect_async,        /* reselect_async */
 239     default_fd_cancel_async           /* cancel_async */
 240 };
 241
 242 static void named_pipe_dump( struct object *obj, int verbose )
 243 {
 244     struct named_pipe *pipe = (struct named_pipe *) obj;
 245     assert( obj->ops == &named_pipe_ops );
 246     fprintf( stderr, "Named pipe " );
 247     dump_object_name( &pipe->obj );
 248     fprintf( stderr, "\n" );
 249 }
 250
 251 static unsigned int named_pipe_map_access( struct object *obj, unsigned int access )
 252 {
 253     if (access & GENERIC_READ)    access |= STANDARD_RIGHTS_READ;
 254     if (access & GENERIC_WRITE)   access |= STANDARD_RIGHTS_WRITE | FILE_CREATE_PIPE_INSTANCE;
 255     if (access & GENERIC_EXECUTE) access |= STANDARD_RIGHTS_EXECUTE;
 256     if (access & GENERIC_ALL)     access |= STANDARD_RIGHTS_ALL;
 257     return access & ~(GENERIC_READ | GENERIC_WRITE | GENERIC_EXECUTE | GENERIC_ALL);
 258 }
 259
 260 static void pipe_server_dump( struct object *obj, int verbose )
 261 {
 262     struct pipe_server *server = (struct pipe_server *) obj;
 263     assert( obj->ops == &pipe_server_ops );
 264     fprintf( stderr, "Named pipe server pipe=%p state=%d\n", server->pipe, server->state );
 265 }
 266
 267 static void pipe_client_dump( struct object *obj, int verbose )
 268 {
 269     struct pipe_client *client = (struct pipe_client *) obj;
 270     assert( obj->ops == &pipe_client_ops );
 271     fprintf( stderr, "Named pipe client server=%p\n", client->server );
 272 }
 273
 274 static void named_pipe_destroy( struct object *obj)
 275 {
 276     struct named_pipe *pipe = (struct named_pipe *) obj;
 277
 278     assert( list_empty( &pipe->servers ) );
 279     assert( !pipe->instances );
 280     free_async_queue( pipe->waiters );
 281 }
 282
 283 static struct fd *pipe_client_get_fd( struct object *obj )
 284 {
 285     struct pipe_client *client = (struct pipe_client *) obj;
 286     if (client->fd)
 287         return (struct fd *) grab_object( client->fd );
 288     set_error( STATUS_PIPE_DISCONNECTED );
 289     return NULL;
 290 }
 291
 292 static struct fd *pipe_server_get_fd( struct object *obj )
 293 {
 294     struct pipe_server *server = (struct pipe_server *) obj;
 295
 296     switch(server->state)
 297     {
 298     case ps_connected_server:
 299     case ps_wait_disconnect:
 300         assert( server->fd );
 301         return (struct fd *) grab_object( server->fd );
 302
 303     case ps_wait_open:
 304     case ps_idle_server:
 305         set_error( STATUS_PIPE_LISTENING );
 306         break;
 307
 308     case ps_disconnected_server:
 309     case ps_wait_connect:
 310         set_error( STATUS_PIPE_DISCONNECTED );
 311         break;
 312     }
 313     return NULL;
 314 }
 315
 316
 317 static void notify_empty( struct pipe_server *server )
 318 {
 319     if (!server->flush_poll)
 320         return;
 321     assert( server->state == ps_connected_server );
 322     assert( server->event );
 323     remove_timeout_user( server->flush_poll );
 324     server->flush_poll = NULL;
 325     set_event( server->event );
 326     release_object( server->event );
 327     server->event = NULL;
 328 }
 329
 330 static void do_disconnect( struct pipe_server *server )
 331 {
 332     /* we may only have a server fd, if the client disconnected */
 333     if (server->client)
 334     {
 335         assert( server->client->server == server );
 336         assert( server->client->fd );
 337         release_object( server->client->fd );
 338         server->client->fd = NULL;
 339     }
 340     assert( server->fd );
 341     shutdown( get_unix_fd( server->fd ), SHUT_RDWR );
 342     release_object( server->fd );
 343     server->fd = NULL;
 344 }
 345
 346 static unsigned int pipe_map_access( struct object *obj, unsigned int access )
 347 {
 348     if (access & GENERIC_READ)    access |= FILE_GENERIC_READ;
 349     if (access & GENERIC_WRITE)   access |= FILE_GENERIC_WRITE;
 350     if (access & GENERIC_EXECUTE) access |= FILE_GENERIC_EXECUTE;
 351     if (access & GENERIC_ALL)     access |= FILE_ALL_ACCESS;
 352     return access & ~(GENERIC_READ | GENERIC_WRITE | GENERIC_EXECUTE | GENERIC_ALL);
 353 }
 354
 355 static void pipe_server_destroy( struct object *obj)
 356 {
 357     struct pipe_server *server = (struct pipe_server *)obj;
 358
 359     assert( obj->ops == &pipe_server_ops );
 360
 361     if (server->fd)
 362     {
 363         notify_empty( server );
 364         do_disconnect( server );
 365     }
 366
 367     if (server->client)
 368     {
 369         server->client->server = NULL;
 370         server->client = NULL;
 371     }
 372
 373     free_async_queue( server->wait_q );
 374
 375     assert( server->pipe->instances );
 376     server->pipe->instances--;
 377
 378     list_remove( &server->entry );
 379     release_object( server->pipe );
 380 }
 381
 382 static void pipe_client_destroy( struct object *obj)
 383 {
 384     struct pipe_client *client = (struct pipe_client *)obj;
 385     struct pipe_server *server = client->server;
 386
 387     assert( obj->ops == &pipe_client_ops );
 388
 389     if (server)
 390     {
 391         notify_empty( server );
 392
 393         switch(server->state)
 394         {
 395         case ps_connected_server:
 396             /* Don't destroy the server's fd here as we can't
 397                do a successful flush without it. */
 398             server->state = ps_wait_disconnect;
 399             release_object( client->fd );
 400             client->fd = NULL;
 401             break;
 402         case ps_disconnected_server:
 403             server->state = ps_wait_connect;
 404             break;
 405         case ps_idle_server:
 406         case ps_wait_open:
 407         case ps_wait_disconnect:
 408         case ps_wait_connect:
 409             assert( 0 );
 410         }
 411         assert( server->client );
 412         server->client = NULL;
 413         client->server = NULL;
 414     }
 415     assert( !client->fd );
 416 }
 417
 418 static void named_pipe_device_dump( struct object *obj, int verbose )
 419 {
 420     assert( obj->ops == &named_pipe_device_ops );
 421     fprintf( stderr, "Named pipe device\n" );
 422 }
 423
 424 static struct fd *named_pipe_device_get_fd( struct object *obj )
 425 {
 426     struct named_pipe_device *device = (struct named_pipe_device *)obj;
 427     return (struct fd *)grab_object( device->fd );
 428 }
 429
 430 static struct object *named_pipe_device_lookup_name( struct object *obj, struct unicode_str *name,
 431                                                      unsigned int attr )
 432 {
 433     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 434     struct object *found;
 435
 436     assert( obj->ops == &named_pipe_device_ops );
 437     assert( device->pipes );
 438
 439     if ((found = find_object( device->pipes, name, attr | OBJ_CASE_INSENSITIVE )))
 440         name->len = 0;
 441
 442     return found;
 443 }
 444
 445 static struct object *named_pipe_device_open_file( struct object *obj, unsigned int access,
 446                                                    unsigned int sharing, unsigned int options )
 447 {
 448     return grab_object( obj );
 449 }
 450
 451 static void named_pipe_device_destroy( struct object *obj )
 452 {
 453     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 454     assert( obj->ops == &named_pipe_device_ops );
 455     if (device->fd) release_object( device->fd );
 456     free( device->pipes );
 457 }
 458
 459 static enum server_fd_type named_pipe_device_get_fd_type( struct fd *fd )
 460 {
 461     return FD_TYPE_DEVICE;
 462 }
 463
 464 void create_named_pipe_device( struct directory *root, const struct unicode_str *name )
 465 {
 466     struct named_pipe_device *dev;
 467
 468     if ((dev = create_named_object_dir( root, name, 0, &named_pipe_device_ops )) &&
 469         get_error() != STATUS_OBJECT_NAME_EXISTS)
 470     {
 471         dev->pipes = NULL;
 472         if (!(dev->fd = alloc_pseudo_fd( &named_pipe_device_fd_ops, &dev->obj )) ||
 473             !(dev->pipes = create_namespace( 7 )))
 474         {
 475             release_object( dev );
 476             dev = NULL;
 477         }
 478     }
 479     if (dev) make_object_static( &dev->obj );
 480 }
 481
 482 static int pipe_data_remaining( struct pipe_server *server )
 483 {
 484     struct pollfd pfd;
 485     int fd;
 486
 487     assert( server->client );
 488
 489     fd = get_unix_fd( server->client->fd );
 490     if (fd < 0)
 491         return 0;
 492     pfd.fd = fd;
 493     pfd.events = POLLIN;
 494     pfd.revents = 0;
 495
 496     if (0 > poll( &pfd, 1, 0 ))
 497         return 0;
 498
 499     return pfd.revents&POLLIN;
 500 }
 501
 502 static void check_flushed( void *arg )
 503 {
 504     struct pipe_server *server = (struct pipe_server*) arg;
 505
 506     assert( server->event );
 507     if (pipe_data_remaining( server ))
 508     {
 509         struct timeval tv = current_time;
 510         add_timeout( &tv, 100 );
 511         server->flush_poll = add_timeout_user( &tv, check_flushed, server );
 512     }
 513     else
 514     {
 515         /* notify_empty( server ); */
 516         server->flush_poll = NULL;
 517         set_event( server->event );
 518         release_object( server->event );
 519         server->event = NULL;
 520     }
 521 }
 522
 523 static void pipe_server_flush( struct fd *fd, struct event **event )
 524 {
 525     struct pipe_server *server = get_fd_user( fd );
 526
 527     if (!server || server->state != ps_connected_server) return;
 528
 529     /* FIXME: if multiple threads flush the same pipe,
 530               maybe should create a list of processes to notify */
 531     if (server->flush_poll) return;
 532
 533     if (pipe_data_remaining( server ))
 534     {
 535         struct timeval tv = current_time;
 536
 537         /* this kind of sux -
 538            there's no unix way to be alerted when a pipe becomes empty */
 539         server->event = create_event( NULL, NULL, 0, 0, 0 );
 540         if (!server->event) return;
 541         add_timeout( &tv, 100 );
 542         server->flush_poll = add_timeout_user( &tv, check_flushed, server );
 543         *event = server->event;
 544     }
 545 }
 546
 547 static void pipe_client_flush( struct fd *fd, struct event **event )
 548 {
 549     /* FIXME: what do we have to do for this? */
 550 }
 551
 552 static inline int is_overlapped( unsigned int options )
 553 {
 554     return !(options & (FILE_SYNCHRONOUS_IO_ALERT | FILE_SYNCHRONOUS_IO_NONALERT));
 555 }
 556
 557 static enum server_fd_type pipe_server_get_fd_type( struct fd *fd )
 558 {
 559     return FD_TYPE_PIPE;
 560 }
 561
 562 static enum server_fd_type pipe_client_get_fd_type( struct fd *fd )
 563 {
 564     return FD_TYPE_PIPE;
 565 }
 566
 567 static struct named_pipe *create_named_pipe( struct directory *root, const struct unicode_str *name,
 568                                              unsigned int attr )
 569 {
 570     struct object *obj;
 571     struct named_pipe *pipe = NULL;
 572     struct unicode_str new_name;
 573
 574     if (!name || !name->len) return alloc_object( &named_pipe_ops );
 575
 576     if (!(obj = find_object_dir( root, name, attr, &new_name )))
 577     {
 578         set_error( STATUS_OBJECT_NAME_INVALID );
 579         return NULL;
 580     }
 581     if (!new_name.len)
 582     {
 583         if (attr & OBJ_OPENIF && obj->ops == &named_pipe_ops)
 584             set_error( STATUS_OBJECT_NAME_EXISTS );
 585         else
 586         {
 587             release_object( obj );
 588             obj = NULL;
 589             if (attr & OBJ_OPENIF)
 590                 set_error( STATUS_OBJECT_TYPE_MISMATCH );
 591             else
 592                 set_error( STATUS_OBJECT_NAME_COLLISION );
 593         }
 594         return (struct named_pipe *)obj;
 595     }
 596
 597     if (obj->ops != &named_pipe_device_ops)
 598         set_error( STATUS_OBJECT_NAME_INVALID );
 599     else
 600     {
 601         struct named_pipe_device *dev = (struct named_pipe_device *)obj;
 602         if ((pipe = create_object( dev->pipes, &named_pipe_ops, &new_name, NULL )))
 603             clear_error();
 604     }
 605
 606     release_object( obj );
 607     return pipe;
 608 }
 609
 610 static struct pipe_server *get_pipe_server_obj( struct process *process,
 611                                 obj_handle_t handle, unsigned int access )
 612 {
 613     struct object *obj;
 614     obj = get_handle_obj( process, handle, access, &pipe_server_ops );
 615     return (struct pipe_server *) obj;
 616 }
 617
 618 static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned int options )
 619 {
 620     struct pipe_server *server;
 621
 622     server = alloc_object( &pipe_server_ops );
 623     if (!server)
 624         return NULL;
 625
 626     server->fd = NULL;
 627     server->pipe = pipe;
 628     server->state = ps_idle_server;
 629     server->client = NULL;
 630     server->flush_poll = NULL;
 631     server->options = options;
 632     server->wait_q = create_async_queue( NULL );
 633
 634     list_add_head( &pipe->servers, &server->entry );
 635     grab_object( pipe );
 636
 637     return server;
 638 }
 639
 640 static struct pipe_client *create_pipe_client( unsigned int flags )
 641 {
 642     struct pipe_client *client;
 643
 644     client = alloc_object( &pipe_client_ops );
 645     if (!client)
 646         return NULL;
 647
 648     client->fd = NULL;
 649     client->server = NULL;
 650     client->flags = flags;
 651
 652     return client;
 653 }
 654
 655 static struct pipe_server *find_available_server( struct named_pipe *pipe )
 656 {
 657     struct pipe_server *server;
 658
 659     LIST_FOR_EACH_ENTRY( server, &pipe->servers, struct pipe_server, entry )
 660     {
 661         if (server->state == ps_idle_server || server->state == ps_wait_open)
 662             return (struct pipe_server *)grab_object( server );
 663     }
 664     return NULL;
 665 }
 666
 667 static struct object *named_pipe_open_file( struct object *obj, unsigned int access,
 668                                             unsigned int sharing, unsigned int options )
 669 {
 670     struct named_pipe *pipe = (struct named_pipe *)obj;
 671     struct pipe_server *server;
 672     struct pipe_client *client;
 673     int fds[2];
 674
 675     if (!(server = find_available_server( pipe )))
 676     {
 677         set_error( STATUS_PIPE_NOT_AVAILABLE );
 678         return NULL;
 679     }
 680
 681     if ((client = create_pipe_client( options )))
 682     {
 683         if (!socketpair( PF_UNIX, SOCK_STREAM, 0, fds ))
 684         {
 685             int res = 0;
 686
 687             assert( !client->fd );
 688             assert( !server->fd );
 689
 690             /* for performance reasons, only set nonblocking mode when using
 691              * overlapped I/O. Otherwise, we will be doing too much busy
 692              * looping */
 693             if (is_overlapped( options ))
 694                 res = fcntl( fds[1], F_SETFL, O_NONBLOCK );
 695             if ((res != -1) && is_overlapped( server->options ))
 696                 res = fcntl( fds[0], F_SETFL, O_NONBLOCK );
 697
 698             if (pipe->insize)
 699             {
 700                 setsockopt( fds[0], SOL_SOCKET, SO_RCVBUF, &pipe->insize, sizeof(pipe->insize) );
 701                 setsockopt( fds[1], SOL_SOCKET, SO_RCVBUF, &pipe->insize, sizeof(pipe->insize) );
 702             }
 703             if (pipe->outsize)
 704             {
 705                 setsockopt( fds[0], SOL_SOCKET, SO_SNDBUF, &pipe->outsize, sizeof(pipe->outsize) );
 706                 setsockopt( fds[1], SOL_SOCKET, SO_SNDBUF, &pipe->outsize, sizeof(pipe->outsize) );
 707             }
 708
 709             client->fd = create_anonymous_fd( &pipe_client_fd_ops, fds[1], &client->obj, options );
 710             server->fd = create_anonymous_fd( &pipe_server_fd_ops, fds[0], &server->obj, server->options );
 711             if (client->fd && server->fd && res != 1)
 712             {
 713                 if (server->state == ps_wait_open)
 714                     async_wake_up( server->wait_q, STATUS_SUCCESS );
 715                 server->state = ps_connected_server;
 716                 server->client = client;
 717                 client->server = server;
 718             }
 719         }
 720         else
 721             file_set_error();
 722     }
 723     release_object( server );
 724     return &client->obj;
 725 }
 726
 727 DECL_HANDLER(create_named_pipe)
 728 {
 729     struct named_pipe *pipe;
 730     struct pipe_server *server;
 731     struct unicode_str name;
 732     struct directory *root = NULL;
 733
 734     reply->handle = 0;
 735     get_req_unicode_str( &name );
 736     if (req->rootdir && !(root = get_directory_obj( current->process, req->rootdir, 0 )))
 737         return;
 738
 739     pipe = create_named_pipe( root, &name, req->attributes | OBJ_OPENIF );
 740
 741     if (root) release_object( root );
 742     if (!pipe) return;
 743
 744     if (get_error() != STATUS_OBJECT_NAME_EXISTS)
 745     {
 746         /* initialize it if it didn't already exist */
 747         pipe->instances = 0;
 748         pipe->waiters = NULL;
 749         list_init( &pipe->servers );
 750         pipe->insize = req->insize;
 751         pipe->outsize = req->outsize;
 752         pipe->maxinstances = req->maxinstances;
 753         pipe->timeout = req->timeout;
 754         pipe->flags = req->flags;
 755     }
 756     else
 757     {
 758         if (pipe->maxinstances <= pipe->instances)
 759         {
 760             set_error( STATUS_INSTANCE_NOT_AVAILABLE );
 761             release_object( pipe );
 762             return;
 763         }
 764         if ((pipe->maxinstances != req->maxinstances) ||
 765             (pipe->timeout != req->timeout) ||
 766             (pipe->flags != req->flags))
 767         {
 768             set_error( STATUS_ACCESS_DENIED );
 769             release_object( pipe );
 770             return;
 771         }
 772         clear_error(); /* clear the name collision */
 773     }
 774
 775     server = create_pipe_server( pipe, req->options );
 776     if (server)
 777     {
 778         reply->handle = alloc_handle( current->process, server, req->access, req->attributes );
 779         server->pipe->instances++;
 780         release_object( server );
 781     }
 782
 783     release_object( pipe );
 784 }
 785
 786 DECL_HANDLER(connect_named_pipe)
 787 {
 788     struct pipe_server *server;
 789     struct async *async;
 790
 791     server = get_pipe_server_obj(current->process, req->handle, 0);
 792     if (!server)
 793         return;
 794
 795     switch(server->state)
 796     {
 797     case ps_idle_server:
 798     case ps_wait_connect:
 799         assert( !server->fd );
 800         server->state = ps_wait_open;
 801         if ((async = create_async( current, server->wait_q, &req->async )))
 802         {
 803             if (server->pipe->waiters) async_wake_up( server->pipe->waiters, STATUS_SUCCESS );
 804             release_object( async );
 805             set_error( STATUS_PENDING );
 806         }
 807         break;
 808     case ps_connected_server:
 809         assert( server->fd );
 810         set_error( STATUS_PIPE_CONNECTED );
 811         break;
 812     case ps_disconnected_server:
 813         set_error( STATUS_PIPE_BUSY );
 814         break;
 815     case ps_wait_disconnect:
 816         set_error( STATUS_NO_DATA_DETECTED );
 817         break;
 818     case ps_wait_open:
 819         set_error( STATUS_INVALID_HANDLE );
 820         break;
 821     }
 822
 823     release_object(server);
 824 }
 825
 826 DECL_HANDLER(wait_named_pipe)
 827 {
 828     struct named_pipe_device *device;
 829     struct named_pipe *pipe;
 830     struct pipe_server *server;
 831     struct unicode_str name;
 832
 833     device = (struct named_pipe_device *)get_handle_obj( current->process, req->handle,
 834                                                          FILE_READ_ATTRIBUTES, &named_pipe_device_ops );
 835     if (!device) return;
 836
 837     get_req_unicode_str( &name );
 838     pipe = (struct named_pipe *)find_object( device->pipes, &name, OBJ_CASE_INSENSITIVE );
 839     release_object( device );
 840     if (!pipe)
 841     {
 842         set_error( STATUS_PIPE_NOT_AVAILABLE );
 843         return;
 844     }
 845     server = find_available_server( pipe );
 846     if (!server)
 847     {
 848         struct async *async;
 849
 850         if (!pipe->waiters && !(pipe->waiters = create_async_queue( NULL )))
 851         {
 852             release_object( pipe );
 853             return;
 854         }
 855
 856         if ((async = create_async( current, pipe->waiters, &req->async )))
 857         {
 858             if (req->timeout != NMPWAIT_WAIT_FOREVER)
 859             {
 860                 struct timeval when = current_time;
 861                 if (req->timeout == NMPWAIT_USE_DEFAULT_WAIT) add_timeout( &when, pipe->timeout );
 862                 else add_timeout( &when, req->timeout );
 863                 async_set_timeout( async, &when, STATUS_TIMEOUT );
 864             }
 865             release_object( async );
 866             set_error( STATUS_PENDING );
 867         }
 868     }
 869     else release_object( server );
 870
 871     release_object( pipe );
 872 }
 873
 874 DECL_HANDLER(disconnect_named_pipe)
 875 {
 876     struct pipe_server *server;
 877
 878     server = get_pipe_server_obj( current->process, req->handle, 0 );
 879     if (!server)
 880         return;
 881     switch(server->state)
 882     {
 883     case ps_connected_server:
 884         assert( server->fd );
 885         assert( server->client );
 886         assert( server->client->fd );
 887
 888         notify_empty( server );
 889
 890         /* Dump the client and server fds, but keep the pointers
 891            around - client loses all waiting data */
 892         server->state = ps_disconnected_server;
 893         do_disconnect( server );
 894         break;
 895
 896     case ps_wait_disconnect:
 897         assert( !server->client );
 898         assert( server->fd );
 899         do_disconnect( server );
 900         server->state = ps_wait_connect;
 901         break;
 902
 903     case ps_idle_server:
 904     case ps_wait_open:
 905     case ps_disconnected_server:
 906     case ps_wait_connect:
 907         set_error( STATUS_PIPE_DISCONNECTED );
 908         break;
 909     }
 910     release_object( server );
 911 }
 912
 913 DECL_HANDLER(get_named_pipe_info)
 914 {
 915     struct pipe_server *server;
 916     struct pipe_client *client = NULL;
 917
 918     server = get_pipe_server_obj( current->process, req->handle, FILE_READ_ATTRIBUTES );
 919     if (!server)
 920     {
 921         clear_error();
 922         client = (struct pipe_client *)get_handle_obj( current->process, req->handle,
 923                                                        FILE_READ_ATTRIBUTES, &pipe_client_ops );
 924         if (!client) return;
 925         server = client->server;
 926     }
 927
 928     reply->flags        = server->pipe->flags;
 929     reply->maxinstances = server->pipe->maxinstances;
 930     reply->instances    = server->pipe->instances;
 931     reply->insize       = server->pipe->insize;
 932     reply->outsize      = server->pipe->outsize;
 933
 934     if (client)
 935         release_object(client);
 936     else
 937     {
 938         reply->flags |= NAMED_PIPE_SERVER_END;
 939         release_object(server);
 940     }
 941 }