Subversion Repositories HelenOS

Rev

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