Index: uspace/lib/c/generic/async.c =================================================================== --- uspace/lib/c/generic/async.c (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/generic/async.c (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -123,4 +123,7 @@ list_t exch_list; + /** Session interface */ + iface_t iface; + /** Exchange management style */ exch_mgmt_t mgmt; @@ -245,4 +248,35 @@ 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 */ @@ -355,4 +389,108 @@ 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; +} + static size_t notification_handler_stksz = FIBRIL_DFLT_STK_SIZE; @@ -380,4 +518,38 @@ */ 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) @@ -959,4 +1131,25 @@ } +static port_t *async_find_port(iface_t iface, port_id_t port_id) +{ + port_t *port = NULL; + + futex_down(&async_futex); + + 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); + + return port; +} + /** Wrapper for client connection fibril. * @@ -1120,7 +1313,28 @@ } - 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), @@ -1286,4 +1500,8 @@ 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(); @@ -1300,4 +1518,5 @@ abort(); + session_ns->iface = 0; session_ns->mgmt = EXCHANGE_ATOMIC; session_ns->phone = PHONE_NS; @@ -1905,4 +2124,5 @@ } + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; @@ -1981,4 +2201,5 @@ } + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; @@ -2052,4 +2273,5 @@ } + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; @@ -2068,4 +2290,55 @@ } +/** 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. * @@ -2086,4 +2359,5 @@ } + sess->iface = 0; sess->mgmt = EXCHANGE_ATOMIC; sess->phone = phone; @@ -2163,5 +2437,9 @@ 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); @@ -2182,6 +2460,6 @@ */ - 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) { @@ -2191,12 +2469,11 @@ 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); @@ -2240,5 +2517,5 @@ atomic_inc(&sess->refcnt); - if (sess->mgmt == EXCHANGE_SERIALIZE) + if (mgmt == EXCHANGE_SERIALIZE) fibril_mutex_lock(&sess->mutex); } @@ -2260,7 +2537,11 @@ 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); @@ -2818,4 +3099,5 @@ } + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; @@ -2867,4 +3149,5 @@ } + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; @@ -2912,4 +3195,5 @@ return NULL; + sess->iface = 0; sess->mgmt = mgmt; sess->phone = phone; Index: uspace/lib/c/generic/loader.c =================================================================== --- uspace/lib/c/generic/loader.c (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/generic/loader.c (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -69,5 +69,6 @@ async_sess_t *sess = - service_connect_blocking(EXCHANGE_SERIALIZE, SERVICE_LOAD, 0); + service_connect_blocking_iface_extended(SERVICE_LOADER, + INTERFACE_LOADER, 0); if (sess == NULL) { free(ldr); Index: uspace/lib/c/generic/ns.c =================================================================== --- uspace/lib/c/generic/ns.c (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/generic/ns.c (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -40,5 +40,5 @@ #include "private/ns.h" -int service_register(sysarg_t service) +int service_register(service_t service) { async_exch_t *exch = async_exchange_begin(session_ns); @@ -119,10 +119,29 @@ } +async_sess_t *service_connect_blocking_iface_extended(service_t service, + iface_t iface, sysarg_t arg3) +{ + async_exch_t *exch = async_exchange_begin(session_ns); + async_sess_t *sess = + async_connect_me_to_blocking_iface(exch, iface, service, arg3); + async_exchange_end(exch); + + if (!sess) + return NULL; + + /* + * FIXME Ugly hack to work around limitation of implementing + * parallel exchanges using multiple connections. Shift out + * first argument for non-initial connections. + */ + async_sess_args_set(sess, iface, arg3, 0); + + return sess; +} + async_sess_t *service_connect_blocking(exch_mgmt_t mgmt, service_t service, sysarg_t arg3) { async_exch_t *exch = async_exchange_begin(session_ns); - if (!exch) - return NULL; async_sess_t *sess = async_connect_me_to_blocking(mgmt, exch, 0, service, arg3); Index: uspace/lib/c/include/async.h =================================================================== --- uspace/lib/c/include/async.h (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/include/async.h (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -162,4 +162,6 @@ extern void async_put_client_data_by_id(task_id_t); +extern int async_create_port(iface_t, async_port_handler_t, void *, + port_id_t *); extern void async_set_fallback_port_handler(async_port_handler_t, void *); extern void async_set_notification_handler_stack_size(size_t); @@ -347,4 +349,6 @@ extern async_sess_t *async_connect_me_to_blocking(exch_mgmt_t, async_exch_t *, sysarg_t, sysarg_t, sysarg_t); +extern async_sess_t *async_connect_me_to_blocking_iface(async_exch_t *, iface_t, + sysarg_t, sysarg_t); extern async_sess_t *async_connect_kbox(task_id_t); Index: uspace/lib/c/include/ipc/services.h =================================================================== --- uspace/lib/c/include/ipc/services.h (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/include/ipc/services.h (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -43,5 +43,5 @@ typedef enum { SERVICE_NONE = 0, - SERVICE_LOAD = FOURCC('l', 'o', 'a', 'd'), + SERVICE_LOADER = FOURCC('l', 'o', 'a', 'd'), SERVICE_VFS = FOURCC('v', 'f', 's', ' '), SERVICE_LOC = FOURCC('l', 'o', 'c', ' '), Index: uspace/lib/c/include/ns.h =================================================================== --- uspace/lib/c/include/ns.h (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/lib/c/include/ns.h (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -41,8 +41,10 @@ #include -extern int service_register(sysarg_t); +extern int service_register(service_t); extern async_sess_t *service_connect(exch_mgmt_t, service_t, sysarg_t); extern async_sess_t *service_connect_iface(exch_mgmt_t, sysarg_t, service_t, sysarg_t); extern async_sess_t *service_connect_blocking(exch_mgmt_t, service_t, sysarg_t); +extern async_sess_t *service_connect_blocking_iface_extended(service_t, iface_t, + sysarg_t); extern async_sess_t *service_connect_blocking_iface(exch_mgmt_t, sysarg_t, service_t, sysarg_t); Index: uspace/srv/loader/main.c =================================================================== --- uspace/srv/loader/main.c (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/srv/loader/main.c (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -443,5 +443,4 @@ int main(int argc, char *argv[]) { - /* Set a handler of incomming connections. */ async_set_fallback_port_handler(ldr_connection, NULL); @@ -452,9 +451,15 @@ return rc; - /* Register at naming service. */ - rc = service_register(SERVICE_LOAD); + /* Create port */ + port_id_t port; + rc = async_create_port(INTERFACE_LOADER, ldr_connection, NULL, &port); if (rc != EOK) return rc; + /* Register at naming service. */ + rc = service_register(SERVICE_LOADER); + if (rc != EOK) + return rc; + async_manager(); Index: uspace/srv/ns/clonable.c =================================================================== --- uspace/srv/ns/clonable.c (revision 57dea627b9006b8560cadda6c4bba014adcbdc35) +++ uspace/srv/ns/clonable.c (revision 566992e1d4649f5d2bb5e0b4c2843589efcd9fb3) @@ -64,5 +64,5 @@ bool service_clonable(service_t service) { - return (service == SERVICE_LOAD); + return (service == SERVICE_LOADER); } @@ -89,5 +89,5 @@ /* Currently we can only handle a single type of clonable service. */ - assert(csr->service == SERVICE_LOAD); + assert(csr->service == SERVICE_LOADER); ipc_answer_0(callid, EOK); @@ -113,5 +113,5 @@ ipc_callid_t callid) { - assert(service == SERVICE_LOAD); + assert(service == SERVICE_LOADER); cs_req_t *csr = malloc(sizeof(cs_req_t));