Subversion Repositories HelenOS

Rev

Rev 3386 | Rev 4327 | 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. /** @addtogroup libc
  30.  * @{
  31.  */
  32. /** @file
  33.  */
  34.  
  35. /**
  36.  * Asynchronous library
  37.  *
  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.
  40.  *
  41.  * You should be able to write very simple multithreaded programs, the async
  42.  * framework will automatically take care of most synchronization problems.
  43.  *
  44.  * Default semantics:
  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 ]
  48.  *
  49.  * Example of use (pseudo C):
  50.  *
  51.  * 1) Multithreaded client application
  52.  *
  53.  * fibril_create(fibril1, ...);
  54.  * fibril_create(fibril2, ...);
  55.  * ...
  56.  *  
  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.  * }
  66.  *
  67.  *
  68.  * 2) Multithreaded server application
  69.  * main()
  70.  * {
  71.  *  async_manager();
  72.  * }
  73.  *
  74.  *
  75.  * my_client_connection(icallid, *icall)
  76.  * {
  77.  *  if (want_refuse) {
  78.  *      ipc_answer_0(icallid, ELIMIT);
  79.  *      return;
  80.  *  }
  81.  *  ipc_answer_0(icallid, EOK);
  82.  *
  83.  *  callid = async_get_call(&call);
  84.  *  handle_call(callid, call);
  85.  *  ipc_answer_2(callid, 1, 2, 3);
  86.  *
  87.  *  callid = async_get_call(&call);
  88.  *  ....
  89.  * }
  90.  *
  91.  */
  92.  
  93. #include <futex.h>
  94. #include <async.h>
  95. #include <fibril.h>
  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>
  102. #include <sys/time.h>
  103. #include <arch/barrier.h>
  104. #include <bool.h>
  105.  
  106. atomic_t async_futex = FUTEX_INITIALIZER;
  107. static hash_table_t conn_hash_table;
  108. static LIST_INITIALIZE(timeout_list);
  109.  
  110. /** Structures of this type represent a waiting fibril. */
  111. typedef struct {
  112.     /** Expiration time. */
  113.     struct timeval expires;    
  114.     /** If true, this struct is in the timeout list. */
  115.     int inlist;
  116.     /** Timeout list link. */
  117.     link_t link;
  118.  
  119.     /** Identification of and link to the waiting fibril. */
  120.     fid_t fid;
  121.     /** If true, this fibril is currently active. */
  122.     int active;
  123.     /** If true, we have timed out. */
  124.     int timedout;
  125. } awaiter_t;
  126.  
  127. typedef struct {
  128.     awaiter_t wdata;
  129.    
  130.     /** If reply was received. */
  131.     int done;
  132.     /** Pointer to where the answer data is stored. */
  133.     ipc_call_t *dataptr;
  134.  
  135.     ipcarg_t retval;
  136. } amsg_t;
  137.  
  138. /**
  139.  * Structures of this type are used to group information about a call and a
  140.  * message queue link.
  141.  */
  142. typedef struct {
  143.     link_t link;
  144.     ipc_callid_t callid;
  145.     ipc_call_t call;
  146. } msg_t;
  147.  
  148. typedef struct {
  149.     awaiter_t wdata;
  150.  
  151.     /** Hash table link. */
  152.     link_t link;
  153.  
  154.     /** Incoming phone hash. */
  155.     ipcarg_t in_phone_hash;    
  156.  
  157.     /** Messages that should be delivered to this fibril. */
  158.     link_t msg_queue;      
  159.                      
  160.     /** Identification of the opening call. */
  161.     ipc_callid_t callid;
  162.     /** Call data of the opening call. */
  163.     ipc_call_t call;
  164.  
  165.     /** Identification of the closing call. */
  166.     ipc_callid_t close_callid;
  167.  
  168.     /** Fibril function that will be used to handle the connection. */
  169.     void (*cfibril)(ipc_callid_t, ipc_call_t *);
  170. } connection_t;
  171.  
  172. /** Identifier of the incoming connection handled by the current fibril. */
  173. __thread connection_t *FIBRIL_connection;
  174.  
  175. /**
  176.  * If true, it is forbidden to use async_req functions and all preemption is
  177.  * disabled.
  178.  */
  179. __thread int _in_interrupt_handler;
  180.  
  181. static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
  182. static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
  183.  
  184. /**
  185.  * Pointer to a fibril function that will be used to handle connections.
  186.  */
  187. static async_client_conn_t client_connection = default_client_connection;
  188. /**
  189.  * Pointer to a fibril function that will be used to handle interrupt
  190.  * notifications.
  191.  */
  192. static async_client_conn_t interrupt_received = default_interrupt_received;
  193.  
  194. /*
  195.  * Getter for _in_interrupt_handler. We need to export the value of this thread
  196.  * local variable to other modules, but the binutils 2.18 linkers die on an
  197.  * attempt to export this symbol in the header file. For now, consider this as a
  198.  * workaround.
  199.  */
  200. bool in_interrupt_handler(void)
  201. {
  202.     return _in_interrupt_handler;
  203. }
  204.  
  205. #define CONN_HASH_TABLE_CHAINS  32
  206.  
  207. /** Compute hash into the connection hash table based on the source phone hash.
  208.  *
  209.  * @param key       Pointer to source phone hash.
  210.  *
  211.  * @return      Index into the connection hash table.
  212.  */
  213. static hash_index_t conn_hash(unsigned long *key)
  214. {
  215.     assert(key);
  216.     return ((*key) >> 4) % CONN_HASH_TABLE_CHAINS;
  217. }
  218.  
  219. /** Compare hash table item with a key.
  220.  *
  221.  * @param key       Array containing the source phone hash as the only item.
  222.  * @param keys      Expected 1 but ignored.
  223.  * @param item      Connection hash table item.
  224.  *
  225.  * @return      True on match, false otherwise.
  226.  */
  227. static int conn_compare(unsigned long key[], hash_count_t keys, link_t *item)
  228. {
  229.     connection_t *hs;
  230.  
  231.     hs = hash_table_get_instance(item, connection_t, link);
  232.    
  233.     return key[0] == hs->in_phone_hash;
  234. }
  235.  
  236. /** Connection hash table removal callback function.
  237.  *
  238.  * This function is called whenever a connection is removed from the connection
  239.  * hash table.
  240.  *
  241.  * @param item      Connection hash table item being removed.
  242.  */
  243. static void conn_remove(link_t *item)
  244. {
  245.     free(hash_table_get_instance(item, connection_t, link));
  246. }
  247.  
  248.  
  249. /** Operations for the connection hash table. */
  250. static hash_table_operations_t conn_hash_table_ops = {
  251.     .hash = conn_hash,
  252.     .compare = conn_compare,
  253.     .remove_callback = conn_remove
  254. };
  255.  
  256. /** Sort in current fibril's timeout request.
  257.  *
  258.  * @param wd        Wait data of the current fibril.
  259.  */
  260. static void insert_timeout(awaiter_t *wd)
  261. {
  262.     link_t *tmp;
  263.     awaiter_t *cur;
  264.  
  265.     wd->timedout = 0;
  266.     wd->inlist = 1;
  267.  
  268.     tmp = timeout_list.next;
  269.     while (tmp != &timeout_list) {
  270.         cur = list_get_instance(tmp, awaiter_t, link);
  271.         if (tv_gteq(&cur->expires, &wd->expires))
  272.             break;
  273.         tmp = tmp->next;
  274.     }
  275.     list_append(&wd->link, tmp);
  276. }
  277.  
  278. /** Try to route a call to an appropriate connection fibril.
  279.  *
  280.  * If the proper connection fibril is found, a message with the call is added to
  281.  * its message queue. If the fibril was not active, it is activated and all
  282.  * timeouts are unregistered.
  283.  *
  284.  * @param callid    Hash of the incoming call.
  285.  * @param call      Data of the incoming call.
  286.  *
  287.  * @return      Zero if the call doesn't match any connection.
  288.  *          One if the call was passed to the respective connection
  289.  *          fibril.
  290.  */
  291. static int route_call(ipc_callid_t callid, ipc_call_t *call)
  292. {
  293.     connection_t *conn;
  294.     msg_t *msg;
  295.     link_t *hlp;
  296.     unsigned long key;
  297.  
  298.     futex_down(&async_futex);
  299.  
  300.     key = call->in_phone_hash;
  301.     hlp = hash_table_find(&conn_hash_table, &key);
  302.     if (!hlp) {
  303.         futex_up(&async_futex);
  304.         return 0;
  305.     }
  306.     conn = hash_table_get_instance(hlp, connection_t, link);
  307.  
  308.     msg = malloc(sizeof(*msg));
  309.     msg->callid = callid;
  310.     msg->call = *call;
  311.     list_append(&msg->link, &conn->msg_queue);
  312.  
  313.     if (IPC_GET_METHOD(*call) == IPC_M_PHONE_HUNGUP)
  314.         conn->close_callid = callid;
  315.    
  316.     /* If the connection fibril is waiting for an event, activate it */
  317.     if (!conn->wdata.active) {
  318.         /* If in timeout list, remove it */
  319.         if (conn->wdata.inlist) {
  320.             conn->wdata.inlist = 0;
  321.             list_remove(&conn->wdata.link);
  322.         }
  323.         conn->wdata.active = 1;
  324.         fibril_add_ready(conn->wdata.fid);
  325.     }
  326.  
  327.     futex_up(&async_futex);
  328.  
  329.     return 1;
  330. }
  331.  
  332. /** Return new incoming message for the current (fibril-local) connection.
  333.  *
  334.  * @param call      Storage where the incoming call data will be stored.
  335.  * @param usecs     Timeout in microseconds. Zero denotes no timeout.
  336.  *
  337.  * @return      If no timeout was specified, then a hash of the
  338.  *          incoming call is returned. If a timeout is specified,
  339.  *          then a hash of the incoming call is returned unless
  340.  *          the timeout expires prior to receiving a message. In
  341.  *          that case zero is returned.
  342.  */
  343. ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
  344. {
  345.     msg_t *msg;
  346.     ipc_callid_t callid;
  347.     connection_t *conn;
  348.    
  349.     assert(FIBRIL_connection);
  350.     /* GCC 4.1.0 coughs on FIBRIL_connection-> dereference,
  351.      * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
  352.      *           I would never expect to find so many errors in
  353.      *           a compiler *($&$(*&$
  354.      */
  355.     conn = FIBRIL_connection;
  356.  
  357.     futex_down(&async_futex);
  358.  
  359.     if (usecs) {
  360.         gettimeofday(&conn->wdata.expires, NULL);
  361.         tv_add(&conn->wdata.expires, usecs);
  362.     } else {
  363.         conn->wdata.inlist = 0;
  364.     }
  365.     /* If nothing in queue, wait until something arrives */
  366.     while (list_empty(&conn->msg_queue)) {
  367.         if (usecs)
  368.             insert_timeout(&conn->wdata);
  369.  
  370.         conn->wdata.active = 0;
  371.         /*
  372.          * Note: the current fibril will be rescheduled either due to a
  373.          * timeout or due to an arriving message destined to it. In the
  374.          * former case, handle_expired_timeouts() and, in the latter
  375.          * case, route_call() will perform the wakeup.
  376.          */
  377.         fibril_switch(FIBRIL_TO_MANAGER);
  378.         /*
  379.          * Futex is up after getting back from async_manager get it
  380.          * again.
  381.          */
  382.         futex_down(&async_futex);
  383.         if (usecs && conn->wdata.timedout &&
  384.             list_empty(&conn->msg_queue)) {
  385.             /* If we timed out -> exit */
  386.             futex_up(&async_futex);
  387.             return 0;
  388.         }
  389.     }
  390.    
  391.     msg = list_get_instance(conn->msg_queue.next, msg_t, link);
  392.     list_remove(&msg->link);
  393.     callid = msg->callid;
  394.     *call = msg->call;
  395.     free(msg);
  396.    
  397.     futex_up(&async_futex);
  398.     return callid;
  399. }
  400.  
  401. /** Default fibril function that gets called to handle new connection.
  402.  *
  403.  * This function is defined as a weak symbol - to be redefined in user code.
  404.  *
  405.  * @param callid    Hash of the incoming call.
  406.  * @param call      Data of the incoming call.
  407.  */
  408. static void default_client_connection(ipc_callid_t callid, ipc_call_t *call)
  409. {
  410.     ipc_answer_0(callid, ENOENT);
  411. }
  412.  
  413. /** Default fibril function that gets called to handle interrupt notifications.
  414.  *
  415.  * @param callid    Hash of the incoming call.
  416.  * @param call      Data of the incoming call.
  417.  */
  418. static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
  419. {
  420. }
  421.  
  422. /** Wrapper for client connection fibril.
  423.  *
  424.  * When a new connection arrives, a fibril with this implementing function is
  425.  * created. It calls client_connection() and does the final cleanup.
  426.  *
  427.  * @param arg       Connection structure pointer.
  428.  *
  429.  * @return      Always zero.
  430.  */
  431. static int connection_fibril(void  *arg)
  432. {
  433.     unsigned long key;
  434.     msg_t *msg;
  435.     int close_answered = 0;
  436.  
  437.     /* Setup fibril-local connection pointer */
  438.     FIBRIL_connection = (connection_t *) arg;
  439.     FIBRIL_connection->cfibril(FIBRIL_connection->callid,
  440.         &FIBRIL_connection->call);
  441.    
  442.     /* Remove myself from the connection hash table */
  443.     futex_down(&async_futex);
  444.     key = FIBRIL_connection->in_phone_hash;
  445.     hash_table_remove(&conn_hash_table, &key, 1);
  446.     futex_up(&async_futex);
  447.    
  448.     /* Answer all remaining messages with EHANGUP */
  449.     while (!list_empty(&FIBRIL_connection->msg_queue)) {
  450.         msg = list_get_instance(FIBRIL_connection->msg_queue.next,
  451.             msg_t, link);
  452.         list_remove(&msg->link);
  453.         if (msg->callid == FIBRIL_connection->close_callid)
  454.             close_answered = 1;
  455.         ipc_answer_0(msg->callid, EHANGUP);
  456.         free(msg);
  457.     }
  458.     if (FIBRIL_connection->close_callid)
  459.         ipc_answer_0(FIBRIL_connection->close_callid, EOK);
  460.    
  461.     return 0;
  462. }
  463.  
  464. /** Create a new fibril for a new connection.
  465.  *
  466.  * Creates new fibril for connection, fills in connection structures and inserts
  467.  * it into the hash table, so that later we can easily do routing of messages to
  468.  * particular fibrils.
  469.  *
  470.  * @param in_phone_hash Identification of the incoming connection.
  471.  * @param callid    Hash of the opening IPC_M_CONNECT_ME_TO call.
  472.  *          If callid is zero, the connection was opened by
  473.  *          accepting the IPC_M_CONNECT_TO_ME call and this function
  474.  *          is called directly by the server.
  475.  * @param call      Call data of the opening call.
  476.  * @param cfibril   Fibril function that should be called upon opening the
  477.  *          connection.
  478.  *
  479.  * @return      New fibril id or NULL on failure.
  480.  */
  481. fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
  482.     ipc_call_t *call, void (*cfibril)(ipc_callid_t, ipc_call_t *))
  483. {
  484.     connection_t *conn;
  485.     unsigned long key;
  486.    
  487.     conn = malloc(sizeof(*conn));
  488.     if (!conn) {
  489.         if (callid)
  490.             ipc_answer_0(callid, ENOMEM);
  491.         return NULL;
  492.     }
  493.     conn->in_phone_hash = in_phone_hash;
  494.     list_initialize(&conn->msg_queue);
  495.     conn->callid = callid;
  496.     conn->close_callid = 0;
  497.     if (call)
  498.         conn->call = *call;
  499.     conn->wdata.active = 1; /* We will activate the fibril ASAP */
  500.     conn->cfibril = cfibril;
  501.    
  502.     conn->wdata.fid = fibril_create(connection_fibril, conn);
  503.     if (!conn->wdata.fid) {
  504.         free(conn);
  505.         if (callid)
  506.             ipc_answer_0(callid, ENOMEM);
  507.         return NULL;
  508.     }
  509.    
  510.     /* Add connection to the connection hash table */
  511.     key = conn->in_phone_hash;
  512.     futex_down(&async_futex);
  513.     hash_table_insert(&conn_hash_table, &key, &conn->link);
  514.     futex_up(&async_futex);
  515.    
  516.     fibril_add_ready(conn->wdata.fid);
  517.    
  518.     return conn->wdata.fid;
  519. }
  520.  
  521. /** Handle a call that was received.
  522.  *
  523.  * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
  524.  * Otherwise the call is routed to its connection fibril.
  525.  *
  526.  * @param callid    Hash of the incoming call.
  527.  * @param call      Data of the incoming call.
  528.  *
  529.  */
  530. static void handle_call(ipc_callid_t callid, ipc_call_t *call)
  531. {
  532.     /* Unrouted call - do some default behaviour */
  533.     if ((callid & IPC_CALLID_NOTIFICATION)) {
  534.         _in_interrupt_handler = 1;
  535.         (*interrupt_received)(callid, call);
  536.         _in_interrupt_handler = 0;
  537.         return;
  538.     }
  539.    
  540.     switch (IPC_GET_METHOD(*call)) {
  541.     case IPC_M_CONNECT_ME_TO:
  542.         /* Open new connection with fibril etc. */
  543.         async_new_connection(IPC_GET_ARG5(*call), callid, call,
  544.             client_connection);
  545.         return;
  546.     }
  547.    
  548.     /* Try to route the call through the connection hash table */
  549.     if (route_call(callid, call))
  550.         return;
  551.    
  552.     /* Unknown call from unknown phone - hang it up */
  553.     ipc_answer_0(callid, EHANGUP);
  554. }
  555.  
  556. /** Fire all timeouts that expired. */
  557. static void handle_expired_timeouts(void)
  558. {
  559.     struct timeval tv;
  560.     awaiter_t *waiter;
  561.     link_t *cur;
  562.  
  563.     gettimeofday(&tv, NULL);
  564.     futex_down(&async_futex);
  565.  
  566.     cur = timeout_list.next;
  567.     while (cur != &timeout_list) {
  568.         waiter = list_get_instance(cur, awaiter_t, link);
  569.         if (tv_gt(&waiter->expires, &tv))
  570.             break;
  571.         cur = cur->next;
  572.         list_remove(&waiter->link);
  573.         waiter->inlist = 0;
  574.         waiter->timedout = 1;
  575.         /*
  576.          * Redundant condition?
  577.          * The fibril should not be active when it gets here.
  578.          */
  579.         if (!waiter->active) {
  580.             waiter->active = 1;
  581.             fibril_add_ready(waiter->fid);
  582.         }
  583.     }
  584.  
  585.     futex_up(&async_futex);
  586. }
  587.  
  588. /** Endless loop dispatching incoming calls and answers.
  589.  *
  590.  * @return      Never returns.
  591.  */
  592. static int async_manager_worker(void)
  593. {
  594.     ipc_call_t call;
  595.     ipc_callid_t callid;
  596.     int timeout;
  597.     awaiter_t *waiter;
  598.     struct timeval tv;
  599.  
  600.     while (1) {
  601.         if (fibril_switch(FIBRIL_FROM_MANAGER)) {
  602.             futex_up(&async_futex);
  603.             /*
  604.              * async_futex is always held when entering a manager
  605.              * fibril.
  606.              */
  607.             continue;
  608.         }
  609.         futex_down(&async_futex);
  610.         if (!list_empty(&timeout_list)) {
  611.             waiter = list_get_instance(timeout_list.next, awaiter_t,
  612.                 link);
  613.             gettimeofday(&tv, NULL);
  614.             if (tv_gteq(&tv, &waiter->expires)) {
  615.                 futex_up(&async_futex);
  616.                 handle_expired_timeouts();
  617.                 continue;
  618.             } else
  619.                 timeout = tv_sub(&waiter->expires, &tv);
  620.         } else
  621.             timeout = SYNCH_NO_TIMEOUT;
  622.         futex_up(&async_futex);
  623.  
  624.         callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
  625.  
  626.         if (!callid) {
  627.             handle_expired_timeouts();
  628.             continue;
  629.         }
  630.  
  631.         if (callid & IPC_CALLID_ANSWERED) {
  632.             continue;
  633.         }
  634.  
  635.         handle_call(callid, &call);
  636.     }
  637.    
  638.     return 0;
  639. }
  640.  
  641. /** Function to start async_manager as a standalone fibril.
  642.  *
  643.  * When more kernel threads are used, one async manager should exist per thread.
  644.  *
  645.  * @param arg       Unused.
  646.  *
  647.  * @return      Never returns.
  648.  */
  649. static int async_manager_fibril(void *arg)
  650. {
  651.     futex_up(&async_futex);
  652.     /*
  653.      * async_futex is always locked when entering manager
  654.      */
  655.     async_manager_worker();
  656.    
  657.     return 0;
  658. }
  659.  
  660. /** Add one manager to manager list. */
  661. void async_create_manager(void)
  662. {
  663.     fid_t fid;
  664.  
  665.     fid = fibril_create(async_manager_fibril, NULL);
  666.     fibril_add_manager(fid);
  667. }
  668.  
  669. /** Remove one manager from manager list */
  670. void async_destroy_manager(void)
  671. {
  672.     fibril_remove_manager();
  673. }
  674.  
  675. /** Initialize the async framework.
  676.  *
  677.  * @return      Zero on success or an error code.
  678.  */
  679. int _async_init(void)
  680. {
  681.     if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
  682.         &conn_hash_table_ops)) {
  683.         printf("%s: cannot create hash table\n", "async");
  684.         return ENOMEM;
  685.     }
  686.    
  687.     return 0;
  688. }
  689.  
  690. /** Reply received callback.
  691.  *
  692.  * This function is called whenever a reply for an asynchronous message sent out
  693.  * by the asynchronous framework is received.
  694.  *
  695.  * Notify the fibril which is waiting for this message that it has arrived.
  696.  *
  697.  * @param private   Pointer to the asynchronous message record.
  698.  * @param retval    Value returned in the answer.
  699.  * @param data      Call data of the answer.
  700.  */
  701. static void reply_received(void *private, int retval, ipc_call_t *data)
  702. {
  703.     amsg_t *msg = (amsg_t *) private;
  704.  
  705.     msg->retval = retval;
  706.  
  707.     futex_down(&async_futex);
  708.     /* Copy data after futex_down, just in case the call was detached */
  709.     if (msg->dataptr)
  710.         *msg->dataptr = *data;
  711.  
  712.     write_barrier();
  713.     /* Remove message from timeout list */
  714.     if (msg->wdata.inlist)
  715.         list_remove(&msg->wdata.link);
  716.     msg->done = 1;
  717.     if (!msg->wdata.active) {
  718.         msg->wdata.active = 1;
  719.         fibril_add_ready(msg->wdata.fid);
  720.     }
  721.     futex_up(&async_futex);
  722. }
  723.  
  724. /** Send message and return id of the sent message.
  725.  *
  726.  * The return value can be used as input for async_wait() to wait for
  727.  * completion.
  728.  *
  729.  * @param phoneid   Handle of the phone that will be used for the send.
  730.  * @param method    Service-defined method.
  731.  * @param arg1      Service-defined payload argument.
  732.  * @param arg2      Service-defined payload argument.
  733.  * @param arg3      Service-defined payload argument.
  734.  * @param arg4      Service-defined payload argument.
  735.  * @param dataptr   If non-NULL, storage where the reply data will be
  736.  *          stored.
  737.  *
  738.  * @return      Hash of the sent message.
  739.  */
  740. aid_t async_send_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  741.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipc_call_t *dataptr)
  742. {
  743.     amsg_t *msg;
  744.    
  745.     msg = malloc(sizeof(*msg));
  746.     msg->done = 0;
  747.     msg->dataptr = dataptr;
  748.    
  749.     /* We may sleep in the next method, but it will use its own mechanism */
  750.     msg->wdata.active = 1;
  751.                
  752.     ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,
  753.         reply_received, !_in_interrupt_handler);
  754.    
  755.     return (aid_t) msg;
  756. }
  757.  
  758. /** Send message and return id of the sent message
  759.  *
  760.  * The return value can be used as input for async_wait() to wait for
  761.  * completion.
  762.  *
  763.  * @param phoneid   Handle of the phone that will be used for the send.
  764.  * @param method    Service-defined method.
  765.  * @param arg1      Service-defined payload argument.
  766.  * @param arg2      Service-defined payload argument.
  767.  * @param arg3      Service-defined payload argument.
  768.  * @param arg4      Service-defined payload argument.
  769.  * @param arg5      Service-defined payload argument.
  770.  * @param dataptr   If non-NULL, storage where the reply data will be
  771.  *          stored.
  772.  *
  773.  * @return      Hash of the sent message.
  774.  */
  775. aid_t async_send_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  776.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5,
  777.     ipc_call_t *dataptr)
  778. {
  779.     amsg_t *msg;
  780.    
  781.     msg = malloc(sizeof(*msg));
  782.     msg->done = 0;
  783.     msg->dataptr = dataptr;
  784.    
  785.     /* We may sleep in next method, but it will use its own mechanism */
  786.     msg->wdata.active = 1;
  787.    
  788.     ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,
  789.         reply_received, !_in_interrupt_handler);
  790.    
  791.     return (aid_t) msg;
  792. }
  793.  
  794. /** Wait for a message sent by the async framework.
  795.  *
  796.  * @param amsgid    Hash of the message to wait for.
  797.  * @param retval    Pointer to storage where the retval of the answer will
  798.  *          be stored.
  799.  */
  800. void async_wait_for(aid_t amsgid, ipcarg_t *retval)
  801. {
  802.     amsg_t *msg = (amsg_t *) amsgid;
  803.  
  804.     futex_down(&async_futex);
  805.     if (msg->done) {
  806.         futex_up(&async_futex);
  807.         goto done;
  808.     }
  809.  
  810.     msg->wdata.fid = fibril_get_id();
  811.     msg->wdata.active = 0;
  812.     msg->wdata.inlist = 0;
  813.     /* Leave the async_futex locked when entering this function */
  814.     fibril_switch(FIBRIL_TO_MANAGER);
  815.     /* futex is up automatically after fibril_switch...*/
  816. done:
  817.     if (retval)
  818.         *retval = msg->retval;
  819.     free(msg);
  820. }
  821.  
  822. /** Wait for a message sent by the async framework, timeout variant.
  823.  *
  824.  * @param amsgid    Hash of the message to wait for.
  825.  * @param retval    Pointer to storage where the retval of the answer will
  826.  *          be stored.
  827.  * @param timeout   Timeout in microseconds.
  828.  *
  829.  * @return      Zero on success, ETIMEOUT if the timeout has expired.
  830.  */
  831. int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
  832. {
  833.     amsg_t *msg = (amsg_t *) amsgid;
  834.  
  835.     /* TODO: Let it go through the event read at least once */
  836.     if (timeout < 0)
  837.         return ETIMEOUT;
  838.  
  839.     futex_down(&async_futex);
  840.     if (msg->done) {
  841.         futex_up(&async_futex);
  842.         goto done;
  843.     }
  844.  
  845.     gettimeofday(&msg->wdata.expires, NULL);
  846.     tv_add(&msg->wdata.expires, timeout);
  847.  
  848.     msg->wdata.fid = fibril_get_id();
  849.     msg->wdata.active = 0;
  850.     insert_timeout(&msg->wdata);
  851.  
  852.     /* Leave the async_futex locked when entering this function */
  853.     fibril_switch(FIBRIL_TO_MANAGER);
  854.     /* futex is up automatically after fibril_switch...*/
  855.  
  856.     if (!msg->done)
  857.         return ETIMEOUT;
  858.  
  859. done:
  860.     if (retval)
  861.         *retval = msg->retval;
  862.     free(msg);
  863.  
  864.     return 0;
  865. }
  866.  
  867. /** Wait for specified time.
  868.  *
  869.  * The current fibril is suspended but the thread continues to execute.
  870.  *
  871.  * @param timeout   Duration of the wait in microseconds.
  872.  */
  873. void async_usleep(suseconds_t timeout)
  874. {
  875.     amsg_t *msg;
  876.    
  877.     msg = malloc(sizeof(*msg));
  878.     if (!msg)
  879.         return;
  880.    
  881.     msg->wdata.fid = fibril_get_id();
  882.     msg->wdata.active = 0;
  883.    
  884.     gettimeofday(&msg->wdata.expires, NULL);
  885.     tv_add(&msg->wdata.expires, timeout);
  886.    
  887.     futex_down(&async_futex);
  888.     insert_timeout(&msg->wdata);
  889.     /* Leave the async_futex locked when entering this function */
  890.     fibril_switch(FIBRIL_TO_MANAGER);
  891.     /* futex is up automatically after fibril_switch()...*/
  892.     free(msg);
  893. }
  894.  
  895. /** Setter for client_connection function pointer.
  896.  *
  897.  * @param conn      Function that will implement a new connection fibril.
  898.  */
  899. void async_set_client_connection(async_client_conn_t conn)
  900. {
  901.     client_connection = conn;
  902. }
  903.  
  904. /** Setter for interrupt_received function pointer.
  905.  *
  906.  * @param conn      Function that will implement a new interrupt
  907.  *          notification fibril.
  908.  */
  909. void async_set_interrupt_received(async_client_conn_t conn)
  910. {
  911.     interrupt_received = conn;
  912. }
  913.  
  914. /** Pseudo-synchronous message sending - fast version.
  915.  *
  916.  * Send message asynchronously and return only after the reply arrives.
  917.  *
  918.  * This function can only transfer 4 register payload arguments. For
  919.  * transferring more arguments, see the slower async_req_slow().
  920.  *
  921.  * @param phoneid   Hash of the phone through which to make the call.
  922.  * @param method    Method of the call.
  923.  * @param arg1      Service-defined payload argument.
  924.  * @param arg2      Service-defined payload argument.
  925.  * @param arg3      Service-defined payload argument.
  926.  * @param arg4      Service-defined payload argument.
  927.  * @param r1        If non-NULL, storage for the 1st reply argument.
  928.  * @param r2        If non-NULL, storage for the 2nd reply argument.
  929.  * @param r3        If non-NULL, storage for the 3rd reply argument.
  930.  * @param r4        If non-NULL, storage for the 4th reply argument.
  931.  * @param r5        If non-NULL, storage for the 5th reply argument.
  932.  * @return      Return code of the reply or a negative error code.
  933.  */
  934. ipcarg_t async_req_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  935.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t *r1, ipcarg_t *r2,
  936.     ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  937. {
  938.     ipc_call_t result;
  939.     ipcarg_t rc;
  940.  
  941.     aid_t eid = async_send_4(phoneid, method, arg1, arg2, arg3, arg4,
  942.         &result);
  943.     async_wait_for(eid, &rc);
  944.     if (r1)
  945.         *r1 = IPC_GET_ARG1(result);
  946.     if (r2)
  947.         *r2 = IPC_GET_ARG2(result);
  948.     if (r3)
  949.         *r3 = IPC_GET_ARG3(result);
  950.     if (r4)
  951.         *r4 = IPC_GET_ARG4(result);
  952.     if (r5)
  953.         *r5 = IPC_GET_ARG5(result);
  954.     return rc;
  955. }
  956.  
  957. /** Pseudo-synchronous message sending - slow version.
  958.  *
  959.  * Send message asynchronously and return only after the reply arrives.
  960.  *
  961.  * @param phoneid   Hash of the phone through which to make the call.
  962.  * @param method    Method of the call.
  963.  * @param arg1      Service-defined payload argument.
  964.  * @param arg2      Service-defined payload argument.
  965.  * @param arg3      Service-defined payload argument.
  966.  * @param arg4      Service-defined payload argument.
  967.  * @param arg5      Service-defined payload argument.
  968.  * @param r1        If non-NULL, storage for the 1st reply argument.
  969.  * @param r2        If non-NULL, storage for the 2nd reply argument.
  970.  * @param r3        If non-NULL, storage for the 3rd reply argument.
  971.  * @param r4        If non-NULL, storage for the 4th reply argument.
  972.  * @param r5        If non-NULL, storage for the 5th reply argument.
  973.  * @return      Return code of the reply or a negative error code.
  974.  */
  975. ipcarg_t async_req_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  976.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5, ipcarg_t *r1,
  977.     ipcarg_t *r2, ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  978. {
  979.     ipc_call_t result;
  980.     ipcarg_t rc;
  981.  
  982.     aid_t eid = async_send_5(phoneid, method, arg1, arg2, arg3, arg4, arg5,
  983.         &result);
  984.     async_wait_for(eid, &rc);
  985.     if (r1)
  986.         *r1 = IPC_GET_ARG1(result);
  987.     if (r2)
  988.         *r2 = IPC_GET_ARG2(result);
  989.     if (r3)
  990.         *r3 = IPC_GET_ARG3(result);
  991.     if (r4)
  992.         *r4 = IPC_GET_ARG4(result);
  993.     if (r5)
  994.         *r5 = IPC_GET_ARG5(result);
  995.     return rc;
  996. }
  997.  
  998. /** @}
  999.  */
  1000.