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 if (get_error() == STATUS_ALERTED
)
343 /* give the client opportunity to complete synchronously. after the
344 * client performs the I/O, it reports the result back to the server
345 * via the set_async_direct_result request. if it turns out that the
346 * I/O request is not actually immediately satiable, the client may
347 * then choose to re-queue the async by reporting STATUS_PENDING
350 * since we're deferring the initial I/O (to the client), we mark the
351 * async as having unknown initial status (unknown_status = 1). note
352 * that we don't reuse async_set_unknown_status() here. this is because
353 * the one responsible for performing the I/O is not the device driver,
354 * but instead the client that requested the I/O in the first place.
356 * also, async_set_unknown_status() would set direct_result to zero
357 * forcing APC_ASYNC_IO to fire in async_terminate(), which is not
358 * useful due to subtle semantic differences between synchronous and
359 * asynchronous completion.
361 async
->unknown_status
= 1;
362 async_terminate( async
, STATUS_ALERTED
);
363 return async
->wait_handle
;
366 async
->initial_status
= get_error();
368 if (!async
->pending
&& NT_ERROR( get_error() ))
370 close_handle( async
->thread
->process
, async
->wait_handle
);
371 async
->wait_handle
= 0;
375 if (get_error() != STATUS_PENDING
)
377 /* status and data are already set and returned */
378 async_terminate( async
, get_error() );
380 else if (async
->iosb
->status
!= STATUS_PENDING
)
382 /* result is already available in iosb, return it */
383 if (async
->iosb
->out_data
)
385 set_reply_data_ptr( async
->iosb
->out_data
, async
->iosb
->out_size
);
386 async
->iosb
->out_data
= NULL
;
390 if (async
->iosb
->status
!= STATUS_PENDING
)
392 if (result
) *result
= async
->iosb
->result
;
397 async
->direct_result
= 0;
399 if (!async
->blocking
)
401 close_handle( async
->thread
->process
, async
->wait_handle
);
402 async
->wait_handle
= 0;
405 async
->initial_status
= async
->iosb
->status
;
406 set_error( async
->iosb
->status
);
407 return async
->wait_handle
;
410 /* complete a request-based async with a pre-allocated buffer */
411 void async_request_complete( struct async
*async
, unsigned int status
, data_size_t result
,
412 data_size_t out_size
, void *out_data
)
414 struct iosb
*iosb
= async_get_iosb( async
);
416 /* the async may have already been canceled */
417 if (iosb
->status
!= STATUS_PENDING
)
419 release_object( iosb
);
424 iosb
->status
= status
;
425 iosb
->result
= result
;
426 iosb
->out_data
= out_data
;
427 iosb
->out_size
= out_size
;
429 release_object( iosb
);
431 async_terminate( async
, status
);
434 /* complete a request-based async */
435 void async_request_complete_alloc( struct async
*async
, unsigned int status
, data_size_t result
,
436 data_size_t out_size
, const void *out_data
)
438 void *out_data_copy
= NULL
;
440 if (out_size
&& !(out_data_copy
= memdup( out_data
, out_size
)))
442 async_terminate( async
, STATUS_NO_MEMORY
);
446 async_request_complete( async
, status
, result
, out_size
, out_data_copy
);
449 /* mark an async as having unknown initial status */
450 void async_set_unknown_status( struct async
*async
)
452 async
->unknown_status
= 1;
453 async
->direct_result
= 0;
456 /* set the timeout of an async operation */
457 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
459 if (async
->timeout
) remove_timeout_user( async
->timeout
);
460 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
461 else async
->timeout
= NULL
;
462 async
->timeout_status
= status
;
465 /* set a callback to be notified when the async is completed */
466 void async_set_completion_callback( struct async
*async
, async_completion_callback func
, void *private )
468 async
->completion_callback
= func
;
469 async
->completion_callback_private
= private;
472 static void add_async_completion( struct async
*async
, apc_param_t cvalue
, unsigned int status
,
473 apc_param_t information
)
475 if (async
->fd
&& !async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
476 if (async
->completion
) add_completion( async
->completion
, async
->comp_key
, cvalue
, status
, information
);
479 /* store the result of the client-side async callback */
480 void async_set_result( struct object
*obj
, unsigned int status
, apc_param_t total
)
482 struct async
*async
= (struct async
*)obj
;
484 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
486 assert( async
->terminated
); /* it must have been woken up if we get a result */
488 if (async
->alerted
&& status
== STATUS_PENDING
) /* restart it */
490 async
->terminated
= 0;
492 async_reselect( async
);
496 if (async
->timeout
) remove_timeout_user( async
->timeout
);
497 async
->timeout
= NULL
;
498 async
->terminated
= 1;
499 if (async
->iosb
) async
->iosb
->status
= status
;
501 /* don't signal completion if the async failed synchronously
502 * this can happen if the initial status was unknown (i.e. for device files)
503 * note that we check the IOSB status here, not the initial status */
504 if (async
->pending
|| !NT_ERROR( status
))
509 memset( &data
, 0, sizeof(data
) );
510 data
.type
= APC_USER
;
511 data
.user
.func
= async
->data
.apc
;
512 data
.user
.args
[0] = async
->data
.apc_context
;
513 data
.user
.args
[1] = async
->data
.iosb
;
514 data
.user
.args
[2] = 0;
515 thread_queue_apc( NULL
, async
->thread
, NULL
, &data
);
517 else if (async
->data
.apc_context
&& (async
->pending
||
518 !(async
->comp_flags
& FILE_SKIP_COMPLETION_PORT_ON_SUCCESS
)))
520 add_async_completion( async
, async
->data
.apc_context
, status
, total
);
523 if (async
->event
) set_event( async
->event
);
524 else if (async
->fd
) set_fd_signaled( async
->fd
, 1 );
527 if (!async
->signaled
)
530 wake_up( &async
->obj
, 0 );
533 if (async
->completion_callback
)
534 async
->completion_callback( async
->completion_callback_private
);
535 async
->completion_callback
= NULL
;
539 list_remove( &async
->queue_entry
);
540 async_reselect( async
);
543 release_object( async
);
548 /* check if an async operation is waiting to be alerted */
549 int async_waiting( struct async_queue
*queue
)
554 if (!(ptr
= list_head( &queue
->queue
))) return 0;
555 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
556 return !async
->terminated
;
559 static int cancel_async( struct process
*process
, struct object
*obj
, struct thread
*thread
, client_ptr_t iosb
)
564 /* FIXME: it would probably be nice to replace the "canceled" flag with a
565 * single LIST_FOR_EACH_ENTRY_SAFE, but currently cancelling an async can
566 * cause other asyncs to be removed via async_reselect() */
569 LIST_FOR_EACH_ENTRY( async
, &process
->asyncs
, struct async
, process_entry
)
571 if (async
->terminated
|| async
->canceled
) continue;
572 if ((!obj
|| (get_fd_user( async
->fd
) == obj
)) &&
573 (!thread
|| async
->thread
== thread
) &&
574 (!iosb
|| async
->data
.iosb
== iosb
))
577 fd_cancel_async( async
->fd
, async
);
585 void cancel_process_asyncs( struct process
*process
)
587 cancel_async( process
, NULL
, NULL
, 0 );
590 /* wake up async operations on the queue */
591 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
593 struct list
*ptr
, *next
;
595 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
597 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
598 async_terminate( async
, status
);
599 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */
603 static void iosb_dump( struct object
*obj
, int verbose
);
604 static void iosb_destroy( struct object
*obj
);
606 static const struct object_ops iosb_ops
=
608 sizeof(struct iosb
), /* size */
610 iosb_dump
, /* dump */
611 no_add_queue
, /* add_queue */
612 NULL
, /* remove_queue */
614 NULL
, /* satisfied */
615 no_signal
, /* signal */
616 no_get_fd
, /* get_fd */
617 default_map_access
, /* map_access */
618 default_get_sd
, /* get_sd */
619 default_set_sd
, /* set_sd */
620 no_get_full_name
, /* get_full_name */
621 no_lookup_name
, /* lookup_name */
622 no_link_name
, /* link_name */
623 NULL
, /* unlink_name */
624 no_open_file
, /* open_file */
625 no_kernel_obj_list
, /* get_kernel_obj_list */
626 no_close_handle
, /* close_handle */
627 iosb_destroy
/* destroy */
630 static void iosb_dump( struct object
*obj
, int verbose
)
632 assert( obj
->ops
== &iosb_ops
);
633 fprintf( stderr
, "I/O status block\n" );
636 static void iosb_destroy( struct object
*obj
)
638 struct iosb
*iosb
= (struct iosb
*)obj
;
640 free( iosb
->in_data
);
641 free( iosb
->out_data
);
644 /* allocate iosb struct */
645 static struct iosb
*create_iosb( const void *in_data
, data_size_t in_size
, data_size_t out_size
)
649 if (!(iosb
= alloc_object( &iosb_ops
))) return NULL
;
651 iosb
->status
= STATUS_PENDING
;
653 iosb
->in_size
= in_size
;
654 iosb
->in_data
= NULL
;
655 iosb
->out_size
= out_size
;
656 iosb
->out_data
= NULL
;
658 if (in_size
&& !(iosb
->in_data
= memdup( in_data
, in_size
)))
660 release_object( iosb
);
667 /* create an async associated with iosb for async-based requests
668 * returned async must be passed to async_handoff */
669 struct async
*create_request_async( struct fd
*fd
, unsigned int comp_flags
, const async_data_t
*data
)
674 if (!(iosb
= create_iosb( get_req_data(), get_req_data_size(), get_reply_max_size() )))
677 async
= create_async( fd
, current
, data
, iosb
);
678 release_object( iosb
);
681 if (!(async
->wait_handle
= alloc_handle( current
->process
, async
, SYNCHRONIZE
, 0 )))
683 release_object( async
);
687 async
->direct_result
= 1;
688 async
->comp_flags
= comp_flags
;
693 struct iosb
*async_get_iosb( struct async
*async
)
695 return async
->iosb
? (struct iosb
*)grab_object( async
->iosb
) : NULL
;
698 struct thread
*async_get_thread( struct async
*async
)
700 return async
->thread
;
703 /* find the first pending async in queue */
704 struct async
*find_pending_async( struct async_queue
*queue
)
707 LIST_FOR_EACH_ENTRY( async
, &queue
->queue
, struct async
, queue_entry
)
708 if (!async
->terminated
) return (struct async
*)grab_object( async
);
712 /* cancels all async I/O */
713 DECL_HANDLER(cancel_async
)
715 struct object
*obj
= get_handle_obj( current
->process
, req
->handle
, 0, NULL
);
716 struct thread
*thread
= req
->only_thread
? current
: NULL
;
720 int count
= cancel_async( current
->process
, obj
, thread
, req
->iosb
);
721 if (!count
&& req
->iosb
) set_error( STATUS_NOT_FOUND
);
722 release_object( obj
);
726 /* get async result from associated iosb */
727 DECL_HANDLER(get_async_result
)
729 struct iosb
*iosb
= NULL
;
732 LIST_FOR_EACH_ENTRY( async
, ¤t
->process
->asyncs
, struct async
, process_entry
)
733 if (async
->data
.user
== req
->user_arg
)
741 set_error( STATUS_INVALID_PARAMETER
);
747 data_size_t size
= min( iosb
->out_size
, get_reply_max_size() );
750 set_reply_data_ptr( iosb
->out_data
, size
);
751 iosb
->out_data
= NULL
;
754 set_error( iosb
->status
);
757 /* notify direct completion of async and close the wait handle if not blocking */
758 DECL_HANDLER(set_async_direct_result
)
760 struct async
*async
= (struct async
*)get_handle_obj( current
->process
, req
->handle
, 0, &async_ops
);
761 unsigned int status
= req
->status
;
765 if (!async
->unknown_status
|| !async
->terminated
|| !async
->alerted
)
767 set_error( STATUS_INVALID_PARAMETER
);
768 release_object( &async
->obj
);
772 async_set_initial_status( async
, status
);
774 if (status
== STATUS_PENDING
)
776 async
->direct_result
= 0;
780 /* if the I/O has completed successfully, the client would have already
781 * set the IOSB. therefore, we can skip waiting on wait_handle and do
782 * async_set_result() directly.
784 async_set_result( &async
->obj
, status
, req
->information
);
786 /* close wait handle here to avoid extra server round trip, if the I/O
787 * either has completed, or is pending and not blocking.
789 if (status
!= STATUS_PENDING
|| !async
->blocking
)
791 close_handle( async
->thread
->process
, async
->wait_handle
);
792 async
->wait_handle
= 0;
795 /* report back to the client whether the wait handle has been closed.
796 * handle will be 0 if closed by us; otherwise the original value is
799 reply
->handle
= async
->wait_handle
;
801 release_object( &async
->obj
);