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
37 struct object obj
; /* object header */
38 struct thread
*thread
; /* owning thread */
39 struct list queue_entry
; /* entry in async queue list */
40 struct async_queue
*queue
; /* queue containing this async */
41 unsigned int status
; /* current status */
42 struct timeout_user
*timeout
;
43 unsigned int timeout_status
; /* status to report upon timeout */
45 async_data_t data
; /* data for async I/O call */
48 static void async_dump( struct object
*obj
, int verbose
);
49 static void async_destroy( struct object
*obj
);
51 static const struct object_ops async_ops
=
53 sizeof(struct async
), /* size */
54 async_dump
, /* dump */
55 no_get_type
, /* get_type */
56 no_add_queue
, /* add_queue */
57 NULL
, /* remove_queue */
60 no_signal
, /* signal */
61 no_get_fd
, /* get_fd */
62 no_map_access
, /* map_access */
63 default_get_sd
, /* get_sd */
64 default_set_sd
, /* set_sd */
65 no_lookup_name
, /* lookup_name */
66 no_open_file
, /* open_file */
67 no_close_handle
, /* close_handle */
68 async_destroy
/* destroy */
74 struct object obj
; /* object header */
75 struct fd
*fd
; /* file descriptor owning this queue */
76 struct list queue
; /* queue of async objects */
79 static void async_queue_dump( struct object
*obj
, int verbose
);
81 static const struct object_ops async_queue_ops
=
83 sizeof(struct async_queue
), /* size */
84 async_queue_dump
, /* dump */
85 no_get_type
, /* get_type */
86 no_add_queue
, /* add_queue */
87 NULL
, /* remove_queue */
90 no_signal
, /* signal */
91 no_get_fd
, /* get_fd */
92 no_map_access
, /* map_access */
93 default_get_sd
, /* get_sd */
94 default_set_sd
, /* set_sd */
95 no_lookup_name
, /* lookup_name */
96 no_open_file
, /* open_file */
97 no_close_handle
, /* close_handle */
98 no_destroy
/* destroy */
102 static inline void async_reselect( struct async
*async
)
104 if (async
->queue
->fd
) fd_reselect_async( async
->queue
->fd
, async
->queue
);
107 static void async_dump( struct object
*obj
, int verbose
)
109 struct async
*async
= (struct async
*)obj
;
110 assert( obj
->ops
== &async_ops
);
111 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
114 static void async_destroy( struct object
*obj
)
116 struct async
*async
= (struct async
*)obj
;
117 assert( obj
->ops
== &async_ops
);
119 list_remove( &async
->queue_entry
);
120 async_reselect( async
);
122 if (async
->timeout
) remove_timeout_user( async
->timeout
);
123 if (async
->event
) release_object( async
->event
);
124 release_object( async
->queue
);
125 release_object( async
->thread
);
128 static void async_queue_dump( struct object
*obj
, int verbose
)
130 struct async_queue
*async_queue
= (struct async_queue
*)obj
;
131 assert( obj
->ops
== &async_queue_ops
);
132 fprintf( stderr
, "Async queue fd=%p\n", async_queue
->fd
);
135 /* notifies client thread of new status of its async request */
136 void async_terminate( struct async
*async
, unsigned int status
)
140 assert( status
!= STATUS_PENDING
);
142 if (async
->status
!= STATUS_PENDING
)
144 /* already terminated, just update status */
145 async
->status
= status
;
149 /* send error completion event */
150 if (status
!= STATUS_ALERTED
&& async
->data
.cvalue
&& async
->queue
->fd
)
151 fd_add_completion( async
->queue
->fd
, async
->data
.cvalue
, status
, 0 );
153 memset( &data
, 0, sizeof(data
) );
154 data
.type
= APC_ASYNC_IO
;
155 data
.async_io
.func
= async
->data
.callback
;
156 data
.async_io
.user
= async
->data
.arg
;
157 data
.async_io
.sb
= async
->data
.iosb
;
158 data
.async_io
.status
= status
;
159 thread_queue_apc( async
->thread
, &async
->obj
, &data
);
160 async
->status
= status
;
161 async_reselect( async
);
162 release_object( async
); /* so that it gets destroyed when the async is done */
165 /* callback for timeout on an async request */
166 static void async_timeout( void *private )
168 struct async
*async
= private;
170 async
->timeout
= NULL
;
171 async_terminate( async
, async
->timeout_status
);
174 /* create a new async queue for a given fd */
175 struct async_queue
*create_async_queue( struct fd
*fd
)
177 struct async_queue
*queue
= alloc_object( &async_queue_ops
);
182 list_init( &queue
->queue
);
187 /* free an async queue, cancelling all async operations */
188 void free_async_queue( struct async_queue
*queue
)
192 async_wake_up( queue
, STATUS_HANDLES_CLOSED
);
193 release_object( queue
);
196 /* create an async on a given queue of a fd */
197 struct async
*create_async( struct thread
*thread
, struct async_queue
*queue
, const async_data_t
*data
)
199 struct event
*event
= NULL
;
202 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
205 if (!(async
= alloc_object( &async_ops
)))
207 if (event
) release_object( event
);
211 async
->thread
= (struct thread
*)grab_object( thread
);
212 async
->event
= event
;
213 async
->status
= STATUS_PENDING
;
215 async
->timeout
= NULL
;
216 async
->queue
= (struct async_queue
*)grab_object( queue
);
218 list_add_tail( &queue
->queue
, &async
->queue_entry
);
219 grab_object( async
);
221 if (queue
->fd
) set_fd_signaled( queue
->fd
, 0 );
222 if (event
) reset_event( event
);
226 /* set the timeout of an async operation */
227 void async_set_timeout( struct async
*async
, timeout_t timeout
, unsigned int status
)
229 if (async
->timeout
) remove_timeout_user( async
->timeout
);
230 if (timeout
!= TIMEOUT_INFINITE
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
231 else async
->timeout
= NULL
;
232 async
->timeout_status
= status
;
235 /* store the result of the client-side async callback */
236 void async_set_result( struct object
*obj
, unsigned int status
, unsigned long total
)
238 struct async
*async
= (struct async
*)obj
;
240 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
242 assert( async
->status
!= STATUS_PENDING
); /* it must have been woken up if we get a result */
244 if (status
== STATUS_PENDING
) /* restart it */
246 status
= async
->status
;
247 async
->status
= STATUS_PENDING
;
248 grab_object( async
);
250 if (status
!= STATUS_ALERTED
) /* it was terminated in the meantime */
251 async_terminate( async
, status
);
253 async_reselect( async
);
257 if (async
->timeout
) remove_timeout_user( async
->timeout
);
258 async
->timeout
= NULL
;
259 async
->status
= status
;
260 if (async
->data
.cvalue
&& async
->queue
->fd
)
261 fd_add_completion( async
->queue
->fd
, async
->data
.cvalue
, status
, total
);
265 data
.type
= APC_USER
;
266 data
.user
.func
= async
->data
.apc
;
267 data
.user
.args
[0] = (unsigned long)async
->data
.arg
;
268 data
.user
.args
[1] = (unsigned long)async
->data
.iosb
;
269 data
.user
.args
[2] = 0;
270 thread_queue_apc( async
->thread
, NULL
, &data
);
272 if (async
->event
) set_event( async
->event
);
273 else if (async
->queue
->fd
) set_fd_signaled( async
->queue
->fd
, 1 );
277 /* check if an async operation is waiting to be alerted */
278 int async_waiting( struct async_queue
*queue
)
283 if (!queue
) return 0;
284 if (!(ptr
= list_head( &queue
->queue
))) return 0;
285 async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
286 return async
->status
== STATUS_PENDING
;
289 /* wake up async operations on the queue */
290 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
292 struct list
*ptr
, *next
;
296 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
298 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
299 async_terminate( async
, status
);
300 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */