Subversion Repositories HelenOS

Rev

Rev 2490 | 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.  * @param call      Call data of the opening call.
  461.  * @param cfibril   Fibril function that should be called upon opening the
  462.  *          connection.
  463.  *
  464.  * @return      New fibril id or NULL on failure.
  465.  */
  466. fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
  467.     ipc_call_t *call, void (*cfibril)(ipc_callid_t, ipc_call_t *))
  468. {
  469.     connection_t *conn;
  470.     unsigned long key;
  471.  
  472.     conn = malloc(sizeof(*conn));
  473.     if (!conn) {
  474.         ipc_answer_fast(callid, ENOMEM, 0, 0);
  475.         return NULL;
  476.     }
  477.     conn->in_phone_hash = in_phone_hash;
  478.     list_initialize(&conn->msg_queue);
  479.     conn->callid = callid;
  480.     conn->close_callid = 0;
  481.     if (call)
  482.         conn->call = *call;
  483.     conn->wdata.active = 1; /* We will activate the fibril ASAP */
  484.     conn->cfibril = cfibril;
  485.  
  486.     conn->wdata.fid = fibril_create(connection_fibril, conn);
  487.     if (!conn->wdata.fid) {
  488.         free(conn);
  489.         ipc_answer_fast(callid, ENOMEM, 0, 0);
  490.         return NULL;
  491.     }
  492.     /* Add connection to the connection hash table */
  493.     key = conn->in_phone_hash;
  494.     futex_down(&async_futex);
  495.     hash_table_insert(&conn_hash_table, &key, &conn->link);
  496.     futex_up(&async_futex);
  497.  
  498.     fibril_add_ready(conn->wdata.fid);
  499.  
  500.     return conn->wdata.fid;
  501. }
  502.  
  503. /** Handle a call that was received.
  504.  *
  505.  * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
  506.  * Otherwise the call is routed to its connection fibril.
  507.  *
  508.  * @param callid    Hash of the incoming call.
  509.  * @param call      Data of the incoming call.
  510.  */
  511. static void handle_call(ipc_callid_t callid, ipc_call_t *call)
  512. {
  513.     /* Unrouted call - do some default behaviour */
  514.     if ((callid & IPC_CALLID_NOTIFICATION)) {
  515.         in_interrupt_handler = 1;
  516.         (*interrupt_received)(callid,call);
  517.         in_interrupt_handler = 0;
  518.         return;
  519.     }      
  520.  
  521.     switch (IPC_GET_METHOD(*call)) {
  522.     case IPC_M_CONNECT_ME_TO:
  523.         /* Open new connection with fibril etc. */
  524.         async_new_connection(IPC_GET_ARG3(*call), callid, call,
  525.             client_connection);
  526.         return;
  527.     }
  528.  
  529.     /* Try to route the call through the connection hash table */
  530.     if (route_call(callid, call))
  531.         return;
  532.  
  533.     /* Unknown call from unknown phone - hang it up */
  534.     ipc_answer_fast(callid, EHANGUP, 0, 0);
  535. }
  536.  
  537. /** Fire all timeouts that expired. */
  538. static void handle_expired_timeouts(void)
  539. {
  540.     struct timeval tv;
  541.     awaiter_t *waiter;
  542.     link_t *cur;
  543.  
  544.     gettimeofday(&tv, NULL);
  545.     futex_down(&async_futex);
  546.  
  547.     cur = timeout_list.next;
  548.     while (cur != &timeout_list) {
  549.         waiter = list_get_instance(cur, awaiter_t, link);
  550.         if (tv_gt(&waiter->expires, &tv))
  551.             break;
  552.         cur = cur->next;
  553.         list_remove(&waiter->link);
  554.         waiter->inlist = 0;
  555.         waiter->timedout = 1;
  556.         /*
  557.          * Redundant condition?
  558.          * The fibril should not be active when it gets here.
  559.          */
  560.         if (!waiter->active) {
  561.             waiter->active = 1;
  562.             fibril_add_ready(waiter->fid);
  563.         }
  564.     }
  565.  
  566.     futex_up(&async_futex);
  567. }
  568.  
  569. /** Endless loop dispatching incoming calls and answers.
  570.  *
  571.  * @return      Never returns.
  572.  */
  573. static int async_manager_worker(void)
  574. {
  575.     ipc_call_t call;
  576.     ipc_callid_t callid;
  577.     int timeout;
  578.     awaiter_t *waiter;
  579.     struct timeval tv;
  580.  
  581.     while (1) {
  582.         if (fibril_schedule_next_adv(FIBRIL_FROM_MANAGER)) {
  583.             futex_up(&async_futex);
  584.             /*
  585.              * async_futex is always held when entering a manager
  586.              * fibril.
  587.              */
  588.             continue;
  589.         }
  590.         futex_down(&async_futex);
  591.         if (!list_empty(&timeout_list)) {
  592.             waiter = list_get_instance(timeout_list.next, awaiter_t,
  593.                 link);
  594.             gettimeofday(&tv, NULL);
  595.             if (tv_gteq(&tv, &waiter->expires)) {
  596.                 futex_up(&async_futex);
  597.                 handle_expired_timeouts();
  598.                 continue;
  599.             } else
  600.                 timeout = tv_sub(&waiter->expires, &tv);
  601.         } else
  602.             timeout = SYNCH_NO_TIMEOUT;
  603.         futex_up(&async_futex);
  604.  
  605.         callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
  606.  
  607.         if (!callid) {
  608.             handle_expired_timeouts();
  609.             continue;
  610.         }
  611.  
  612.         if (callid & IPC_CALLID_ANSWERED) {
  613.             continue;
  614.         }
  615.  
  616.         handle_call(callid, &call);
  617.     }
  618.    
  619.     return 0;
  620. }
  621.  
  622. /** Function to start async_manager as a standalone fibril.
  623.  *
  624.  * When more kernel threads are used, one async manager should exist per thread.
  625.  *
  626.  * @param arg       Unused.
  627.  *
  628.  * @return      Never returns.
  629.  */
  630. static int async_manager_fibril(void *arg)
  631. {
  632.     futex_up(&async_futex);
  633.     /*
  634.      * async_futex is always locked when entering manager
  635.      */
  636.     async_manager_worker();
  637.    
  638.     return 0;
  639. }
  640.  
  641. /** Add one manager to manager list. */
  642. void async_create_manager(void)
  643. {
  644.     fid_t fid;
  645.  
  646.     fid = fibril_create(async_manager_fibril, NULL);
  647.     fibril_add_manager(fid);
  648. }
  649.  
  650. /** Remove one manager from manager list */
  651. void async_destroy_manager(void)
  652. {
  653.     fibril_remove_manager();
  654. }
  655.  
  656. /** Initialize the async framework.
  657.  *
  658.  * @return      Zero on success or an error code.
  659.  */
  660. int _async_init(void)
  661. {
  662.     if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
  663.         &conn_hash_table_ops)) {
  664.         printf("%s: cannot create hash table\n", "async");
  665.         return ENOMEM;
  666.     }
  667.    
  668.     return 0;
  669. }
  670.  
  671. /** Reply received callback.
  672.  *
  673.  * This function is called whenever a reply for an asynchronous message sent out
  674.  * by the asynchronous framework is received.
  675.  *
  676.  * Notify the fibril which is waiting for this message that it has arrived.
  677.  *
  678.  * @param private   Pointer to the asynchronous message record.
  679.  * @param retval    Value returned in the answer.
  680.  * @param data      Call data of the answer.
  681.  */
  682. static void reply_received(void *private, int retval, ipc_call_t *data)
  683. {
  684.     amsg_t *msg = (amsg_t *) private;
  685.  
  686.     msg->retval = retval;
  687.  
  688.     futex_down(&async_futex);
  689.     /* Copy data after futex_down, just in case the call was detached */
  690.     if (msg->dataptr)
  691.         *msg->dataptr = *data;
  692.  
  693.     write_barrier();
  694.     /* Remove message from timeout list */
  695.     if (msg->wdata.inlist)
  696.         list_remove(&msg->wdata.link);
  697.     msg->done = 1;
  698.     if (!msg->wdata.active) {
  699.         msg->wdata.active = 1;
  700.         fibril_add_ready(msg->wdata.fid);
  701.     }
  702.     futex_up(&async_futex);
  703. }
  704.  
  705. /** Send message and return id of the sent message.
  706.  *
  707.  * The return value can be used as input for async_wait() to wait for
  708.  * completion.
  709.  *
  710.  * @param phoneid   Handle of the phone that will be used for the send.
  711.  * @param method    Service-defined method.
  712.  * @param arg1      Service-defined payload argument.
  713.  * @param arg2      Service-defined payload argument.
  714.  * @param dataptr   If non-NULL, storage where the reply data will be
  715.  *          stored.
  716.  *
  717.  * @return      Hash of the sent message.
  718.  */
  719. aid_t async_send_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
  720.     ipc_call_t *dataptr)
  721. {
  722.     amsg_t *msg;
  723.  
  724.     if (in_interrupt_handler) {
  725.         printf("Cannot send asynchronous request in interrupt "
  726.             "handler.\n");
  727.         _exit(1);
  728.     }
  729.  
  730.     msg = malloc(sizeof(*msg));
  731.     msg->done = 0;
  732.     msg->dataptr = dataptr;
  733.  
  734.     /* We may sleep in the next method, but it will use its own mechanism */
  735.     msg->wdata.active = 1;
  736.                
  737.     ipc_call_async_2(phoneid, method, arg1, arg2, msg, reply_received, 1);
  738.  
  739.     return (aid_t) msg;
  740. }
  741.  
  742. /** Send message and return id of the sent message
  743.  *
  744.  * The return value can be used as input for async_wait() to wait for
  745.  * completion.
  746.  *
  747.  * @param phoneid   Handle of the phone that will be used for the send.
  748.  * @param method    Service-defined method.
  749.  * @param arg1      Service-defined payload argument.
  750.  * @param arg2      Service-defined payload argument.
  751.  * @param arg3      Service-defined payload argument.
  752.  * @param dataptr   If non-NULL, storage where the reply data will be
  753.  *          stored.
  754.  *
  755.  * @return      Hash of the sent message.
  756.  */
  757. aid_t async_send_3(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
  758.     ipcarg_t arg3, ipc_call_t *dataptr)
  759. {
  760.     amsg_t *msg;
  761.  
  762.     if (in_interrupt_handler) {
  763.         printf("Cannot send asynchronous request in interrupt "
  764.             "handler.\n");
  765.         _exit(1);
  766.     }
  767.  
  768.     msg = malloc(sizeof(*msg));
  769.     msg->done = 0;
  770.     msg->dataptr = dataptr;
  771.  
  772.     /* We may sleep in next method, but it will use its own mechanism */
  773.     msg->wdata.active = 1;
  774.                
  775.     ipc_call_async_3(phoneid, method, arg1, arg2, arg3, msg, reply_received,
  776.         1);
  777.  
  778.     return (aid_t) msg;
  779. }
  780.  
  781. /** Wait for a message sent by the async framework.
  782.  *
  783.  * @param amsgid    Hash of the message to wait for.
  784.  * @param retval    Pointer to storage where the retval of the answer will
  785.  *          be stored.
  786.  */
  787. void async_wait_for(aid_t amsgid, ipcarg_t *retval)
  788. {
  789.     amsg_t *msg = (amsg_t *) amsgid;
  790.  
  791.     futex_down(&async_futex);
  792.     if (msg->done) {
  793.         futex_up(&async_futex);
  794.         goto done;
  795.     }
  796.  
  797.     msg->wdata.fid = fibril_get_id();
  798.     msg->wdata.active = 0;
  799.     msg->wdata.inlist = 0;
  800.     /* Leave the async_futex locked when entering this function */
  801.     fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
  802.     /* futex is up automatically after fibril_schedule_next...*/
  803. done:
  804.     if (retval)
  805.         *retval = msg->retval;
  806.     free(msg);
  807. }
  808.  
  809. /** Wait for a message sent by the async framework, timeout variant.
  810.  *
  811.  * @param amsgid    Hash of the message to wait for.
  812.  * @param retval    Pointer to storage where the retval of the answer will
  813.  *          be stored.
  814.  * @param timeout   Timeout in microseconds.
  815.  *
  816.  * @return      Zero on success, ETIMEOUT if the timeout has expired.
  817.  */
  818. int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
  819. {
  820.     amsg_t *msg = (amsg_t *) amsgid;
  821.  
  822.     /* TODO: Let it go through the event read at least once */
  823.     if (timeout < 0)
  824.         return ETIMEOUT;
  825.  
  826.     futex_down(&async_futex);
  827.     if (msg->done) {
  828.         futex_up(&async_futex);
  829.         goto done;
  830.     }
  831.  
  832.     gettimeofday(&msg->wdata.expires, NULL);
  833.     tv_add(&msg->wdata.expires, timeout);
  834.  
  835.     msg->wdata.fid = fibril_get_id();
  836.     msg->wdata.active = 0;
  837.     insert_timeout(&msg->wdata);
  838.  
  839.     /* Leave the async_futex locked when entering this function */
  840.     fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
  841.     /* futex is up automatically after fibril_schedule_next...*/
  842.  
  843.     if (!msg->done)
  844.         return ETIMEOUT;
  845.  
  846. done:
  847.     if (retval)
  848.         *retval = msg->retval;
  849.     free(msg);
  850.  
  851.     return 0;
  852. }
  853.  
  854. /** Wait for specified time.
  855.  *
  856.  * The current fibril is suspended but the thread continues to execute.
  857.  *
  858.  * @param timeout   Duration of the wait in microseconds.
  859.  */
  860. void async_usleep(suseconds_t timeout)
  861. {
  862.     amsg_t *msg;
  863.    
  864.     if (in_interrupt_handler) {
  865.         printf("Cannot call async_usleep in interrupt handler.\n");
  866.         _exit(1);
  867.     }
  868.  
  869.     msg = malloc(sizeof(*msg));
  870.     if (!msg)
  871.         return;
  872.  
  873.     msg->wdata.fid = fibril_get_id();
  874.     msg->wdata.active = 0;
  875.  
  876.     gettimeofday(&msg->wdata.expires, NULL);
  877.     tv_add(&msg->wdata.expires, timeout);
  878.  
  879.     futex_down(&async_futex);
  880.     insert_timeout(&msg->wdata);
  881.     /* Leave the async_futex locked when entering this function */
  882.     fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
  883.     /* futex is up automatically after fibril_schedule_next_adv()...*/
  884.     free(msg);
  885. }
  886.  
  887. /** Setter for client_connection function pointer.
  888.  *
  889.  * @param conn      Function that will implement a new connection fibril.
  890.  */
  891. void async_set_client_connection(async_client_conn_t conn)
  892. {
  893.     client_connection = conn;
  894. }
  895.  
  896. /** Setter for interrupt_received function pointer.
  897.  *
  898.  * @param conn      Function that will implement a new interrupt
  899.  *          notification fibril.
  900.  */
  901. void async_set_interrupt_received(async_client_conn_t conn)
  902. {
  903.     interrupt_received = conn;
  904. }
  905.  
  906. /* Primitive functions for simple communication */
  907. void async_msg_3(int phoneid, ipcarg_t method, ipcarg_t arg1,
  908.          ipcarg_t arg2, ipcarg_t arg3)
  909. {
  910.     ipc_call_async_3(phoneid, method, arg1, arg2, arg3, NULL, NULL,
  911.         !in_interrupt_handler);
  912. }
  913.  
  914. void async_msg_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2)
  915. {
  916.     ipc_call_async_2(phoneid, method, arg1, arg2, NULL, NULL,
  917.         !in_interrupt_handler);
  918. }
  919.  
  920. /** @}
  921.  */
  922.  
  923.