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 unsigned int status
; /* current status */
46 struct timeout_user
*timeout
;
47 unsigned int timeout_status
; /* status to report upon timeout */
49 async_data_t data
; /* data for async I/O call */
50 struct iosb
*iosb
; /* I/O status block */
51 obj_handle_t wait_handle
; /* pre-allocated wait handle */
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 struct completion
*completion
; /* completion associated with fd */
56 apc_param_t comp_key
; /* completion key associated with fd */
57 unsigned int comp_flags
; /* completion flags */
60 static void async_dump( struct object
*obj
, int verbose
);
61 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
);
62 static void async_satisfied( struct object
* obj
, struct wait_queue_entry
*entry
);
63 static void async_destroy( struct object
*obj
);
65 static const struct object_ops async_ops
=
67 sizeof(struct async
), /* size */
69 async_dump
, /* dump */
70 add_queue
, /* add_queue */
71 remove_queue
, /* remove_queue */
72 async_signaled
, /* signaled */
73 async_satisfied
, /* satisfied */
74 no_signal
, /* signal */
75 no_get_fd
, /* get_fd */
76 default_map_access
, /* map_access */
77 default_get_sd
, /* get_sd */
78 default_set_sd
, /* set_sd */
79 no_get_full_name
, /* get_full_name */
80 no_lookup_name
, /* lookup_name */
81 no_link_name
, /* link_name */
82 NULL
, /* unlink_name */
83 no_open_file
, /* open_file */
84 no_kernel_obj_list
, /* get_kernel_obj_list */
85 no_close_handle
, /* close_handle */
86 async_destroy
/* destroy */
89 static inline void async_reselect( struct async
*async
)
91 if (async
->queue
&& async
->fd
) fd_reselect_async( async
->fd
, async
->queue
);
94 static void async_dump( struct object
*obj
, int verbose
)
96 struct async
*async
= (struct async
*)obj
;
97 assert( obj
->ops
== &async_ops
);
98 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
101 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
)
103 struct async
*async
= (struct async
*)obj
;
104 assert( obj
->ops
== &async_ops
);
105 return async
->signaled
;
108 static void async_satisfied( struct object
*obj
, struct wait_queue_entry
*entry
)
110 struct async
*async
= (struct async
*)obj
;
111 assert( obj
->ops
== &async_ops
);
113 if (async
->direct_result
)
115 async_set_result( &async
->obj
, async
->iosb
->status
, async
->iosb
->result
);
116 async
->direct_result
= 0;
119 /* close wait handle here to avoid extra server round trip */
120 if (async
->wait_handle
)
122 close_handle( async
->thread
->process
, async
->wait_handle
);
123 async
->wait_handle
= 0;
126 if (async
->status
== STATUS_PENDING
) make_wait_abandoned( entry
);
129 static void async_destroy( struct object
*obj
)
131 struct async
*async
= (struct async
*)obj
;
132 assert( obj
->ops
== &async_ops
);
134 list_remove( &async
->process_entry
);
138 list_remove( &async
->queue_entry
);
139 async_reselect( async
);
141 else if (async
->fd
) release_object( async
->fd
);
143 if (async
->timeout
) remove_timeout_user( async
->timeout
);
144 if (async
->completion
) release_object( async
->completion
);
145 if (async
->event
) release_object( async
->event
);
146 if (async
->iosb
) release_object( async
->iosb
);
147 release_object( async
->thread
);
150 /* notifies client thread of new status of its async request */
151 void async_terminate( struct async
*async
, unsigned int status
)
153 assert( status
!= STATUS_PENDING
);
155 if (async
->status
!= STATUS_PENDING
)
157 /* already terminated, just update status */
158 async
->status
= status
;
162 async
->status
= status
;
163 if (async
->iosb
&& async
->iosb
->status
== STATUS_PENDING
) async
->iosb
->status
= status
;
165 if (!async
->direct_result
)
167 if (async
->data
.user
)
171 memset( &data
, 0, sizeof(data
) );
172 data
.type
= APC_ASYNC_IO
;
173 data
.async_io
.user
= async
->data
.user
;
174 data
.async_io
.sb
= async
->data
.iosb
;
175 data
.async_io
.status
= status
;
176 thread_queue_apc( async
->thread
->process
, async
->thread
, &async
->obj
, &data
);
178 else async_set_result( &async
->obj
, STATUS_SUCCESS
, 0 );
181 async_reselect( async
);
182 if (async
->queue
) release_object( async
); /* so that it gets destroyed when the async is done */
185 /* callback for timeout on an async request */
186 static void async_timeout( void *private )
188 struct async
*async
= private;
190 async
->timeout
= NULL
;
191 async_terminate( async
, async
->timeout_status
);
194 /* free an async queue, cancelling all async operations */
195 void free_async_queue( struct async_queue
*queue
)
197 struct async
*async
, *next
;
199 LIST_FOR_EACH_ENTRY_SAFE( async
, next
, &queue
->queue
, struct async
, queue_entry
)
201 grab_object( &async
->obj
);
202 if (!async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
204 async_terminate( async
, STATUS_HANDLES_CLOSED
);
206 release_object( &async
->obj
);
210 void queue_async( struct async_queue
*queue
, struct async
*async
)
212 /* fd will be set to NULL in free_async_queue when fd is destroyed */
213 release_object( async
->fd
);
215 async
->queue
= queue
;
216 grab_object( async
);
217 list_add_tail( &queue
->queue
, &async
->queue_entry
);
219 set_fd_signaled( async
->fd
, 0 );
222 /* create an async on a given queue of a fd */
223 struct async
*create_async( struct fd
*fd
, struct thread
*thread
, const async_data_t
*data
, struct iosb
*iosb
)
225 struct event
*event
= NULL
;
228 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
231 if (!(async
= alloc_object( &async_ops
)))
233 if (event
) release_object( event
);
237 async
->thread
= (struct thread
*)grab_object( thread
);
238 async
->event
= event
;
239 async
->status
= STATUS_PENDING
;
241 async
->timeout
= NULL
;
243 async
->fd
= (struct fd
*)grab_object( fd
);
246 async
->wait_handle
= 0;
247 async
->direct_result
= 0;
248 async
->completion
= fd_get_completion( fd
, &async
->comp_key
);
249 async
->comp_flags
= 0;
251 if (iosb
) async
->iosb
= (struct iosb
*)grab_object( iosb
);
252 else async
->iosb
= NULL
;
254 list_add_head( &thread
->process
->asyncs
, &async
->process_entry
);
255 if (event
) reset_event( event
);
257 if (async
->completion
&& data
->apc
)
259 release_object( async
);
260 set_error( STATUS_INVALID_PARAMETER
);
267 void set_async_pending( struct async
*async
, int signal
)
269 if (async
->status
== STATUS_PENDING
)
272 if (signal
&& !async
->signaled
)
275 wake_up( &async
->obj
, 0 );
280 /* create an async associated with iosb for async-based requests
281 * returned async must be passed to async_handoff */
282 struct async
*create_request_async( struct fd
*fd
, unsigned int comp_flags
, const async_data_t
*data
)
287 if (!(iosb
= create_iosb( get_req_data(), get_req_data_size(), get_reply_max_size() )))
290 async
= create_async( fd
, current
, data
, iosb
);
291 release_object( iosb
);
294 if (!(async
->wait_handle
= alloc_handle( current
->process
, async
, SYNCHRONIZE
, 0 )))
296 release_object( async
);
300 async
->direct_result
= 1;
301 async
->comp_flags
= comp_flags
;
306 /* return async object status and wait handle to client */
307 obj_handle_t
async_handoff( struct async
*async
, int success
, data_size_t
*result
, int force_blocking
)
311 if (get_error() == STATUS_PENDING
)
313 /* we don't know the result yet, so client needs to wait */
314 async
->direct_result
= 0;
315 return async
->wait_handle
;
317 close_handle( async
->thread
->process
, async
->wait_handle
);
318 async
->wait_handle
= 0;
322 if (get_error() != STATUS_PENDING
)
324 /* status and data are already set and returned */
325 async_terminate( async
, get_error() );
327 else if (async
->iosb
->status
!= STATUS_PENDING
)
329 /* result is already available in iosb, return it */
330 if (async
->iosb
->out_data
)
332 set_reply_data_ptr( async
->iosb
->out_data
, async
->iosb
->out_size
);
333 async
->iosb
->out_data
= NULL
;
337 if (async
->iosb
->status
!= STATUS_PENDING
)
339 if (result
) *result
= async
->iosb
->result
;
344 async
->direct_result
= 0;
346 if (!force_blocking
&& async
->fd
&& is_fd_overlapped( async
->fd
))
348 close_handle( async
->thread
->process
, async
->wait_handle
);
349 async
->wait_handle
= 0;
352 set_error( async
->iosb
->status
);
353 return async
->wait_handle
;
356 /* set the timeout of an async operation */
357 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
359 if (async
->timeout
) remove_timeout_user( async
->timeout
);
360 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
361 else async
->timeout
= NULL
;
362 async
->timeout_status
= status
;
365 static void add_async_completion( struct async
*async
, apc_param_t cvalue
, unsigned int status
,
366 apc_param_t information
)
368 if (async
->fd
&& !async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
369 if (async
->completion
) add_completion( async
->completion
, async
->comp_key
, cvalue
, status
, information
);
372 /* store the result of the client-side async callback */
373 void async_set_result( struct object
*obj
, unsigned int status
, apc_param_t total
)
375 struct async
*async
= (struct async
*)obj
;
377 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
379 assert( async
->status
!= STATUS_PENDING
); /* it must have been woken up if we get a result */
381 if (status
== STATUS_PENDING
) /* restart it */
383 status
= async
->status
;
384 async
->status
= STATUS_PENDING
;
385 grab_object( async
);
387 if (status
!= STATUS_ALERTED
) /* it was terminated in the meantime */
388 async_terminate( async
, status
);
390 async_reselect( async
);
394 if (async
->timeout
) remove_timeout_user( async
->timeout
);
395 async
->timeout
= NULL
;
396 async
->status
= status
;
397 if (status
== STATUS_MORE_PROCESSING_REQUIRED
) return; /* don't report the completion */
402 memset( &data
, 0, sizeof(data
) );
403 data
.type
= APC_USER
;
404 data
.user
.user
.func
= async
->data
.apc
;
405 data
.user
.user
.args
[0] = async
->data
.apc_context
;
406 data
.user
.user
.args
[1] = async
->data
.iosb
;
407 data
.user
.user
.args
[2] = 0;
408 thread_queue_apc( NULL
, async
->thread
, NULL
, &data
);
410 else if (async
->data
.apc_context
&& (async
->pending
||
411 !(async
->comp_flags
& FILE_SKIP_COMPLETION_PORT_ON_SUCCESS
)))
413 add_async_completion( async
, async
->data
.apc_context
, status
, total
);
416 if (async
->event
) set_event( async
->event
);
417 else if (async
->fd
) set_fd_signaled( async
->fd
, 1 );
418 if (!async
->signaled
)
421 wake_up( &async
->obj
, 0 );
426 /* check if an async operation is waiting to be alerted */
427 int async_waiting( struct async_queue
*queue
)
432 if (!(ptr
= list_head( &queue
->queue
))) return 0;
433 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
434 return async
->status
== STATUS_PENDING
;
437 static int cancel_async( struct process
*process
, struct object
*obj
, struct thread
*thread
, client_ptr_t iosb
)
443 LIST_FOR_EACH_ENTRY( async
, &process
->asyncs
, struct async
, process_entry
)
445 if (async
->status
== STATUS_CANCELLED
) continue;
446 if ((!obj
|| (async
->fd
&& get_fd_user( async
->fd
) == obj
)) &&
447 (!thread
|| async
->thread
== thread
) &&
448 (!iosb
|| async
->data
.iosb
== iosb
))
450 async_terminate( async
, STATUS_CANCELLED
);
458 void cancel_process_asyncs( struct process
*process
)
460 cancel_async( process
, NULL
, NULL
, 0 );
463 /* wake up async operations on the queue */
464 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
466 struct list
*ptr
, *next
;
468 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
470 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
471 async_terminate( async
, status
);
472 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */
476 static void iosb_dump( struct object
*obj
, int verbose
);
477 static void iosb_destroy( struct object
*obj
);
479 static const struct object_ops iosb_ops
=
481 sizeof(struct iosb
), /* size */
483 iosb_dump
, /* dump */
484 no_add_queue
, /* add_queue */
485 NULL
, /* remove_queue */
487 NULL
, /* satisfied */
488 no_signal
, /* signal */
489 no_get_fd
, /* get_fd */
490 default_map_access
, /* map_access */
491 default_get_sd
, /* get_sd */
492 default_set_sd
, /* set_sd */
493 no_get_full_name
, /* get_full_name */
494 no_lookup_name
, /* lookup_name */
495 no_link_name
, /* link_name */
496 NULL
, /* unlink_name */
497 no_open_file
, /* open_file */
498 no_kernel_obj_list
, /* get_kernel_obj_list */
499 no_close_handle
, /* close_handle */
500 iosb_destroy
/* destroy */
503 static void iosb_dump( struct object
*obj
, int verbose
)
505 assert( obj
->ops
== &iosb_ops
);
506 fprintf( stderr
, "I/O status block\n" );
509 static void iosb_destroy( struct object
*obj
)
511 struct iosb
*iosb
= (struct iosb
*)obj
;
513 free( iosb
->in_data
);
514 free( iosb
->out_data
);
517 /* allocate iosb struct */
518 struct iosb
*create_iosb( const void *in_data
, data_size_t in_size
, data_size_t out_size
)
522 if (!(iosb
= alloc_object( &iosb_ops
))) return NULL
;
524 iosb
->status
= STATUS_PENDING
;
526 iosb
->in_size
= in_size
;
527 iosb
->in_data
= NULL
;
528 iosb
->out_size
= out_size
;
529 iosb
->out_data
= NULL
;
531 if (in_size
&& !(iosb
->in_data
= memdup( in_data
, in_size
)))
533 release_object( iosb
);
540 struct iosb
*async_get_iosb( struct async
*async
)
542 return async
->iosb
? (struct iosb
*)grab_object( async
->iosb
) : NULL
;
545 struct thread
*async_get_thread( struct async
*async
)
547 return async
->thread
;
550 int async_is_blocking( struct async
*async
)
552 return !async
->event
&& !async
->data
.apc
&& !async
->data
.apc_context
;
555 /* find the first pending async in queue */
556 struct async
*find_pending_async( struct async_queue
*queue
)
559 LIST_FOR_EACH_ENTRY( async
, &queue
->queue
, struct async
, queue_entry
)
560 if (async
->status
== STATUS_PENDING
) return (struct async
*)grab_object( async
);
564 /* cancels all async I/O */
565 DECL_HANDLER(cancel_async
)
567 struct object
*obj
= get_handle_obj( current
->process
, req
->handle
, 0, NULL
);
568 struct thread
*thread
= req
->only_thread
? current
: NULL
;
572 int count
= cancel_async( current
->process
, obj
, thread
, req
->iosb
);
573 if (!count
&& req
->iosb
) set_error( STATUS_NOT_FOUND
);
574 release_object( obj
);
578 /* get async result from associated iosb */
579 DECL_HANDLER(get_async_result
)
581 struct iosb
*iosb
= NULL
;
584 LIST_FOR_EACH_ENTRY( async
, ¤t
->process
->asyncs
, struct async
, process_entry
)
585 if (async
->data
.user
== req
->user_arg
)
593 set_error( STATUS_INVALID_PARAMETER
);
599 data_size_t size
= min( iosb
->out_size
, get_reply_max_size() );
602 set_reply_data_ptr( iosb
->out_data
, size
);
603 iosb
->out_data
= NULL
;
606 reply
->size
= iosb
->result
;
607 set_error( iosb
->status
);