Index: uspace/lib/c/generic/async/server.c =================================================================== --- uspace/lib/c/generic/async/server.c (revision a55d76b113225868d13a698012a31aae993d243d) +++ uspace/lib/c/generic/async/server.c (revision c407b987d06b7c0ba17e0044f01f5989c5eaeb4b) @@ -181,5 +181,9 @@ /* Notification data */ typedef struct { - ht_link_t link; + /** notification_hash_table link */ + ht_link_t htlink; + + /** notification_queue link */ + link_t qlink; /** Notification method */ @@ -189,6 +193,20 @@ async_notification_handler_t handler; - /** Notification data */ - void *data; + /** Notification handler argument */ + void *arg; + + /** Data of the most recent notification. */ + ipc_call_t calldata; + + /** + * How many notifications with this `imethod` arrived since it was last + * handled. If `count` > 1, `calldata` only holds the data for the most + * recent such notification, all the older data being lost. + * + * `async_spawn_notification_handler()` can be used to increase the + * number of notifications that can be processed simultaneously, + * reducing the likelihood of losing them when the handler blocks. + */ + long count; } notification_t; @@ -224,5 +242,11 @@ static hash_table_t client_hash_table; static hash_table_t conn_hash_table; + +// TODO: lockfree notification_queue? +static futex_t notification_futex = FUTEX_INITIALIZER; static hash_table_t notification_hash_table; +static LIST_INITIALIZE(notification_queue); +static FIBRIL_SEMAPHORE_INITIALIZE(notification_semaphore, 0); + static LIST_INITIALIZE(timeout_list); @@ -556,5 +580,5 @@ { notification_t *notification = - hash_table_get_inst(item, notification_t, link); + hash_table_get_inst(item, notification_t, htlink); return notification_key_hash(¬ification->imethod); } @@ -564,5 +588,5 @@ sysarg_t id = *(sysarg_t *) key; notification_t *notification = - hash_table_get_inst(item, notification_t, link); + hash_table_get_inst(item, notification_t, htlink); return id == notification->imethod; } @@ -663,31 +687,144 @@ } -/** Process notification. +/** Function implementing the notification handler fibril. Never returns. */ +static errno_t notification_fibril_func(void *arg) +{ + (void) arg; + + while (true) { + fibril_semaphore_down(¬ification_semaphore); + + futex_lock(¬ification_futex); + + /* + * The semaphore ensures that if we get this far, + * the queue must be non-empty. + */ + assert(!list_empty(¬ification_queue)); + + notification_t *notification = list_get_instance( + list_first(¬ification_queue), notification_t, qlink); + list_remove(¬ification->qlink); + + async_notification_handler_t handler = notification->handler; + void *arg = notification->arg; + ipc_call_t calldata = notification->calldata; + long count = notification->count; + + notification->count = 0; + + futex_unlock(¬ification_futex); + + // FIXME: Pass count to the handler. It might be important. + (void) count; + + if (handler) + handler(&calldata, arg); + } + + /* Not reached. */ + return EOK; +} + +/** + * Creates a new dedicated fibril for handling notifications. + * By default, there is one such fibril. This function can be used to + * create more in order to increase the number of notification that can + * be processed concurrently. + * + * Currently, there is no way to destroy those fibrils after they are created. + */ +errno_t async_spawn_notification_handler(void) +{ + fid_t f = fibril_create(notification_fibril_func, NULL); + if (f == 0) + return ENOMEM; + + fibril_add_ready(f); + return EOK; +} + +/** Queue notification. * * @param call Data of the incoming call. * */ -static void process_notification(ipc_call_t *call) -{ - async_notification_handler_t handler = NULL; - void *data = NULL; - +static void queue_notification(ipc_call_t *call) +{ assert(call); - futex_down(&async_futex); + futex_lock(¬ification_futex); ht_link_t *link = hash_table_find(¬ification_hash_table, &IPC_GET_IMETHOD(*call)); - if (link) { - notification_t *notification = - hash_table_get_inst(link, notification_t, link); - handler = notification->handler; - data = notification->data; - } - - futex_up(&async_futex); - - if (handler) - handler(call, data); + if (!link) { + /* Invalid notification. */ + // TODO: Make sure this can't happen and turn it into assert. + futex_unlock(¬ification_futex); + return; + } + + notification_t *notification = + hash_table_get_inst(link, notification_t, htlink); + + notification->count++; + notification->calldata = *call; + + if (link_in_use(¬ification->qlink)) { + /* Notification already queued. */ + futex_unlock(¬ification_futex); + return; + } + + list_append(¬ification->qlink, ¬ification_queue); + futex_unlock(¬ification_futex); + + fibril_semaphore_up(¬ification_semaphore); +} + +/** + * Creates a new notification structure and inserts it into the hash table. + * + * @param handler Function to call when notification is received. + * @param arg Argument for the handler function. + * @return The newly created notification structure. + */ +static notification_t *notification_create(async_notification_handler_t handler, void *arg) +{ + notification_t *notification = calloc(1, sizeof(notification_t)); + if (!notification) + return NULL; + + notification->handler = handler; + notification->arg = arg; + + fid_t fib = 0; + + futex_lock(¬ification_futex); + + if (notification_avail == 0) { + /* Attempt to create the first handler fibril. */ + fib = fibril_create(notification_fibril_func, NULL); + if (fib == 0) { + futex_unlock(¬ification_futex); + free(notification); + return NULL; + } + } + + sysarg_t imethod = notification_avail; + notification_avail++; + + notification->imethod = imethod; + hash_table_insert(¬ification_hash_table, ¬ification->htlink); + + futex_unlock(¬ification_futex); + + if (imethod == 0) { + assert(fib); + fibril_add_ready(fib); + } + + return notification; } @@ -707,24 +844,11 @@ void *data, const irq_code_t *ucode, cap_irq_handle_t *handle) { - notification_t *notification = - (notification_t *) malloc(sizeof(notification_t)); + notification_t *notification = notification_create(handler, data); if (!notification) return ENOMEM; - futex_down(&async_futex); - - sysarg_t imethod = notification_avail; - notification_avail++; - - notification->imethod = imethod; - notification->handler = handler; - notification->data = data; - - hash_table_insert(¬ification_hash_table, ¬ification->link); - - futex_up(&async_futex); - cap_irq_handle_t ihandle; - errno_t rc = ipc_irq_subscribe(inr, imethod, ucode, &ihandle); + errno_t rc = ipc_irq_subscribe(inr, notification->imethod, ucode, + &ihandle); if (rc == EOK && handle != NULL) { *handle = ihandle; @@ -760,23 +884,9 @@ async_notification_handler_t handler, void *data) { - notification_t *notification = - (notification_t *) malloc(sizeof(notification_t)); + notification_t *notification = notification_create(handler, data); if (!notification) return ENOMEM; - futex_down(&async_futex); - - sysarg_t imethod = notification_avail; - notification_avail++; - - notification->imethod = imethod; - notification->handler = handler; - notification->data = data; - - hash_table_insert(¬ification_hash_table, ¬ification->link); - - futex_up(&async_futex); - - return ipc_event_subscribe(evno, imethod); + return ipc_event_subscribe(evno, notification->imethod); } @@ -793,23 +903,9 @@ async_notification_handler_t handler, void *data) { - notification_t *notification = - (notification_t *) malloc(sizeof(notification_t)); + notification_t *notification = notification_create(handler, data); if (!notification) return ENOMEM; - futex_down(&async_futex); - - sysarg_t imethod = notification_avail; - notification_avail++; - - notification->imethod = imethod; - notification->handler = handler; - notification->data = data; - - hash_table_insert(¬ification_hash_table, ¬ification->link); - - futex_up(&async_futex); - - return ipc_event_task_subscribe(evno, imethod); + return ipc_event_task_subscribe(evno, notification->imethod); } @@ -973,25 +1069,5 @@ /* Kernel notification */ if ((chandle == CAP_NIL) && (call->flags & IPC_CALL_NOTIF)) { - fibril_t *fibril = (fibril_t *) __tcb_get()->fibril_data; - unsigned oldsw = fibril->switches; - - process_notification(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); - } - + queue_notification(call); return; } Index: uspace/lib/c/generic/fibril.c =================================================================== --- uspace/lib/c/generic/fibril.c (revision a55d76b113225868d13a698012a31aae993d243d) +++ uspace/lib/c/generic/fibril.c (revision c407b987d06b7c0ba17e0044f01f5989c5eaeb4b) @@ -116,6 +116,4 @@ fibril->waits_for = NULL; - fibril->switches = 0; - /* * We are called before __tcb_set(), so we need to use @@ -206,5 +204,4 @@ break; case FIBRIL_TO_MANAGER: - srcf->switches++; /* * Don't put the current fibril into any list, it should Index: uspace/lib/c/include/async.h =================================================================== --- uspace/lib/c/include/async.h (revision a55d76b113225868d13a698012a31aae993d243d) +++ uspace/lib/c/include/async.h (revision c407b987d06b7c0ba17e0044f01f5989c5eaeb4b) @@ -492,4 +492,6 @@ sysarg_t, sysarg_t, sysarg_t); +errno_t async_spawn_notification_handler(void); + #endif Index: uspace/lib/c/include/fibril.h =================================================================== --- uspace/lib/c/include/fibril.h (revision a55d76b113225868d13a698012a31aae993d243d) +++ uspace/lib/c/include/fibril.h (revision c407b987d06b7c0ba17e0044f01f5989c5eaeb4b) @@ -72,6 +72,4 @@ fibril_owner_info_t *waits_for; - - unsigned int switches; } fibril_t;