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 struct timeout_user
*timeout
;
42 unsigned int timeout_status
; /* status to report upon timeout */
44 async_data_t data
; /* data for async I/O call */
47 static void async_dump( struct object
*obj
, int verbose
);
48 static void async_destroy( struct object
*obj
);
50 static const struct object_ops async_ops
=
52 sizeof(struct async
), /* size */
53 async_dump
, /* dump */
54 no_add_queue
, /* add_queue */
55 NULL
, /* remove_queue */
58 no_signal
, /* signal */
59 no_get_fd
, /* get_fd */
60 no_map_access
, /* map_access */
61 no_lookup_name
, /* lookup_name */
62 no_open_file
, /* open_file */
63 no_close_handle
, /* close_handle */
64 async_destroy
/* destroy */
70 struct object obj
; /* object header */
71 struct fd
*fd
; /* file descriptor owning this queue */
72 struct list queue
; /* queue of async objects */
75 static void async_queue_dump( struct object
*obj
, int verbose
);
77 static const struct object_ops async_queue_ops
=
79 sizeof(struct async_queue
), /* size */
80 async_queue_dump
, /* dump */
81 no_add_queue
, /* add_queue */
82 NULL
, /* remove_queue */
85 no_signal
, /* signal */
86 no_get_fd
, /* get_fd */
87 no_map_access
, /* map_access */
88 no_lookup_name
, /* lookup_name */
89 no_open_file
, /* open_file */
90 no_close_handle
, /* close_handle */
91 no_destroy
/* destroy */
95 static void async_dump( struct object
*obj
, int verbose
)
97 struct async
*async
= (struct async
*)obj
;
98 assert( obj
->ops
== &async_ops
);
99 fprintf( stderr
, "Async thread=%p\n", async
->thread
);
102 static void async_destroy( struct object
*obj
)
104 struct async
*async
= (struct async
*)obj
;
105 assert( obj
->ops
== &async_ops
);
107 if (async
->timeout
) remove_timeout_user( async
->timeout
);
108 if (async
->event
) release_object( async
->event
);
109 release_object( async
->queue
);
111 release_object( async
->thread
);
114 static void async_queue_dump( struct object
*obj
, int verbose
)
116 struct async_queue
*async_queue
= (struct async_queue
*)obj
;
117 assert( obj
->ops
== &async_queue_ops
);
118 fprintf( stderr
, "Async queue fd=%p\n", async_queue
->fd
);
121 /* notifies client thread of new status of its async request */
122 /* destroys the server side of it */
123 static void async_terminate( struct async
*async
, unsigned int status
)
127 memset( &data
, 0, sizeof(data
) );
128 data
.type
= APC_ASYNC_IO
;
129 data
.async_io
.func
= async
->data
.callback
;
130 data
.async_io
.user
= async
->data
.arg
;
131 data
.async_io
.sb
= async
->data
.iosb
;
132 data
.async_io
.status
= status
;
133 thread_queue_apc( async
->thread
, &async
->obj
, &data
);
135 if (async
->timeout
) remove_timeout_user( async
->timeout
);
136 async
->timeout
= NULL
;
137 list_remove( &async
->queue_entry
);
138 release_object( async
);
141 /* callback for timeout on an async request */
142 static void async_timeout( void *private )
144 struct async
*async
= private;
146 async
->timeout
= NULL
;
147 async_terminate( async
, async
->timeout_status
);
150 /* create a new async queue for a given fd */
151 struct async_queue
*create_async_queue( struct fd
*fd
)
153 struct async_queue
*queue
= alloc_object( &async_queue_ops
);
158 list_init( &queue
->queue
);
163 /* free an async queue, cancelling all async operations */
164 void free_async_queue( struct async_queue
*queue
)
168 async_wake_up( queue
, STATUS_HANDLES_CLOSED
);
169 release_object( queue
);
172 /* create an async on a given queue of a fd */
173 struct async
*create_async( struct thread
*thread
, struct async_queue
*queue
, const async_data_t
*data
)
175 struct event
*event
= NULL
;
178 if (data
->event
&& !(event
= get_event_obj( thread
->process
, data
->event
, EVENT_MODIFY_STATE
)))
181 if (!(async
= alloc_object( &async_ops
)))
183 if (event
) release_object( event
);
187 async
->thread
= (struct thread
*)grab_object( thread
);
188 async
->event
= event
;
190 async
->timeout
= NULL
;
191 async
->queue
= (struct async_queue
*)grab_object( queue
);
193 list_add_tail( &queue
->queue
, &async
->queue_entry
);
194 grab_object( async
);
196 if (queue
->fd
) set_fd_signaled( queue
->fd
, 0 );
197 if (event
) reset_event( event
);
201 /* set the timeout of an async operation */
202 void async_set_timeout( struct async
*async
, const struct timeval
*timeout
, unsigned int status
)
204 if (async
->timeout
) remove_timeout_user( async
->timeout
);
205 if (timeout
) async
->timeout
= add_timeout_user( timeout
, async_timeout
, async
);
206 else async
->timeout
= NULL
;
207 async
->timeout_status
= status
;
210 /* store the result of the client-side async callback */
211 void async_set_result( struct object
*obj
, unsigned int status
)
213 struct async
*async
= (struct async
*)obj
;
215 if (obj
->ops
!= &async_ops
) return; /* in case the client messed up the APC results */
217 if (status
== STATUS_PENDING
)
219 /* FIXME: restart the async operation */
226 data
.type
= APC_USER
;
227 data
.user
.func
= async
->data
.apc
;
228 data
.user
.args
[0] = (unsigned long)async
->data
.apc_arg
;
229 data
.user
.args
[1] = (unsigned long)async
->data
.iosb
;
230 data
.user
.args
[2] = 0;
231 thread_queue_apc( async
->thread
, NULL
, &data
);
233 if (async
->event
) set_event( async
->event
);
234 else if (async
->queue
->fd
) set_fd_signaled( async
->queue
->fd
, 1 );
238 /* check if an async operation is waiting to be alerted */
239 int async_waiting( struct async_queue
*queue
)
241 return queue
&& !list_empty( &queue
->queue
);
244 /* wake up async operations on the queue */
245 void async_wake_up( struct async_queue
*queue
, unsigned int status
)
247 struct list
*ptr
, *next
;
251 LIST_FOR_EACH_SAFE( ptr
, next
, &queue
->queue
)
253 struct async
*async
= LIST_ENTRY( ptr
, struct async
, queue_entry
);
254 async_terminate( async
, status
);
255 if (status
== STATUS_ALERTED
) break; /* only wake up the first one */