Changes in uspace/lib/c/generic/async.c [b72efe8:36e2b55] in mainline

File:

• uspace/lib/c/generic/async.c (modified) (29 diffs)

uspace/lib/c/generic/async.c

@@ -98 +98 @@
 #include <ipc/ipc.h>
 #include <async.h>
+#include "private/async.h"
 #undef LIBC_ASYNC_C_
 
@@ -107 +108 @@
 #include <errno.h>
 #include <sys/time.h>
-#include <arch/barrier.h>
+#include <libarch/barrier.h>
 #include <bool.h>
 #include <malloc.h>
 #include <mem.h>
 #include <stdlib.h>
-#include "private/async.h"
+#include <macros.h>
 
 #define CLIENT_HASH_TABLE_BUCKETS  32
@@ -138 +139 @@
         link_t link;
         
-        sysarg_t in_task_hash;
+        task_id_t in_task_id;
         atomic_t refcnt;
         void *data;
@@ -150 +151 @@
         link_t link;
         
-        /** Incoming client task hash. */
-        sysarg_t in_task_hash;
+        /** Incoming client task ID. */
+        task_id_t in_task_id;
         
         /** Incoming phone hash. */
@@ -203 +204 @@
 }
 
-void *async_get_client_data(void)
-{
-        assert(fibril_connection);
-        return fibril_connection->client->data;
-}
-
 /** Default fibril function that gets called to handle new connection.
  *
  * This function is defined as a weak symbol - to be redefined in user code.
  *
- * @param callid        Hash of the incoming call.
- * @param call          Data of the incoming call.
- * @param arg           Local argument
+ * @param callid Hash of the incoming call.
+ * @param call   Data of the incoming call.
+ * @param arg    Local argument
  *
  */
@@ -228 +223 @@
  * This function is defined as a weak symbol - to be redefined in user code.
  *
- * @param callid        Hash of the incoming call.
- * @param call          Data of the incoming call.
- * @param arg           Local argument.
+ * @param callid Hash of the incoming call.
+ * @param call   Data of the incoming call.
+ * @param arg    Local argument.
  *
  */
@@ -289 +284 @@
 {
         assert(key);
+        assert(keys == 2);
         assert(item);
         
         client_t *client = hash_table_get_instance(item, client_t, link);
-        return (key[0] == client->in_task_hash);
+        return (key[0] == LOWER32(client->in_task_id) &&
+            (key[1] == UPPER32(client->in_task_id)));
 }
 
@@ -580 +577 @@
 }
 
+static client_t *async_client_get(task_id_t client_id, bool create)
+{
+        unsigned long key[2] = {
+                LOWER32(client_id),
+                UPPER32(client_id),
+        };
+        client_t *client = NULL;
+
+        futex_down(&async_futex);
+        link_t *lnk = hash_table_find(&client_hash_table, key);
+        if (lnk) {
+                client = hash_table_get_instance(lnk, client_t, link);
+                atomic_inc(&client->refcnt);
+        } else if (create) {
+                client = malloc(sizeof(client_t));
+                if (client) {
+                        client->in_task_id = client_id;
+                        client->data = async_client_data_create();
+                
+                        atomic_set(&client->refcnt, 1);
+                        hash_table_insert(&client_hash_table, key, &client->link);
+                }
+        }
+
+        futex_up(&async_futex);
+        return client;
+}
+
+static void async_client_put(client_t *client)
+{
+        bool destroy;
+        unsigned long key[2] = {
+                LOWER32(client->in_task_id),
+                UPPER32(client->in_task_id)
+        };
+        
+        futex_down(&async_futex);
+        
+        if (atomic_predec(&client->refcnt) == 0) {
+                hash_table_remove(&client_hash_table, key, 2);
+                destroy = true;
+        } else
+                destroy = false;
+        
+        futex_up(&async_futex);
+        
+        if (destroy) {
+                if (client->data)
+                        async_client_data_destroy(client->data);
+                
+                free(client);
+        }
+}
+
+void *async_get_client_data(void)
+{
+        assert(fibril_connection);
+        return fibril_connection->client->data;
+}
+
+void *async_get_client_data_by_id(task_id_t client_id)
+{
+        client_t *client = async_client_get(client_id, false);
+        if (!client)
+                return NULL;
+        if (!client->data) {
+                async_client_put(client);
+                return NULL;
+        }
+
+        return client->data;
+}
+
+void async_put_client_data_by_id(task_id_t client_id)
+{
+        client_t *client = async_client_get(client_id, false);
+
+        assert(client);
+        assert(client->data);
+
+        /* Drop the reference we got in async_get_client_data_by_hash(). */
+        async_client_put(client);
+
+        /* Drop our own reference we got at the beginning of this function. */
+        async_client_put(client);
+}
+
 /** Wrapper for client connection fibril.
  *
@@ -598 +682 @@
          */
         fibril_connection = (connection_t *) arg;
-        
-        futex_down(&async_futex);
         
         /*
@@ -606 +688 @@
          * hash in a new tracking structure.
          */
-        
-        unsigned long key = fibril_connection->in_task_hash;
-        link_t *lnk = hash_table_find(&client_hash_table, &key);
-        
-        client_t *client;
-        
-        if (lnk) {
-                client = hash_table_get_instance(lnk, client_t, link);
-                atomic_inc(&client->refcnt);
-        } else {
-                client = malloc(sizeof(client_t));
-                if (!client) {
-                        ipc_answer_0(fibril_connection->callid, ENOMEM);
-                        futex_up(&async_futex);
-                        return 0;
-                }
-                
-                client->in_task_hash = fibril_connection->in_task_hash;
-                client->data = async_client_data_create();
-                
-                atomic_set(&client->refcnt, 1);
-                hash_table_insert(&client_hash_table, &key, &client->link);
-        }
-        
-        futex_up(&async_futex);
-        
+
+        client_t *client = async_client_get(fibril_connection->in_task_id, true);
+        if (!client) {
+                ipc_answer_0(fibril_connection->callid, ENOMEM);
+                return 0;
+        }
+
         fibril_connection->client = client;
         
@@ -643 +706 @@
          * Remove the reference for this client task connection.
          */
-        bool destroy;
-        
-        futex_down(&async_futex);
-        
-        if (atomic_predec(&client->refcnt) == 0) {
-                hash_table_remove(&client_hash_table, &key, 1);
-                destroy = true;
-        } else
-                destroy = false;
-        
-        futex_up(&async_futex);
-        
-        if (destroy) {
-                if (client->data)
-                        async_client_data_destroy(client->data);
-                
-                free(client);
-        }
+        async_client_put(client);
         
         /*
@@ -666 +712 @@
          */
         futex_down(&async_futex);
-        key = fibril_connection->in_phone_hash;
+        unsigned long key = fibril_connection->in_phone_hash;
         hash_table_remove(&conn_hash_table, &key, 1);
         futex_up(&async_futex);
@@ -700 +746 @@
  * particular fibrils.
  *
- * @param in_task_hash  Identification of the incoming connection.
+ * @param in_task_id    Identification of the incoming connection.
  * @param in_phone_hash Identification of the incoming connection.
  * @param callid        Hash of the opening IPC_M_CONNECT_ME_TO call.
@@ -709 +755 @@
  * @param cfibril       Fibril function that should be called upon opening the
  *                      connection.
- * @param carg          Extra argument to pass to the connection fibril
+ * @param carg          Extra argument to pass to the connection fibril
  *
  * @return New fibril id or NULL on failure.
  *
  */
-fid_t async_new_connection(sysarg_t in_task_hash, sysarg_t in_phone_hash,
+fid_t async_new_connection(task_id_t in_task_id, sysarg_t in_phone_hash,
     ipc_callid_t callid, ipc_call_t *call,
     async_client_conn_t cfibril, void *carg)
@@ -726 +772 @@
         }
         
-        conn->in_task_hash = in_task_hash;
+        conn->in_task_id = in_task_id;
         conn->in_phone_hash = in_phone_hash;
         list_initialize(&conn->msg_queue);
@@ -785 +831 @@
         case IPC_M_CONNECT_ME_TO:
                 /* Open new connection with fibril, etc. */
-                async_new_connection(call->in_task_hash, IPC_GET_ARG5(*call),
+                async_new_connection(call->in_task_id, IPC_GET_ARG5(*call),
                     callid, call, client_connection, NULL);
                 return;
@@ -933 +979 @@
 {
         if (!hash_table_create(&client_hash_table, CLIENT_HASH_TABLE_BUCKETS,
-            1, &client_hash_table_ops))
+            2, &client_hash_table_ops))
                 abort();
         
@@ -949 +995 @@
         session_ns->arg2 = 0;
         session_ns->arg3 = 0;
+        
+        fibril_mutex_initialize(&session_ns->remote_state_mtx);
+        session_ns->remote_state_data = NULL;
         
         list_initialize(&session_ns->exch_list);
@@ -1426 +1475 @@
                 return ENOENT;
         
-        sysarg_t task_hash;
         sysarg_t phone_hash;
-        int rc = async_req_3_5(exch, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3,
-            NULL, NULL, NULL, &task_hash, &phone_hash);
+        sysarg_t rc;
+
+        aid_t req;
+        ipc_call_t answer;
+        req = async_send_3(exch, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3,
+            &answer);
+        async_wait_for(req, &rc);
         if (rc != EOK)
-                return rc;
-        
+                return (int) rc;
+
+        phone_hash = IPC_GET_ARG5(answer);
+
         if (client_receiver != NULL)
-                async_new_connection(task_hash, phone_hash, 0, NULL,
+                async_new_connection(answer.in_task_id, phone_hash, 0, NULL,
                     client_receiver, carg);
         
@@ -1509 +1564 @@
         sess->arg3 = 0;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -1590 +1648 @@
         sess->arg3 = arg3;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -1595 +1656 @@
         
         return sess;
+}
+
+/** Set arguments for new connections.
+ *
+ * FIXME This is an ugly hack to work around the problem that parallel
+ * exchanges are implemented using parallel connections. When we create
+ * a callback session, the framework does not know arguments for the new
+ * connections.
+ *
+ * The proper solution seems to be to implement parallel exchanges using
+ * tagging.
+ */
+void async_sess_args_set(async_sess_t *sess, sysarg_t arg1, sysarg_t arg2,
+    sysarg_t arg3)
+{
+        sess->arg1 = arg1;
+        sess->arg2 = arg2;
+        sess->arg3 = arg3;
 }
 
@@ -1640 +1719 @@
         sess->arg3 = arg3;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -1671 +1753 @@
         sess->arg3 = 0;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -1692 +1777 @@
 int async_hangup(async_sess_t *sess)
 {
+        async_exch_t *exch;
+        
         assert(sess);
         
         if (atomic_get(&sess->refcnt) > 0)
                 return EBUSY;
+        
+        fibril_mutex_lock(&async_sess_mutex);
         
         int rc = async_hangup_internal(sess->phone);
         if (rc == EOK)
                 free(sess);
+        
+        while (!list_empty(&sess->exch_list)) {
+                exch = (async_exch_t *)
+                    list_get_instance(list_first(&sess->exch_list),
+                    async_exch_t, sess_link);
+                
+                list_remove(&exch->sess_link);
+                list_remove(&exch->global_link);
+                async_hangup_internal(exch->phone);
+                free(exch);
+        }
+        
+        fibril_mutex_unlock(&async_sess_mutex);
         
         return rc;
@@ -2334 +2436 @@
         sess->arg3 = 0;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -2380 +2485 @@
         sess->arg3 = 0;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -2422 +2530 @@
         sess->arg3 = 0;
         
+        fibril_mutex_initialize(&sess->remote_state_mtx);
+        sess->remote_state_data = NULL;
+        
         list_initialize(&sess->exch_list);
         fibril_mutex_initialize(&sess->mutex);
@@ -2429 +2540 @@
 }
 
+int async_state_change_start(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2,
+    sysarg_t arg3, async_exch_t *other_exch)
+{
+        return async_req_5_0(exch, IPC_M_STATE_CHANGE_AUTHORIZE,
+            arg1, arg2, arg3, 0, other_exch->phone);
+}
+
+bool async_state_change_receive(ipc_callid_t *callid, sysarg_t *arg1,
+    sysarg_t *arg2, sysarg_t *arg3)
+{
+        assert(callid);
+
+        ipc_call_t call;
+        *callid = async_get_call(&call);
+
+        if (IPC_GET_IMETHOD(call) != IPC_M_STATE_CHANGE_AUTHORIZE)
+                return false;
+        
+        if (arg1)
+                *arg1 = IPC_GET_ARG1(call);
+        if (arg2)
+                *arg2 = IPC_GET_ARG2(call);
+        if (arg3)
+                *arg3 = IPC_GET_ARG3(call);
+
+        return true;
+}
+
+int async_state_change_finalize(ipc_callid_t callid, async_exch_t *other_exch)
+{
+        return ipc_answer_1(callid, EOK, other_exch->phone);
+}
+
+/** Lock and get session remote state
+ *
+ * Lock and get the local replica of the remote state
+ * in stateful sessions. The call should be paired
+ * with async_remote_state_release*().
+ *
+ * @param[in] sess Stateful session.
+ *
+ * @return Local replica of the remote state.
+ *
+ */
+void *async_remote_state_acquire(async_sess_t *sess)
+{
+        fibril_mutex_lock(&sess->remote_state_mtx);
+        return sess->remote_state_data;
+}
+
+/** Update the session remote state
+ *
+ * Update the local replica of the remote state
+ * in stateful sessions. The remote state must
+ * be already locked.
+ *
+ * @param[in] sess  Stateful session.
+ * @param[in] state New local replica of the remote state.
+ *
+ */
+void async_remote_state_update(async_sess_t *sess, void *state)
+{
+        assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
+        sess->remote_state_data = state;
+}
+
+/** Release the session remote state
+ *
+ * Unlock the local replica of the remote state
+ * in stateful sessions.
+ *
+ * @param[in] sess Stateful session.
+ *
+ */
+void async_remote_state_release(async_sess_t *sess)
+{
+        assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
+        
+        fibril_mutex_unlock(&sess->remote_state_mtx);
+}
+
+/** Release the session remote state and end an exchange
+ *
+ * Unlock the local replica of the remote state
+ * in stateful sessions. This is convenience function
+ * which gets the session pointer from the exchange
+ * and also ends the exchange.
+ *
+ * @param[in] exch Stateful session's exchange.
+ *
+ */
+void async_remote_state_release_exchange(async_exch_t *exch)
+{
+        if (exch == NULL)
+                return;
+        
+        async_sess_t *sess = exch->sess;
+        assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
+        
+        async_exchange_end(exch);
+        fibril_mutex_unlock(&sess->remote_state_mtx);
+}
+
 /** @}
  */