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 unsigned int status
; /* current status */
45 struct timeout_user
*timeout
;
46 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 */
53 static void async_dump( struct object
*obj
, int verbose
);
54 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
);
55 static void async_destroy( struct object
*obj
);
57 static const struct object_ops async_ops
=
59 sizeof(struct async
), /* size */
60 async_dump
, /* dump */
61 no_get_type
, /* get_type */
62 add_queue
, /* add_queue */
63 remove_queue
, /* remove_queue */
64 async_signaled
, /* signaled */
65 no_satisfied
, /* satisfied */
66 no_signal
, /* signal */
67 no_get_fd
, /* get_fd */
68 no_map_access
, /* map_access */
69 default_get_sd
, /* get_sd */
70 default_set_sd
, /* set_sd */
71 no_lookup_name
, /* lookup_name */
72 no_link_name
, /* link_name */
73 NULL
, /* unlink_name */
74 no_open_file
, /* open_file */
75 no_close_handle
, /* close_handle */
76 async_destroy
/* destroy */
82 struct object obj
; /* object header */
83 struct fd
*fd
; /* file descriptor owning this queue */
84 struct completion
*completion
; /* completion associated with a recently closed file descriptor */
85 apc_param_t comp_key
; /* completion key associated with a recently closed file descriptor */
86 struct list queue
; /* queue of async objects */
89 static void async_queue_dump( struct object
*obj
, int verbose
);
90 static void async_queue_destroy( struct object
*obj
);
92 static const struct object_ops async_queue_ops
=
94 sizeof(struct async_queue
), /* size */
95 async_queue_dump
, /* dump */
96 no_get_type
, /* get_type */
97 no_add_queue
, /* add_queue */
98 NULL
, /* remove_queue */
100 NULL
, /* satisfied */
101 no_signal
, /* signal */
102 no_get_fd
, /* get_fd */
103 no_map_access
, /* map_access */
104 default_get_sd
, /* get_sd */
105 default_set_sd
, /* set_sd */
106 no_lookup_name
, /* lookup_name */
107 no_link_name
, /* link_name */
108 NULL
, /* unlink_name */
109 no_open_file
, /* open_file */
110 no_close_handle
, /* close_handle */
111 async_queue_destroy
/* destroy */
115 static inline void async_reselect( struct async
*async
)
117 if (async
->queue
&& async
->queue
->fd
) fd_reselect_async( async
->queue
->fd
, async
->queue
);
120 static void async_dump( struct object
*obj
, int verbose
)
122 struct async
*async
= (struct async
*)obj
;
123 assert( obj
->ops
== &async_ops
);
124 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
127 static int async_signaled( struct object
*obj
, struct wait_queue_entry
*entry
)
129 struct async
*async
= (struct async
*)obj
;
130 assert( obj
->ops
== &async_ops
);
131 return async
->signaled
;
134 static void async_destroy( struct object
*obj
)
136 struct async
*async
= (struct async
*)obj
;
137 assert( obj
->ops
== &async_ops
);
139 list_remove( &async
->process_entry
);
143 list_remove( &async
->queue_entry
);
144 async_reselect( async
);
145 release_object( async
->queue
);
148 if (async
->timeout
) remove_timeout_user( async
->timeout
);
149 if (async
->event
) release_object( async
->event
);
150 if (async
->iosb
) release_object( async
->iosb
);
151 release_object( async
->thread
);
154 static void async_queue_dump( struct object
*obj
, int verbose
)
156 struct async_queue
*async_queue
= (struct async_queue
*)obj
;
157 assert( obj
->ops
== &async_queue_ops
);
158 fprintf( stderr
, "Async queue fd=%p\n", async_queue
->fd
);
161 static void async_queue_destroy( struct object
*obj
)
163 struct async_queue
*async_queue
= (struct async_queue
*)obj
;
164 assert( obj
->ops
== &async_queue_ops
);
165 if (async_queue
->completion
) release_object( async_queue
->completion
);
168 /* notifies client thread of new status of its async request */
169 void async_terminate( struct async
*async
, unsigned int status
)
171 assert( status
!= STATUS_PENDING
);
173 if (async
->status
!= STATUS_PENDING
)
175 /* already terminated, just update status */
176 async
->status
= status
;
180 async
->status
= status
;
181 if (async
->iosb
&& async
->iosb
->status
== STATUS_PENDING
) async
->iosb
->status
= status
;
183 if (async
->data
.user
)
187 memset( &data
, 0, sizeof(data
) );
188 data
.type
= APC_ASYNC_IO
;
189 data
.async_io
.user
= async
->data
.user
;
190 data
.async_io
.sb
= async
->data
.iosb
;
191 data
.async_io
.status
= status
;
192 thread_queue_apc( async
->thread
, &async
->obj
, &data
);
194 else async_set_result( &async
->obj
, STATUS_SUCCESS
, 0 );
196 async_reselect( async
);
197 if (async
->queue
) release_object( async
); /* so that it gets destroyed when the async is done */
200 /* callback for timeout on an async request */
201 static void async_timeout( void *private )
203 struct async
*async
= private;
205 async
->timeout
= NULL
;
206 async_terminate( async
, async
->timeout_status
);
209 /* create a new async queue for a given fd */
210 struct async_queue
*create_async_queue( struct fd
*fd
)
212 struct async_queue
*queue
= alloc_object( &async_queue_ops
);
217 queue
->completion
= NULL
;
218 list_init( &queue
->queue
);
223 /* free an async queue, cancelling all async operations */
224 void free_async_queue( struct async_queue
*queue
)
227 if (queue
->fd
) queue
->completion
= fd_get_completion( queue
->fd
, &queue
->comp_key
);
229 async_wake_up( queue
, STATUS_HANDLES_CLOSED
);
230 release_object( queue
);
233 void queue_async( struct async_queue
*queue
, struct async
*async
)
235 async
->queue
= (struct async_queue
*)grab_object( queue
);
237 grab_object( async
);
238 list_add_tail( &queue
->queue
, &async
->queue_entry
);
240 if (queue
->fd
) set_fd_signaled( queue
->fd
, 0 );
243 /* create an async on a given queue of a fd */
244 struct async
*create_async( struct thread
*thread
, const async_data_t
*data
, struct iosb
*iosb
)
246 struct event
*event
= NULL
;
249 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
252 if (!(async
= alloc_object( &async_ops
)))
254 if (event
) release_object( event
);
258 async
->thread
= (struct thread
*)grab_object( thread
);
259 async
->event
= event
;
260 async
->status
= STATUS_PENDING
;
262 async
->timeout
= NULL
;
266 if (iosb
) async
->iosb
= (struct iosb
*)grab_object( iosb
);
267 else async
->iosb
= NULL
;
269 list_add_head( &thread
->process
->asyncs
, &async
->process_entry
);
270 if (event
) reset_event( event
);
275 /* create an async associated with iosb for async-based requests */
276 struct async
*create_request_async( struct thread
*thread
, const async_data_t
*data
)
281 if (!(iosb
= create_iosb( get_req_data(), get_req_data_size(), get_reply_max_size() )))
284 async
= create_async( current
, data
, iosb
);
285 release_object( iosb
);
289 /* set the timeout of an async operation */
290 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
292 if (async
->timeout
) remove_timeout_user( async
->timeout
);
293 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
294 else async
->timeout
= NULL
;
295 async
->timeout_status
= status
;
298 static void add_async_completion( struct async_queue
*queue
, apc_param_t cvalue
, unsigned int status
,
299 apc_param_t information
)
304 struct completion
*completion
= fd_get_completion( queue
->fd
, &ckey
);
308 add_completion( completion
, ckey
, cvalue
, status
, information
);
309 release_object( completion
);
312 else if (queue
->completion
) add_completion( queue
->completion
, queue
->comp_key
,
313 cvalue
, status
, information
);
316 /* store the result of the client-side async callback */
317 void async_set_result( struct object
*obj
, unsigned int status
, apc_param_t total
)
319 struct async
*async
= (struct async
*)obj
;
321 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
323 assert( async
->status
!= STATUS_PENDING
); /* it must have been woken up if we get a result */
325 if (status
== STATUS_PENDING
) /* restart it */
327 status
= async
->status
;
328 async
->status
= STATUS_PENDING
;
329 grab_object( async
);
331 if (status
!= STATUS_ALERTED
) /* it was terminated in the meantime */
332 async_terminate( async
, status
);
334 async_reselect( async
);
338 if (async
->timeout
) remove_timeout_user( async
->timeout
);
339 async
->timeout
= NULL
;
340 async
->status
= status
;
341 if (status
== STATUS_MORE_PROCESSING_REQUIRED
) return; /* don't report the completion */
343 if (async
->queue
&& !async
->data
.apc
&& async
->data
.apc_context
)
344 add_async_completion( async
->queue
, async
->data
.apc_context
, status
, total
);
348 memset( &data
, 0, sizeof(data
) );
349 data
.type
= APC_USER
;
350 data
.user
.func
= async
->data
.apc
;
351 data
.user
.args
[0] = async
->data
.apc_context
;
352 data
.user
.args
[1] = async
->data
.iosb
;
353 data
.user
.args
[2] = 0;
354 thread_queue_apc( async
->thread
, NULL
, &data
);
356 if (async
->event
) set_event( async
->event
);
357 else if (async
->queue
&& async
->queue
->fd
) set_fd_signaled( async
->queue
->fd
, 1 );
359 wake_up( &async
->obj
, 0 );
363 /* check if there are any queued async operations */
364 int async_queued( struct async_queue
*queue
)
366 return queue
&& list_head( &queue
->queue
);
369 /* check if an async operation is waiting to be alerted */
370 int async_waiting( struct async_queue
*queue
)
375 if (!queue
) return 0;
376 if (!(ptr
= list_head( &queue
->queue
))) return 0;
377 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
378 return async
->status
== STATUS_PENDING
;
381 static int cancel_async( struct process
*process
, struct object
*obj
, struct thread
*thread
, client_ptr_t iosb
)
387 LIST_FOR_EACH_ENTRY( async
, &process
->asyncs
, struct async
, process_entry
)
389 if (async
->status
== STATUS_CANCELLED
) continue;
390 if ((!obj
|| (async
->queue
&& async
->queue
->fd
&& get_fd_user( async
->queue
->fd
) == obj
)) &&
391 (!thread
|| async
->thread
== thread
) &&
392 (!iosb
|| async
->data
.iosb
== iosb
))
394 async_terminate( async
, STATUS_CANCELLED
);
402 void cancel_process_asyncs( struct process
*process
)
404 cancel_async( process
, NULL
, NULL
, 0 );
407 /* wake up async operations on the queue */
408 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
410 struct list
*ptr
, *next
;
414 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
416 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
417 async_terminate( async
, status
);
418 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */
422 static void iosb_dump( struct object
*obj
, int verbose
);
423 static void iosb_destroy( struct object
*obj
);
425 static const struct object_ops iosb_ops
=
427 sizeof(struct iosb
), /* size */
428 iosb_dump
, /* dump */
429 no_get_type
, /* get_type */
430 no_add_queue
, /* add_queue */
431 NULL
, /* remove_queue */
433 NULL
, /* satisfied */
434 no_signal
, /* signal */
435 no_get_fd
, /* get_fd */
436 no_map_access
, /* map_access */
437 default_get_sd
, /* get_sd */
438 default_set_sd
, /* set_sd */
439 no_lookup_name
, /* lookup_name */
440 no_link_name
, /* link_name */
441 NULL
, /* unlink_name */
442 no_open_file
, /* open_file */
443 no_close_handle
, /* close_handle */
444 iosb_destroy
/* destroy */
447 static void iosb_dump( struct object
*obj
, int verbose
)
449 assert( obj
->ops
== &iosb_ops
);
450 fprintf( stderr
, "I/O status block\n" );
453 static void iosb_destroy( struct object
*obj
)
455 struct iosb
*iosb
= (struct iosb
*)obj
;
457 free( iosb
->in_data
);
458 free( iosb
->out_data
);
461 /* allocate iosb struct */
462 struct iosb
*create_iosb( const void *in_data
, data_size_t in_size
, data_size_t out_size
)
466 if (!(iosb
= alloc_object( &iosb_ops
))) return NULL
;
468 iosb
->status
= STATUS_PENDING
;
470 iosb
->in_size
= in_size
;
471 iosb
->in_data
= NULL
;
472 iosb
->out_size
= out_size
;
473 iosb
->out_data
= NULL
;
475 if (in_size
&& !(iosb
->in_data
= memdup( in_data
, in_size
)))
477 release_object( iosb
);
484 struct iosb
*async_get_iosb( struct async
*async
)
486 return async
->iosb
? (struct iosb
*)grab_object( async
->iosb
) : NULL
;
489 int async_is_blocking( struct async
*async
)
491 return !async
->event
&& !async
->data
.apc
&& !async
->data
.apc_context
;
494 /* find the first pending async in queue */
495 struct async
*find_pending_async( struct async_queue
*queue
)
498 if (queue
) LIST_FOR_EACH_ENTRY( async
, &queue
->queue
, struct async
, queue_entry
)
499 if (async
->status
== STATUS_PENDING
) return (struct async
*)grab_object( async
);
503 /* cancels all async I/O */
504 DECL_HANDLER(cancel_async
)
506 struct object
*obj
= get_handle_obj( current
->process
, req
->handle
, 0, NULL
);
507 struct thread
*thread
= req
->only_thread
? current
: NULL
;
511 int count
= cancel_async( current
->process
, obj
, thread
, req
->iosb
);
512 if (!count
&& req
->iosb
) set_error( STATUS_NOT_FOUND
);
513 release_object( obj
);
517 /* get async result from associated iosb */
518 DECL_HANDLER(get_async_result
)
520 struct iosb
*iosb
= NULL
;
523 LIST_FOR_EACH_ENTRY( async
, ¤t
->process
->asyncs
, struct async
, process_entry
)
524 if (async
->data
.user
== req
->user_arg
)
532 set_error( STATUS_INVALID_PARAMETER
);
538 data_size_t size
= min( iosb
->out_size
, get_reply_max_size() );
541 set_reply_data_ptr( iosb
->out_data
, size
);
542 iosb
->out_data
= NULL
;
545 reply
->size
= iosb
->result
;
546 set_error( iosb
->status
);