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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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 list          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 list         waiters;     /* list of clients waiting to connect */
 100 };
 101
 102 struct named_pipe_device
 103 {
 104     struct object       obj;         /* object header */
 105     struct namespace   *pipes;       /* named pipe namespace */
 106 };
 107
 108 static void named_pipe_dump( struct object *obj, int verbose );
 109 static void named_pipe_destroy( struct object *obj );
 110
 111 static const struct object_ops named_pipe_ops =
 112 {
 113     sizeof(struct named_pipe),    /* size */
 114     named_pipe_dump,              /* dump */
 115     no_add_queue,                 /* add_queue */
 116     NULL,                         /* remove_queue */
 117     NULL,                         /* signaled */
 118     NULL,                         /* satisfied */
 119     no_signal,                    /* signal */
 120     no_get_fd,                    /* get_fd */
 121     no_lookup_name,               /* lookup_name */
 122     no_close_handle,              /* close_handle */
 123     named_pipe_destroy            /* destroy */
 124 };
 125
 126 /* server end functions */
 127 static void pipe_server_dump( struct object *obj, int verbose );
 128 static struct fd *pipe_server_get_fd( struct object *obj );
 129 static void pipe_server_destroy( struct object *obj);
 130 static int pipe_server_flush( struct fd *fd, struct event **event );
 131 static int pipe_server_get_info( struct fd *fd );
 132
 133 static const struct object_ops pipe_server_ops =
 134 {
 135     sizeof(struct pipe_server),   /* size */
 136     pipe_server_dump,             /* dump */
 137     default_fd_add_queue,         /* add_queue */
 138     default_fd_remove_queue,      /* remove_queue */
 139     default_fd_signaled,          /* signaled */
 140     no_satisfied,                 /* satisfied */
 141     no_signal,                    /* signal */
 142     pipe_server_get_fd,           /* get_fd */
 143     no_lookup_name,               /* lookup_name */
 144     no_close_handle,              /* close_handle */
 145     pipe_server_destroy           /* destroy */
 146 };
 147
 148 static const struct fd_ops pipe_server_fd_ops =
 149 {
 150     default_fd_get_poll_events,     /* get_poll_events */
 151     default_poll_event,           /* poll_event */
 152     pipe_server_flush,            /* flush */
 153     pipe_server_get_info,         /* get_file_info */
 154     default_fd_queue_async,       /* queue_async */
 155     default_fd_cancel_async,      /* cancel_async */
 156 };
 157
 158 /* client end functions */
 159 static void pipe_client_dump( struct object *obj, int verbose );
 160 static struct fd *pipe_client_get_fd( struct object *obj );
 161 static void pipe_client_destroy( struct object *obj );
 162 static int pipe_client_flush( struct fd *fd, struct event **event );
 163 static int pipe_client_get_info( struct fd *fd );
 164
 165 static const struct object_ops pipe_client_ops =
 166 {
 167     sizeof(struct pipe_client),   /* size */
 168     pipe_client_dump,             /* dump */
 169     default_fd_add_queue,         /* add_queue */
 170     default_fd_remove_queue,      /* remove_queue */
 171     default_fd_signaled,          /* signaled */
 172     no_satisfied,                 /* satisfied */
 173     no_signal,                    /* signal */
 174     pipe_client_get_fd,           /* get_fd */
 175     no_lookup_name,               /* lookup_name */
 176     no_close_handle,              /* close_handle */
 177     pipe_client_destroy           /* destroy */
 178 };
 179
 180 static const struct fd_ops pipe_client_fd_ops =
 181 {
 182     default_fd_get_poll_events,   /* get_poll_events */
 183     default_poll_event,           /* poll_event */
 184     pipe_client_flush,            /* flush */
 185     pipe_client_get_info,         /* get_file_info */
 186     default_fd_queue_async,       /* queue_async */
 187     default_fd_cancel_async       /* cancel_async */
 188 };
 189
 190 static void named_pipe_device_dump( struct object *obj, int verbose );
 191 static struct object *named_pipe_device_lookup_name( struct object *obj,
 192     struct unicode_str *name, unsigned int attr );
 193 static void named_pipe_device_destroy( struct object *obj );
 194
 195 static const struct object_ops named_pipe_device_ops =
 196 {
 197     sizeof(struct named_pipe_device), /* size */
 198     named_pipe_device_dump,           /* dump */
 199     no_add_queue,                     /* add_queue */
 200     NULL,                             /* remove_queue */
 201     NULL,                             /* signaled */
 202     no_satisfied,                     /* satisfied */
 203     no_signal,                        /* signal */
 204     no_get_fd,                        /* get_fd */
 205     named_pipe_device_lookup_name,    /* lookup_name */
 206     no_close_handle,                  /* close_handle */
 207     named_pipe_device_destroy         /* destroy */
 208 };
 209
 210 static void named_pipe_dump( struct object *obj, int verbose )
 211 {
 212     struct named_pipe *pipe = (struct named_pipe *) obj;
 213     assert( obj->ops == &named_pipe_ops );
 214     fprintf( stderr, "Named pipe " );
 215     dump_object_name( &pipe->obj );
 216     fprintf( stderr, "\n" );
 217 }
 218
 219 static void pipe_server_dump( struct object *obj, int verbose )
 220 {
 221     struct pipe_server *server = (struct pipe_server *) obj;
 222     assert( obj->ops == &pipe_server_ops );
 223     fprintf( stderr, "Named pipe server pipe=%p state=%d\n", server->pipe, server->state );
 224 }
 225
 226 static void pipe_client_dump( struct object *obj, int verbose )
 227 {
 228     struct pipe_client *client = (struct pipe_client *) obj;
 229     assert( obj->ops == &pipe_client_ops );
 230     fprintf( stderr, "Named pipe client server=%p\n", client->server );
 231 }
 232
 233 static void named_pipe_destroy( struct object *obj)
 234 {
 235     struct named_pipe *pipe = (struct named_pipe *) obj;
 236
 237     assert( list_empty( &pipe->servers ) );
 238     assert( !pipe->instances );
 239     async_terminate_queue( &pipe->waiters, STATUS_HANDLES_CLOSED );
 240 }
 241
 242 static struct fd *pipe_client_get_fd( struct object *obj )
 243 {
 244     struct pipe_client *client = (struct pipe_client *) obj;
 245     if (client->fd)
 246         return (struct fd *) grab_object( client->fd );
 247     set_error( STATUS_PIPE_DISCONNECTED );
 248     return NULL;
 249 }
 250
 251 static struct fd *pipe_server_get_fd( struct object *obj )
 252 {
 253     struct pipe_server *server = (struct pipe_server *) obj;
 254
 255     switch(server->state)
 256     {
 257     case ps_connected_server:
 258     case ps_wait_disconnect:
 259         assert( server->fd );
 260         return (struct fd *) grab_object( server->fd );
 261
 262     case ps_wait_open:
 263     case ps_idle_server:
 264         set_error( STATUS_PIPE_LISTENING );
 265         break;
 266
 267     case ps_disconnected_server:
 268     case ps_wait_connect:
 269         set_error( STATUS_PIPE_DISCONNECTED );
 270         break;
 271     }
 272     return NULL;
 273 }
 274
 275
 276 static void notify_empty( struct pipe_server *server )
 277 {
 278     if (!server->flush_poll)
 279         return;
 280     assert( server->state == ps_connected_server );
 281     assert( server->event );
 282     remove_timeout_user( server->flush_poll );
 283     server->flush_poll = NULL;
 284     set_event( server->event );
 285     release_object( server->event );
 286     server->event = NULL;
 287 }
 288
 289 static void do_disconnect( struct pipe_server *server )
 290 {
 291     /* we may only have a server fd, if the client disconnected */
 292     if (server->client)
 293     {
 294         assert( server->client->server == server );
 295         assert( server->client->fd );
 296         release_object( server->client->fd );
 297         server->client->fd = NULL;
 298     }
 299     assert( server->fd );
 300     release_object( server->fd );
 301     server->fd = NULL;
 302 }
 303
 304 static void pipe_server_destroy( struct object *obj)
 305 {
 306     struct pipe_server *server = (struct pipe_server *)obj;
 307
 308     assert( obj->ops == &pipe_server_ops );
 309
 310     if (server->fd)
 311     {
 312         notify_empty( server );
 313         do_disconnect( server );
 314     }
 315
 316     if (server->client)
 317     {
 318         server->client->server = NULL;
 319         server->client = NULL;
 320     }
 321
 322     async_terminate_head( &server->wait_q, STATUS_HANDLES_CLOSED );
 323
 324     assert( server->pipe->instances );
 325     server->pipe->instances--;
 326
 327     list_remove( &server->entry );
 328     release_object( server->pipe );
 329 }
 330
 331 static void pipe_client_destroy( struct object *obj)
 332 {
 333     struct pipe_client *client = (struct pipe_client *)obj;
 334     struct pipe_server *server = client->server;
 335
 336     assert( obj->ops == &pipe_client_ops );
 337
 338     if (server)
 339     {
 340         notify_empty( server );
 341
 342         switch(server->state)
 343         {
 344         case ps_connected_server:
 345             /* Don't destroy the server's fd here as we can't
 346                do a successful flush without it. */
 347             server->state = ps_wait_disconnect;
 348             release_object( client->fd );
 349             client->fd = NULL;
 350             break;
 351         case ps_disconnected_server:
 352             server->state = ps_wait_connect;
 353             break;
 354         case ps_idle_server:
 355         case ps_wait_open:
 356         case ps_wait_disconnect:
 357         case ps_wait_connect:
 358             assert( 0 );
 359         }
 360         assert( server->client );
 361         server->client = NULL;
 362         client->server = NULL;
 363     }
 364     assert( !client->fd );
 365 }
 366
 367 static void named_pipe_device_dump( struct object *obj, int verbose )
 368 {
 369     assert( obj->ops == &named_pipe_device_ops );
 370     fprintf( stderr, "Named pipe device\n" );
 371 }
 372
 373 static struct object *named_pipe_device_lookup_name( struct object *obj, struct unicode_str *name,
 374                                                      unsigned int attr )
 375 {
 376     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 377     struct object *found;
 378
 379     assert( obj->ops == &named_pipe_device_ops );
 380     assert( device->pipes );
 381
 382     if ((found = find_object( device->pipes, name, attr | OBJ_CASE_INSENSITIVE )))
 383         name->len = 0;
 384
 385     return found;
 386 }
 387
 388 static void named_pipe_device_destroy( struct object *obj )
 389 {
 390     struct named_pipe_device *device = (struct named_pipe_device*)obj;
 391     assert( obj->ops == &named_pipe_device_ops );
 392     free( device->pipes );
 393 }
 394
 395 /* this will be deleted as soon an we fix wait_named_pipe */
 396 static struct named_pipe_device *named_pipe_device;
 397
 398 struct named_pipe_device *create_named_pipe_device( struct directory *root,
 399                                                     const struct unicode_str *name )
 400 {
 401     struct named_pipe_device *dev;
 402
 403     if ((dev = create_named_object_dir( root, name, 0, &named_pipe_device_ops )) &&
 404         get_error() != STATUS_OBJECT_NAME_EXISTS)
 405     {
 406         if (!(dev->pipes = create_namespace( 7 )))
 407         {
 408             release_object( dev );
 409             dev = NULL;
 410         }
 411     }
 412     named_pipe_device = dev;
 413     return dev;
 414 }
 415
 416 static int pipe_data_remaining( struct pipe_server *server )
 417 {
 418     struct pollfd pfd;
 419     int fd;
 420
 421     assert( server->client );
 422
 423     fd = get_unix_fd( server->client->fd );
 424     if (fd < 0)
 425         return 0;
 426     pfd.fd = fd;
 427     pfd.events = POLLIN;
 428     pfd.revents = 0;
 429
 430     if (0 > poll( &pfd, 1, 0 ))
 431         return 0;
 432
 433     return pfd.revents&POLLIN;
 434 }
 435
 436 static void check_flushed( void *arg )
 437 {
 438     struct pipe_server *server = (struct pipe_server*) arg;
 439
 440     assert( server->event );
 441     if (pipe_data_remaining( server ))
 442     {
 443         struct timeval tv;
 444
 445         gettimeofday( &tv, NULL );
 446         add_timeout( &tv, 100 );
 447         server->flush_poll = add_timeout_user( &tv, check_flushed, server );
 448     }
 449     else
 450     {
 451         /* notify_empty( server ); */
 452         server->flush_poll = NULL;
 453         set_event( server->event );
 454         release_object( server->event );
 455         server->event = NULL;
 456     }
 457 }
 458
 459 static int pipe_server_flush( struct fd *fd, struct event **event )
 460 {
 461     struct pipe_server *server = get_fd_user( fd );
 462
 463     if (!server)
 464         return 0;
 465
 466     if (server->state != ps_connected_server)
 467         return 0;
 468
 469     /* FIXME: if multiple threads flush the same pipe,
 470               maybe should create a list of processes to notify */
 471     if (server->flush_poll)
 472         return 0;
 473
 474     if (pipe_data_remaining( server ))
 475     {
 476         struct timeval tv;
 477
 478         /* this kind of sux -
 479            there's no unix way to be alerted when a pipe becomes empty */
 480         server->event = create_event( NULL, NULL, 0, 0, 0 );
 481         if (!server->event)
 482             return 0;
 483         gettimeofday( &tv, NULL );
 484         add_timeout( &tv, 100 );
 485         server->flush_poll = add_timeout_user( &tv, check_flushed, server );
 486         *event = server->event;
 487     }
 488
 489     return 0;
 490 }
 491
 492 static int pipe_client_flush( struct fd *fd, struct event **event )
 493 {
 494     /* FIXME: what do we have to do for this? */
 495     return 0;
 496 }
 497
 498 static inline int is_overlapped( unsigned int options )
 499 {
 500     return !(options & (FILE_SYNCHRONOUS_IO_ALERT | FILE_SYNCHRONOUS_IO_NONALERT));
 501 }
 502
 503 static int pipe_server_get_info( struct fd *fd )
 504 {
 505     struct pipe_server *server = get_fd_user( fd );
 506     int flags = FD_FLAG_AVAILABLE;
 507
 508     if (is_overlapped( server->options )) flags |= FD_FLAG_OVERLAPPED;
 509
 510     return flags;
 511 }
 512
 513 static int pipe_client_get_info( struct fd *fd )
 514 {
 515     struct pipe_client *client = get_fd_user( fd );
 516     int flags = FD_FLAG_AVAILABLE;
 517
 518     if (is_overlapped( client->flags )) flags |= FD_FLAG_OVERLAPPED;
 519
 520     return flags;
 521 }
 522
 523 static struct named_pipe *create_named_pipe( struct directory *root, const struct unicode_str *name,
 524                                              unsigned int attr )
 525 {
 526     struct object *obj;
 527     struct named_pipe *pipe = NULL;
 528     struct unicode_str new_name;
 529
 530     if (!name || !name->len) return alloc_object( &named_pipe_ops );
 531
 532     if (!(obj = find_object_dir( root, name, attr, &new_name ))) return NULL;
 533     if (!new_name.len)
 534     {
 535         if (attr & OBJ_OPENIF && obj->ops == &named_pipe_ops)
 536             set_error( STATUS_OBJECT_NAME_EXISTS );
 537         else
 538         {
 539             release_object( obj );
 540             obj = NULL;
 541             if (attr & OBJ_OPENIF)
 542                 set_error( STATUS_OBJECT_TYPE_MISMATCH );
 543             else
 544                 set_error( STATUS_OBJECT_NAME_COLLISION );
 545         }
 546         return (struct named_pipe *)obj;
 547     }
 548
 549     if (obj->ops != &named_pipe_device_ops)
 550         set_error( STATUS_OBJECT_TYPE_MISMATCH );
 551     else
 552     {
 553         struct named_pipe_device *dev = (struct named_pipe_device *)obj;
 554         if ((pipe = create_object( dev->pipes, &named_pipe_ops, &new_name, NULL )))
 555             clear_error();
 556     }
 557
 558     release_object( obj );
 559     return pipe;
 560 }
 561
 562 static struct pipe_server *get_pipe_server_obj( struct process *process,
 563                                 obj_handle_t handle, unsigned int access )
 564 {
 565     struct object *obj;
 566     obj = get_handle_obj( process, handle, access, &pipe_server_ops );
 567     return (struct pipe_server *) obj;
 568 }
 569
 570 static struct pipe_server *create_pipe_server( struct named_pipe *pipe, unsigned int options )
 571 {
 572     struct pipe_server *server;
 573
 574     server = alloc_object( &pipe_server_ops );
 575     if (!server)
 576         return NULL;
 577
 578     server->fd = NULL;
 579     server->pipe = pipe;
 580     server->state = ps_idle_server;
 581     server->client = NULL;
 582     server->flush_poll = NULL;
 583     server->options = options;
 584     list_init( &server->wait_q );
 585
 586     list_add_head( &pipe->servers, &server->entry );
 587     grab_object( pipe );
 588
 589     return server;
 590 }
 591
 592 static struct pipe_client *create_pipe_client( struct pipe_server *server, unsigned int flags )
 593 {
 594     struct pipe_client *client;
 595
 596     client = alloc_object( &pipe_client_ops );
 597     if (!client)
 598         return NULL;
 599
 600     client->fd = NULL;
 601     client->server = server;
 602     client->flags = flags;
 603
 604     return client;
 605 }
 606
 607 static inline struct pipe_server *find_server( struct named_pipe *pipe, enum pipe_state state )
 608 {
 609     struct pipe_server *server;
 610
 611     LIST_FOR_EACH_ENTRY( server, &pipe->servers, struct pipe_server, entry )
 612     {
 613         if (server->state == state) return (struct pipe_server *)grab_object( server );
 614     }
 615     return NULL;
 616 }
 617
 618 static inline struct pipe_server *find_server2( struct named_pipe *pipe,
 619                                                 enum pipe_state state1, enum pipe_state state2 )
 620 {
 621     struct pipe_server *server;
 622
 623     LIST_FOR_EACH_ENTRY( server, &pipe->servers, struct pipe_server, entry )
 624     {
 625         if (server->state == state1 || server->state == state2)
 626             return (struct pipe_server *)grab_object( server );
 627     }
 628     return NULL;
 629 }
 630
 631 DECL_HANDLER(create_named_pipe)
 632 {
 633     struct named_pipe *pipe;
 634     struct pipe_server *server;
 635     struct unicode_str name;
 636     struct directory *root = NULL;
 637
 638     reply->handle = 0;
 639     get_req_unicode_str( &name );
 640     if (req->rootdir && !(root = get_directory_obj( current->process, req->rootdir, 0 )))
 641         return;
 642
 643     pipe = create_named_pipe( root, &name, req->attributes | OBJ_OPENIF );
 644
 645     if (root) release_object( root );
 646     if (!pipe) return;
 647
 648     if (get_error() != STATUS_OBJECT_NAME_EXISTS)
 649     {
 650         /* initialize it if it didn't already exist */
 651         pipe->instances = 0;
 652         list_init( &pipe->servers );
 653         list_init( &pipe->waiters );
 654         pipe->insize = req->insize;
 655         pipe->outsize = req->outsize;
 656         pipe->maxinstances = req->maxinstances;
 657         pipe->timeout = req->timeout;
 658         pipe->flags = req->flags;
 659     }
 660     else
 661     {
 662         if (pipe->maxinstances <= pipe->instances)
 663         {
 664             set_error( STATUS_INSTANCE_NOT_AVAILABLE );
 665             release_object( pipe );
 666             return;
 667         }
 668         if ((pipe->maxinstances != req->maxinstances) ||
 669             (pipe->timeout != req->timeout) ||
 670             (pipe->flags != req->flags))
 671         {
 672             set_error( STATUS_ACCESS_DENIED );
 673             release_object( pipe );
 674             return;
 675         }
 676         set_error( 0 );  /* clear the name collision */
 677     }
 678
 679     server = create_pipe_server( pipe, req->options );
 680     if (server)
 681     {
 682         reply->handle = alloc_handle( current->process, server,
 683                                       req->access, req->attributes & OBJ_INHERIT );
 684         server->pipe->instances++;
 685         release_object( server );
 686     }
 687
 688     release_object( pipe );
 689 }
 690
 691 DECL_HANDLER(open_named_pipe)
 692 {
 693     struct pipe_server *server;
 694     struct pipe_client *client;
 695     struct unicode_str name;
 696     struct directory *root = NULL;
 697     struct named_pipe *pipe;
 698     int fds[2];
 699
 700     get_req_unicode_str( &name );
 701     if (req->rootdir && !(root = get_directory_obj( current->process, req->rootdir, 0 )))
 702         return;
 703
 704     pipe = open_object_dir( root, &name, req->attributes, &named_pipe_ops );
 705
 706     if (root) release_object( root );
 707     if (!pipe) return;
 708
 709     server = find_server2( pipe, ps_idle_server, ps_wait_open );
 710     release_object( pipe );
 711
 712     if (!server)
 713     {
 714         set_error( STATUS_PIPE_NOT_AVAILABLE );
 715         return;
 716     }
 717
 718     client = create_pipe_client( server, req->flags );
 719     if (client)
 720     {
 721         if (!socketpair( PF_UNIX, SOCK_STREAM, 0, fds ))
 722         {
 723             int res = 0;
 724
 725             assert( !client->fd );
 726             assert( !server->fd );
 727
 728             /* for performance reasons, only set nonblocking mode when using
 729              * overlapped I/O. Otherwise, we will be doing too much busy
 730              * looping */
 731             if (is_overlapped( req->flags ))
 732                 res = fcntl( fds[1], F_SETFL, O_NONBLOCK );
 733             if ((res != -1) && is_overlapped( server->options ))
 734                 res = fcntl( fds[0], F_SETFL, O_NONBLOCK );
 735
 736             client->fd = create_anonymous_fd( &pipe_client_fd_ops,
 737                                             fds[1], &client->obj );
 738             server->fd = create_anonymous_fd( &pipe_server_fd_ops,
 739                                             fds[0], &server->obj );
 740             if (client->fd && server->fd && res != 1)
 741             {
 742                 if (server->state == ps_wait_open)
 743                     async_terminate_head( &server->wait_q, STATUS_SUCCESS );
 744                 assert( list_empty( &server->wait_q ) );
 745                 server->state = ps_connected_server;
 746                 server->client = client;
 747                 client->server = server;
 748                 reply->handle = alloc_handle( current->process, client,
 749                                               req->access, req->attributes & OBJ_INHERIT );
 750             }
 751         }
 752         else
 753             file_set_error();
 754
 755         release_object( client );
 756     }
 757     release_object( server );
 758 }
 759
 760 DECL_HANDLER(connect_named_pipe)
 761 {
 762     struct pipe_server *server;
 763
 764     server = get_pipe_server_obj(current->process, req->handle, 0);
 765     if (!server)
 766         return;
 767
 768     switch(server->state)
 769     {
 770     case ps_idle_server:
 771     case ps_wait_connect:
 772         assert( !server->fd );
 773         server->state = ps_wait_open;
 774         create_async( current, NULL, &server->wait_q,
 775                       req->func, req->event, NULL );
 776         async_terminate_queue( &server->pipe->waiters, STATUS_SUCCESS );
 777         break;
 778     case ps_connected_server:
 779         assert( server->fd );
 780         set_error( STATUS_PIPE_CONNECTED );
 781         break;
 782     case ps_disconnected_server:
 783         set_error( STATUS_PIPE_BUSY );
 784         break;
 785     case ps_wait_disconnect:
 786         set_error( STATUS_NO_DATA_DETECTED );
 787         break;
 788     case ps_wait_open:
 789         set_error( STATUS_INVALID_HANDLE );
 790         break;
 791     }
 792
 793     release_object(server);
 794 }
 795
 796 DECL_HANDLER(wait_named_pipe)
 797 {
 798     struct named_pipe *pipe;
 799     struct pipe_server *server;
 800     struct unicode_str name;
 801
 802     get_req_unicode_str( &name );
 803     pipe = (struct named_pipe *)find_object( named_pipe_device->pipes, &name, OBJ_CASE_INSENSITIVE );
 804     if (!pipe)
 805     {
 806         set_error( STATUS_PIPE_NOT_AVAILABLE );
 807         return;
 808     }
 809     server = find_server( pipe, ps_wait_open );
 810     if (server)
 811     {
 812         /* there's already a server waiting for a client to connect */
 813         thread_queue_apc( current, NULL, req->func, APC_ASYNC_IO,
 814                           1, req->overlapped, NULL, (void *)STATUS_SUCCESS );
 815         release_object( server );
 816     }
 817     else
 818     {
 819         int timeout;
 820         if (req->timeout == NMPWAIT_USE_DEFAULT_WAIT)
 821             timeout = pipe->timeout;
 822         else
 823             timeout = req->timeout;
 824
 825         if (req->timeout == NMPWAIT_WAIT_FOREVER)
 826             create_async( current, NULL, &pipe->waiters,
 827                           req->func, req->overlapped, NULL );
 828         else
 829             create_async( current, &timeout, &pipe->waiters,
 830                           req->func, req->overlapped, NULL );
 831     }
 832
 833     release_object( pipe );
 834 }
 835
 836 DECL_HANDLER(disconnect_named_pipe)
 837 {
 838     struct pipe_server *server;
 839
 840     reply->fd = -1;
 841     server = get_pipe_server_obj( current->process, req->handle, 0 );
 842     if (!server)
 843         return;
 844     switch(server->state)
 845     {
 846     case ps_connected_server:
 847         assert( server->fd );
 848         assert( server->client );
 849         assert( server->client->fd );
 850
 851         notify_empty( server );
 852
 853         /* Dump the client and server fds, but keep the pointers
 854            around - client loses all waiting data */
 855         server->state = ps_disconnected_server;
 856         do_disconnect( server );
 857         reply->fd = flush_cached_fd( current->process, req->handle );
 858         break;
 859
 860     case ps_wait_disconnect:
 861         assert( !server->client );
 862         assert( server->fd );
 863         do_disconnect( server );
 864         server->state = ps_wait_connect;
 865         reply->fd = flush_cached_fd( current->process, req->handle );
 866         break;
 867
 868     case ps_idle_server:
 869     case ps_wait_open:
 870     case ps_disconnected_server:
 871     case ps_wait_connect:
 872         set_error( STATUS_PIPE_DISCONNECTED );
 873         break;
 874     }
 875     release_object( server );
 876 }
 877
 878 DECL_HANDLER(get_named_pipe_info)
 879 {
 880     struct pipe_server *server;
 881
 882     server = get_pipe_server_obj( current->process, req->handle, 0 );
 883     if (!server)
 884         return;
 885
 886     reply->flags        = server->pipe->flags;
 887     reply->maxinstances = server->pipe->maxinstances;
 888     reply->insize       = server->pipe->insize;
 889     reply->outsize      = server->pipe->outsize;
 890
 891     release_object(server);
 892 }