Subversion Repositories HelenOS

Rev

Rev 2635 | 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.     /* Add connection to the connection hash table */
  510.     key = conn->in_phone_hash;
  511.     futex_down(&async_futex);
  512.     hash_table_insert(&conn_hash_table, &key, &conn->link);
  513.     futex_up(&async_futex);
  514.  
  515.     fibril_add_ready(conn->wdata.fid);
  516.  
  517.     return conn->wdata.fid;
  518. }
  519.  
  520. /** Handle a call that was received.
  521.  *
  522.  * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
  523.  * Otherwise the call is routed to its connection fibril.
  524.  *
  525.  * @param callid    Hash of the incoming call.
  526.  * @param call      Data of the incoming call.
  527.  */
  528. static void handle_call(ipc_callid_t callid, ipc_call_t *call)
  529. {
  530.     /* Unrouted call - do some default behaviour */
  531.     if ((callid & IPC_CALLID_NOTIFICATION)) {
  532.         _in_interrupt_handler = 1;
  533.         (*interrupt_received)(callid, call);
  534.         _in_interrupt_handler = 0;
  535.         return;
  536.     }      
  537.  
  538.     switch (IPC_GET_METHOD(*call)) {
  539.     case IPC_M_CONNECT_ME_TO:
  540.         /* Open new connection with fibril etc. */
  541.         async_new_connection(IPC_GET_ARG5(*call), callid, call,
  542.             client_connection);
  543.         return;
  544.     }
  545.  
  546.     /* Try to route the call through the connection hash table */
  547.     if (route_call(callid, call))
  548.         return;
  549.  
  550.     /* Unknown call from unknown phone - hang it up */
  551.     ipc_answer_0(callid, EHANGUP);
  552. }
  553.  
  554. /** Fire all timeouts that expired. */
  555. static void handle_expired_timeouts(void)
  556. {
  557.     struct timeval tv;
  558.     awaiter_t *waiter;
  559.     link_t *cur;
  560.  
  561.     gettimeofday(&tv, NULL);
  562.     futex_down(&async_futex);
  563.  
  564.     cur = timeout_list.next;
  565.     while (cur != &timeout_list) {
  566.         waiter = list_get_instance(cur, awaiter_t, link);
  567.         if (tv_gt(&waiter->expires, &tv))
  568.             break;
  569.         cur = cur->next;
  570.         list_remove(&waiter->link);
  571.         waiter->inlist = 0;
  572.         waiter->timedout = 1;
  573.         /*
  574.          * Redundant condition?
  575.          * The fibril should not be active when it gets here.
  576.          */
  577.         if (!waiter->active) {
  578.             waiter->active = 1;
  579.             fibril_add_ready(waiter->fid);
  580.         }
  581.     }
  582.  
  583.     futex_up(&async_futex);
  584. }
  585.  
  586. /** Endless loop dispatching incoming calls and answers.
  587.  *
  588.  * @return      Never returns.
  589.  */
  590. static int async_manager_worker(void)
  591. {
  592.     ipc_call_t call;
  593.     ipc_callid_t callid;
  594.     int timeout;
  595.     awaiter_t *waiter;
  596.     struct timeval tv;
  597.  
  598.     while (1) {
  599.         if (fibril_switch(FIBRIL_FROM_MANAGER)) {
  600.             futex_up(&async_futex);
  601.             /*
  602.              * async_futex is always held when entering a manager
  603.              * fibril.
  604.              */
  605.             continue;
  606.         }
  607.         futex_down(&async_futex);
  608.         if (!list_empty(&timeout_list)) {
  609.             waiter = list_get_instance(timeout_list.next, awaiter_t,
  610.                 link);
  611.             gettimeofday(&tv, NULL);
  612.             if (tv_gteq(&tv, &waiter->expires)) {
  613.                 futex_up(&async_futex);
  614.                 handle_expired_timeouts();
  615.                 continue;
  616.             } else
  617.                 timeout = tv_sub(&waiter->expires, &tv);
  618.         } else
  619.             timeout = SYNCH_NO_TIMEOUT;
  620.         futex_up(&async_futex);
  621.  
  622.         callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
  623.  
  624.         if (!callid) {
  625.             handle_expired_timeouts();
  626.             continue;
  627.         }
  628.  
  629.         if (callid & IPC_CALLID_ANSWERED) {
  630.             continue;
  631.         }
  632.  
  633.         handle_call(callid, &call);
  634.     }
  635.    
  636.     return 0;
  637. }
  638.  
  639. /** Function to start async_manager as a standalone fibril.
  640.  *
  641.  * When more kernel threads are used, one async manager should exist per thread.
  642.  *
  643.  * @param arg       Unused.
  644.  *
  645.  * @return      Never returns.
  646.  */
  647. static int async_manager_fibril(void *arg)
  648. {
  649.     futex_up(&async_futex);
  650.     /*
  651.      * async_futex is always locked when entering manager
  652.      */
  653.     async_manager_worker();
  654.    
  655.     return 0;
  656. }
  657.  
  658. /** Add one manager to manager list. */
  659. void async_create_manager(void)
  660. {
  661.     fid_t fid;
  662.  
  663.     fid = fibril_create(async_manager_fibril, NULL);
  664.     fibril_add_manager(fid);
  665. }
  666.  
  667. /** Remove one manager from manager list */
  668. void async_destroy_manager(void)
  669. {
  670.     fibril_remove_manager();
  671. }
  672.  
  673. /** Initialize the async framework.
  674.  *
  675.  * @return      Zero on success or an error code.
  676.  */
  677. int _async_init(void)
  678. {
  679.     if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
  680.         &conn_hash_table_ops)) {
  681.         printf("%s: cannot create hash table\n", "async");
  682.         return ENOMEM;
  683.     }
  684.    
  685.     return 0;
  686. }
  687.  
  688. /** Reply received callback.
  689.  *
  690.  * This function is called whenever a reply for an asynchronous message sent out
  691.  * by the asynchronous framework is received.
  692.  *
  693.  * Notify the fibril which is waiting for this message that it has arrived.
  694.  *
  695.  * @param private   Pointer to the asynchronous message record.
  696.  * @param retval    Value returned in the answer.
  697.  * @param data      Call data of the answer.
  698.  */
  699. static void reply_received(void *private, int retval, ipc_call_t *data)
  700. {
  701.     amsg_t *msg = (amsg_t *) private;
  702.  
  703.     msg->retval = retval;
  704.  
  705.     futex_down(&async_futex);
  706.     /* Copy data after futex_down, just in case the call was detached */
  707.     if (msg->dataptr)
  708.         *msg->dataptr = *data;
  709.  
  710.     write_barrier();
  711.     /* Remove message from timeout list */
  712.     if (msg->wdata.inlist)
  713.         list_remove(&msg->wdata.link);
  714.     msg->done = 1;
  715.     if (!msg->wdata.active) {
  716.         msg->wdata.active = 1;
  717.         fibril_add_ready(msg->wdata.fid);
  718.     }
  719.     futex_up(&async_futex);
  720. }
  721.  
  722. /** Send message and return id of the sent message.
  723.  *
  724.  * The return value can be used as input for async_wait() to wait for
  725.  * completion.
  726.  *
  727.  * @param phoneid   Handle of the phone that will be used for the send.
  728.  * @param method    Service-defined method.
  729.  * @param arg1      Service-defined payload argument.
  730.  * @param arg2      Service-defined payload argument.
  731.  * @param arg3      Service-defined payload argument.
  732.  * @param arg4      Service-defined payload argument.
  733.  * @param dataptr   If non-NULL, storage where the reply data will be
  734.  *          stored.
  735.  *
  736.  * @return      Hash of the sent message.
  737.  */
  738. aid_t async_send_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  739.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipc_call_t *dataptr)
  740. {
  741.     amsg_t *msg;
  742.  
  743.     if (_in_interrupt_handler) {
  744.         printf("Cannot send asynchronous request in interrupt "
  745.             "handler.\n");
  746.         _exit(1);
  747.     }
  748.  
  749.     msg = malloc(sizeof(*msg));
  750.     msg->done = 0;
  751.     msg->dataptr = dataptr;
  752.  
  753.     /* We may sleep in the next method, but it will use its own mechanism */
  754.     msg->wdata.active = 1;
  755.                
  756.     ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,
  757.         reply_received, 1);
  758.  
  759.     return (aid_t) msg;
  760. }
  761.  
  762. /** Send message and return id of the sent message
  763.  *
  764.  * The return value can be used as input for async_wait() to wait for
  765.  * completion.
  766.  *
  767.  * @param phoneid   Handle of the phone that will be used for the send.
  768.  * @param method    Service-defined method.
  769.  * @param arg1      Service-defined payload argument.
  770.  * @param arg2      Service-defined payload argument.
  771.  * @param arg3      Service-defined payload argument.
  772.  * @param arg4      Service-defined payload argument.
  773.  * @param arg5      Service-defined payload argument.
  774.  * @param dataptr   If non-NULL, storage where the reply data will be
  775.  *          stored.
  776.  *
  777.  * @return      Hash of the sent message.
  778.  */
  779. aid_t async_send_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  780.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5,
  781.     ipc_call_t *dataptr)
  782. {
  783.     amsg_t *msg;
  784.  
  785.     if (_in_interrupt_handler) {
  786.         printf("Cannot send asynchronous request in interrupt "
  787.             "handler.\n");
  788.         _exit(1);
  789.     }
  790.  
  791.     msg = malloc(sizeof(*msg));
  792.     msg->done = 0;
  793.     msg->dataptr = dataptr;
  794.  
  795.     /* We may sleep in next method, but it will use its own mechanism */
  796.     msg->wdata.active = 1;
  797.  
  798.     ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,
  799.         reply_received, 1);
  800.  
  801.     return (aid_t) msg;
  802. }
  803.  
  804. /** Wait for a message sent by the async framework.
  805.  *
  806.  * @param amsgid    Hash of the message to wait for.
  807.  * @param retval    Pointer to storage where the retval of the answer will
  808.  *          be stored.
  809.  */
  810. void async_wait_for(aid_t amsgid, ipcarg_t *retval)
  811. {
  812.     amsg_t *msg = (amsg_t *) amsgid;
  813.  
  814.     futex_down(&async_futex);
  815.     if (msg->done) {
  816.         futex_up(&async_futex);
  817.         goto done;
  818.     }
  819.  
  820.     msg->wdata.fid = fibril_get_id();
  821.     msg->wdata.active = 0;
  822.     msg->wdata.inlist = 0;
  823.     /* Leave the async_futex locked when entering this function */
  824.     fibril_switch(FIBRIL_TO_MANAGER);
  825.     /* futex is up automatically after fibril_switch...*/
  826. done:
  827.     if (retval)
  828.         *retval = msg->retval;
  829.     free(msg);
  830. }
  831.  
  832. /** Wait for a message sent by the async framework, timeout variant.
  833.  *
  834.  * @param amsgid    Hash of the message to wait for.
  835.  * @param retval    Pointer to storage where the retval of the answer will
  836.  *          be stored.
  837.  * @param timeout   Timeout in microseconds.
  838.  *
  839.  * @return      Zero on success, ETIMEOUT if the timeout has expired.
  840.  */
  841. int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
  842. {
  843.     amsg_t *msg = (amsg_t *) amsgid;
  844.  
  845.     /* TODO: Let it go through the event read at least once */
  846.     if (timeout < 0)
  847.         return ETIMEOUT;
  848.  
  849.     futex_down(&async_futex);
  850.     if (msg->done) {
  851.         futex_up(&async_futex);
  852.         goto done;
  853.     }
  854.  
  855.     gettimeofday(&msg->wdata.expires, NULL);
  856.     tv_add(&msg->wdata.expires, timeout);
  857.  
  858.     msg->wdata.fid = fibril_get_id();
  859.     msg->wdata.active = 0;
  860.     insert_timeout(&msg->wdata);
  861.  
  862.     /* Leave the async_futex locked when entering this function */
  863.     fibril_switch(FIBRIL_TO_MANAGER);
  864.     /* futex is up automatically after fibril_switch...*/
  865.  
  866.     if (!msg->done)
  867.         return ETIMEOUT;
  868.  
  869. done:
  870.     if (retval)
  871.         *retval = msg->retval;
  872.     free(msg);
  873.  
  874.     return 0;
  875. }
  876.  
  877. /** Wait for specified time.
  878.  *
  879.  * The current fibril is suspended but the thread continues to execute.
  880.  *
  881.  * @param timeout   Duration of the wait in microseconds.
  882.  */
  883. void async_usleep(suseconds_t timeout)
  884. {
  885.     amsg_t *msg;
  886.    
  887.     if (_in_interrupt_handler) {
  888.         printf("Cannot call async_usleep in interrupt handler.\n");
  889.         _exit(1);
  890.     }
  891.  
  892.     msg = malloc(sizeof(*msg));
  893.     if (!msg)
  894.         return;
  895.  
  896.     msg->wdata.fid = fibril_get_id();
  897.     msg->wdata.active = 0;
  898.  
  899.     gettimeofday(&msg->wdata.expires, NULL);
  900.     tv_add(&msg->wdata.expires, timeout);
  901.  
  902.     futex_down(&async_futex);
  903.     insert_timeout(&msg->wdata);
  904.     /* Leave the async_futex locked when entering this function */
  905.     fibril_switch(FIBRIL_TO_MANAGER);
  906.     /* futex is up automatically after fibril_switch()...*/
  907.     free(msg);
  908. }
  909.  
  910. /** Setter for client_connection function pointer.
  911.  *
  912.  * @param conn      Function that will implement a new connection fibril.
  913.  */
  914. void async_set_client_connection(async_client_conn_t conn)
  915. {
  916.     client_connection = conn;
  917. }
  918.  
  919. /** Setter for interrupt_received function pointer.
  920.  *
  921.  * @param conn      Function that will implement a new interrupt
  922.  *          notification fibril.
  923.  */
  924. void async_set_interrupt_received(async_client_conn_t conn)
  925. {
  926.     interrupt_received = conn;
  927. }
  928.  
  929. /** Pseudo-synchronous message sending - fast version.
  930.  *
  931.  * Send message asynchronously and return only after the reply arrives.
  932.  *
  933.  * This function can only transfer 4 register payload arguments. For
  934.  * transferring more arguments, see the slower async_req_slow().
  935.  *
  936.  * @param phoneid   Hash of the phone through which to make the call.
  937.  * @param method    Method of the call.
  938.  * @param arg1      Service-defined payload argument.
  939.  * @param arg2      Service-defined payload argument.
  940.  * @param arg3      Service-defined payload argument.
  941.  * @param arg4      Service-defined payload argument.
  942.  * @param r1        If non-NULL, storage for the 1st reply argument.
  943.  * @param r2        If non-NULL, storage for the 2nd reply argument.
  944.  * @param r3        If non-NULL, storage for the 3rd reply argument.
  945.  * @param r4        If non-NULL, storage for the 4th reply argument.
  946.  * @param r5        If non-NULL, storage for the 5th reply argument.
  947.  * @return      Return code of the reply or a negative error code.
  948.  */
  949. ipcarg_t async_req_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  950.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t *r1, ipcarg_t *r2,
  951.     ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  952. {
  953.     ipc_call_t result;
  954.     ipcarg_t rc;
  955.  
  956.     aid_t eid = async_send_4(phoneid, method, arg1, arg2, arg3, arg4,
  957.         &result);
  958.     async_wait_for(eid, &rc);
  959.     if (r1)
  960.         *r1 = IPC_GET_ARG1(result);
  961.     if (r2)
  962.         *r2 = IPC_GET_ARG2(result);
  963.     if (r3)
  964.         *r3 = IPC_GET_ARG3(result);
  965.     if (r4)
  966.         *r4 = IPC_GET_ARG4(result);
  967.     if (r5)
  968.         *r5 = IPC_GET_ARG5(result);
  969.     return rc;
  970. }
  971.  
  972. /** Pseudo-synchronous message sending - slow version.
  973.  *
  974.  * Send message asynchronously and return only after the reply arrives.
  975.  *
  976.  * @param phoneid   Hash of the phone through which to make the call.
  977.  * @param method    Method of the call.
  978.  * @param arg1      Service-defined payload argument.
  979.  * @param arg2      Service-defined payload argument.
  980.  * @param arg3      Service-defined payload argument.
  981.  * @param arg4      Service-defined payload argument.
  982.  * @param arg5      Service-defined payload argument.
  983.  * @param r1        If non-NULL, storage for the 1st reply argument.
  984.  * @param r2        If non-NULL, storage for the 2nd reply argument.
  985.  * @param r3        If non-NULL, storage for the 3rd reply argument.
  986.  * @param r4        If non-NULL, storage for the 4th reply argument.
  987.  * @param r5        If non-NULL, storage for the 5th reply argument.
  988.  * @return      Return code of the reply or a negative error code.
  989.  */
  990. ipcarg_t async_req_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  991.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5, ipcarg_t *r1,
  992.     ipcarg_t *r2, ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  993. {
  994.     ipc_call_t result;
  995.     ipcarg_t rc;
  996.  
  997.     aid_t eid = async_send_5(phoneid, method, arg1, arg2, arg3, arg4, arg5,
  998.         &result);
  999.     async_wait_for(eid, &rc);
  1000.     if (r1)
  1001.         *r1 = IPC_GET_ARG1(result);
  1002.     if (r2)
  1003.         *r2 = IPC_GET_ARG2(result);
  1004.     if (r3)
  1005.         *r3 = IPC_GET_ARG3(result);
  1006.     if (r4)
  1007.         *r4 = IPC_GET_ARG4(result);
  1008.     if (r5)
  1009.         *r5 = IPC_GET_ARG5(result);
  1010.     return rc;
  1011. }
  1012.  
  1013. /** @}
  1014.  */
  1015.