Subversion Repositories HelenOS

Rev

Rev 1547 | Rev 1610 | Go to most recent revision | Blame | Compare with Previous | Last modification | View Log | Download | RSS feed

  1. /*
  2.  * Copyright (C) 2006 Ondrej Palkovsky
  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.
  27.  */
  28.  
  29. /**
  30.  * Asynchronous library
  31.  *
  32.  * The aim of this library is facilitating writing programs utilizing
  33.  * the asynchronous nature of Helenos IPC, yet using a normal way
  34.  * of programming.
  35.  *
  36.  * You should be able to write very simple multithreaded programs,
  37.  * the async framework will automatically take care of most synchronization
  38.  * problems.
  39.  *
  40.  * Default semantics:
  41.  * - send() - send asynchronously. If the kernel refuses to send more
  42.  *            messages, [ try to get responses from kernel, if nothing
  43.  *            found, might try synchronous ]
  44.  *
  45.  * Example of use:
  46.  *
  47.  * 1) Multithreaded client application
  48.  *  create_thread(thread1);
  49.  *  create_thread(thread2);
  50.  *  ...
  51.  *  
  52.  *  thread1() {
  53.  *        conn = ipc_connect_me_to();
  54.  *        c1 = send(conn);
  55.  *        c2 = send(conn);
  56.  *        wait_for(c1);
  57.  *        wait_for(c2);
  58.  *  }
  59.  *
  60.  *
  61.  * 2) Multithreaded server application
  62.  * main() {
  63.  *      async_manager();
  64.  * }
  65.  *
  66.  *
  67.  * client_connection(icallid, *icall) {
  68.  *       if (want_refuse) {
  69.  *           ipc_answer_fast(icallid, ELIMIT, 0, 0);
  70.  *           return;
  71.  *       }
  72.  *       ipc_answer_fast(icallid, 0, 0, 0);
  73.  *
  74.  *       callid = async_get_call(&call);
  75.  *       handle(callid, call);
  76.  *       ipc_answer_fast(callid, 1,2,3);
  77.  *
  78.  *       callid = async_get_call(&call);
  79.  *       ....
  80.  * }
  81.  *
  82.  * TODO: Detaching/joining dead psthreads?
  83.  */
  84. #include <futex.h>
  85. #include <async.h>
  86. #include <psthread.h>
  87. #include <stdio.h>
  88. #include <libadt/hash_table.h>
  89. #include <libadt/list.h>
  90. #include <ipc/ipc.h>
  91. #include <assert.h>
  92. #include <errno.h>
  93. #include <time.h>
  94. #include <arch/barrier.h>
  95.  
  96. atomic_t async_futex = FUTEX_INITIALIZER;
  97. static hash_table_t conn_hash_table;
  98. static LIST_INITIALIZE(timeout_list);
  99.  
  100. typedef struct {
  101.     struct timeval expires;      /**< Expiration time for waiting thread */
  102.     int inlist;             /**< If true, this struct is in timeout list */
  103.     link_t link;
  104.  
  105.     pstid_t ptid;                /**< Thread waiting for this message */
  106.     int active;                  /**< If this thread is currently active */
  107.     int timedout;                /**< If true, we timed out */
  108. } awaiter_t;
  109.  
  110. typedef struct {
  111.     awaiter_t wdata;
  112.  
  113.     int done;                    /**< If reply was received */
  114.     ipc_call_t *dataptr;         /**< Pointer where the answer data
  115.                       *   is stored */
  116.     ipcarg_t retval;
  117. } amsg_t;
  118.  
  119. typedef struct {
  120.     link_t link;
  121.     ipc_callid_t callid;
  122.     ipc_call_t call;
  123. } msg_t;
  124.  
  125. typedef struct {
  126.     awaiter_t wdata;
  127.  
  128.     link_t link;             /**< Hash table link */
  129.     ipcarg_t in_phone_hash;  /**< Incoming phone hash. */
  130.     link_t msg_queue;        /**< Messages that should be delivered to this thread */
  131.     /* Structures for connection opening packet */
  132.     ipc_callid_t callid;
  133.     ipc_call_t call;
  134.     void (*cthread)(ipc_callid_t,ipc_call_t *);
  135. } connection_t;
  136.  
  137.  
  138. __thread connection_t *PS_connection;
  139.  
  140. static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
  141. static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
  142. static async_client_conn_t client_connection = default_client_connection;
  143. static async_client_conn_t interrupt_received = default_interrupt_received;
  144.  
  145. /** Add microseconds to give timeval */
  146. static void tv_add(struct timeval *tv, suseconds_t usecs)
  147. {
  148.     tv->tv_sec += usecs / 1000000;
  149.     tv->tv_usec += usecs % 1000000;
  150.     if (tv->tv_usec > 1000000) {
  151.         tv->tv_sec++;
  152.         tv->tv_usec -= 1000000;
  153.     }
  154. }
  155.  
  156. /** Subtract 2 timevals, return microseconds difference */
  157. static suseconds_t tv_sub(struct timeval *tv1, struct timeval *tv2)
  158. {
  159.     suseconds_t result;
  160.  
  161.     result = tv1->tv_usec - tv2->tv_usec;
  162.     result += (tv1->tv_sec - tv2->tv_sec) * 1000000;
  163.  
  164.     return result;
  165. }
  166.  
  167. /** Compare timeval
  168.  *
  169.  * @return 1 if tv1 > tv2, otherwise 0
  170.  */
  171. static int tv_gt(struct timeval *tv1, struct timeval *tv2)
  172. {
  173.     if (tv1->tv_sec > tv2->tv_sec)
  174.         return 1;
  175.     if (tv1->tv_sec == tv2->tv_sec && tv1->tv_usec > tv2->tv_usec)
  176.         return 1;
  177.     return 0;
  178. }
  179. static int tv_gteq(struct timeval *tv1, struct timeval *tv2)
  180. {
  181.     if (tv1->tv_sec > tv2->tv_sec)
  182.         return 1;
  183.     if (tv1->tv_sec == tv2->tv_sec && tv1->tv_usec >= tv2->tv_usec)
  184.         return 1;
  185.     return 0;
  186. }
  187.  
  188. /* Hash table functions */
  189. #define CONN_HASH_TABLE_CHAINS  32
  190.  
  191. static hash_index_t conn_hash(unsigned long *key)
  192. {
  193.     assert(key);
  194.     return ((*key) >> 4) % CONN_HASH_TABLE_CHAINS;
  195. }
  196.  
  197. static int conn_compare(unsigned long key[], hash_count_t keys, link_t *item)
  198. {
  199.     connection_t *hs;
  200.  
  201.     hs = hash_table_get_instance(item, connection_t, link);
  202.    
  203.     return key[0] == hs->in_phone_hash;
  204. }
  205.  
  206. static void conn_remove(link_t *item)
  207. {
  208.     free(hash_table_get_instance(item, connection_t, link));
  209. }
  210.  
  211.  
  212. /** Operations for NS hash table. */
  213. static hash_table_operations_t conn_hash_table_ops = {
  214.     .hash = conn_hash,
  215.     .compare = conn_compare,
  216.     .remove_callback = conn_remove
  217. };
  218.  
  219. /** Insert sort timeout msg into timeouts list
  220.  *
  221.  */
  222. static void insert_timeout(awaiter_t *wd)
  223. {
  224.     link_t *tmp;
  225.     awaiter_t *cur;
  226.  
  227.     wd->timedout = 0;
  228.  
  229.     tmp = timeout_list.next;
  230.     while (tmp != &timeout_list) {
  231.         cur = list_get_instance(tmp, awaiter_t, link);
  232.         if (tv_gteq(&cur->expires, &wd->expires))
  233.             break;
  234.         tmp = tmp->next;
  235.     }
  236.     list_append(&wd->link, tmp);
  237. }
  238.  
  239. /*************************************************/
  240.  
  241. /** Try to route a call to an appropriate connection thread
  242.  *
  243.  */
  244. static int route_call(ipc_callid_t callid, ipc_call_t *call)
  245. {
  246.     connection_t *conn;
  247.     msg_t *msg;
  248.     link_t *hlp;
  249.     unsigned long key;
  250.  
  251.     futex_down(&async_futex);
  252.  
  253.     key = call->in_phone_hash;
  254.     hlp = hash_table_find(&conn_hash_table, &key);
  255.     if (!hlp) {
  256.         futex_up(&async_futex);
  257.         return 0;
  258.     }
  259.     conn = hash_table_get_instance(hlp, connection_t, link);
  260.  
  261.     msg = malloc(sizeof(*msg));
  262.     msg->callid = callid;
  263.     msg->call = *call;
  264.     list_append(&msg->link, &conn->msg_queue);
  265.    
  266.     /* If the call is waiting for event, run it */
  267.     if (!conn->wdata.active) {
  268.         /* If in timeout list, remove it */
  269.         if (conn->wdata.inlist) {
  270.             conn->wdata.inlist = 0;
  271.             list_remove(&conn->wdata.link);
  272.         }
  273.         conn->wdata.active = 1;
  274.         psthread_add_ready(conn->wdata.ptid);
  275.     }
  276.  
  277.     futex_up(&async_futex);
  278.  
  279.     return 1;
  280. }
  281.  
  282. /** Return new incoming message for current(thread-local) connection */
  283. ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
  284. {
  285.     msg_t *msg;
  286.     ipc_callid_t callid;
  287.     connection_t *conn;
  288.    
  289.     assert(PS_connection);
  290.     /* GCC 4.1.0 coughs on PS_connection-> dereference,
  291.      * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
  292.      *           I would never expect to find so many errors in
  293.      *           compiler *($&$(*&$
  294.      */
  295.     conn = PS_connection;
  296.  
  297.     if (usecs < 0) /* TODO: let it get through the ipc_call once */
  298.         return 0;
  299.  
  300.     futex_down(&async_futex);
  301.  
  302.     if (usecs) {
  303.         gettimeofday(&conn->wdata.expires, NULL);
  304.         tv_add(&conn->wdata.expires, usecs);
  305.     } else {
  306.         conn->wdata.inlist = 0;
  307.     }
  308.     /* If nothing in queue, wait until something appears */
  309.     while (list_empty(&conn->msg_queue)) {
  310.         if (usecs) {
  311.             conn->wdata.inlist = 1;
  312.             insert_timeout(&conn->wdata);
  313.         }
  314.         conn->wdata.active = 0;
  315.         psthread_schedule_next_adv(PS_TO_MANAGER);
  316.         /* Futex is up after getting back from async_manager
  317.          * get it again */
  318.         futex_down(&async_futex);
  319.         if (usecs && conn->wdata.timedout && \
  320.             list_empty(&conn->msg_queue)) {
  321.             /* If we timed out-> exit */
  322.             futex_up(&async_futex);
  323.             return 0;
  324.         }
  325.     }
  326.    
  327.     msg = list_get_instance(conn->msg_queue.next, msg_t, link);
  328.     list_remove(&msg->link);
  329.     callid = msg->callid;
  330.     *call = msg->call;
  331.     free(msg);
  332.    
  333.     futex_up(&async_futex);
  334.     return callid;
  335. }
  336.  
  337. /** Thread function that gets created on new connection
  338.  *
  339.  * This function is defined as a weak symbol - to be redefined in
  340.  * user code.
  341.  */
  342. static void default_client_connection(ipc_callid_t callid, ipc_call_t *call)
  343. {
  344.     ipc_answer_fast(callid, ENOENT, 0, 0);
  345. }
  346. static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
  347. {
  348. }
  349.  
  350. /** Wrapper for client connection thread
  351.  *
  352.  * When new connection arrives, thread with this function is created.
  353.  * It calls client_connection and does final cleanup.
  354.  *
  355.  * @parameter arg Connection structure pointer
  356.  */
  357. static int connection_thread(void  *arg)
  358. {
  359.     unsigned long key;
  360.     msg_t *msg;
  361.  
  362.     /* Setup thread local connection pointer */
  363.     PS_connection = (connection_t *)arg;
  364.     PS_connection->cthread(PS_connection->callid, &PS_connection->call);
  365.  
  366.     /* Remove myself from connection hash table */
  367.     futex_down(&async_futex);
  368.     key = PS_connection->in_phone_hash;
  369.     hash_table_remove(&conn_hash_table, &key, 1);
  370.     futex_up(&async_futex);
  371.     /* Answer all remaining messages with ehangup */
  372.     while (!list_empty(&PS_connection->msg_queue)) {
  373.         msg = list_get_instance(PS_connection->msg_queue.next, msg_t, link);
  374.         list_remove(&msg->link);
  375.         ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
  376.         free(msg);
  377.     }
  378. }
  379.  
  380. /** Create new thread for a new connection
  381.  *
  382.  * Creates new thread for connection, fills in connection
  383.  * structures and inserts it into the hash table, so that
  384.  * later we can easily do routing of messages to particular
  385.  * threads.
  386.  *
  387.  * @param in_phone_hash Identification of the incoming connection
  388.  * @param callid Callid of the IPC_M_CONNECT_ME_TO packet
  389.  * @param call Call data of the opening packet
  390.  * @param cthread Thread function that should be called upon
  391.  *                opening the connection
  392.  * @return New thread id
  393.  */
  394. pstid_t async_new_connection(ipcarg_t in_phone_hash,ipc_callid_t callid,
  395.                  ipc_call_t *call,
  396.                  void (*cthread)(ipc_callid_t,ipc_call_t *))
  397. {
  398.     pstid_t ptid;
  399.     connection_t *conn;
  400.     unsigned long key;
  401.  
  402.     conn = malloc(sizeof(*conn));
  403.     if (!conn) {
  404.         ipc_answer_fast(callid, ENOMEM, 0, 0);
  405.         return NULL;
  406.     }
  407.     conn->in_phone_hash = in_phone_hash;
  408.     list_initialize(&conn->msg_queue);
  409.     conn->callid = callid;
  410.     if (call)
  411.         conn->call = *call;
  412.     conn->wdata.active = 1; /* We will activate it asap */
  413.     conn->cthread = cthread;
  414.  
  415.     conn->wdata.ptid = psthread_create(connection_thread, conn);
  416.     if (!conn->wdata.ptid) {
  417.         free(conn);
  418.         ipc_answer_fast(callid, ENOMEM, 0, 0);
  419.         return NULL;
  420.     }
  421.     /* Add connection to hash table */
  422.     key = conn->in_phone_hash;
  423.     futex_down(&async_futex);
  424.     hash_table_insert(&conn_hash_table, &key, &conn->link);
  425.     futex_up(&async_futex);
  426.  
  427.     psthread_add_ready(conn->wdata.ptid);
  428.  
  429.     return conn->wdata.ptid;
  430. }
  431.  
  432. /** Handle call that was received */
  433. static void handle_call(ipc_callid_t callid, ipc_call_t *call)
  434. {
  435.     /* Unrouted call - do some default behaviour */
  436.     switch (IPC_GET_METHOD(*call)) {
  437.     case IPC_M_INTERRUPT:
  438.         (*interrupt_received)(callid,call);
  439.         return;
  440.     case IPC_M_CONNECT_ME_TO:
  441.         /* Open new connection with thread etc. */
  442.         async_new_connection(IPC_GET_ARG3(*call), callid, call, client_connection);
  443.         return;
  444.     }
  445.  
  446.     /* Try to route call through connection tables */
  447.     if (route_call(callid, call))
  448.         return;
  449.  
  450.     /* Unknown call from unknown phone - hang it up */
  451.     ipc_answer_fast(callid, EHANGUP, 0, 0);
  452. }
  453.  
  454. /** Fire all timeouts that expired
  455.  *
  456.  */
  457. static void handle_expired_timeouts(void)
  458. {
  459.     struct timeval tv;
  460.     awaiter_t *waiter;
  461.     link_t *cur;
  462.  
  463.     gettimeofday(&tv,NULL);
  464.     futex_down(&async_futex);
  465.  
  466.     cur = timeout_list.next;
  467.     while (cur != &timeout_list) {
  468.         waiter = list_get_instance(cur,awaiter_t,link);
  469.         if (tv_gt(&waiter->expires, &tv))
  470.             break;
  471.         cur = cur->next;
  472.         list_remove(&waiter->link);
  473.         waiter->inlist = 0;
  474.         waiter->timedout = 1;
  475.         /* Redundant condition? The thread should not
  476.          * be active when it gets here.
  477.          */
  478.         if (!waiter->active) {
  479.             waiter->active = 1;
  480.             psthread_add_ready(waiter->ptid);
  481.         }
  482.     }
  483.  
  484.     futex_up(&async_futex);
  485. }
  486.  
  487. /** Endless loop dispatching incoming calls and answers */
  488. int async_manager(void)
  489. {
  490.     ipc_call_t call;
  491.     ipc_callid_t callid;
  492.     int timeout;
  493.     awaiter_t *waiter;
  494.     struct timeval tv;
  495.  
  496.     while (1) {
  497.         if (psthread_schedule_next_adv(PS_FROM_MANAGER)) {
  498.             futex_up(&async_futex); /* async_futex is always held
  499.                         * when entering manager thread
  500.                         */
  501.             continue;
  502.         }
  503.         futex_down(&async_futex);
  504.         if (!list_empty(&timeout_list)) {
  505.             waiter = list_get_instance(timeout_list.next,awaiter_t,link);
  506.             gettimeofday(&tv,NULL);
  507.             if (tv_gteq(&tv, &waiter->expires)) {
  508.                 futex_up(&async_futex);
  509.                 handle_expired_timeouts();
  510.                 continue;
  511.             } else
  512.                 timeout = tv_sub(&waiter->expires, &tv);
  513.         } else
  514.             timeout = SYNCH_NO_TIMEOUT;
  515.         futex_up(&async_futex);
  516.  
  517.         callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
  518.  
  519.         if (!callid) {
  520.             handle_expired_timeouts();
  521.             continue;
  522.         }
  523.  
  524.         if (callid & IPC_CALLID_ANSWERED)
  525.             continue;
  526.  
  527.         handle_call(callid, &call);
  528.     }
  529. }
  530.  
  531. /** Function to start async_manager as a standalone thread
  532.  *
  533.  * When more kernel threads are used, one async manager should
  534.  * exist per thread. The particular implementation may change,
  535.  * currently one async_manager is started automatically per kernel
  536.  * thread except main thread.
  537.  */
  538. static int async_manager_thread(void *arg)
  539. {
  540.     futex_up(&async_futex); /* async_futex is always locked when entering
  541.                 * manager */
  542.     async_manager();
  543. }
  544.  
  545. /** Add one manager to manager list */
  546. void async_create_manager(void)
  547. {
  548.     pstid_t ptid;
  549.  
  550.     ptid = psthread_create(async_manager_thread, NULL);
  551.     psthread_add_manager(ptid);
  552. }
  553.  
  554. /** Remove one manager from manager list */
  555. void async_destroy_manager(void)
  556. {
  557.     psthread_remove_manager();
  558. }
  559.  
  560. /** Initialize internal structures needed for async manager */
  561. int _async_init(void)
  562. {
  563.     if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1, &conn_hash_table_ops)) {
  564.         printf("%s: cannot create hash table\n", "async");
  565.         return ENOMEM;
  566.     }
  567.    
  568. }
  569.  
  570. /** IPC handler for messages in async framework
  571.  *
  572.  * Notify thread that is waiting for this message, that it arrived
  573.  */
  574. static void reply_received(void *private, int retval,
  575.                ipc_call_t *data)
  576. {
  577.     amsg_t *msg = (amsg_t *) private;
  578.  
  579.     msg->retval = retval;
  580.  
  581.     futex_down(&async_futex);
  582.     /* Copy data after futex_down, just in case the
  583.      * call was detached
  584.      */
  585.     if (msg->dataptr)
  586.         *msg->dataptr = *data;
  587.  
  588.     write_barrier();
  589.     /* Remove message from timeout list */
  590.     if (msg->wdata.inlist)
  591.         list_remove(&msg->wdata.link);
  592.     msg->done = 1;
  593.     if (! msg->wdata.active) {
  594.         msg->wdata.active = 1;
  595.         psthread_add_ready(msg->wdata.ptid);
  596.     }
  597.     futex_up(&async_futex);
  598. }
  599.  
  600. /** Send message and return id of the sent message
  601.  *
  602.  * The return value can be used as input for async_wait() to wait
  603.  * for completion.
  604.  */
  605. aid_t async_send_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
  606.            ipc_call_t *dataptr)
  607. {
  608.     amsg_t *msg;
  609.  
  610.     msg = malloc(sizeof(*msg));
  611.     msg->done = 0;
  612.     msg->dataptr = dataptr;
  613.  
  614.     msg->wdata.active = 1; /* We may sleep in next method, but it
  615.                 * will use it's own mechanism */
  616.     ipc_call_async_2(phoneid,method,arg1,arg2,msg,reply_received,1);
  617.  
  618.     return (aid_t) msg;
  619. }
  620.  
  621. /** Send message and return id of the sent message
  622.  *
  623.  * The return value can be used as input for async_wait() to wait
  624.  * for completion.
  625.  */
  626. aid_t async_send_3(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
  627.            ipcarg_t arg3, ipc_call_t *dataptr)
  628. {
  629.     amsg_t *msg;
  630.  
  631.     msg = malloc(sizeof(*msg));
  632.     msg->done = 0;
  633.     msg->dataptr = dataptr;
  634.  
  635.     msg->wdata.active = 1; /* We may sleep in next method, but it
  636.                 * will use it's own mechanism */
  637.     ipc_call_async_3(phoneid,method,arg1,arg2,arg3, msg,reply_received,1);
  638.  
  639.     return (aid_t) msg;
  640. }
  641.  
  642. /** Wait for a message sent by async framework
  643.  *
  644.  * @param amsgid Message ID to wait for
  645.  * @param retval Pointer to variable where will be stored retval
  646.  *               of the answered message. If NULL, it is ignored.
  647.  *
  648.  */
  649. void async_wait_for(aid_t amsgid, ipcarg_t *retval)
  650. {
  651.     amsg_t *msg = (amsg_t *) amsgid;
  652.     connection_t *conn;
  653.  
  654.     futex_down(&async_futex);
  655.     if (msg->done) {
  656.         futex_up(&async_futex);
  657.         goto done;
  658.     }
  659.  
  660.     msg->wdata.ptid = psthread_get_id();
  661.     msg->wdata.active = 0;
  662.     msg->wdata.inlist = 0;
  663.     /* Leave locked async_futex when entering this function */
  664.     psthread_schedule_next_adv(PS_TO_MANAGER);
  665.     /* futex is up automatically after psthread_schedule_next...*/
  666. done:
  667.     if (retval)
  668.         *retval = msg->retval;
  669.     free(msg);
  670. }
  671.  
  672. /** Wait for a message sent by async framework with timeout
  673.  *
  674.  * @param amsgid Message ID to wait for
  675.  * @param retval Pointer to variable where will be stored retval
  676.  *               of the answered message. If NULL, it is ignored.
  677.  * @param timeout Timeout in usecs
  678.  * @return 0 on success, ETIMEOUT if timeout expired
  679.  *
  680.  */
  681. int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
  682. {
  683.     amsg_t *msg = (amsg_t *) amsgid;
  684.     connection_t *conn;
  685.  
  686.     /* TODO: Let it go through the event read at least once */
  687.     if (timeout < 0)
  688.         return ETIMEOUT;
  689.  
  690.     futex_down(&async_futex);
  691.     if (msg->done) {
  692.         futex_up(&async_futex);
  693.         goto done;
  694.     }
  695.  
  696.     gettimeofday(&msg->wdata.expires, NULL);
  697.     tv_add(&msg->wdata.expires, timeout);
  698.  
  699.     msg->wdata.ptid = psthread_get_id();
  700.     msg->wdata.active = 0;
  701.     msg->wdata.inlist = 1;
  702.  
  703.     insert_timeout(&msg->wdata);
  704.  
  705.     /* Leave locked async_futex when entering this function */
  706.     psthread_schedule_next_adv(PS_TO_MANAGER);
  707.     /* futex is up automatically after psthread_schedule_next...*/
  708.  
  709.     if (!msg->done)
  710.         return ETIMEOUT;
  711.  
  712. done:
  713.     if (retval)
  714.         *retval = msg->retval;
  715.     free(msg);
  716.  
  717.     return 0;
  718. }
  719.  
  720. /** Wait specified time, but in the meantime handle incoming events
  721.  *
  722.  * @param timeout Time in microseconds to wait
  723.  */
  724. void async_usleep(suseconds_t timeout)
  725. {
  726.     amsg_t *msg;
  727.    
  728.     msg = malloc(sizeof(*msg));
  729.     if (!msg)
  730.         return;
  731.  
  732.     msg->wdata.ptid = psthread_get_id();
  733.     msg->wdata.inlist = 1;
  734.     msg->wdata.active = 0;
  735.  
  736.     gettimeofday(&msg->wdata.expires, NULL);
  737.     tv_add(&msg->wdata.expires, timeout);
  738.  
  739.     futex_down(&async_futex);
  740.     insert_timeout(&msg->wdata);
  741.     /* Leave locked async_futex when entering this function */
  742.     psthread_schedule_next_adv(PS_TO_MANAGER);
  743.     /* futex is up automatically after psthread_schedule_next...*/
  744.     free(msg);
  745. }
  746.  
  747. /** Set function that is called, IPC_M_CONNECT_ME_TO is received
  748.  *
  749.  * @param conn Function that will form new psthread.
  750.  */
  751. void async_set_client_connection(async_client_conn_t conn)
  752. {
  753.     client_connection = conn;
  754. }
  755. void async_set_interrupt_received(async_client_conn_t conn)
  756. {
  757.     interrupt_received = conn;
  758. }
  759.