source: mainline/uspace/lib/libc/generic/async.c@ c7509e5

lfn serial ticket/834-toolchain-update topic/msim-upgrade topic/simplify-dev-export
Last change on this file since c7509e5 was c7509e5, checked in by Jakub Jermar <jakub@…>, 18 years ago

Add comment clarifying how a fibril waiting for a message gets rescheduled.

  • Property mode set to 100644
File size: 23.2 KB
RevLine 
[06502f7d]1/*
[df4ed85]2 * Copyright (c) 2006 Ondrej Palkovsky
[06502f7d]3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 *
9 * - Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * - Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 * - The name of the author may not be used to endorse or promote products
15 * derived from this software without specific prior written permission.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
18 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
19 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
20 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
21 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
22 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
26 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
[b2951e2]27 */
28
[a46da63]29/** @addtogroup libc
[b2951e2]30 * @{
31 */
32/** @file
[450cd3a]33 */
[06502f7d]34
[80649a91]35/**
36 * Asynchronous library
37 *
[9591265]38 * The aim of this library is facilitating writing programs utilizing the
39 * asynchronous nature of HelenOS IPC, yet using a normal way of programming.
[80649a91]40 *
[9591265]41 * You should be able to write very simple multithreaded programs, the async
42 * framework will automatically take care of most synchronization problems.
[80649a91]43 *
44 * Default semantics:
[9591265]45 * - async_send_*(): send asynchronously. If the kernel refuses to send
46 * more messages, [ try to get responses from kernel, if
47 * nothing found, might try synchronous ]
[80649a91]48 *
[9591265]49 * Example of use (pseudo C):
[80649a91]50 *
51 * 1) Multithreaded client application
[9591265]52 *
53 * fibril_create(fibril1, ...);
54 * fibril_create(fibril2, ...);
55 * ...
[80649a91]56 *
[9591265]57 * int fibril1(void *arg)
58 * {
59 * conn = ipc_connect_me_to();
60 * c1 = async_send(conn);
61 * c2 = async_send(conn);
62 * async_wait_for(c1);
63 * async_wait_for(c2);
64 * ...
65 * }
[80649a91]66 *
67 *
68 * 2) Multithreaded server application
[9591265]69 * main()
70 * {
71 * async_manager();
[80649a91]72 * }
73 *
74 *
[36c9234]75 * my_client_connection(icallid, *icall)
[9591265]76 * {
77 * if (want_refuse) {
78 * ipc_answer_fast(icallid, ELIMIT, 0, 0);
79 * return;
80 * }
81 * ipc_answer_fast(icallid, EOK, 0, 0);
[80649a91]82 *
[9591265]83 * callid = async_get_call(&call);
84 * handle_call(callid, call);
85 * ipc_answer_fast(callid, 1, 2, 3);
[53ca318]86 *
[9591265]87 * callid = async_get_call(&call);
88 * ....
[80649a91]89 * }
[a2cd194]90 *
[80649a91]91 */
[9591265]92
[80649a91]93#include <futex.h>
94#include <async.h>
[bc1f1c2]95#include <fibril.h>
[80649a91]96#include <stdio.h>
97#include <libadt/hash_table.h>
98#include <libadt/list.h>
99#include <ipc/ipc.h>
100#include <assert.h>
101#include <errno.h>
[daa90e8]102#include <sys/time.h>
[c042bdd]103#include <arch/barrier.h>
[80649a91]104
[fc42b28]105atomic_t async_futex = FUTEX_INITIALIZER;
[80649a91]106static hash_table_t conn_hash_table;
[c042bdd]107static LIST_INITIALIZE(timeout_list);
[80649a91]108
[e70bfa5]109/** Structures of this type represent a waiting fibril. */
[01ff41c]110typedef struct {
[e70bfa5]111 /** Expiration time. */
[bc1f1c2]112 struct timeval expires;
113 /** If true, this struct is in the timeout list. */
114 int inlist;
[e70bfa5]115 /** Timeout list link. */
[49d072e]116 link_t link;
117
[36c9234]118 /** Identification of and link to the waiting fibril. */
[bc1f1c2]119 fid_t fid;
[e70bfa5]120 /** If true, this fibril is currently active. */
[bc1f1c2]121 int active;
[e70bfa5]122 /** If true, we have timed out. */
[bc1f1c2]123 int timedout;
[49d072e]124} awaiter_t;
125
126typedef struct {
127 awaiter_t wdata;
[e70bfa5]128
129 /** If reply was received. */
130 int done;
131 /** Pointer to where the answer data is stored. */
132 ipc_call_t *dataptr;
[49d072e]133
[01ff41c]134 ipcarg_t retval;
135} amsg_t;
136
[36c9234]137/**
138 * Structures of this type are used to group information about a call and a
139 * message queue link.
140 */
[80649a91]141typedef struct {
142 link_t link;
143 ipc_callid_t callid;
144 ipc_call_t call;
145} msg_t;
146
147typedef struct {
[49d072e]148 awaiter_t wdata;
149
[e70bfa5]150 /** Hash table link. */
151 link_t link;
152
153 /** Incoming phone hash. */
154 ipcarg_t in_phone_hash;
155
156 /** Messages that should be delivered to this fibril. */
157 link_t msg_queue;
158
159 /** Identification of the opening call. */
[80649a91]160 ipc_callid_t callid;
[e70bfa5]161 /** Call data of the opening call. */
[80649a91]162 ipc_call_t call;
[e70bfa5]163
164 /** Identification of the closing call. */
165 ipc_callid_t close_callid;
166
167 /** Fibril function that will be used to handle the connection. */
[bc1f1c2]168 void (*cfibril)(ipc_callid_t, ipc_call_t *);
[80649a91]169} connection_t;
170
[bc1f1c2]171/** Identifier of the incoming connection handled by the current fibril. */
172__thread connection_t *FIBRIL_connection;
[e70bfa5]173
[36c9234]174/**
175 * If true, it is forbidden to use async_req functions and all preemption is
176 * disabled.
177 */
[085bd54]178__thread int in_interrupt_handler;
[80649a91]179
[da0c91e7]180static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
[51dbadf3]181static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
[36c9234]182
183/**
184 * Pointer to a fibril function that will be used to handle connections.
185 */
[da0c91e7]186static async_client_conn_t client_connection = default_client_connection;
[36c9234]187/**
188 * Pointer to a fibril function that will be used to handle interrupt
189 * notifications.
190 */
[51dbadf3]191static async_client_conn_t interrupt_received = default_interrupt_received;
[da0c91e7]192
[a2cd194]193#define CONN_HASH_TABLE_CHAINS 32
[80649a91]194
[e70bfa5]195/** Compute hash into the connection hash table based on the source phone hash.
196 *
197 * @param key Pointer to source phone hash.
198 *
199 * @return Index into the connection hash table.
200 */
[80649a91]201static hash_index_t conn_hash(unsigned long *key)
[450cd3a]202{
[80649a91]203 assert(key);
[a2cd194]204 return ((*key) >> 4) % CONN_HASH_TABLE_CHAINS;
[450cd3a]205}
[06502f7d]206
[e70bfa5]207/** Compare hash table item with a key.
208 *
209 * @param key Array containing the source phone hash as the only item.
210 * @param keys Expected 1 but ignored.
211 * @param item Connection hash table item.
212 *
213 * @return True on match, false otherwise.
214 */
[80649a91]215static int conn_compare(unsigned long key[], hash_count_t keys, link_t *item)
[450cd3a]216{
[80649a91]217 connection_t *hs;
218
219 hs = hash_table_get_instance(item, connection_t, link);
220
221 return key[0] == hs->in_phone_hash;
[450cd3a]222}
[06502f7d]223
[e70bfa5]224/** Connection hash table removal callback function.
225 *
226 * This function is called whenever a connection is removed from the connection
227 * hash table.
228 *
229 * @param item Connection hash table item being removed.
230 */
[80649a91]231static void conn_remove(link_t *item)
[450cd3a]232{
[80649a91]233 free(hash_table_get_instance(item, connection_t, link));
[450cd3a]234}
235
[80649a91]236
[e70bfa5]237/** Operations for the connection hash table. */
[80649a91]238static hash_table_operations_t conn_hash_table_ops = {
239 .hash = conn_hash,
240 .compare = conn_compare,
241 .remove_callback = conn_remove
242};
243
[e70bfa5]244/** Sort in current fibril's timeout request.
[49d072e]245 *
[e70bfa5]246 * @param wd Wait data of the current fibril.
[49d072e]247 */
248static void insert_timeout(awaiter_t *wd)
249{
250 link_t *tmp;
251 awaiter_t *cur;
252
253 wd->timedout = 0;
[085bd54]254 wd->inlist = 1;
[49d072e]255
256 tmp = timeout_list.next;
257 while (tmp != &timeout_list) {
258 cur = list_get_instance(tmp, awaiter_t, link);
259 if (tv_gteq(&cur->expires, &wd->expires))
260 break;
261 tmp = tmp->next;
262 }
263 list_append(&wd->link, tmp);
264}
265
[e70bfa5]266/** Try to route a call to an appropriate connection fibril.
[80649a91]267 *
[36c9234]268 * If the proper connection fibril is found, a message with the call is added to
269 * its message queue. If the fibril was not active, it is activated and all
270 * timeouts are unregistered.
271 *
272 * @param callid Hash of the incoming call.
273 * @param call Data of the incoming call.
274 *
275 * @return Zero if the call doesn't match any connection.
276 * One if the call was passed to the respective connection
277 * fibril.
[80649a91]278 */
279static int route_call(ipc_callid_t callid, ipc_call_t *call)
[450cd3a]280{
[80649a91]281 connection_t *conn;
282 msg_t *msg;
283 link_t *hlp;
284 unsigned long key;
285
[01ff41c]286 futex_down(&async_futex);
[80649a91]287
288 key = call->in_phone_hash;
289 hlp = hash_table_find(&conn_hash_table, &key);
290 if (!hlp) {
[01ff41c]291 futex_up(&async_futex);
[80649a91]292 return 0;
[450cd3a]293 }
[80649a91]294 conn = hash_table_get_instance(hlp, connection_t, link);
295
296 msg = malloc(sizeof(*msg));
297 msg->callid = callid;
298 msg->call = *call;
299 list_append(&msg->link, &conn->msg_queue);
[41269bd]300
301 if (IPC_GET_METHOD(*call) == IPC_M_PHONE_HUNGUP)
302 conn->close_callid = callid;
[80649a91]303
[36c9234]304 /* If the connection fibril is waiting for an event, activate it */
[49d072e]305 if (!conn->wdata.active) {
306 /* If in timeout list, remove it */
307 if (conn->wdata.inlist) {
308 conn->wdata.inlist = 0;
309 list_remove(&conn->wdata.link);
310 }
311 conn->wdata.active = 1;
[bc1f1c2]312 fibril_add_ready(conn->wdata.fid);
[80649a91]313 }
314
[01ff41c]315 futex_up(&async_futex);
[80649a91]316
317 return 1;
318}
319
[e70bfa5]320/** Return new incoming message for the current (fibril-local) connection.
321 *
322 * @param call Storage where the incoming call data will be stored.
323 * @param usecs Timeout in microseconds. Zero denotes no timeout.
324 *
325 * @return If no timeout was specified, then a hash of the
326 * incoming call is returned. If a timeout is specified,
327 * then a hash of the incoming call is returned unless
328 * the timeout expires prior to receiving a message. In
329 * that case zero is returned.
330 */
[49d072e]331ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
[80649a91]332{
333 msg_t *msg;
334 ipc_callid_t callid;
[6c46350]335 connection_t *conn;
[80649a91]336
[bc1f1c2]337 assert(FIBRIL_connection);
338 /* GCC 4.1.0 coughs on FIBRIL_connection-> dereference,
[6c46350]339 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
340 * I would never expect to find so many errors in
[36c9234]341 * a compiler *($&$(*&$
[6c46350]342 */
[bc1f1c2]343 conn = FIBRIL_connection;
[c0e674a]344
[01ff41c]345 futex_down(&async_futex);
[80649a91]346
[49d072e]347 if (usecs) {
[6c46350]348 gettimeofday(&conn->wdata.expires, NULL);
349 tv_add(&conn->wdata.expires, usecs);
[49d072e]350 } else {
[6c46350]351 conn->wdata.inlist = 0;
[49d072e]352 }
[e70bfa5]353 /* If nothing in queue, wait until something arrives */
[6c46350]354 while (list_empty(&conn->msg_queue)) {
[085bd54]355 if (usecs)
[6c46350]356 insert_timeout(&conn->wdata);
[085bd54]357
[6c46350]358 conn->wdata.active = 0;
[c7509e5]359 /*
360 * Note: the current fibril will be rescheduled either due to a
361 * timeout or due to an arriving message destined to it. In the
362 * former case, handle_expired_timeouts() and, in the latter
363 * case, route_call() will perform the wakeup.
364 */
[bc1f1c2]365 fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
[e70bfa5]366 /*
367 * Futex is up after getting back from async_manager get it
368 * again.
[c7509e5]369 */
[49d072e]370 futex_down(&async_futex);
[bc1f1c2]371 if (usecs && conn->wdata.timedout &&
[6c46350]372 list_empty(&conn->msg_queue)) {
[e70bfa5]373 /* If we timed out -> exit */
[49d072e]374 futex_up(&async_futex);
375 return 0;
376 }
[450cd3a]377 }
378
[6c46350]379 msg = list_get_instance(conn->msg_queue.next, msg_t, link);
[80649a91]380 list_remove(&msg->link);
381 callid = msg->callid;
382 *call = msg->call;
383 free(msg);
384
[01ff41c]385 futex_up(&async_futex);
[80649a91]386 return callid;
387}
388
[36c9234]389/** Default fibril function that gets called to handle new connection.
[a2cd194]390 *
[e70bfa5]391 * This function is defined as a weak symbol - to be redefined in user code.
[36c9234]392 *
393 * @param callid Hash of the incoming call.
394 * @param call Data of the incoming call.
[a2cd194]395 */
[da0c91e7]396static void default_client_connection(ipc_callid_t callid, ipc_call_t *call)
[80649a91]397{
[a2cd194]398 ipc_answer_fast(callid, ENOENT, 0, 0);
[80649a91]399}
[36c9234]400
401/** Default fibril function that gets called to handle interrupt notifications.
402 *
403 * @param callid Hash of the incoming call.
404 * @param call Data of the incoming call.
405 */
[51dbadf3]406static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
[44c6d88d]407{
408}
409
[f2f0392]410/** Wrapper for client connection fibril.
411 *
[36c9234]412 * When a new connection arrives, a fibril with this implementing function is
[f2f0392]413 * created. It calls client_connection() and does the final cleanup.
[a2cd194]414 *
[36c9234]415 * @param arg Connection structure pointer.
[a2cd194]416 *
[f2f0392]417 * @return Always zero.
[a2cd194]418 */
[bc1f1c2]419static int connection_fibril(void *arg)
[80649a91]420{
[a2cd194]421 unsigned long key;
422 msg_t *msg;
[41269bd]423 int close_answered = 0;
[a2cd194]424
[f2f0392]425 /* Setup fibril-local connection pointer */
[bc1f1c2]426 FIBRIL_connection = (connection_t *) arg;
427 FIBRIL_connection->cfibril(FIBRIL_connection->callid,
428 &FIBRIL_connection->call);
[a46da63]429
[36c9234]430 /* Remove myself from the connection hash table */
[01ff41c]431 futex_down(&async_futex);
[bc1f1c2]432 key = FIBRIL_connection->in_phone_hash;
[a2cd194]433 hash_table_remove(&conn_hash_table, &key, 1);
[01ff41c]434 futex_up(&async_futex);
[a46da63]435
[36c9234]436 /* Answer all remaining messages with EHANGUP */
[bc1f1c2]437 while (!list_empty(&FIBRIL_connection->msg_queue)) {
438 msg = list_get_instance(FIBRIL_connection->msg_queue.next,
439 msg_t, link);
[a2cd194]440 list_remove(&msg->link);
[bc1f1c2]441 if (msg->callid == FIBRIL_connection->close_callid)
[41269bd]442 close_answered = 1;
[a2cd194]443 ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
444 free(msg);
445 }
[bc1f1c2]446 if (FIBRIL_connection->close_callid)
447 ipc_answer_fast(FIBRIL_connection->close_callid, 0, 0, 0);
[a46da63]448
449 return 0;
[80649a91]450}
451
[f2f0392]452/** Create a new fibril for a new connection.
[80649a91]453 *
[f2f0392]454 * Creates new fibril for connection, fills in connection structures and inserts
455 * it into the hash table, so that later we can easily do routing of messages to
456 * particular fibrils.
[53ca318]457 *
[36c9234]458 * @param in_phone_hash Identification of the incoming connection.
459 * @param callid Hash of the opening IPC_M_CONNECT_ME_TO call.
460 * @param call Call data of the opening call.
461 * @param cfibril Fibril function that should be called upon opening the
462 * connection.
463 *
464 * @return New fibril id or NULL on failure.
[80649a91]465 */
[bc1f1c2]466fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
467 ipc_call_t *call, void (*cfibril)(ipc_callid_t, ipc_call_t *))
[80649a91]468{
469 connection_t *conn;
470 unsigned long key;
471
472 conn = malloc(sizeof(*conn));
473 if (!conn) {
474 ipc_answer_fast(callid, ENOMEM, 0, 0);
[53ca318]475 return NULL;
[80649a91]476 }
[44c6d88d]477 conn->in_phone_hash = in_phone_hash;
[80649a91]478 list_initialize(&conn->msg_queue);
479 conn->callid = callid;
[41269bd]480 conn->close_callid = 0;
[eaf34f7]481 if (call)
482 conn->call = *call;
[36c9234]483 conn->wdata.active = 1; /* We will activate the fibril ASAP */
[bc1f1c2]484 conn->cfibril = cfibril;
[49d072e]485
[bc1f1c2]486 conn->wdata.fid = fibril_create(connection_fibril, conn);
487 if (!conn->wdata.fid) {
[80649a91]488 free(conn);
489 ipc_answer_fast(callid, ENOMEM, 0, 0);
[53ca318]490 return NULL;
[80649a91]491 }
[36c9234]492 /* Add connection to the connection hash table */
[80649a91]493 key = conn->in_phone_hash;
[01ff41c]494 futex_down(&async_futex);
[80649a91]495 hash_table_insert(&conn_hash_table, &key, &conn->link);
[01ff41c]496 futex_up(&async_futex);
[80649a91]497
[bc1f1c2]498 fibril_add_ready(conn->wdata.fid);
[53ca318]499
[bc1f1c2]500 return conn->wdata.fid;
[80649a91]501}
502
[36c9234]503/** Handle a call that was received.
504 *
505 * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
506 * Otherwise the call is routed to its connection fibril.
507 *
508 * @param callid Hash of the incoming call.
509 * @param call Data of the incoming call.
510 */
[80649a91]511static void handle_call(ipc_callid_t callid, ipc_call_t *call)
512{
[44c6d88d]513 /* Unrouted call - do some default behaviour */
[15039b67]514 if ((callid & IPC_CALLID_NOTIFICATION)) {
[085bd54]515 in_interrupt_handler = 1;
[51dbadf3]516 (*interrupt_received)(callid,call);
[085bd54]517 in_interrupt_handler = 0;
[44c6d88d]518 return;
[15039b67]519 }
520
521 switch (IPC_GET_METHOD(*call)) {
[80649a91]522 case IPC_M_CONNECT_ME_TO:
[f2f0392]523 /* Open new connection with fibril etc. */
[bc1f1c2]524 async_new_connection(IPC_GET_ARG3(*call), callid, call,
525 client_connection);
[44c6d88d]526 return;
[80649a91]527 }
[44c6d88d]528
[36c9234]529 /* Try to route the call through the connection hash table */
[44c6d88d]530 if (route_call(callid, call))
531 return;
532
533 /* Unknown call from unknown phone - hang it up */
534 ipc_answer_fast(callid, EHANGUP, 0, 0);
[450cd3a]535}
536
[f2f0392]537/** Fire all timeouts that expired. */
[c042bdd]538static void handle_expired_timeouts(void)
539{
540 struct timeval tv;
[49d072e]541 awaiter_t *waiter;
[c042bdd]542 link_t *cur;
543
[36c9234]544 gettimeofday(&tv, NULL);
[c042bdd]545 futex_down(&async_futex);
546
547 cur = timeout_list.next;
548 while (cur != &timeout_list) {
[bc1f1c2]549 waiter = list_get_instance(cur, awaiter_t, link);
[49d072e]550 if (tv_gt(&waiter->expires, &tv))
[c042bdd]551 break;
552 cur = cur->next;
[49d072e]553 list_remove(&waiter->link);
554 waiter->inlist = 0;
555 waiter->timedout = 1;
[36c9234]556 /*
557 * Redundant condition?
558 * The fibril should not be active when it gets here.
[c042bdd]559 */
[49d072e]560 if (!waiter->active) {
561 waiter->active = 1;
[bc1f1c2]562 fibril_add_ready(waiter->fid);
[c042bdd]563 }
564 }
565
566 futex_up(&async_futex);
567}
568
[36c9234]569/** Endless loop dispatching incoming calls and answers.
570 *
571 * @return Never returns.
572 */
[085bd54]573static int async_manager_worker(void)
[80649a91]574{
575 ipc_call_t call;
576 ipc_callid_t callid;
[0b99e40]577 int timeout;
[49d072e]578 awaiter_t *waiter;
[c042bdd]579 struct timeval tv;
[80649a91]580
581 while (1) {
[bc1f1c2]582 if (fibril_schedule_next_adv(FIBRIL_FROM_MANAGER)) {
[a46da63]583 futex_up(&async_futex);
[36c9234]584 /*
585 * async_futex is always held when entering a manager
586 * fibril.
[a46da63]587 */
[80649a91]588 continue;
589 }
[c042bdd]590 futex_down(&async_futex);
591 if (!list_empty(&timeout_list)) {
[bc1f1c2]592 waiter = list_get_instance(timeout_list.next, awaiter_t,
593 link);
594 gettimeofday(&tv, NULL);
[49d072e]595 if (tv_gteq(&tv, &waiter->expires)) {
[6c46350]596 futex_up(&async_futex);
[c042bdd]597 handle_expired_timeouts();
598 continue;
599 } else
[49d072e]600 timeout = tv_sub(&waiter->expires, &tv);
[c042bdd]601 } else
[0b99e40]602 timeout = SYNCH_NO_TIMEOUT;
[c042bdd]603 futex_up(&async_futex);
604
[2d22049]605 callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
[0b99e40]606
607 if (!callid) {
[c042bdd]608 handle_expired_timeouts();
[0b99e40]609 continue;
610 }
[80649a91]611
[085bd54]612 if (callid & IPC_CALLID_ANSWERED) {
[80649a91]613 continue;
[085bd54]614 }
[01ff41c]615
[80649a91]616 handle_call(callid, &call);
617 }
[a46da63]618
619 return 0;
[80649a91]620}
621
[36c9234]622/** Function to start async_manager as a standalone fibril.
[a2cd194]623 *
[36c9234]624 * When more kernel threads are used, one async manager should exist per thread.
625 *
626 * @param arg Unused.
627 *
628 * @return Never returns.
[a2cd194]629 */
[9591265]630static int async_manager_fibril(void *arg)
[80649a91]631{
[a46da63]632 futex_up(&async_futex);
[36c9234]633 /*
634 * async_futex is always locked when entering manager
635 */
[085bd54]636 async_manager_worker();
[a46da63]637
638 return 0;
[80649a91]639}
[450cd3a]640
[36c9234]641/** Add one manager to manager list. */
[80649a91]642void async_create_manager(void)
[450cd3a]643{
[bc1f1c2]644 fid_t fid;
[80649a91]645
[9591265]646 fid = fibril_create(async_manager_fibril, NULL);
[bc1f1c2]647 fibril_add_manager(fid);
[80649a91]648}
649
650/** Remove one manager from manager list */
651void async_destroy_manager(void)
652{
[bc1f1c2]653 fibril_remove_manager();
[80649a91]654}
655
[36c9234]656/** Initialize the async framework.
657 *
658 * @return Zero on success or an error code.
659 */
[80649a91]660int _async_init(void)
661{
[bc1f1c2]662 if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
663 &conn_hash_table_ops)) {
[80649a91]664 printf("%s: cannot create hash table\n", "async");
665 return ENOMEM;
666 }
667
[a46da63]668 return 0;
[450cd3a]669}
[01ff41c]670
[36c9234]671/** Reply received callback.
[01ff41c]672 *
[36c9234]673 * This function is called whenever a reply for an asynchronous message sent out
674 * by the asynchronous framework is received.
675 *
676 * Notify the fibril which is waiting for this message that it has arrived.
677 *
678 * @param private Pointer to the asynchronous message record.
679 * @param retval Value returned in the answer.
680 * @param data Call data of the answer.
[01ff41c]681 */
[f2f0392]682static void reply_received(void *private, int retval, ipc_call_t *data)
[01ff41c]683{
684 amsg_t *msg = (amsg_t *) private;
685
686 msg->retval = retval;
687
688 futex_down(&async_futex);
[36c9234]689 /* Copy data after futex_down, just in case the call was detached */
[01ff41c]690 if (msg->dataptr)
691 *msg->dataptr = *data;
[0b99e40]692
[c042bdd]693 write_barrier();
694 /* Remove message from timeout list */
[49d072e]695 if (msg->wdata.inlist)
696 list_remove(&msg->wdata.link);
[01ff41c]697 msg->done = 1;
[36c9234]698 if (!msg->wdata.active) {
[49d072e]699 msg->wdata.active = 1;
[bc1f1c2]700 fibril_add_ready(msg->wdata.fid);
[01ff41c]701 }
702 futex_up(&async_futex);
703}
704
[36c9234]705/** Send message and return id of the sent message.
706 *
707 * The return value can be used as input for async_wait() to wait for
708 * completion.
[01ff41c]709 *
[36c9234]710 * @param phoneid Handle of the phone that will be used for the send.
711 * @param method Service-defined method.
712 * @param arg1 Service-defined payload argument.
713 * @param arg2 Service-defined payload argument.
714 * @param dataptr If non-NULL, storage where the reply data will be
715 * stored.
716 *
717 * @return Hash of the sent message.
[01ff41c]718 */
719aid_t async_send_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
[f2f0392]720 ipc_call_t *dataptr)
[01ff41c]721{
722 amsg_t *msg;
723
[085bd54]724 if (in_interrupt_handler) {
[bc1f1c2]725 printf("Cannot send asynchronous request in interrupt "
726 "handler.\n");
[085bd54]727 _exit(1);
728 }
729
[01ff41c]730 msg = malloc(sizeof(*msg));
731 msg->done = 0;
732 msg->dataptr = dataptr;
[49d072e]733
[36c9234]734 /* We may sleep in the next method, but it will use its own mechanism */
735 msg->wdata.active = 1;
736
[bc1f1c2]737 ipc_call_async_2(phoneid, method, arg1, arg2, msg, reply_received, 1);
[01ff41c]738
739 return (aid_t) msg;
740}
741
[90f5d64]742/** Send message and return id of the sent message
743 *
[36c9234]744 * The return value can be used as input for async_wait() to wait for
745 * completion.
746 *
747 * @param phoneid Handle of the phone that will be used for the send.
748 * @param method Service-defined method.
749 * @param arg1 Service-defined payload argument.
750 * @param arg2 Service-defined payload argument.
751 * @param arg3 Service-defined payload argument.
752 * @param dataptr If non-NULL, storage where the reply data will be
753 * stored.
754 *
755 * @return Hash of the sent message.
[90f5d64]756 */
757aid_t async_send_3(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
[f2f0392]758 ipcarg_t arg3, ipc_call_t *dataptr)
[90f5d64]759{
760 amsg_t *msg;
761
[085bd54]762 if (in_interrupt_handler) {
[36c9234]763 printf("Cannot send asynchronous request in interrupt "
764 "handler.\n");
[085bd54]765 _exit(1);
766 }
767
[90f5d64]768 msg = malloc(sizeof(*msg));
769 msg->done = 0;
770 msg->dataptr = dataptr;
771
[36c9234]772 /* We may sleep in next method, but it will use its own mechanism */
773 msg->wdata.active = 1;
774
[bc1f1c2]775 ipc_call_async_3(phoneid, method, arg1, arg2, arg3, msg, reply_received,
776 1);
[90f5d64]777
778 return (aid_t) msg;
779}
780
[36c9234]781/** Wait for a message sent by the async framework.
[01ff41c]782 *
[36c9234]783 * @param amsgid Hash of the message to wait for.
784 * @param retval Pointer to storage where the retval of the answer will
785 * be stored.
[01ff41c]786 */
787void async_wait_for(aid_t amsgid, ipcarg_t *retval)
788{
789 amsg_t *msg = (amsg_t *) amsgid;
790
791 futex_down(&async_futex);
792 if (msg->done) {
793 futex_up(&async_futex);
794 goto done;
795 }
796
[bc1f1c2]797 msg->wdata.fid = fibril_get_id();
[49d072e]798 msg->wdata.active = 0;
799 msg->wdata.inlist = 0;
[36c9234]800 /* Leave the async_futex locked when entering this function */
[bc1f1c2]801 fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
802 /* futex is up automatically after fibril_schedule_next...*/
[01ff41c]803done:
804 if (retval)
805 *retval = msg->retval;
806 free(msg);
807}
[0b99e40]808
[36c9234]809/** Wait for a message sent by the async framework, timeout variant.
[c042bdd]810 *
[36c9234]811 * @param amsgid Hash of the message to wait for.
812 * @param retval Pointer to storage where the retval of the answer will
813 * be stored.
814 * @param timeout Timeout in microseconds.
[c042bdd]815 *
[36c9234]816 * @return Zero on success, ETIMEOUT if the timeout has expired.
[c042bdd]817 */
818int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
819{
820 amsg_t *msg = (amsg_t *) amsgid;
[0b99e40]821
[86029498]822 /* TODO: Let it go through the event read at least once */
823 if (timeout < 0)
824 return ETIMEOUT;
825
[c042bdd]826 futex_down(&async_futex);
827 if (msg->done) {
828 futex_up(&async_futex);
829 goto done;
830 }
[0b99e40]831
[49d072e]832 gettimeofday(&msg->wdata.expires, NULL);
833 tv_add(&msg->wdata.expires, timeout);
834
[bc1f1c2]835 msg->wdata.fid = fibril_get_id();
[49d072e]836 msg->wdata.active = 0;
837 insert_timeout(&msg->wdata);
[0b99e40]838
[36c9234]839 /* Leave the async_futex locked when entering this function */
[bc1f1c2]840 fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
841 /* futex is up automatically after fibril_schedule_next...*/
[0b99e40]842
[c042bdd]843 if (!msg->done)
844 return ETIMEOUT;
[0b99e40]845
[c042bdd]846done:
847 if (retval)
848 *retval = msg->retval;
849 free(msg);
850
851 return 0;
852}
[0b99e40]853
[36c9234]854/** Wait for specified time.
[44c6d88d]855 *
[36c9234]856 * The current fibril is suspended but the thread continues to execute.
857 *
858 * @param timeout Duration of the wait in microseconds.
[44c6d88d]859 */
860void async_usleep(suseconds_t timeout)
861{
862 amsg_t *msg;
863
[085bd54]864 if (in_interrupt_handler) {
865 printf("Cannot call async_usleep in interrupt handler.\n");
866 _exit(1);
867 }
868
[44c6d88d]869 msg = malloc(sizeof(*msg));
870 if (!msg)
871 return;
872
[bc1f1c2]873 msg->wdata.fid = fibril_get_id();
[49d072e]874 msg->wdata.active = 0;
[44c6d88d]875
[49d072e]876 gettimeofday(&msg->wdata.expires, NULL);
877 tv_add(&msg->wdata.expires, timeout);
[44c6d88d]878
879 futex_down(&async_futex);
[49d072e]880 insert_timeout(&msg->wdata);
[36c9234]881 /* Leave the async_futex locked when entering this function */
[bc1f1c2]882 fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
[f2f0392]883 /* futex is up automatically after fibril_schedule_next_adv()...*/
[44c6d88d]884 free(msg);
885}
[da0c91e7]886
[36c9234]887/** Setter for client_connection function pointer.
[da0c91e7]888 *
[36c9234]889 * @param conn Function that will implement a new connection fibril.
[da0c91e7]890 */
891void async_set_client_connection(async_client_conn_t conn)
892{
893 client_connection = conn;
894}
[36c9234]895
896/** Setter for interrupt_received function pointer.
897 *
898 * @param conn Function that will implement a new interrupt
899 * notification fibril.
900 */
[51dbadf3]901void async_set_interrupt_received(async_client_conn_t conn)
902{
903 interrupt_received = conn;
904}
[085bd54]905
906/* Primitive functions for simple communication */
907void async_msg_3(int phoneid, ipcarg_t method, ipcarg_t arg1,
908 ipcarg_t arg2, ipcarg_t arg3)
909{
[bc1f1c2]910 ipc_call_async_3(phoneid, method, arg1, arg2, arg3, NULL, NULL,
911 !in_interrupt_handler);
[085bd54]912}
913
914void async_msg_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2)
915{
[bc1f1c2]916 ipc_call_async_2(phoneid, method, arg1, arg2, NULL, NULL,
917 !in_interrupt_handler);
[085bd54]918}
[b2951e2]919
[a46da63]920/** @}
[b2951e2]921 */
[36c9234]922
Note: See TracBrowser for help on using the repository browser.