server: Hold a pointer to the queue from the async operations.

server/async.c

@@ -36,7 +36,8 @@ struct async
 {
     struct object        obj;             /* object header */
     struct thread       *thread;          /* owning thread */
-    struct list          queue_entry;     /* entry in file descriptor queue */
+    struct list          queue_entry;     /* entry in async queue list */
+    struct async_queue  *queue;           /* queue containing this async */
     struct timeout_user *timeout;
     unsigned int         timeout_status;  /* status to report upon timeout */
     struct event        *event;
@@ -72,7 +73,6 @@ struct async_queue
 };
 
 static void async_queue_dump( struct object *obj, int verbose );
-static void async_queue_destroy( struct object *obj );
 
 static const struct object_ops async_queue_ops =
 {
@@ -88,7 +88,7 @@ static const struct object_ops async_queue_ops =
     no_lookup_name,                  /* lookup_name */
     no_open_file,                    /* open_file */
     no_close_handle,                 /* close_handle */
-    async_queue_destroy              /* destroy */
+    no_destroy                       /* destroy */
 };
 
 
@@ -106,6 +106,8 @@ static void async_destroy( struct object *obj )
 
     if (async->timeout) remove_timeout_user( async->timeout );
     if (async->event) release_object( async->event );
+    release_object( async->queue );
+    async->queue = NULL;
     release_object( async->thread );
 }
 
@@ -116,14 +118,6 @@ static void async_queue_dump( struct object *obj, int verbose )
     fprintf( stderr, "Async queue fd=%p\n", async_queue->fd );
 }
 
-static void async_queue_destroy( struct object *obj )
-{
-    struct async_queue *async_queue = (struct async_queue *)obj;
-    assert( obj->ops == &async_queue_ops );
-
-    async_wake_up( async_queue, STATUS_HANDLES_CLOSED );
-}
-
 /* notifies client thread of new status of its async request */
 /* destroys the server side of it */
 static void async_terminate( struct async *async, unsigned int status )
@@ -166,6 +160,15 @@ struct async_queue *create_async_queue( struct fd *fd )
     return queue;
 }
 
+/* free an async queue, cancelling all async operations */
+void free_async_queue( struct async_queue *queue )
+{
+    if (!queue) return;
+    queue->fd = NULL;
+    async_wake_up( queue, STATUS_HANDLES_CLOSED );
+    release_object( queue );
+}
+
 /* create an async on a given queue of a fd */
 struct async *create_async( struct thread *thread, struct async_queue *queue, const async_data_t *data )
 {
@@ -185,6 +188,7 @@ struct async *create_async( struct thread *thread, struct async_queue *queue, co
     async->event = event;
     async->data = *data;
     async->timeout = NULL;
+    async->queue = (struct async_queue *)grab_object( queue );
 
     list_add_tail( &queue->queue, &async->queue_entry );
     grab_object( async );

server/fd.c

@@ -1286,9 +1286,9 @@ static void fd_destroy( struct object *obj )
 {
     struct fd *fd = (struct fd *)obj;
 
-    if (fd->read_q) release_object( fd->read_q );
-    if (fd->write_q) release_object( fd->write_q );
-    if (fd->wait_q) release_object( fd->wait_q );
+    free_async_queue( fd->read_q );
+    free_async_queue( fd->write_q );
+    free_async_queue( fd->wait_q );
 
     remove_fd_locks( fd );
     list_remove( &fd->inode_entry );

server/file.h

@@ -124,6 +124,7 @@ extern struct object *create_serial( struct fd *fd, unsigned int options );
 
 /* async I/O functions */
 extern struct async_queue *create_async_queue( struct fd *fd );
+extern void free_async_queue( struct async_queue *queue );
 extern struct async *create_async( struct thread *thread, struct async_queue *queue,
                                    const async_data_t *data );
 extern void async_set_timeout( struct async *async, const struct timeval *timeout,

server/named_pipe.c

@@ -274,7 +274,7 @@ static void named_pipe_destroy( struct object *obj)
 
     assert( list_empty( &pipe->servers ) );
     assert( !pipe->instances );
-    if (pipe->waiters) release_object( pipe->waiters );
+    free_async_queue( pipe->waiters );
 }
 
 static struct fd *pipe_client_get_fd( struct object *obj )
@@ -366,7 +366,7 @@ static void pipe_server_destroy( struct object *obj)
         server->client = NULL;
     }
 
-    release_object( server->wait_q );
+    free_async_queue( server->wait_q );
 
     assert( server->pipe->instances );
     server->pipe->instances--;

server/sock.c

@@ -581,8 +581,8 @@ static void sock_destroy( struct object *obj )
     if ( sock->deferred )
         release_object( sock->deferred );
 
-    if (sock->read_q) release_object( sock->read_q );
-    if (sock->write_q) release_object( sock->write_q );
+    free_async_queue( sock->read_q );
+    free_async_queue( sock->write_q );
     if (sock->event) release_object( sock->event );
     if (sock->fd)
     {