Changeset c407b98 in mainline

Timestamp:

2018-06-13T17:14:51Z (6 years ago)

Author:

Jiří Zárevúcky <jiri.zarevucky@…>

Parents:

a55d76b1

git-author:

Jiří Zárevúcky <jiri.zarevucky@…> (2018-06-13 17:09:21)

git-committer:

Jiří Zárevúcky <jiri.zarevucky@…> (2018-06-13 17:14:51)

Message:

Add notification queue to the async framework.

Instead of running notification handlers in the same fibril that received it,
forcing us to allocate a new fibril when the handler blocks, we instead queue
the notifications, and allow an arbitrary but fixed number of dedicated fibrils
handle them.

Although a service can increase the number of handler fibrils to reduce latency,
there are now no dynamic allocations due to received notifications.
When the same notification is received again while the older instance is still
in queue, the new notification overwrites the old and increments a counter
of received notifications.

The counter is currently unused, because passing it to the handler would
require extensive changes to user code, but it should be straightforward
to make use of it should the need arise.

Location:

uspace/lib/c

Files:

• generic/async/server.c (modified) (10 diffs)
• generic/fibril.c (modified) (2 diffs)
• include/async.h (modified) (1 diff)
• include/fibril.h (modified) (1 diff)

uspace/lib/c/generic/async/server.c

@@ -181 +181 @@
 /* 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 +193 @@
         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 +242 @@
 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 +580 @@
 {
         notification_t *notification =
-            hash_table_get_inst(item, notification_t, link);
+            hash_table_get_inst(item, notification_t, htlink);
         return notification_key_hash(&notification->imethod);
 }
@@ -564 +588 @@
         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 +687 @@
 }
 
-/** 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(&notification_semaphore);
+
+                futex_lock(&notification_futex);
+
+                /*
+                 * The semaphore ensures that if we get this far,
+                 * the queue must be non-empty.
+                 */
+                assert(!list_empty(&notification_queue));
+
+                notification_t *notification = list_get_instance(
+                    list_first(&notification_queue), notification_t, qlink);
+                list_remove(&notification->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(&notification_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(&notification_futex);
 
         ht_link_t *link = hash_table_find(&notification_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(&notification_futex);
+                return;
+        }
+
+        notification_t *notification =
+            hash_table_get_inst(link, notification_t, htlink);
+
+        notification->count++;
+        notification->calldata = *call;
+
+        if (link_in_use(&notification->qlink)) {
+                /* Notification already queued. */
+                futex_unlock(&notification_futex);
+                return;
+        }
+
+        list_append(&notification->qlink, &notification_queue);
+        futex_unlock(&notification_futex);
+
+        fibril_semaphore_up(&notification_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(&notification_futex);
+
+        if (notification_avail == 0) {
+                /* Attempt to create the first handler fibril. */
+                fib = fibril_create(notification_fibril_func, NULL);
+                if (fib == 0) {
+                        futex_unlock(&notification_futex);
+                        free(notification);
+                        return NULL;
+                }
+        }
+
+        sysarg_t imethod = notification_avail;
+        notification_avail++;
+
+        notification->imethod = imethod;
+        hash_table_insert(&notification_hash_table, &notification->htlink);
+
+        futex_unlock(&notification_futex);
+
+        if (imethod == 0) {
+                assert(fib);
+                fibril_add_ready(fib);
+        }
+
+        return notification;
 }
 
@@ -707 +844 @@
     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(&notification_hash_table, &notification->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 +884 @@
     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(&notification_hash_table, &notification->link);
-
-        futex_up(&async_futex);
-
-        return ipc_event_subscribe(evno, imethod);
+        return ipc_event_subscribe(evno, notification->imethod);
 }
 
@@ -793 +903 @@
     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(&notification_hash_table, &notification->link);
-
-        futex_up(&async_futex);
-
-        return ipc_event_task_subscribe(evno, imethod);
+        return ipc_event_task_subscribe(evno, notification->imethod);
 }
 
@@ -973 +1069 @@
         /* 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;
         }

uspace/lib/c/generic/fibril.c

@@ -116 +116 @@
         fibril->waits_for = NULL;
 
-        fibril->switches = 0;
-
         /*
          * We are called before __tcb_set(), so we need to use
@@ -206 +204 @@
                 break;
         case FIBRIL_TO_MANAGER:
-                srcf->switches++;
                 /*
                  * Don't put the current fibril into any list, it should

uspace/lib/c/include/async.h

@@ -492 +492 @@
     sysarg_t, sysarg_t, sysarg_t);
 
+errno_t async_spawn_notification_handler(void);
+
 #endif
 

uspace/lib/c/include/fibril.h

@@ -72 +72 @@
 
         fibril_owner_info_t *waits_for;
-
-        unsigned int switches;
 } fibril_t;