Subversion Repositories HelenOS

Rev

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