Subversion Repositories HelenOS

Rev

Rev 4407 | Rev 4509 | 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 <libadt/hash_table.h>
  99. #include <libadt/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. __thread 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.             = list_get_instance(FIBRIL_connection->msg_queue.next, msg_t, link);
  555.        
  556.         list_remove(&msg->link);
  557.         ipc_answer_0(msg->callid, EHANGUP);
  558.         free(msg);
  559.     }
  560.    
  561.     if (FIBRIL_connection->close_callid)
  562.         ipc_answer_0(FIBRIL_connection->close_callid, EOK);
  563.    
  564.     return 0;
  565. }
  566.  
  567. /** Create a new fibril for a new connection.
  568.  *
  569.  * Create new fibril for connection, fill in connection structures and inserts
  570.  * it into the hash table, so that later we can easily do routing of messages to
  571.  * particular fibrils.
  572.  *
  573.  * @param in_phone_hash Identification of the incoming connection.
  574.  * @param callid        Hash of the opening IPC_M_CONNECT_ME_TO call.
  575.  *                      If callid is zero, the connection was opened by
  576.  *                      accepting the IPC_M_CONNECT_TO_ME call and this function
  577.  *                      is called directly by the server.
  578.  * @param call          Call data of the opening call.
  579.  * @param cfibril       Fibril function that should be called upon opening the
  580.  *                      connection.
  581.  *
  582.  * @return New fibril id or NULL on failure.
  583.  *
  584.  */
  585. fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
  586.     ipc_call_t *call, void (*cfibril)(ipc_callid_t, ipc_call_t *))
  587. {
  588.     connection_t *conn = malloc(sizeof(*conn));
  589.     if (!conn) {
  590.         if (callid)
  591.             ipc_answer_0(callid, ENOMEM);
  592.         return NULL;
  593.     }
  594.    
  595.     conn->in_phone_hash = in_phone_hash;
  596.     list_initialize(&conn->msg_queue);
  597.     conn->callid = callid;
  598.     conn->close_callid = false;
  599.    
  600.     if (call)
  601.         conn->call = *call;
  602.    
  603.     /* We will activate the fibril ASAP */
  604.     conn->wdata.active = true;
  605.     conn->cfibril = cfibril;
  606.     conn->wdata.fid = fibril_create(connection_fibril, conn);
  607.    
  608.     if (!conn->wdata.fid) {
  609.         free(conn);
  610.         if (callid)
  611.             ipc_answer_0(callid, ENOMEM);
  612.         return NULL;
  613.     }
  614.    
  615.     /* Add connection to the connection hash table */
  616.     unsigned long key = conn->in_phone_hash;
  617.    
  618.     futex_down(&async_futex);
  619.     hash_table_insert(&conn_hash_table, &key, &conn->link);
  620.     futex_up(&async_futex);
  621.    
  622.     fibril_add_ready(conn->wdata.fid);
  623.    
  624.     return conn->wdata.fid;
  625. }
  626.  
  627. /** Handle a call that was received.
  628.  *
  629.  * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
  630.  * Otherwise the call is routed to its connection fibril.
  631.  *
  632.  * @param callid Hash of the incoming call.
  633.  * @param call   Data of the incoming call.
  634.  *
  635.  */
  636. static void handle_call(ipc_callid_t callid, ipc_call_t *call)
  637. {
  638.     /* Unrouted call - do some default behaviour */
  639.     if ((callid & IPC_CALLID_NOTIFICATION)) {
  640.         process_notification(callid, call);
  641.         goto out;
  642.     }
  643.    
  644.     switch (IPC_GET_METHOD(*call)) {
  645.     case IPC_M_CONNECT_ME:
  646.     case IPC_M_CONNECT_ME_TO:
  647.         /* Open new connection with fibril etc. */
  648.         async_new_connection(IPC_GET_ARG5(*call), callid, call,
  649.             client_connection);
  650.         goto out;
  651.     }
  652.    
  653.     /* Try to route the call through the connection hash table */
  654.     if (route_call(callid, call))
  655.         goto out;
  656.    
  657.     /* Unknown call from unknown phone - hang it up */
  658.     ipc_answer_0(callid, EHANGUP);
  659.     return;
  660.    
  661. out:
  662.     process_pending();
  663. }
  664.  
  665. /** Fire all timeouts that expired. */
  666. static void handle_expired_timeouts(void)
  667. {
  668.     struct timeval tv;
  669.     gettimeofday(&tv, NULL);
  670.    
  671.     futex_down(&async_futex);
  672.    
  673.     link_t *cur = timeout_list.next;
  674.     while (cur != &timeout_list) {
  675.         awaiter_t *waiter = list_get_instance(cur, awaiter_t, link);
  676.        
  677.         if (tv_gt(&waiter->expires, &tv))
  678.             break;
  679.        
  680.         cur = cur->next;
  681.        
  682.         list_remove(&waiter->link);
  683.         waiter->inlist = false;
  684.         waiter->timedout = true;
  685.        
  686.         /*
  687.          * Redundant condition?
  688.          * The fibril should not be active when it gets here.
  689.          */
  690.         if (!waiter->active) {
  691.             waiter->active = true;
  692.             fibril_add_ready(waiter->fid);
  693.         }
  694.     }
  695.    
  696.     futex_up(&async_futex);
  697. }
  698.  
  699. /** Endless loop dispatching incoming calls and answers.
  700.  *
  701.  * @return Never returns.
  702.  *
  703.  */
  704. static int async_manager_worker(void)
  705. {
  706.     while (true) {
  707.         if (fibril_switch(FIBRIL_FROM_MANAGER)) {
  708.             futex_up(&async_futex);
  709.             /*
  710.              * async_futex is always held when entering a manager
  711.              * fibril.
  712.              */
  713.             continue;
  714.         }
  715.        
  716.         futex_down(&async_futex);
  717.        
  718.         suseconds_t timeout;
  719.         if (!list_empty(&timeout_list)) {
  720.             awaiter_t *waiter
  721.                 = list_get_instance(timeout_list.next, awaiter_t, link);
  722.            
  723.             struct timeval tv;
  724.             gettimeofday(&tv, NULL);
  725.            
  726.             if (tv_gteq(&tv, &waiter->expires)) {
  727.                 futex_up(&async_futex);
  728.                 handle_expired_timeouts();
  729.                 continue;
  730.             } else
  731.                 timeout = tv_sub(&waiter->expires, &tv);
  732.         } else
  733.             timeout = SYNCH_NO_TIMEOUT;
  734.        
  735.         futex_up(&async_futex);
  736.        
  737.         ipc_call_t call;
  738.         ipc_callid_t callid
  739.             = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
  740.        
  741.         if (!callid) {
  742.             handle_expired_timeouts();
  743.             continue;
  744.         }
  745.        
  746.         if (callid & IPC_CALLID_ANSWERED)
  747.             continue;
  748.        
  749.         handle_call(callid, &call);
  750.     }
  751.    
  752.     return 0;
  753. }
  754.  
  755. /** Function to start async_manager as a standalone fibril.
  756.  *
  757.  * When more kernel threads are used, one async manager should exist per thread.
  758.  *
  759.  * @param arg Unused.
  760.  * @return Never returns.
  761.  *
  762.  */
  763. static int async_manager_fibril(void *arg)
  764. {
  765.     futex_up(&async_futex);
  766.    
  767.     /*
  768.      * async_futex is always locked when entering manager
  769.      */
  770.     async_manager_worker();
  771.    
  772.     return 0;
  773. }
  774.  
  775. /** Add one manager to manager list. */
  776. void async_create_manager(void)
  777. {
  778.     fid_t fid = fibril_create(async_manager_fibril, NULL);
  779.     fibril_add_manager(fid);
  780. }
  781.  
  782. /** Remove one manager from manager list */
  783. void async_destroy_manager(void)
  784. {
  785.     fibril_remove_manager();
  786. }
  787.  
  788. /** Initialize the async framework.
  789.  *
  790.  * @return Zero on success or an error code.
  791.  */
  792. int _async_init(void)
  793. {
  794.     if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
  795.         &conn_hash_table_ops)) {
  796.         printf("%s: cannot create hash table\n", "async");
  797.         return ENOMEM;
  798.     }
  799.    
  800.     return 0;
  801. }
  802.  
  803. /** Reply received callback.
  804.  *
  805.  * This function is called whenever a reply for an asynchronous message sent out
  806.  * by the asynchronous framework is received.
  807.  *
  808.  * Notify the fibril which is waiting for this message that it has arrived.
  809.  *
  810.  * @param arg    Pointer to the asynchronous message record.
  811.  * @param retval Value returned in the answer.
  812.  * @param data   Call data of the answer.
  813.  */
  814. static void reply_received(void *arg, int retval, ipc_call_t *data)
  815. {
  816.     futex_down(&async_futex);
  817.    
  818.     amsg_t *msg = (amsg_t *) arg;
  819.     msg->retval = retval;
  820.    
  821.     /* Copy data after futex_down, just in case the call was detached */
  822.     if ((msg->dataptr) && (data))
  823.         *msg->dataptr = *data;
  824.    
  825.     write_barrier();
  826.    
  827.     /* Remove message from timeout list */
  828.     if (msg->wdata.inlist)
  829.         list_remove(&msg->wdata.link);
  830.    
  831.     msg->done = true;
  832.     if (!msg->wdata.active) {
  833.         msg->wdata.active = true;
  834.         fibril_add_ready(msg->wdata.fid);
  835.     }
  836.    
  837.     futex_up(&async_futex);
  838. }
  839.  
  840. /** Send message and return id of the sent message.
  841.  *
  842.  * The return value can be used as input for async_wait() to wait for
  843.  * completion.
  844.  *
  845.  * @param phoneid Handle of the phone that will be used for the send.
  846.  * @param method  Service-defined method.
  847.  * @param arg1    Service-defined payload argument.
  848.  * @param arg2    Service-defined payload argument.
  849.  * @param arg3    Service-defined payload argument.
  850.  * @param arg4    Service-defined payload argument.
  851.  * @param dataptr If non-NULL, storage where the reply data will be
  852.  *                stored.
  853.  *
  854.  * @return Hash of the sent message or 0 on error.
  855.  *
  856.  */
  857. aid_t async_send_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  858.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipc_call_t *dataptr)
  859. {
  860.     amsg_t *msg = malloc(sizeof(*msg));
  861.    
  862.     if (!msg)
  863.         return 0;
  864.    
  865.     msg->done = false;
  866.     msg->dataptr = dataptr;
  867.    
  868.     msg->wdata.inlist = false;
  869.     /* We may sleep in the next method, but it will use its own mechanism */
  870.     msg->wdata.active = true;
  871.    
  872.     ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,
  873.         reply_received, true);
  874.    
  875.     return (aid_t) msg;
  876. }
  877.  
  878. /** Send message and return id of the sent message
  879.  *
  880.  * The return value can be used as input for async_wait() to wait for
  881.  * completion.
  882.  *
  883.  * @param phoneid Handle of the phone that will be used for the send.
  884.  * @param method  Service-defined method.
  885.  * @param arg1    Service-defined payload argument.
  886.  * @param arg2    Service-defined payload argument.
  887.  * @param arg3    Service-defined payload argument.
  888.  * @param arg4    Service-defined payload argument.
  889.  * @param arg5    Service-defined payload argument.
  890.  * @param dataptr If non-NULL, storage where the reply data will be
  891.  *                stored.
  892.  *
  893.  * @return Hash of the sent message or 0 on error.
  894.  *
  895.  */
  896. aid_t async_send_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  897.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5,
  898.     ipc_call_t *dataptr)
  899. {
  900.     amsg_t *msg = malloc(sizeof(*msg));
  901.    
  902.     if (!msg)
  903.         return 0;
  904.    
  905.     msg->done = false;
  906.     msg->dataptr = dataptr;
  907.    
  908.     msg->wdata.inlist = false;
  909.     /* We may sleep in next method, but it will use its own mechanism */
  910.     msg->wdata.active = true;
  911.    
  912.     ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,
  913.         reply_received, true);
  914.    
  915.     return (aid_t) msg;
  916. }
  917.  
  918. /** Wait for a message sent by the async framework.
  919.  *
  920.  * @param amsgid Hash of the message to wait for.
  921.  * @param retval Pointer to storage where the retval of the answer will
  922.  *               be stored.
  923.  *
  924.  */
  925. void async_wait_for(aid_t amsgid, ipcarg_t *retval)
  926. {
  927.     amsg_t *msg = (amsg_t *) amsgid;
  928.    
  929.     futex_down(&async_futex);
  930.     if (msg->done) {
  931.         futex_up(&async_futex);
  932.         goto done;
  933.     }
  934.    
  935.     msg->wdata.fid = fibril_get_id();
  936.     msg->wdata.active = false;
  937.     msg->wdata.inlist = false;
  938.    
  939.     /* Leave the async_futex locked when entering this function */
  940.     fibril_switch(FIBRIL_TO_MANAGER);
  941.    
  942.     /* Futex is up automatically after fibril_switch */
  943.    
  944. done:
  945.     if (retval)
  946.         *retval = msg->retval;
  947.    
  948.     free(msg);
  949. }
  950.  
  951. /** Wait for a message sent by the async framework, timeout variant.
  952.  *
  953.  * @param amsgid  Hash of the message to wait for.
  954.  * @param retval  Pointer to storage where the retval of the answer will
  955.  *                be stored.
  956.  * @param timeout Timeout in microseconds.
  957.  *
  958.  * @return Zero on success, ETIMEOUT if the timeout has expired.
  959.  *
  960.  */
  961. int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
  962. {
  963.     amsg_t *msg = (amsg_t *) amsgid;
  964.    
  965.     /* TODO: Let it go through the event read at least once */
  966.     if (timeout < 0)
  967.         return ETIMEOUT;
  968.    
  969.     futex_down(&async_futex);
  970.     if (msg->done) {
  971.         futex_up(&async_futex);
  972.         goto done;
  973.     }
  974.    
  975.     gettimeofday(&msg->wdata.expires, NULL);
  976.     tv_add(&msg->wdata.expires, timeout);
  977.    
  978.     msg->wdata.fid = fibril_get_id();
  979.     msg->wdata.active = false;
  980.     insert_timeout(&msg->wdata);
  981.    
  982.     /* Leave the async_futex locked when entering this function */
  983.     fibril_switch(FIBRIL_TO_MANAGER);
  984.    
  985.     /* Futex is up automatically after fibril_switch */
  986.    
  987.     if (!msg->done)
  988.         return ETIMEOUT;
  989.    
  990. done:
  991.     if (retval)
  992.         *retval = msg->retval;
  993.    
  994.     free(msg);
  995.    
  996.     return 0;
  997. }
  998.  
  999. /** Wait for specified time.
  1000.  *
  1001.  * The current fibril is suspended but the thread continues to execute.
  1002.  *
  1003.  * @param timeout Duration of the wait in microseconds.
  1004.  *
  1005.  */
  1006. void async_usleep(suseconds_t timeout)
  1007. {
  1008.     amsg_t *msg = malloc(sizeof(*msg));
  1009.    
  1010.     if (!msg)
  1011.         return;
  1012.    
  1013.     msg->wdata.fid = fibril_get_id();
  1014.     msg->wdata.active = false;
  1015.    
  1016.     gettimeofday(&msg->wdata.expires, NULL);
  1017.     tv_add(&msg->wdata.expires, timeout);
  1018.    
  1019.     futex_down(&async_futex);
  1020.    
  1021.     insert_timeout(&msg->wdata);
  1022.    
  1023.     /* Leave the async_futex locked when entering this function */
  1024.     fibril_switch(FIBRIL_TO_MANAGER);
  1025.    
  1026.     /* Futex is up automatically after fibril_switch() */
  1027.    
  1028.     free(msg);
  1029. }
  1030.  
  1031. /** Setter for client_connection function pointer.
  1032.  *
  1033.  * @param conn Function that will implement a new connection fibril.
  1034.  *
  1035.  */
  1036. void async_set_client_connection(async_client_conn_t conn)
  1037. {
  1038.     client_connection = conn;
  1039. }
  1040.  
  1041. /** Setter for interrupt_received function pointer.
  1042.  *
  1043.  * @param intr Function that will implement a new interrupt
  1044.  *             notification fibril.
  1045.  */
  1046. void async_set_interrupt_received(async_client_conn_t intr)
  1047. {
  1048.     interrupt_received = intr;
  1049. }
  1050.  
  1051. /** Setter for pending function pointer.
  1052.  *
  1053.  * @param pend Function that will implement a new pending
  1054.  *             operations fibril.
  1055.  */
  1056. void async_set_pending(async_pending_t pend)
  1057. {
  1058.     pending = pend;
  1059. }
  1060.  
  1061. /** Pseudo-synchronous message sending - fast version.
  1062.  *
  1063.  * Send message asynchronously and return only after the reply arrives.
  1064.  *
  1065.  * This function can only transfer 4 register payload arguments. For
  1066.  * transferring more arguments, see the slower async_req_slow().
  1067.  *
  1068.  * @param phoneid Hash of the phone through which to make the call.
  1069.  * @param method  Method of the call.
  1070.  * @param arg1    Service-defined payload argument.
  1071.  * @param arg2    Service-defined payload argument.
  1072.  * @param arg3    Service-defined payload argument.
  1073.  * @param arg4    Service-defined payload argument.
  1074.  * @param r1      If non-NULL, storage for the 1st reply argument.
  1075.  * @param r2      If non-NULL, storage for the 2nd reply argument.
  1076.  * @param r3      If non-NULL, storage for the 3rd reply argument.
  1077.  * @param r4      If non-NULL, storage for the 4th reply argument.
  1078.  * @param r5      If non-NULL, storage for the 5th reply argument.
  1079.  *
  1080.  * @return Return code of the reply or a negative error code.
  1081.  *
  1082.  */
  1083. ipcarg_t async_req_fast(int phoneid, ipcarg_t method, ipcarg_t arg1,
  1084.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t *r1, ipcarg_t *r2,
  1085.     ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  1086. {
  1087.     ipc_call_t result;
  1088.     aid_t eid = async_send_4(phoneid, method, arg1, arg2, arg3, arg4,
  1089.         &result);
  1090.    
  1091.     ipcarg_t rc;
  1092.     async_wait_for(eid, &rc);
  1093.    
  1094.     if (r1)
  1095.         *r1 = IPC_GET_ARG1(result);
  1096.    
  1097.     if (r2)
  1098.         *r2 = IPC_GET_ARG2(result);
  1099.    
  1100.     if (r3)
  1101.         *r3 = IPC_GET_ARG3(result);
  1102.    
  1103.     if (r4)
  1104.         *r4 = IPC_GET_ARG4(result);
  1105.    
  1106.     if (r5)
  1107.         *r5 = IPC_GET_ARG5(result);
  1108.    
  1109.     return rc;
  1110. }
  1111.  
  1112. /** Pseudo-synchronous message sending - slow version.
  1113.  *
  1114.  * Send message asynchronously and return only after the reply arrives.
  1115.  *
  1116.  * @param phoneid Hash of the phone through which to make the call.
  1117.  * @param method  Method of the call.
  1118.  * @param arg1    Service-defined payload argument.
  1119.  * @param arg2    Service-defined payload argument.
  1120.  * @param arg3    Service-defined payload argument.
  1121.  * @param arg4    Service-defined payload argument.
  1122.  * @param arg5    Service-defined payload argument.
  1123.  * @param r1      If non-NULL, storage for the 1st reply argument.
  1124.  * @param r2      If non-NULL, storage for the 2nd reply argument.
  1125.  * @param r3      If non-NULL, storage for the 3rd reply argument.
  1126.  * @param r4      If non-NULL, storage for the 4th reply argument.
  1127.  * @param r5      If non-NULL, storage for the 5th reply argument.
  1128.  *
  1129.  * @return Return code of the reply or a negative error code.
  1130.  *
  1131.  */
  1132. ipcarg_t async_req_slow(int phoneid, ipcarg_t method, ipcarg_t arg1,
  1133.     ipcarg_t arg2, ipcarg_t arg3, ipcarg_t arg4, ipcarg_t arg5, ipcarg_t *r1,
  1134.     ipcarg_t *r2, ipcarg_t *r3, ipcarg_t *r4, ipcarg_t *r5)
  1135. {
  1136.     ipc_call_t result;
  1137.     aid_t eid = async_send_5(phoneid, method, arg1, arg2, arg3, arg4, arg5,
  1138.         &result);
  1139.    
  1140.     ipcarg_t rc;
  1141.     async_wait_for(eid, &rc);
  1142.    
  1143.     if (r1)
  1144.         *r1 = IPC_GET_ARG1(result);
  1145.    
  1146.     if (r2)
  1147.         *r2 = IPC_GET_ARG2(result);
  1148.    
  1149.     if (r3)
  1150.         *r3 = IPC_GET_ARG3(result);
  1151.    
  1152.     if (r4)
  1153.         *r4 = IPC_GET_ARG4(result);
  1154.    
  1155.     if (r5)
  1156.         *r5 = IPC_GET_ARG5(result);
  1157.    
  1158.     return rc;
  1159. }
  1160.  
  1161. /** @}
  1162.  */
  1163.