Changes in uspace/lib/c/generic/async.c [7f9d97f3:c170438] in mainline

File:

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

uspace/lib/c/generic/async.c

@@ -77 +77 @@
  *   }
  *
- *   my_client_connection(icallid, *icall)
+ *   port_handler(icallid, *icall)
  *   {
  *     if (want_refuse) {
@@ -123 +123 @@
         list_t exch_list;
         
+        /** Session interface */
+        iface_t iface;
+        
         /** Exchange management style */
         exch_mgmt_t mgmt;
@@ -189 +192 @@
         /** If reply was received. */
         bool done;
-
+        
         /** If the message / reply should be discarded on arrival. */
         bool forget;
-
+        
         /** If already destroyed. */
         bool destroyed;
@@ -232 +235 @@
         /** Identification of the opening call. */
         ipc_callid_t callid;
+        
         /** Call data of the opening call. */
         ipc_call_t call;
-        /** Local argument or NULL if none. */
-        void *carg;
         
         /** Identification of the closing call. */
@@ -241 +243 @@
         
         /** Fibril function that will be used to handle the connection. */
-        async_client_conn_t cfibril;
+        async_port_handler_t handler;
+        
+        /** Client data */
+        void *data;
 } connection_t;
+
+/** Interface data */
+typedef struct {
+        ht_link_t link;
+        
+        /** Interface ID */
+        iface_t iface;
+        
+        /** Futex protecting the hash table */
+        futex_t futex;
+        
+        /** Interface ports */
+        hash_table_t port_hash_table;
+        
+        /** Next available port ID */
+        port_id_t port_id_avail;
+} interface_t;
+
+/* Port data */
+typedef struct {
+        ht_link_t link;
+        
+        /** Port ID */
+        port_id_t id;
+        
+        /** Port connection handler */
+        async_port_handler_t handler;
+        
+        /** Client data */
+        void *data;
+} port_t;
 
 /* Notification data */
@@ -264 +300 @@
 {
         struct timeval tv = { 0, 0 };
-
+        
         to->inlist = false;
         to->occurred = false;
@@ -287 +323 @@
 static amsg_t *amsg_create(void)
 {
-        amsg_t *msg;
-
-        msg = malloc(sizeof(amsg_t));
+        amsg_t *msg = malloc(sizeof(amsg_t));
         if (msg) {
                 msg->done = false;
@@ -298 +332 @@
                 awaiter_initialize(&msg->wdata);
         }
-
+        
         return msg;
 }
@@ -335 +369 @@
 }
 
-/** Default fibril function that gets called to handle new connection.
- *
- * This function is defined as a weak symbol - to be redefined in user code.
+/** Default fallback fibril function.
+ *
+ * This fallback fibril function gets called on incomming
+ * connections that do not have a specific handler defined.
  *
  * @param callid Hash of the incoming call.
@@ -344 +379 @@
  *
  */
-static void default_client_connection(ipc_callid_t callid, ipc_call_t *call,
+static void default_fallback_port_handler(ipc_callid_t callid, ipc_call_t *call,
     void *arg)
 {
@@ -350 +385 @@
 }
 
-static async_client_conn_t client_connection = default_client_connection;
-static size_t notification_handler_stksz = FIBRIL_DFLT_STK_SIZE;
-
-/** Setter for client_connection function pointer.
- *
- * @param conn Function that will implement a new connection fibril.
- *
- */
-void async_set_client_connection(async_client_conn_t conn)
-{
-        assert(client_connection == default_client_connection);
-        client_connection = conn;
-}
-
-/** Set the stack size for the notification handler notification fibrils.
- *
- * @param size Stack size in bytes.
- */
-void async_set_notification_handler_stack_size(size_t size)
-{
-        notification_handler_stksz = size;
+static async_port_handler_t fallback_port_handler =
+    default_fallback_port_handler;
+static void *fallback_port_data = NULL;
+
+static hash_table_t interface_hash_table;
+
+static size_t interface_key_hash(void *key)
+{
+        iface_t iface = *(iface_t *) key;
+        return iface;
+}
+
+static size_t interface_hash(const ht_link_t *item)
+{
+        interface_t *interface = hash_table_get_inst(item, interface_t, link);
+        return interface_key_hash(&interface->iface);
+}
+
+static bool interface_key_equal(void *key, const ht_link_t *item)
+{
+        iface_t iface = *(iface_t *) key;
+        interface_t *interface = hash_table_get_inst(item, interface_t, link);
+        return iface == interface->iface;
+}
+
+/** Operations for the port hash table. */
+static hash_table_ops_t interface_hash_table_ops = {
+        .hash = interface_hash,
+        .key_hash = interface_key_hash,
+        .key_equal = interface_key_equal,
+        .equal = NULL,
+        .remove_callback = NULL
+};
+
+static size_t port_key_hash(void *key)
+{
+        port_id_t port_id = *(port_id_t *) key;
+        return port_id;
+}
+
+static size_t port_hash(const ht_link_t *item)
+{
+        port_t *port = hash_table_get_inst(item, port_t, link);
+        return port_key_hash(&port->id);
+}
+
+static bool port_key_equal(void *key, const ht_link_t *item)
+{
+        port_id_t port_id = *(port_id_t *) key;
+        port_t *port = hash_table_get_inst(item, port_t, link);
+        return port_id == port->id;
+}
+
+/** Operations for the port hash table. */
+static hash_table_ops_t port_hash_table_ops = {
+        .hash = port_hash,
+        .key_hash = port_key_hash,
+        .key_equal = port_key_equal,
+        .equal = NULL,
+        .remove_callback = NULL
+};
+
+static interface_t *async_new_interface(iface_t iface)
+{
+        interface_t *interface =
+            (interface_t *) malloc(sizeof(interface_t));
+        if (!interface)
+                return NULL;
+        
+        bool ret = hash_table_create(&interface->port_hash_table, 0, 0,
+            &port_hash_table_ops);
+        if (!ret) {
+                free(interface);
+                return NULL;
+        }
+        
+        interface->iface = iface;
+        futex_initialize(&interface->futex, 1);
+        interface->port_id_avail = 0;
+        
+        hash_table_insert(&interface_hash_table, &interface->link);
+        
+        return interface;
+}
+
+static port_t *async_new_port(interface_t *interface,
+    async_port_handler_t handler, void *data)
+{
+        port_t *port = (port_t *) malloc(sizeof(port_t));
+        if (!port)
+                return NULL;
+        
+        futex_down(&interface->futex);
+        
+        port_id_t id = interface->port_id_avail;
+        interface->port_id_avail++;
+        
+        port->id = id;
+        port->handler = handler;
+        port->data = data;
+        
+        hash_table_insert(&interface->port_hash_table, &port->link);
+        
+        futex_up(&interface->futex);
+        
+        return port;
 }
 
@@ -387 +507 @@
  */
 static FIBRIL_CONDVAR_INITIALIZE(avail_phone_cv);
+
+int async_create_port(iface_t iface, async_port_handler_t handler,
+    void *data, port_id_t *port_id)
+{
+        if ((iface & IFACE_MOD_MASK) == IFACE_MOD_CALLBACK)
+                return EINVAL;
+        
+        interface_t *interface;
+        
+        futex_down(&async_futex);
+        
+        ht_link_t *link = hash_table_find(&interface_hash_table, &iface);
+        if (link)
+                interface = hash_table_get_inst(link, interface_t, link);
+        else
+                interface = async_new_interface(iface);
+        
+        if (!interface) {
+                futex_up(&async_futex);
+                return ENOMEM;
+        }
+        
+        port_t *port = async_new_port(interface, handler, data);
+        if (!port) {
+                futex_up(&async_futex);
+                return ENOMEM;
+        }
+        
+        *port_id = port->id;
+        
+        futex_up(&async_futex);
+        
+        return EOK;
+}
+
+void async_set_fallback_port_handler(async_port_handler_t handler, void *data)
+{
+        assert(handler != NULL);
+        
+        fallback_port_handler = handler;
+        fallback_port_data = data;
+}
 
 static hash_table_t client_hash_table;
@@ -457 +619 @@
         .remove_callback = NULL
 };
+
+static client_t *async_client_get(task_id_t client_id, bool create)
+{
+        client_t *client = NULL;
+        
+        futex_down(&async_futex);
+        ht_link_t *link = hash_table_find(&client_hash_table, &client_id);
+        if (link) {
+                client = hash_table_get_inst(link, 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, &client->link);
+                }
+        }
+        
+        futex_up(&async_futex);
+        return client;
+}
+
+static void async_client_put(client_t *client)
+{
+        bool destroy;
+        
+        futex_down(&async_futex);
+        
+        if (atomic_predec(&client->refcnt) == 0) {
+                hash_table_remove(&client_hash_table, &client->in_task_id);
+                destroy = true;
+        } else
+                destroy = false;
+        
+        futex_up(&async_futex);
+        
+        if (destroy) {
+                if (client->data)
+                        async_client_data_destroy(client->data);
+                
+                free(client);
+        }
+}
+
+/** Wrapper for client connection fibril.
+ *
+ * When a new connection arrives, a fibril with this implementing
+ * function is created.
+ *
+ * @param arg Connection structure pointer.
+ *
+ * @return Always zero.
+ *
+ */
+static int connection_fibril(void *arg)
+{
+        assert(arg);
+        
+        /*
+         * Setup fibril-local connection pointer.
+         */
+        fibril_connection = (connection_t *) arg;
+        
+        /*
+         * Add our reference for the current connection in the client task
+         * tracking structure. If this is the first reference, create and
+         * hash in a new tracking structure.
+         */
+        
+        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;
+        
+        /*
+         * Call the connection handler function.
+         */
+        fibril_connection->handler(fibril_connection->callid,
+            &fibril_connection->call, fibril_connection->data);
+        
+        /*
+         * Remove the reference for this client task connection.
+         */
+        async_client_put(client);
+        
+        /*
+         * Remove myself from the connection hash table.
+         */
+        futex_down(&async_futex);
+        hash_table_remove(&conn_hash_table, &fibril_connection->in_phone_hash);
+        futex_up(&async_futex);
+        
+        /*
+         * Answer all remaining messages with EHANGUP.
+         */
+        while (!list_empty(&fibril_connection->msg_queue)) {
+                msg_t *msg =
+                    list_get_instance(list_first(&fibril_connection->msg_queue),
+                    msg_t, link);
+                
+                list_remove(&msg->link);
+                ipc_answer_0(msg->callid, EHANGUP);
+                free(msg);
+        }
+        
+        /*
+         * If the connection was hung-up, answer the last call,
+         * i.e. IPC_M_PHONE_HUNGUP.
+         */
+        if (fibril_connection->close_callid)
+                ipc_answer_0(fibril_connection->close_callid, EOK);
+        
+        free(fibril_connection);
+        return 0;
+}
+
+/** Create a new fibril for a new connection.
+ *
+ * Create new fibril for connection, fill in connection structures
+ * and insert it into the hash table, so that later we can easily
+ * do routing of messages to particular fibrils.
+ *
+ * @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.
+ *                      If callid is zero, the connection was opened by
+ *                      accepting the IPC_M_CONNECT_TO_ME call and this
+ *                      function is called directly by the server.
+ * @param call          Call data of the opening call.
+ * @param handler       Connection handler.
+ * @param data          Client argument to pass to the connection handler.
+ *
+ * @return New fibril id or NULL on failure.
+ *
+ */
+static 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_port_handler_t handler,
+    void *data)
+{
+        connection_t *conn = malloc(sizeof(*conn));
+        if (!conn) {
+                if (callid)
+                        ipc_answer_0(callid, ENOMEM);
+                
+                return (uintptr_t) NULL;
+        }
+        
+        conn->in_task_id = in_task_id;
+        conn->in_phone_hash = in_phone_hash;
+        list_initialize(&conn->msg_queue);
+        conn->callid = callid;
+        conn->close_callid = 0;
+        conn->handler = handler;
+        conn->data = data;
+        
+        if (call)
+                conn->call = *call;
+        
+        /* We will activate the fibril ASAP */
+        conn->wdata.active = true;
+        conn->wdata.fid = fibril_create(connection_fibril, conn);
+        
+        if (conn->wdata.fid == 0) {
+                free(conn);
+                
+                if (callid)
+                        ipc_answer_0(callid, ENOMEM);
+                
+                return (uintptr_t) NULL;
+        }
+        
+        /* Add connection to the connection hash table */
+        
+        futex_down(&async_futex);
+        hash_table_insert(&conn_hash_table, &conn->link);
+        futex_up(&async_futex);
+        
+        fibril_add_ready(conn->wdata.fid);
+        
+        return conn->wdata.fid;
+}
+
+/** Wrapper for making IPC_M_CONNECT_TO_ME calls using the async framework.
+ *
+ * Ask through phone for a new connection to some service.
+ *
+ * @param exch    Exchange for sending the message.
+ * @param iface   Callback interface.
+ * @param arg1    User defined argument.
+ * @param arg2    User defined argument.
+ * @param handler Callback handler.
+ * @param data    Handler data.
+ * @param port_id ID of the newly created port.
+ *
+ * @return Zero on success or a negative error code.
+ *
+ */
+int async_create_callback_port(async_exch_t *exch, iface_t iface, sysarg_t arg1,
+    sysarg_t arg2, async_port_handler_t handler, void *data, port_id_t *port_id)
+{
+        if ((iface & IFACE_MOD_CALLBACK) != IFACE_MOD_CALLBACK)
+                return EINVAL;
+        
+        if (exch == NULL)
+                return ENOENT;
+        
+        ipc_call_t answer;
+        aid_t req = async_send_3(exch, IPC_M_CONNECT_TO_ME, iface, arg1, arg2,
+            &answer);
+        
+        sysarg_t ret;
+        async_wait_for(req, &ret);
+        if (ret != EOK)
+                return (int) ret;
+        
+        sysarg_t phone_hash = IPC_GET_ARG5(answer);
+        interface_t *interface;
+        
+        futex_down(&async_futex);
+        
+        ht_link_t *link = hash_table_find(&interface_hash_table, &iface);
+        if (link)
+                interface = hash_table_get_inst(link, interface_t, link);
+        else
+                interface = async_new_interface(iface);
+        
+        if (!interface) {
+                futex_up(&async_futex);
+                return ENOMEM;
+        }
+        
+        port_t *port = async_new_port(interface, handler, data);
+        if (!port) {
+                futex_up(&async_futex);
+                return ENOMEM;
+        }
+        
+        *port_id = port->id;
+        
+        futex_up(&async_futex);
+        
+        fid_t fid = async_new_connection(answer.in_task_id, phone_hash,
+            0, NULL, handler, data);
+        if (fid == (uintptr_t) NULL)
+                return ENOMEM;
+        
+        return EOK;
+}
 
 static size_t notification_key_hash(void *key)
@@ -571 +987 @@
 }
 
-/** Notification fibril.
- *
- * When a notification arrives, a fibril with this implementing function is
- * created. It calls the corresponding notification handler and does the final
- * cleanup.
- *
- * @param arg Message structure pointer.
- *
- * @return Always zero.
- *
- */
-static int notification_fibril(void *arg)
-{
-        assert(arg);
-        
-        msg_t *msg = (msg_t *) arg;
+/** Process notification.
+ *
+ * @param callid Hash of the incoming call.
+ * @param call   Data of the incoming call.
+ */
+static void process_notification(ipc_callid_t callid, ipc_call_t *call)
+{
         async_notification_handler_t handler = NULL;
         void *data = NULL;
+
+        assert(call);
         
         futex_down(&async_futex);
         
         ht_link_t *link = hash_table_find(&notification_hash_table,
-            &IPC_GET_IMETHOD(msg->call));
+            &IPC_GET_IMETHOD(*call));
         if (link) {
                 notification_t *notification =
@@ -604 +1013 @@
         
         if (handler)
-                handler(msg->callid, &msg->call, data);
-        
-        free(msg);
-        return 0;
-}
-
-/** Process notification.
- *
- * A new fibril is created which would process the notification.
- *
- * @param callid Hash of the incoming call.
- * @param call   Data of the incoming call.
- *
- * @return False if an error occured.
- *         True if the call was passed to the notification fibril.
- *
- */
-static bool process_notification(ipc_callid_t callid, ipc_call_t *call)
-{
-        assert(call);
-        
-        futex_down(&async_futex);
-        
-        msg_t *msg = malloc(sizeof(*msg));
-        if (!msg) {
-                futex_up(&async_futex);
-                return false;
-        }
-        
-        msg->callid = callid;
-        msg->call = *call;
-        
-        fid_t fid = fibril_create_generic(notification_fibril, msg,
-            notification_handler_stksz);
-        if (fid == 0) {
-                free(msg);
-                futex_up(&async_futex);
-                return false;
-        }
-        
-        fibril_add_ready(fid);
-        
-        futex_up(&async_futex);
-        return true;
+                handler(callid, call, data);
 }
 
@@ -866 +1232 @@
         }
         
-        msg_t *msg = list_get_instance(list_first(&conn->msg_queue), msg_t, link);
+        msg_t *msg = list_get_instance(list_first(&conn->msg_queue),
+            msg_t, link);
         list_remove(&msg->link);
         
@@ -877 +1244 @@
 }
 
-static client_t *async_client_get(task_id_t client_id, bool create)
-{
-        client_t *client = NULL;
-
-        futex_down(&async_futex);
-        ht_link_t *link = hash_table_find(&client_hash_table, &client_id);
-        if (link) {
-                client = hash_table_get_inst(link, 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, &client->link);
-                }
-        }
-
-        futex_up(&async_futex);
-        return client;
-}
-
-static void async_client_put(client_t *client)
-{
-        bool destroy;
-
-        futex_down(&async_futex);
-        
-        if (atomic_predec(&client->refcnt) == 0) {
-                hash_table_remove(&client_hash_table, &client->in_task_id);
-                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)
 {
@@ -934 +1255 @@
         if (!client)
                 return NULL;
+        
         if (!client->data) {
                 async_client_put(client);
                 return NULL;
         }
-
+        
         return client->data;
 }
@@ -945 +1267 @@
 {
         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.
- *
- * When a new connection arrives, a fibril with this implementing function is
- * created. It calls client_connection() and does the final cleanup.
- *
- * @param arg Connection structure pointer.
- *
- * @return Always zero.
- *
- */
-static int connection_fibril(void *arg)
-{
-        assert(arg);
-        
-        /*
-         * Setup fibril-local connection pointer.
-         */
-        fibril_connection = (connection_t *) arg;
-        
-        /*
-         * Add our reference for the current connection in the client task
-         * tracking structure. If this is the first reference, create and
-         * hash in a new tracking structure.
-         */
-
-        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;
-        
-        /*
-         * Call the connection handler function.
-         */
-        fibril_connection->cfibril(fibril_connection->callid,
-            &fibril_connection->call, fibril_connection->carg);
-        
-        /*
-         * Remove the reference for this client task connection.
-         */
-        async_client_put(client);
-        
-        /*
-         * Remove myself from the connection hash table.
-         */
+static port_t *async_find_port(iface_t iface, port_id_t port_id)
+{
+        port_t *port = NULL;
+        
         futex_down(&async_futex);
-        hash_table_remove(&conn_hash_table, &fibril_connection->in_phone_hash);
+        
+        ht_link_t *link = hash_table_find(&interface_hash_table, &iface);
+        if (link) {
+                interface_t *interface =
+                    hash_table_get_inst(link, interface_t, link);
+                
+                link = hash_table_find(&interface->port_hash_table, &port_id);
+                if (link)
+                        port = hash_table_get_inst(link, port_t, link);
+        }
+        
         futex_up(&async_futex);
         
-        /*
-         * Answer all remaining messages with EHANGUP.
-         */
-        while (!list_empty(&fibril_connection->msg_queue)) {
-                msg_t *msg =
-                    list_get_instance(list_first(&fibril_connection->msg_queue),
-                    msg_t, link);
-                
-                list_remove(&msg->link);
-                ipc_answer_0(msg->callid, EHANGUP);
-                free(msg);
-        }
-        
-        /*
-         * If the connection was hung-up, answer the last call,
-         * i.e. IPC_M_PHONE_HUNGUP.
-         */
-        if (fibril_connection->close_callid)
-                ipc_answer_0(fibril_connection->close_callid, EOK);
-        
-        free(fibril_connection);
-        return 0;
-}
-
-/** Create a new fibril for a new connection.
- *
- * Create new fibril for connection, fill in connection structures and insert
- * it into the hash table, so that later we can easily do routing of messages to
- * particular fibrils.
- *
- * @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.
- *                      If callid is zero, the connection was opened by
- *                      accepting the IPC_M_CONNECT_TO_ME call and this function
- *                      is called directly by the server.
- * @param call          Call data of the opening call.
- * @param cfibril       Fibril function that should be called upon opening the
- *                      connection.
- * @param carg          Extra argument to pass to the connection fibril
- *
- * @return New fibril id or NULL on failure.
- *
- */
-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)
-{
-        connection_t *conn = malloc(sizeof(*conn));
-        if (!conn) {
-                if (callid)
-                        ipc_answer_0(callid, ENOMEM);
-                
-                return (uintptr_t) NULL;
-        }
-        
-        conn->in_task_id = in_task_id;
-        conn->in_phone_hash = in_phone_hash;
-        list_initialize(&conn->msg_queue);
-        conn->callid = callid;
-        conn->close_callid = 0;
-        conn->carg = carg;
-        
-        if (call)
-                conn->call = *call;
-        
-        /* We will activate the fibril ASAP */
-        conn->wdata.active = true;
-        conn->cfibril = cfibril;
-        conn->wdata.fid = fibril_create(connection_fibril, conn);
-        
-        if (conn->wdata.fid == 0) {
-                free(conn);
-                
-                if (callid)
-                        ipc_answer_0(callid, ENOMEM);
-                
-                return (uintptr_t) NULL;
-        }
-        
-        /* Add connection to the connection hash table */
-        
-        futex_down(&async_futex);
-        hash_table_insert(&conn_hash_table, &conn->link);
-        futex_up(&async_futex);
-        
-        fibril_add_ready(conn->wdata.fid);
-        
-        return conn->wdata.fid;
+        return port;
 }
 
@@ -1111 +1312 @@
         assert(call);
         
-        /* Unrouted call - take some default action */
+        /* Kernel notification */
         if ((callid & IPC_CALLID_NOTIFICATION)) {
+                fibril_t *fibril = (fibril_t *) __tcb_get()->fibril_data;
+                unsigned oldsw = fibril->switches;
+
                 process_notification(callid, call);
+
+                if (oldsw != fibril->switches) {
+                        /*
+                         * The notification handler did not execute atomically
+                         * and so the current manager fibril assumed the role of
+                         * a notification fibril. While waiting for its
+                         * resources, it switched to another manager fibril that
+                         * had already existed or it created a new one. We
+                         * therefore know there is at least yet another
+                         * manager fibril that can take over. We now kill the
+                         * current 'notification' fibril to prevent fibril
+                         * population explosion.
+                         */
+                        futex_down(&async_futex);
+                        fibril_switch(FIBRIL_FROM_DEAD);
+                }
                 return;
         }
         
-        switch (IPC_GET_IMETHOD(*call)) {
-        case IPC_M_CLONE_ESTABLISH:
-        case IPC_M_CONNECT_ME_TO:
+        /* New connection */
+        if (IPC_GET_IMETHOD(*call) == IPC_M_CONNECT_ME_TO) {
+                iface_t iface = (iface_t) IPC_GET_ARG1(*call);
+                sysarg_t in_phone_hash = IPC_GET_ARG5(*call);
+                
+                async_notification_handler_t handler = fallback_port_handler;
+                void *data = fallback_port_data;
+                
+                // TODO: Currently ignores all ports but the first one
+                port_t *port = async_find_port(iface, 0);
+                if (port) {
+                        handler = port->handler;
+                        data = port->data;
+                }
+                
+                async_new_connection(call->in_task_id, in_phone_hash, callid,
+                    call, handler, data);
+                return;
+        }
+        
+        /* Cloned connection */
+        if (IPC_GET_IMETHOD(*call) == IPC_M_CLONE_ESTABLISH) {
+                // TODO: Currently ignores ports altogether
+                
                 /* Open new connection with fibril, etc. */
                 async_new_connection(call->in_task_id, IPC_GET_ARG5(*call),
-                    callid, call, client_connection, NULL);
+                    callid, call, fallback_port_handler, fallback_port_data);
                 return;
         }
@@ -1267 +1508 @@
 void async_create_manager(void)
 {
-        fid_t fid = fibril_create(async_manager_fibril, NULL);
+        fid_t fid = fibril_create_generic(async_manager_fibril, NULL, PAGE_SIZE);
         if (fid != 0)
                 fibril_add_manager(fid);
@@ -1283 +1524 @@
 void __async_init(void)
 {
+        if (!hash_table_create(&interface_hash_table, 0, 0,
+            &interface_hash_table_ops))
+                abort();
+        
         if (!hash_table_create(&client_hash_table, 0, 0, &client_hash_table_ops))
                 abort();
@@ -1297 +1542 @@
                 abort();
         
+        session_ns->iface = 0;
         session_ns->mgmt = EXCHANGE_ATOMIC;
         session_ns->phone = PHONE_NS;
@@ -1343 +1589 @@
         
         msg->done = true;
-
+        
         if (msg->forget) {
                 assert(msg->wdata.active);
@@ -1351 +1597 @@
                 fibril_add_ready(msg->wdata.fid);
         }
-
+        
         futex_up(&async_futex);
 }
@@ -1386 +1632 @@
         
         ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4, msg,
-            reply_received, true);
+            reply_received);
         
         return (aid_t) msg;
@@ -1424 +1670 @@
         
         ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4, arg5,
-            msg, reply_received, true);
+            msg, reply_received);
         
         return (aid_t) msg;
@@ -1443 +1689 @@
         
         futex_down(&async_futex);
-
+        
         assert(!msg->forget);
         assert(!msg->destroyed);
-
+        
         if (msg->done) {
                 futex_up(&async_futex);
@@ -1487 +1733 @@
         
         amsg_t *msg = (amsg_t *) amsgid;
-
+        
         futex_down(&async_futex);
-
+        
         assert(!msg->forget);
         assert(!msg->destroyed);
-
+        
         if (msg->done) {
                 futex_up(&async_futex);
@@ -1504 +1750 @@
         if (timeout < 0)
                 timeout = 0;
-
+        
         getuptime(&msg->wdata.to_event.expires);
         tv_add_diff(&msg->wdata.to_event.expires, timeout);
@@ -1557 +1803 @@
 {
         amsg_t *msg = (amsg_t *) amsgid;
-
+        
         assert(msg);
         assert(!msg->forget);
         assert(!msg->destroyed);
-
+        
         futex_down(&async_futex);
+        
         if (msg->done) {
                 amsg_destroy(msg);
@@ -1569 +1816 @@
                 msg->forget = true;
         }
+        
         futex_up(&async_futex);
 }
@@ -1711 +1959 @@
 {
         if (exch != NULL)
-                ipc_call_async_0(exch->phone, imethod, NULL, NULL, true);
+                ipc_call_async_0(exch->phone, imethod, NULL, NULL);
 }
 
@@ -1717 +1965 @@
 {
         if (exch != NULL)
-                ipc_call_async_1(exch->phone, imethod, arg1, NULL, NULL, true);
+                ipc_call_async_1(exch->phone, imethod, arg1, NULL, NULL);
 }
 
@@ -1724 +1972 @@
 {
         if (exch != NULL)
-                ipc_call_async_2(exch->phone, imethod, arg1, arg2, NULL, NULL,
-                    true);
+                ipc_call_async_2(exch->phone, imethod, arg1, arg2, NULL, NULL);
 }
 
@@ -1733 +1980 @@
         if (exch != NULL)
                 ipc_call_async_3(exch->phone, imethod, arg1, arg2, arg3, NULL,
-                    NULL, true);
+                    NULL);
 }
 
@@ -1741 +1988 @@
         if (exch != NULL)
                 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4,
-                    NULL, NULL, true);
+                    NULL, NULL);
 }
 
@@ -1749 +1996 @@
         if (exch != NULL)
                 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4,
-                    arg5, NULL, NULL, true);
+                    arg5, NULL, NULL);
 }
 
@@ -1814 +2061 @@
  * @param arg2            User defined argument.
  * @param arg3            User defined argument.
- * @param client_receiver Connection handing routine.
  *
  * @return Zero on success or a negative error code.
@@ -1820 +2066 @@
  */
 int async_connect_to_me(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2,
-    sysarg_t arg3, async_client_conn_t client_receiver, void *carg)
+    sysarg_t arg3)
 {
         if (exch == NULL)
                 return ENOENT;
         
-        sysarg_t phone_hash;
+        ipc_call_t answer;
+        aid_t req = async_send_3(exch, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3,
+            &answer);
+        
         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 (int) rc;
-
-        phone_hash = IPC_GET_ARG5(answer);
-
-        if (client_receiver != NULL)
-                async_new_connection(answer.in_task_id, phone_hash, 0, NULL,
-                    client_receiver, carg);
         
         return EOK;
@@ -1881 +2119 @@
         
         ipc_call_async_0(exch->phone, IPC_M_CLONE_ESTABLISH, msg,
-            reply_received, true);
+            reply_received);
         
         sysarg_t rc;
@@ -1900 +2138 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -1929 +2168 @@
         
         ipc_call_async_4(phone, IPC_M_CONNECT_ME_TO, arg1, arg2, arg3, arg4,
-            msg, reply_received, true);
+            msg, reply_received);
         
         sysarg_t rc;
@@ -1969 +2208 @@
         int phone = async_connect_me_to_internal(exch->phone, arg1, arg2, arg3,
             0);
-        
         if (phone < 0) {
                 errno = phone;
@@ -1976 +2214 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -1992 +2231 @@
 }
 
+/** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
+ *
+ * Ask through phone for a new connection to some service and block until
+ * success.
+ *
+ * @param exch  Exchange for sending the message.
+ * @param iface Connection interface.
+ * @param arg2  User defined argument.
+ * @param arg3  User defined argument.
+ *
+ * @return New session on success or NULL on error.
+ *
+ */
+async_sess_t *async_connect_me_to_iface(async_exch_t *exch, iface_t iface,
+    sysarg_t arg2, sysarg_t arg3)
+{
+        if (exch == NULL) {
+                errno = ENOENT;
+                return NULL;
+        }
+        
+        async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
+        if (sess == NULL) {
+                errno = ENOMEM;
+                return NULL;
+        }
+        
+        int phone = async_connect_me_to_internal(exch->phone, iface, arg2,
+            arg3, 0);
+        if (phone < 0) {
+                errno = phone;
+                free(sess);
+                return NULL;
+        }
+        
+        sess->iface = iface;
+        sess->phone = phone;
+        sess->arg1 = iface;
+        sess->arg2 = arg2;
+        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);
+        atomic_set(&sess->refcnt, 0);
+        
+        return sess;
+}
+
 /** Set arguments for new connections.
  *
@@ -2047 +2337 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -2063 +2354 @@
 }
 
+/** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
+ *
+ * Ask through phone for a new connection to some service and block until
+ * success.
+ *
+ * @param exch  Exchange for sending the message.
+ * @param iface Connection interface.
+ * @param arg2  User defined argument.
+ * @param arg3  User defined argument.
+ *
+ * @return New session on success or NULL on error.
+ *
+ */
+async_sess_t *async_connect_me_to_blocking_iface(async_exch_t *exch, iface_t iface,
+    sysarg_t arg2, sysarg_t arg3)
+{
+        if (exch == NULL) {
+                errno = ENOENT;
+                return NULL;
+        }
+        
+        async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
+        if (sess == NULL) {
+                errno = ENOMEM;
+                return NULL;
+        }
+        
+        int phone = async_connect_me_to_internal(exch->phone, iface, arg2,
+            arg3, IPC_FLAG_BLOCKING);
+        if (phone < 0) {
+                errno = phone;
+                free(sess);
+                return NULL;
+        }
+        
+        sess->iface = iface;
+        sess->phone = phone;
+        sess->arg1 = iface;
+        sess->arg2 = arg2;
+        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);
+        atomic_set(&sess->refcnt, 0);
+        
+        return sess;
+}
+
 /** Connect to a task specified by id.
  *
@@ -2081 +2423 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = EXCHANGE_ATOMIC;
         sess->phone = phone;
@@ -2158 +2501 @@
                 return NULL;
         
-        async_exch_t *exch;
+        exch_mgmt_t mgmt = sess->mgmt;
+        if (sess->iface != 0)
+                mgmt = sess->iface & IFACE_EXCHANGE_MASK;
+        
+        async_exch_t *exch = NULL;
         
         fibril_mutex_lock(&async_sess_mutex);
@@ -2177 +2524 @@
                  */
                 
-                if ((sess->mgmt == EXCHANGE_ATOMIC) ||
-                    (sess->mgmt == EXCHANGE_SERIALIZE)) {
+                if ((mgmt == EXCHANGE_ATOMIC) ||
+                    (mgmt == EXCHANGE_SERIALIZE)) {
                         exch = (async_exch_t *) malloc(sizeof(async_exch_t));
                         if (exch != NULL) {
@@ -2186 +2533 @@
                                 exch->phone = sess->phone;
                         }
-                } else {  /* EXCHANGE_PARALLEL */
+                } else if (mgmt == EXCHANGE_PARALLEL) {
+                        int phone;
+                        
+                retry:
                         /*
                          * Make a one-time attempt to connect a new data phone.
                          */
-                        
-                        int phone;
-                        
-retry:
                         phone = async_connect_me_to_internal(sess->phone, sess->arg1,
                             sess->arg2, sess->arg3, 0);
@@ -2235 +2581 @@
                 atomic_inc(&sess->refcnt);
                 
-                if (sess->mgmt == EXCHANGE_SERIALIZE)
+                if (mgmt == EXCHANGE_SERIALIZE)
                         fibril_mutex_lock(&sess->mutex);
         }
@@ -2255 +2601 @@
         assert(sess != NULL);
         
+        exch_mgmt_t mgmt = sess->mgmt;
+        if (sess->iface != 0)
+                mgmt = sess->iface & IFACE_EXCHANGE_MASK;
+        
         atomic_dec(&sess->refcnt);
         
-        if (sess->mgmt == EXCHANGE_SERIALIZE)
+        if (mgmt == EXCHANGE_SERIALIZE)
                 fibril_mutex_unlock(&sess->mutex);
         
@@ -2694 +3044 @@
         }
         
-        void *_data;
+        void *arg_data;
         
         if (nullterm)
-                _data = malloc(size + 1);
+                arg_data = malloc(size + 1);
         else
-                _data = malloc(size);
-        
-        if (_data == NULL) {
+                arg_data = malloc(size);
+        
+        if (arg_data == NULL) {
                 ipc_answer_0(callid, ENOMEM);
                 return ENOMEM;
         }
         
-        int rc = async_data_write_finalize(callid, _data, size);
+        int rc = async_data_write_finalize(callid, arg_data, size);
         if (rc != EOK) {
-                free(_data);
+                free(arg_data);
                 return rc;
         }
         
         if (nullterm)
-                ((char *) _data)[size] = 0;
-        
-        *data = _data;
+                ((char *) arg_data)[size] = 0;
+        
+        *data = arg_data;
         if (received != NULL)
                 *received = size;
@@ -2813 +3163 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -2862 +3213 @@
         }
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -2907 +3259 @@
                 return NULL;
         
+        sess->iface = 0;
         sess->mgmt = mgmt;
         sess->phone = phone;
@@ -2934 +3287 @@
 {
         assert(callid);
-
+        
         ipc_call_t call;
         *callid = async_get_call(&call);
-
+        
         if (IPC_GET_IMETHOD(call) != IPC_M_STATE_CHANGE_AUTHORIZE)
                 return false;
@@ -2947 +3300 @@
         if (arg3)
                 *arg3 = IPC_GET_ARG3(call);
-
+        
         return true;
 }