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 */
50 async_data_t data
; /* data for async I/O call */
51 struct iosb
*iosb
; /* I/O status block */
52 obj_handle_t wait_handle
; /* pre-allocated wait handle */
53 int direct_result
; /* a flag if we're passing result directly from request instead of APC */
54 struct completion
*completion
; /* completion associated with fd */
55 apc_param_t comp_key
; /* completion key associated with fd */
58 static void async_dump( struct object
*obj
, int verbose
);
59 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
);
60 static void async_satisfied( struct object
* obj
, struct wait_queue_entry
*entry
);
61 static void async_destroy( struct object
*obj
);
63 static const struct object_ops async_ops
=
65 sizeof(struct async
), /* size */
66 async_dump
, /* dump */
67 no_get_type
, /* get_type */
68 add_queue
, /* add_queue */
69 remove_queue
, /* remove_queue */
70 async_signaled
, /* signaled */
71 async_satisfied
, /* satisfied */
72 no_signal
, /* signal */
73 no_get_fd
, /* get_fd */
74 no_map_access
, /* map_access */
75 default_get_sd
, /* get_sd */
76 default_set_sd
, /* set_sd */
77 no_lookup_name
, /* lookup_name */
78 no_link_name
, /* link_name */
79 NULL
, /* unlink_name */
80 no_open_file
, /* open_file */
81 no_close_handle
, /* close_handle */
82 async_destroy
/* destroy */
85 static inline void async_reselect( struct async
*async
)
87 if (async
->queue
&& async
->fd
) fd_reselect_async( async
->fd
, async
->queue
);
90 static void async_dump( struct object
*obj
, int verbose
)
92 struct async
*async
= (struct async
*)obj
;
93 assert( obj
->ops
== &async_ops
);
94 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
97 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
)
99 struct async
*async
= (struct async
*)obj
;
100 assert( obj
->ops
== &async_ops
);
101 return async
->signaled
;
104 static void async_satisfied( struct object
*obj
, struct wait_queue_entry
*entry
)
106 struct async
*async
= (struct async
*)obj
;
107 assert( obj
->ops
== &async_ops
);
109 if (async
->direct_result
)
111 async_set_result( &async
->obj
, async
->iosb
->status
, async
->iosb
->result
);
112 async
->direct_result
= 0;
115 /* close wait handle here to avoid extra server round trip */
116 if (async
->wait_handle
)
118 close_handle( async
->thread
->process
, async
->wait_handle
);
119 async
->wait_handle
= 0;
123 static void async_destroy( struct object
*obj
)
125 struct async
*async
= (struct async
*)obj
;
126 assert( obj
->ops
== &async_ops
);
128 list_remove( &async
->process_entry
);
132 list_remove( &async
->queue_entry
);
133 async_reselect( async
);
135 else if (async
->fd
) release_object( async
->fd
);
137 if (async
->timeout
) remove_timeout_user( async
->timeout
);
138 if (async
->completion
) release_object( async
->completion
);
139 if (async
->event
) release_object( async
->event
);
140 if (async
->iosb
) release_object( async
->iosb
);
141 release_object( async
->thread
);
144 /* notifies client thread of new status of its async request */
145 void async_terminate( struct async
*async
, unsigned int status
)
147 assert( status
!= STATUS_PENDING
);
149 if (async
->status
!= STATUS_PENDING
)
151 /* already terminated, just update status */
152 async
->status
= status
;
156 async
->status
= status
;
157 if (async
->iosb
&& async
->iosb
->status
== STATUS_PENDING
) async
->iosb
->status
= status
;
159 if (!async
->direct_result
)
161 if (async
->data
.user
)
165 memset( &data
, 0, sizeof(data
) );
166 data
.type
= APC_ASYNC_IO
;
167 data
.async_io
.user
= async
->data
.user
;
168 data
.async_io
.sb
= async
->data
.iosb
;
169 data
.async_io
.status
= status
;
170 thread_queue_apc( async
->thread
, &async
->obj
, &data
);
172 else async_set_result( &async
->obj
, STATUS_SUCCESS
, 0 );
175 async_reselect( async
);
176 if (async
->queue
) release_object( async
); /* so that it gets destroyed when the async is done */
179 /* callback for timeout on an async request */
180 static void async_timeout( void *private )
182 struct async
*async
= private;
184 async
->timeout
= NULL
;
185 async_terminate( async
, async
->timeout_status
);
188 /* free an async queue, cancelling all async operations */
189 void free_async_queue( struct async_queue
*queue
)
191 struct async
*async
, *next
;
193 LIST_FOR_EACH_ENTRY_SAFE( async
, next
, &queue
->queue
, struct async
, queue_entry
)
195 async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
197 async_terminate( async
, STATUS_HANDLES_CLOSED
);
202 void queue_async( struct async_queue
*queue
, struct async
*async
)
204 /* fd will be set to NULL in free_async_queue when fd is destroyed */
205 release_object( async
->fd
);
207 async
->queue
= queue
;
208 grab_object( async
);
209 list_add_tail( &queue
->queue
, &async
->queue_entry
);
211 if (async
->fd
) set_fd_signaled( async
->fd
, 0 );
214 /* create an async on a given queue of a fd */
215 struct async
*create_async( struct fd
*fd
, struct thread
*thread
, const async_data_t
*data
, struct iosb
*iosb
)
217 struct event
*event
= NULL
;
220 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
223 if (!(async
= alloc_object( &async_ops
)))
225 if (event
) release_object( event
);
229 async
->thread
= (struct thread
*)grab_object( thread
);
230 async
->event
= event
;
231 async
->status
= STATUS_PENDING
;
233 async
->timeout
= NULL
;
235 async
->fd
= (struct fd
*)grab_object( fd
);
237 async
->wait_handle
= 0;
238 async
->direct_result
= 0;
239 async
->completion
= NULL
;
241 if (iosb
) async
->iosb
= (struct iosb
*)grab_object( iosb
);
242 else async
->iosb
= NULL
;
244 list_add_head( &thread
->process
->asyncs
, &async
->process_entry
);
245 if (event
) reset_event( event
);
250 /* create an async associated with iosb for async-based requests
251 * returned async must be passed to async_handoff */
252 struct async
*create_request_async( struct fd
*fd
, const async_data_t
*data
)
257 if (!(iosb
= create_iosb( get_req_data(), get_req_data_size(), get_reply_max_size() )))
260 async
= create_async( fd
, current
, data
, iosb
);
261 release_object( iosb
);
264 if (!(async
->wait_handle
= alloc_handle( current
->process
, async
, SYNCHRONIZE
, 0 )))
266 release_object( async
);
269 async
->direct_result
= 1;
274 /* return async object status and wait handle to client */
275 obj_handle_t
async_handoff( struct async
*async
, int success
, data_size_t
*result
)
279 close_handle( async
->thread
->process
, async
->wait_handle
);
280 async
->wait_handle
= 0;
284 if (get_error() != STATUS_PENDING
)
286 /* status and data are already set and returned */
287 async_terminate( async
, get_error() );
289 else if (async
->iosb
->status
!= STATUS_PENDING
)
291 /* result is already available in iosb, return it */
292 if (async
->iosb
->out_data
)
294 set_reply_data_ptr( async
->iosb
->out_data
, async
->iosb
->out_size
);
295 async
->iosb
->out_data
= NULL
;
299 if (async
->iosb
->status
!= STATUS_PENDING
)
301 if (result
) *result
= async
->iosb
->result
;
306 async
->direct_result
= 0;
307 if (!async_is_blocking( async
))
309 close_handle( async
->thread
->process
, async
->wait_handle
);
310 async
->wait_handle
= 0;
313 set_error( async
->iosb
->status
);
314 return async
->wait_handle
;
317 /* set the timeout of an async operation */
318 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
320 if (async
->timeout
) remove_timeout_user( async
->timeout
);
321 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
322 else async
->timeout
= NULL
;
323 async
->timeout_status
= status
;
326 static void add_async_completion( struct async
*async
, apc_param_t cvalue
, unsigned int status
,
327 apc_param_t information
)
329 if (async
->fd
&& !async
->completion
) async
->completion
= fd_get_completion( async
->fd
, &async
->comp_key
);
330 if (async
->completion
) add_completion( async
->completion
, async
->comp_key
, cvalue
, status
, information
);
333 /* store the result of the client-side async callback */
334 void async_set_result( struct object
*obj
, unsigned int status
, apc_param_t total
)
336 struct async
*async
= (struct async
*)obj
;
338 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
340 assert( async
->status
!= STATUS_PENDING
); /* it must have been woken up if we get a result */
342 if (status
== STATUS_PENDING
) /* restart it */
344 status
= async
->status
;
345 async
->status
= STATUS_PENDING
;
346 grab_object( async
);
348 if (status
!= STATUS_ALERTED
) /* it was terminated in the meantime */
349 async_terminate( async
, status
);
351 async_reselect( async
);
355 if (async
->timeout
) remove_timeout_user( async
->timeout
);
356 async
->timeout
= NULL
;
357 async
->status
= status
;
358 if (status
== STATUS_MORE_PROCESSING_REQUIRED
) return; /* don't report the completion */
363 memset( &data
, 0, sizeof(data
) );
364 data
.type
= APC_USER
;
365 data
.user
.func
= async
->data
.apc
;
366 data
.user
.args
[0] = async
->data
.apc_context
;
367 data
.user
.args
[1] = async
->data
.iosb
;
368 data
.user
.args
[2] = 0;
369 thread_queue_apc( async
->thread
, NULL
, &data
);
371 else if (async
->data
.apc_context
)
372 add_async_completion( async
, async
->data
.apc_context
, status
, total
);
374 if (async
->event
) set_event( async
->event
);
375 else if (async
->fd
) set_fd_signaled( async
->fd
, 1 );
376 if (!async
->signaled
)
379 wake_up( &async
->obj
, 0 );
384 /* check if an async operation is waiting to be alerted */
385 int async_waiting( struct async_queue
*queue
)
390 if (!(ptr
= list_head( &queue
->queue
))) return 0;
391 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
392 return async
->status
== STATUS_PENDING
;
395 static int cancel_async( struct process
*process
, struct object
*obj
, struct thread
*thread
, client_ptr_t iosb
)
401 LIST_FOR_EACH_ENTRY( async
, &process
->asyncs
, struct async
, process_entry
)
403 if (async
->status
== STATUS_CANCELLED
) continue;
404 if ((!obj
|| (async
->fd
&& get_fd_user( async
->fd
) == obj
)) &&
405 (!thread
|| async
->thread
== thread
) &&
406 (!iosb
|| async
->data
.iosb
== iosb
))
408 async_terminate( async
, STATUS_CANCELLED
);
416 void cancel_process_asyncs( struct process
*process
)
418 cancel_async( process
, NULL
, NULL
, 0 );
421 /* wake up async operations on the queue */
422 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
424 struct list
*ptr
, *next
;
426 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
428 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
429 async_terminate( async
, status
);
430 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */
434 static void iosb_dump( struct object
*obj
, int verbose
);
435 static void iosb_destroy( struct object
*obj
);
437 static const struct object_ops iosb_ops
=
439 sizeof(struct iosb
), /* size */
440 iosb_dump
, /* dump */
441 no_get_type
, /* get_type */
442 no_add_queue
, /* add_queue */
443 NULL
, /* remove_queue */
445 NULL
, /* satisfied */
446 no_signal
, /* signal */
447 no_get_fd
, /* get_fd */
448 no_map_access
, /* map_access */
449 default_get_sd
, /* get_sd */
450 default_set_sd
, /* set_sd */
451 no_lookup_name
, /* lookup_name */
452 no_link_name
, /* link_name */
453 NULL
, /* unlink_name */
454 no_open_file
, /* open_file */
455 no_close_handle
, /* close_handle */
456 iosb_destroy
/* destroy */
459 static void iosb_dump( struct object
*obj
, int verbose
)
461 assert( obj
->ops
== &iosb_ops
);
462 fprintf( stderr
, "I/O status block\n" );
465 static void iosb_destroy( struct object
*obj
)
467 struct iosb
*iosb
= (struct iosb
*)obj
;
469 free( iosb
->in_data
);
470 free( iosb
->out_data
);
473 /* allocate iosb struct */
474 struct iosb
*create_iosb( const void *in_data
, data_size_t in_size
, data_size_t out_size
)
478 if (!(iosb
= alloc_object( &iosb_ops
))) return NULL
;
480 iosb
->status
= STATUS_PENDING
;
482 iosb
->in_size
= in_size
;
483 iosb
->in_data
= NULL
;
484 iosb
->out_size
= out_size
;
485 iosb
->out_data
= NULL
;
487 if (in_size
&& !(iosb
->in_data
= memdup( in_data
, in_size
)))
489 release_object( iosb
);
496 struct iosb
*async_get_iosb( struct async
*async
)
498 return async
->iosb
? (struct iosb
*)grab_object( async
->iosb
) : NULL
;
501 int async_is_blocking( struct async
*async
)
503 return !async
->event
&& !async
->data
.apc
&& !async
->data
.apc_context
;
506 /* find the first pending async in queue */
507 struct async
*find_pending_async( struct async_queue
*queue
)
510 LIST_FOR_EACH_ENTRY( async
, &queue
->queue
, struct async
, queue_entry
)
511 if (async
->status
== STATUS_PENDING
) return (struct async
*)grab_object( async
);
515 /* cancels all async I/O */
516 DECL_HANDLER(cancel_async
)
518 struct object
*obj
= get_handle_obj( current
->process
, req
->handle
, 0, NULL
);
519 struct thread
*thread
= req
->only_thread
? current
: NULL
;
523 int count
= cancel_async( current
->process
, obj
, thread
, req
->iosb
);
524 if (!count
&& req
->iosb
) set_error( STATUS_NOT_FOUND
);
525 release_object( obj
);
529 /* get async result from associated iosb */
530 DECL_HANDLER(get_async_result
)
532 struct iosb
*iosb
= NULL
;
535 LIST_FOR_EACH_ENTRY( async
, ¤t
->process
->asyncs
, struct async
, process_entry
)
536 if (async
->data
.user
== req
->user_arg
)
544 set_error( STATUS_INVALID_PARAMETER
);
550 data_size_t size
= min( iosb
->out_size
, get_reply_max_size() );
553 set_reply_data_ptr( iosb
->out_data
, size
);
554 iosb
->out_data
= NULL
;
557 reply
->size
= iosb
->result
;
558 set_error( iosb
->status
);