server: Add a separate function to set the timeout of an async I/O operation.

server/async.c

@@ -166,8 +166,7 @@ struct async_queue *create_async_queue( struct fd *fd )
 }
 
 /* create an async on a given queue of a fd */
-struct async *create_async( struct thread *thread, const struct timeval *timeout,
-                            struct async_queue *queue, const async_data_t *data )
+struct async *create_async( struct thread *thread, struct async_queue *queue, const async_data_t *data )
 {
     struct event *event = NULL;
     struct async *async;
@@ -184,16 +183,23 @@ struct async *create_async( struct thread *thread, const struct timeval *timeout
     async->thread = (struct thread *)grab_object( thread );
     async->event = event;
     async->data = *data;
+    async->timeout = NULL;
 
     list_add_tail( &queue->queue, &async->queue_entry );
-
-    if (timeout) async->timeout = add_timeout_user( timeout, async_timeout, async );
-    else async->timeout = NULL;
+    grab_object( async );
 
     if (event) reset_event( event );
     return async;
 }
 
+/* set the timeout of an async operation */
+void async_set_timeout( struct async *async, const struct timeval *timeout )
+{
+    if (async->timeout) remove_timeout_user( async->timeout );
+    if (timeout) async->timeout = add_timeout_user( timeout, async_timeout, async );
+    else async->timeout = NULL;
+}
+
 /* store the result of the client-side async callback */
 void async_set_result( struct object *obj, unsigned int status )
 {

server/change.c

@@ -1058,6 +1058,7 @@ DECL_HANDLER(read_directory_changes)
 {
     struct event *event = NULL;
     struct dir *dir;
+    struct async *async;
 
     if (!req->filter)
     {
@@ -1079,7 +1080,7 @@ DECL_HANDLER(read_directory_changes)
     dir->event = event;
 
     /* requests don't timeout */
-    if (!fd_queue_async_timeout( dir->fd, &req->async, ASYNC_TYPE_WAIT, 0, NULL )) goto end;
+    if (!(async = fd_queue_async( dir->fd, &req->async, ASYNC_TYPE_WAIT, 0 ))) goto end;
 
     /* assign it once */
     if (!dir->filter)
@@ -1103,6 +1104,7 @@ DECL_HANDLER(read_directory_changes)
     if (!inotify_adjust_changes( dir ))
         dnotify_adjust_changes( dir );
 
+    release_object( async );
     set_error(STATUS_PENDING);
 
 end:

server/fd.c

@@ -1708,38 +1708,37 @@ void default_poll_event( struct fd *fd, int event )
     wake_up( fd->user, 0 );
 }
 
-int fd_queue_async_timeout( struct fd *fd, const async_data_t *data, int type, int count,
-                            const struct timeval *timeout )
+struct async *fd_queue_async( struct fd *fd, const async_data_t *data, int type, int count )
 {
     struct async_queue *queue;
+    struct async *async;
 
     switch (type)
     {
     case ASYNC_TYPE_READ:
-        if (!fd->read_q && !(fd->read_q = create_async_queue( fd ))) return 0;
+        if (!fd->read_q && !(fd->read_q = create_async_queue( fd ))) return NULL;
         queue = fd->read_q;
         break;
     case ASYNC_TYPE_WRITE:
-        if (!fd->write_q && !(fd->write_q = create_async_queue( fd ))) return 0;
+        if (!fd->write_q && !(fd->write_q = create_async_queue( fd ))) return NULL;
         queue = fd->write_q;
         break;
     case ASYNC_TYPE_WAIT:
-        if (!fd->wait_q && !(fd->wait_q = create_async_queue( fd ))) return 0;
+        if (!fd->wait_q && !(fd->wait_q = create_async_queue( fd ))) return NULL;
         queue = fd->wait_q;
         break;
     default:
         assert(0);
     }
 
-    if (!create_async( current, timeout, queue, data )) return 0;
-    set_error( STATUS_PENDING );
-
-    if (!fd->inode)
-        set_fd_events( fd, fd->fd_ops->get_poll_events( fd ) );
-    else  /* regular files are always ready for read and write */
-        if (type != ASYNC_TYPE_WAIT) async_wake_up( queue, STATUS_ALERTED );
-
-    return 1;
+    if ((async = create_async( current, queue, data )))
+    {
+        if (!fd->inode)
+            set_fd_events( fd, fd->fd_ops->get_poll_events( fd ) );
+        else  /* regular files are always ready for read and write */
+            if (type != ASYNC_TYPE_WAIT) async_wake_up( queue, STATUS_ALERTED );
+    }
+    return async;
 }
 
 void fd_async_wake_up( struct fd *fd, int type, unsigned int status )
@@ -1763,6 +1762,7 @@ void fd_async_wake_up( struct fd *fd, int type, unsigned int status )
 void default_fd_queue_async( struct fd *fd, const async_data_t *data, int type, int count )
 {
     int flags;
+    struct async *async;
 
     fd->fd_ops->get_file_info( fd, &flags );
     if (!(flags & (FD_FLAG_OVERLAPPED|FD_FLAG_TIMEOUT)))
@@ -1770,7 +1770,11 @@ void default_fd_queue_async( struct fd *fd, const async_data_t *data, int type,
         set_error( STATUS_INVALID_HANDLE );
         return;
     }
-    fd_queue_async_timeout( fd, data, type, count, NULL );
+    if ((async = fd_queue_async( fd, data, type, count )))
+    {
+        release_object( async );
+        set_error( STATUS_PENDING );
+    }
 }
 
 void default_fd_cancel_async( struct fd *fd )

server/file.h

@@ -69,8 +69,7 @@ extern void default_fd_remove_queue( struct object *obj, struct wait_queue_entry
 extern int default_fd_signaled( struct object *obj, struct thread *thread );
 extern int default_fd_get_poll_events( struct fd *fd );
 extern void default_poll_event( struct fd *fd, int event );
-extern int fd_queue_async_timeout( struct fd *fd, const async_data_t *data, int type,
-                                   int count, const struct timeval *timeout );
+extern struct async *fd_queue_async( struct fd *fd, const async_data_t *data, int type, int count );
 extern void fd_async_wake_up( struct fd *fd, int type, unsigned int status );
 extern void default_fd_queue_async( struct fd *fd, const async_data_t *data, int type, int count );
 extern void default_fd_cancel_async( struct fd *fd );
@@ -125,8 +124,9 @@ 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 struct async *create_async( struct thread *thread, const struct timeval *timeout,
-                                   struct async_queue *queue, const async_data_t *data );
+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 );
 extern void async_set_result( struct object *obj, unsigned int status );
 extern int async_waiting( struct async_queue *queue );
 extern void async_wake_up( struct async_queue *queue, unsigned int status );

server/mailslot.c

@@ -282,6 +282,7 @@ static struct object *mailslot_open_file( struct object *obj, unsigned int acces
 static void mailslot_queue_async( struct fd *fd, const async_data_t *data, int type, int count )
 {
     struct mailslot *mailslot = get_fd_user( fd );
+    struct async *async;
 
     assert(mailslot->obj.ops == &mailslot_ops);
 
@@ -292,13 +293,17 @@ static void mailslot_queue_async( struct fd *fd, const async_data_t *data, int t
         return;
     }
 
-    if (mailslot->read_timeout != -1)
+    if ((async = fd_queue_async( fd, data, type, count )))
     {
-        struct timeval when = current_time;
-        add_timeout( &when, max(1,mailslot->read_timeout) );
-        fd_queue_async_timeout( fd, data, type, count, &when );
+        if (mailslot->read_timeout != -1)
+        {
+            struct timeval when = current_time;
+            add_timeout( &when, max(1,mailslot->read_timeout) );
+            async_set_timeout( async, &when );
+        }
+        release_object( async );
+        set_error( STATUS_PENDING );
     }
-    else fd_queue_async_timeout( fd, data, type, count, NULL );
 }
 
 static void mailslot_device_dump( struct object *obj, int verbose )

server/named_pipe.c

@@ -793,6 +793,7 @@ DECL_HANDLER(create_named_pipe)
 DECL_HANDLER(connect_named_pipe)
 {
     struct pipe_server *server;
+    struct async *async;
 
     server = get_pipe_server_obj(current->process, req->handle, 0);
     if (!server)
@@ -804,9 +805,12 @@ DECL_HANDLER(connect_named_pipe)
     case ps_wait_connect:
         assert( !server->fd );
         server->state = ps_wait_open;
-        create_async( current, NULL, server->wait_q, &req->async );
-        if (server->pipe->waiters) async_wake_up( server->pipe->waiters, STATUS_SUCCESS );
-        set_error( STATUS_PENDING );
+        if ((async = create_async( current, server->wait_q, &req->async )))
+        {
+            if (server->pipe->waiters) async_wake_up( server->pipe->waiters, STATUS_SUCCESS );
+            release_object( async );
+            set_error( STATUS_PENDING );
+        }
         break;
     case ps_connected_server:
         assert( server->fd );
@@ -848,23 +852,25 @@ DECL_HANDLER(wait_named_pipe)
     server = find_available_server( pipe );
     if (!server)
     {
+        struct async *async;
+
         if (!pipe->waiters && !(pipe->waiters = create_async_queue( NULL )))
         {
             release_object( pipe );
             return;
         }
-        if (req->timeout == NMPWAIT_WAIT_FOREVER)
-        {
-            if (create_async( current, NULL, pipe->waiters, &req->async ))
-                set_error( STATUS_PENDING );
-        }
-        else
+
+        if ((async = create_async( current, pipe->waiters, &req->async )))
         {
-            struct timeval when = current_time;
-            if (req->timeout == NMPWAIT_USE_DEFAULT_WAIT) add_timeout( &when, pipe->timeout );
-            else add_timeout( &when, req->timeout );
-            if (create_async( current, &when, pipe->waiters, &req->async ))
-                set_error( STATUS_PENDING );
+            if (req->timeout != NMPWAIT_WAIT_FOREVER)
+            {
+                struct timeval when = current_time;
+                if (req->timeout == NMPWAIT_USE_DEFAULT_WAIT) add_timeout( &when, pipe->timeout );
+                else add_timeout( &when, req->timeout );
+                async_set_timeout( async, &when );
+            }
+            release_object( async );
+            set_error( STATUS_PENDING );
         }
     }
     else release_object( server );

server/serial.c

@@ -198,8 +198,8 @@ static enum server_fd_type serial_get_info( struct fd *fd, int *flags )
 static void serial_queue_async( struct fd *fd, const async_data_t *data, int type, int count )
 {
     struct serial *serial = get_fd_user( fd );
-    struct timeval when = current_time;
     int timeout = 0;
+    struct async *async;
 
     assert(serial->obj.ops == &serial_ops);
 
@@ -213,8 +213,17 @@ static void serial_queue_async( struct fd *fd, const async_data_t *data, int typ
         break;
     }
 
-    add_timeout( &when, timeout );
-    fd_queue_async_timeout( fd, data, type, count, timeout ? &when : NULL );
+    if ((async = fd_queue_async( fd, data, type, count )))
+    {
+        if (timeout)
+        {
+            struct timeval when = current_time;
+            add_timeout( &when, timeout );
+            async_set_timeout( async, &when );
+        }
+        release_object( async );
+        set_error( STATUS_PENDING );
+    }
 }
 
 static void serial_flush( struct fd *fd, struct event **event )

server/sock.c

@@ -546,7 +546,9 @@ static void sock_queue_async( struct fd *fd, const async_data_t *data, int type,
     }
     else
     {
-        if (!create_async( current, NULL, queue, data )) return;
+        struct async *async;
+        if (!(async = create_async( current, queue, data ))) return;
+        release_object( async );
         set_error( STATUS_PENDING );
     }