Subversion Repositories HelenOS

Rev

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