2 * Server-side async I/O support
4 * Copyright (C) 2007 Alexandre Julliard
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
27 #define WIN32_NO_STATUS
39 struct object obj
; /* object header */
40 struct thread
*thread
; /* owning thread */
41 struct list queue_entry
; /* entry in async queue list */
42 struct list process_entry
; /* entry in process list */
43 struct async_queue
*queue
; /* queue containing this async */
44 struct fd
*fd
; /* fd associated with an unqueued async */
45 struct timeout_user
*timeout
;
46 unsigned int timeout_status
; /* status to report upon timeout */
48 async_data_t data
; /* data for async I/O call */
49 struct iosb
*iosb
; /* I/O status block */
50 obj_handle_t wait_handle
; /* pre-allocated wait handle */
51 unsigned int initial_status
; /* status returned from initial request */
52 unsigned int signaled
:1;
53 unsigned int pending
:1; /* request successfully queued, but pending */
54 unsigned int direct_result
:1;/* a flag if we're passing result directly from request instead of APC */
55 unsigned int alerted
:1; /* fd is signaled, but we are waiting for client-side I/O */
56 unsigned int terminated
:1; /* async has been terminated */
57 unsigned int canceled
:1; /* have we already queued cancellation for this async? */
58 unsigned int unknown_status
:1; /* initial status is not known yet */
59 unsigned int blocking
:1; /* async is blocking */
60 struct completion
*completion
; /* completion associated with fd */
61 apc_param_t comp_key
; /* completion key associated with fd */
62 unsigned int comp_flags
; /* completion flags */
63 async_completion_callback completion_callback
; /* callback to be called on completion */
64 void *completion_callback_private
; /* argument to completion_callback */
67 static void async_dump( struct object
*obj
, int verbose
);
68 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
);
69 static void async_satisfied( struct object
* obj
, struct wait_queue_entry
*entry
);
70 static void async_destroy( struct object
*obj
);
72 static const struct object_ops async_ops
=
74 sizeof(struct async
), /* size */
76 async_dump
, /* dump */
77 add_queue
, /* add_queue */
78 remove_queue
, /* remove_queue */
79 async_signaled
, /* signaled */
80 async_satisfied
, /* satisfied */
81 no_signal
, /* signal */
82 no_get_fd
, /* get_fd */
83 default_map_access
, /* map_access */
84 default_get_sd
, /* get_sd */
85 default_set_sd
, /* set_sd */
86 no_get_full_name
, /* get_full_name */
87 no_lookup_name
, /* lookup_name */
88 no_link_name
, /* link_name */
89 NULL
, /* unlink_name */
90 no_open_file
, /* open_file */
91 no_kernel_obj_list
, /* get_kernel_obj_list */
92 no_close_handle
, /* close_handle */
93 async_destroy
/* destroy */
96 static inline void async_reselect( struct async
*async
)
98 if (async
->queue
&& async
->fd
) fd_reselect_async( async
->fd
, async
->queue
);
101 static void async_dump( struct object
*obj
, int verbose
)
103 struct async
*async
= (struct async
*)obj
;
104 assert( obj
->ops
== &async_ops
);
105 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
108 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
)
110 struct async
*async
= (struct async
*)obj
;
111 assert( obj
->ops
== &async_ops
);
112 return async
->signaled
;
115 static void async_satisfied( struct object
*obj
, struct wait_queue_entry
*entry
)
117 struct async
*async
= (struct async
*)obj
;
118 assert( obj
->ops
== &async_ops
);
120 /* we only return an async handle for asyncs created via create_request_async() */
121 assert( async
->iosb
);
123 if (async
->direct_result
)
125 async_set_result( &async
->obj
, async
->iosb
->status
, async
->iosb
->result
);
126 async
->direct_result
= 0;
129 if (async
->initial_status
== STATUS_PENDING
&& async
->blocking
)
130 set_wait_status( entry
, async
->iosb
->status
);
132 set_wait_status( entry
, async
->initial_status
);
134 /* close wait handle here to avoid extra server round trip */
135 if (async
->wait_handle
)
137 close_handle( async
->thread
->process
, async
->wait_handle
);
138 async
->wait_handle
= 0;
142 static void async_destroy( struct object
*obj
)
144 struct async
*async
= (struct async
*)obj
;
145 assert( obj
->ops
== &async_ops
);
147 list_remove( &async
->process_entry
);
151 list_remove( &async
->queue_entry
);
152 async_reselect( async
);
154 else if (async
->fd
) release_object( async
->fd
);
156 if (async
->timeout
) remove_timeout_user( async
->timeout
);
157 if (async
->completion
) release_object( async
->completion
);
158 if (async
->event
) release_object( async
->event
);
159 if (async
->iosb
) release_object( async
->iosb
);
160 release_object( async
->thread
);
163 /* notifies client thread of new status of its async request */
164 void async_terminate( struct async
*async
, unsigned int status
)
166 struct iosb
*iosb
= async
->iosb
;
168 if (async
->terminated
) return;
170 async
->terminated
= 1;
171 if (async
->iosb
&& async
->iosb
->status
== STATUS_PENDING
) async
->iosb
->status
= status
;
172 if (status
== STATUS_ALERTED
)
175 /* if no APC could be queued (e.g. the process is terminated),
176 * thread_queue_apc() may trigger async_set_result(), which may drop the
177 * last reference to the async, so grab a temporary reference here */
178 grab_object( async
);
180 if (!async
->direct_result
)
184 memset( &data
, 0, sizeof(data
) );
185 data
.type
= APC_ASYNC_IO
;
186 data
.async_io
.user
= async
->data
.user
;
187 data
.async_io
.result
= iosb
? iosb
->result
: 0;
189 /* this can happen if the initial status was unknown (i.e. for device
190 * files). the client should not fill the IOSB in this case; pass it as
191 * NULL to communicate that.
192 * note that we check the IOSB status and not the initial status */
193 if (NT_ERROR( status
) && (!is_fd_overlapped( async
->fd
) || !async
->pending
))
194 data
.async_io
.sb
= 0;
196 data
.async_io
.sb
= async
->data
.iosb
;
198 /* if there is output data, the client needs to make an extra request
199 * to retrieve it; use STATUS_ALERTED to signal this case */
200 if (iosb
&& iosb
->out_data
)
201 data
.async_io
.status
= STATUS_ALERTED
;
203 data
.async_io
.status
= status
;
205 thread_queue_apc( async
->thread
->process
, async
->thread
, &async
->obj
, &data
);
208 async_reselect( async
);
210 release_object( async
);
213 /* callback for timeout on an async request */
214 static void async_timeout( void *private )
216 struct async
*async
= private;
218 async
->timeout
= NULL
;
219 async_terminate( async
, async
->timeout_status
);
222 /* free an async queue, cancelling all async operations */
223 void free_async_queue( struct async_queue
*queue
)
225 struct async
*async
, *next
;
227 LIST_FOR_EACH_ENTRY_SAFE( async
, next
, &queue
->queue
, struct async
, queue_entry
)
229 if (!async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
231 async_terminate( async
, STATUS_HANDLES_CLOSED
);
233 release_object( &async
->obj
);
237 void queue_async( struct async_queue
*queue
, struct async
*async
)
239 /* fd will be set to NULL in free_async_queue when fd is destroyed */
240 release_object( async
->fd
);
242 async
->queue
= queue
;
243 grab_object( async
);
244 list_add_tail( &queue
->queue
, &async
->queue_entry
);
246 set_fd_signaled( async
->fd
, 0 );
249 /* create an async on a given queue of a fd */
250 struct async
*create_async( struct fd
*fd
, struct thread
*thread
, const async_data_t
*data
, struct iosb
*iosb
)
252 struct event
*event
= NULL
;
255 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
258 if (!(async
= alloc_object( &async_ops
)))
260 if (event
) release_object( event
);
264 async
->thread
= (struct thread
*)grab_object( thread
);
265 async
->event
= event
;
267 async
->timeout
= NULL
;
269 async
->fd
= (struct fd
*)grab_object( fd
);
270 async
->initial_status
= STATUS_PENDING
;
273 async
->wait_handle
= 0;
274 async
->direct_result
= 0;
276 async
->terminated
= 0;
278 async
->unknown_status
= 0;
279 async
->blocking
= !is_fd_overlapped( fd
);
280 async
->completion
= fd_get_completion( fd
, &async
->comp_key
);
281 async
->comp_flags
= 0;
282 async
->completion_callback
= NULL
;
283 async
->completion_callback_private
= NULL
;
285 if (iosb
) async
->iosb
= (struct iosb
*)grab_object( iosb
);
286 else async
->iosb
= NULL
;
288 list_add_head( &thread
->process
->asyncs
, &async
->process_entry
);
289 if (event
) reset_event( event
);
291 if (async
->completion
&& data
->apc
)
293 release_object( async
);
294 set_error( STATUS_INVALID_PARAMETER
);
301 /* set the initial status of an async whose status was previously unknown
302 * the initial status may be STATUS_PENDING */
303 void async_set_initial_status( struct async
*async
, unsigned int status
)
305 assert( async
->unknown_status
);
306 if (!async
->terminated
)
308 async
->initial_status
= status
;
309 async
->unknown_status
= 0;
313 void set_async_pending( struct async
*async
)
315 if (!async
->terminated
)
319 void async_wake_obj( struct async
*async
)
321 assert( !async
->unknown_status
);
322 if (!async
->blocking
)
325 wake_up( &async
->obj
, 0 );
329 /* return async object status and wait handle to client */
330 obj_handle_t
async_handoff( struct async
*async
, data_size_t
*result
, int force_blocking
)
332 async
->blocking
= force_blocking
|| async
->blocking
;
334 if (async
->unknown_status
)
336 /* even the initial status is not known yet */
337 set_error( STATUS_PENDING
);
338 return async
->wait_handle
;
341 async
->initial_status
= get_error();
343 if (!async
->pending
&& NT_ERROR( get_error() ))
345 close_handle( async
->thread
->process
, async
->wait_handle
);
346 async
->wait_handle
= 0;
350 if (get_error() != STATUS_PENDING
)
352 /* status and data are already set and returned */
353 async_terminate( async
, get_error() );
355 else if (async
->iosb
->status
!= STATUS_PENDING
)
357 /* result is already available in iosb, return it */
358 if (async
->iosb
->out_data
)
360 set_reply_data_ptr( async
->iosb
->out_data
, async
->iosb
->out_size
);
361 async
->iosb
->out_data
= NULL
;
365 if (async
->iosb
->status
!= STATUS_PENDING
)
367 if (result
) *result
= async
->iosb
->result
;
372 async
->direct_result
= 0;
374 if (!async
->blocking
)
376 close_handle( async
->thread
->process
, async
->wait_handle
);
377 async
->wait_handle
= 0;
380 async
->initial_status
= async
->iosb
->status
;
381 set_error( async
->iosb
->status
);
382 return async
->wait_handle
;
385 /* complete a request-based async with a pre-allocated buffer */
386 void async_request_complete( struct async
*async
, unsigned int status
, data_size_t result
,
387 data_size_t out_size
, void *out_data
)
389 struct iosb
*iosb
= async_get_iosb( async
);
391 /* the async may have already been canceled */
392 if (iosb
->status
!= STATUS_PENDING
)
394 release_object( iosb
);
399 iosb
->status
= status
;
400 iosb
->result
= result
;
401 iosb
->out_data
= out_data
;
402 iosb
->out_size
= out_size
;
404 release_object( iosb
);
406 async_terminate( async
, status
);
409 /* complete a request-based async */
410 void async_request_complete_alloc( struct async
*async
, unsigned int status
, data_size_t result
,
411 data_size_t out_size
, const void *out_data
)
413 void *out_data_copy
= NULL
;
415 if (out_size
&& !(out_data_copy
= memdup( out_data
, out_size
)))
417 async_terminate( async
, STATUS_NO_MEMORY
);
421 async_request_complete( async
, status
, result
, out_size
, out_data_copy
);
424 /* mark an async as having unknown initial status */
425 void async_set_unknown_status( struct async
*async
)
427 async
->unknown_status
= 1;
428 async
->direct_result
= 0;
431 /* set the timeout of an async operation */
432 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
434 if (async
->timeout
) remove_timeout_user( async
->timeout
);
435 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
436 else async
->timeout
= NULL
;
437 async
->timeout_status
= status
;
440 /* set a callback to be notified when the async is completed */
441 void async_set_completion_callback( struct async
*async
, async_completion_callback func
, void *private )
443 async
->completion_callback
= func
;
444 async
->completion_callback_private
= private;
447 static void add_async_completion( struct async
*async
, apc_param_t cvalue
, unsigned int status
,
448 apc_param_t information
)
450 if (async
->fd
&& !async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
451 if (async
->completion
) add_completion( async
->completion
, async
->comp_key
, cvalue
, status
, information
);
454 /* store the result of the client-side async callback */
455 void async_set_result( struct object
*obj
, unsigned int status
, apc_param_t total
)
457 struct async
*async
= (struct async
*)obj
;
459 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
461 assert( async
->terminated
); /* it must have been woken up if we get a result */
463 if (async
->alerted
&& status
== STATUS_PENDING
) /* restart it */
465 async
->terminated
= 0;
467 async_reselect( async
);
471 if (async
->timeout
) remove_timeout_user( async
->timeout
);
472 async
->timeout
= NULL
;
473 async
->terminated
= 1;
474 if (async
->iosb
) async
->iosb
->status
= status
;
476 /* don't signal completion if the async failed synchronously
477 * this can happen if the initial status was unknown (i.e. for device files)
478 * note that we check the IOSB status here, not the initial status */
479 if (async
->pending
|| !NT_ERROR( status
))
484 memset( &data
, 0, sizeof(data
) );
485 data
.type
= APC_USER
;
486 data
.user
.func
= async
->data
.apc
;
487 data
.user
.args
[0] = async
->data
.apc_context
;
488 data
.user
.args
[1] = async
->data
.iosb
;
489 data
.user
.args
[2] = 0;
490 thread_queue_apc( NULL
, async
->thread
, NULL
, &data
);
492 else if (async
->data
.apc_context
&& (async
->pending
||
493 !(async
->comp_flags
& FILE_SKIP_COMPLETION_PORT_ON_SUCCESS
)))
495 add_async_completion( async
, async
->data
.apc_context
, status
, total
);
498 if (async
->event
) set_event( async
->event
);
499 else if (async
->fd
) set_fd_signaled( async
->fd
, 1 );
502 if (!async
->signaled
)
505 wake_up( &async
->obj
, 0 );
508 if (async
->completion_callback
)
509 async
->completion_callback( async
->completion_callback_private
);
510 async
->completion_callback
= NULL
;
514 list_remove( &async
->queue_entry
);
515 async_reselect( async
);
518 release_object( async
);
523 /* check if an async operation is waiting to be alerted */
524 int async_waiting( struct async_queue
*queue
)
529 if (!(ptr
= list_head( &queue
->queue
))) return 0;
530 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
531 return !async
->terminated
;
534 static int cancel_async( struct process
*process
, struct object
*obj
, struct thread
*thread
, client_ptr_t iosb
)
539 /* FIXME: it would probably be nice to replace the "canceled" flag with a
540 * single LIST_FOR_EACH_ENTRY_SAFE, but currently cancelling an async can
541 * cause other asyncs to be removed via async_reselect() */
544 LIST_FOR_EACH_ENTRY( async
, &process
->asyncs
, struct async
, process_entry
)
546 if (async
->terminated
|| async
->canceled
) continue;
547 if ((!obj
|| (get_fd_user( async
->fd
) == obj
)) &&
548 (!thread
|| async
->thread
== thread
) &&
549 (!iosb
|| async
->data
.iosb
== iosb
))
552 fd_cancel_async( async
->fd
, async
);
560 void cancel_process_asyncs( struct process
*process
)
562 cancel_async( process
, NULL
, NULL
, 0 );
565 /* wake up async operations on the queue */
566 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
568 struct list
*ptr
, *next
;
570 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
572 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
573 async_terminate( async
, status
);
574 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */
578 static void iosb_dump( struct object
*obj
, int verbose
);
579 static void iosb_destroy( struct object
*obj
);
581 static const struct object_ops iosb_ops
=
583 sizeof(struct iosb
), /* size */
585 iosb_dump
, /* dump */
586 no_add_queue
, /* add_queue */
587 NULL
, /* remove_queue */
589 NULL
, /* satisfied */
590 no_signal
, /* signal */
591 no_get_fd
, /* get_fd */
592 default_map_access
, /* map_access */
593 default_get_sd
, /* get_sd */
594 default_set_sd
, /* set_sd */
595 no_get_full_name
, /* get_full_name */
596 no_lookup_name
, /* lookup_name */
597 no_link_name
, /* link_name */
598 NULL
, /* unlink_name */
599 no_open_file
, /* open_file */
600 no_kernel_obj_list
, /* get_kernel_obj_list */
601 no_close_handle
, /* close_handle */
602 iosb_destroy
/* destroy */
605 static void iosb_dump( struct object
*obj
, int verbose
)
607 assert( obj
->ops
== &iosb_ops
);
608 fprintf( stderr
, "I/O status block\n" );
611 static void iosb_destroy( struct object
*obj
)
613 struct iosb
*iosb
= (struct iosb
*)obj
;
615 free( iosb
->in_data
);
616 free( iosb
->out_data
);
619 /* allocate iosb struct */
620 static struct iosb
*create_iosb( const void *in_data
, data_size_t in_size
, data_size_t out_size
)
624 if (!(iosb
= alloc_object( &iosb_ops
))) return NULL
;
626 iosb
->status
= STATUS_PENDING
;
628 iosb
->in_size
= in_size
;
629 iosb
->in_data
= NULL
;
630 iosb
->out_size
= out_size
;
631 iosb
->out_data
= NULL
;
633 if (in_size
&& !(iosb
->in_data
= memdup( in_data
, in_size
)))
635 release_object( iosb
);
642 /* create an async associated with iosb for async-based requests
643 * returned async must be passed to async_handoff */
644 struct async
*create_request_async( struct fd
*fd
, unsigned int comp_flags
, const async_data_t
*data
)
649 if (!(iosb
= create_iosb( get_req_data(), get_req_data_size(), get_reply_max_size() )))
652 async
= create_async( fd
, current
, data
, iosb
);
653 release_object( iosb
);
656 if (!(async
->wait_handle
= alloc_handle( current
->process
, async
, SYNCHRONIZE
, 0 )))
658 release_object( async
);
662 async
->direct_result
= 1;
663 async
->comp_flags
= comp_flags
;
668 struct iosb
*async_get_iosb( struct async
*async
)
670 return async
->iosb
? (struct iosb
*)grab_object( async
->iosb
) : NULL
;
673 struct thread
*async_get_thread( struct async
*async
)
675 return async
->thread
;
678 /* find the first pending async in queue */
679 struct async
*find_pending_async( struct async_queue
*queue
)
682 LIST_FOR_EACH_ENTRY( async
, &queue
->queue
, struct async
, queue_entry
)
683 if (!async
->terminated
) return (struct async
*)grab_object( async
);
687 /* cancels all async I/O */
688 DECL_HANDLER(cancel_async
)
690 struct object
*obj
= get_handle_obj( current
->process
, req
->handle
, 0, NULL
);
691 struct thread
*thread
= req
->only_thread
? current
: NULL
;
695 int count
= cancel_async( current
->process
, obj
, thread
, req
->iosb
);
696 if (!count
&& req
->iosb
) set_error( STATUS_NOT_FOUND
);
697 release_object( obj
);
701 /* get async result from associated iosb */
702 DECL_HANDLER(get_async_result
)
704 struct iosb
*iosb
= NULL
;
707 LIST_FOR_EACH_ENTRY( async
, ¤t
->process
->asyncs
, struct async
, process_entry
)
708 if (async
->data
.user
== req
->user_arg
)
716 set_error( STATUS_INVALID_PARAMETER
);
722 data_size_t size
= min( iosb
->out_size
, get_reply_max_size() );
725 set_reply_data_ptr( iosb
->out_data
, size
);
726 iosb
->out_data
= NULL
;
729 set_error( iosb
->status
);