Subversion Repositories HelenOS

Rev

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