ioq: add probes tracing and documentation

ioq tracing will allow users to notice when devices are saturated
(from a cmogstored POV) and increase aio_threads if necessary.

3 files changed, 107 insertions, 10 deletions

diff --git a/ioq.c b/ioq.c
index e88d7f3..4f62cc5 100644
--- a/ioq.c
+++ b/ioq.c
@@ -3,13 +3,41 @@
  * License: GPLv3 or later (see COPYING for details)
  */
 #include "cmogstored.h"
+#include "trace.h"
 /*
  * This is a semaphore-like API with explicit queueing and activation,
  * so contended scheduling/wakeups happen via epoll/kqueue and there
- * is never blocking of threads (other than the mutex)
+ * is never blocking of threads (other than the mutex which only protects
+ * small memory-only operations)
  *
  * The main operations are mog_ioq_ready and mog_ioq_next
  *
+ * Flow:
+ *
+ *     mog_ioq_ready ---> true --> normal dispatch --> mog_ioq_next
+ *          |    ^                 (mog_ioq_unblock)         |
+ *          |    |                                           |
+ *          |     `-------<--------\                         |
+ *        false                    |                         |
+ *          |                      |                         V
+ *          |                      |                         |
+ *          V                      |                         |
+ *   SIMPLEQ_INSERT_TAIL(push)     ^                         |
+ *          ||                     |                         V
+ *          VV                     |                         /
+ *           \\                    |                        /
+ *            \\                   |                       /
+ *             \\                  |                      V
+ *              `===(wait for)==========>===> SIMPLEQ_{FIRST,REMOVE_HEAD}(pop)
+ *                                 |                   |
+ *                                 |                   V
+ *                                 |                   |
+ *                                 ^       add to kqueue/epoll ready list
+ *                                 |                   |
+ *                                 |                   V
+ *                                 |                   /
+ *                                 `---------<---------'
+ *
  * mog_ioq_next is automatically called when releases a regular file.
  */
 __thread struct mog_ioq *mog_ioq_current;
@@ -53,9 +81,9 @@ static inline void ioq_set_contended(struct mog_ioq *ioq)
  * Adding the mfd to an epoll/kqueue watch list in the same thread/context
  * where this function returns true is a guaranteed bug.
  *
- * client_mfd is the client socket, not the open (regular) file
+ * mfd is the client socket, not the open (regular) file
  */
-bool mog_ioq_ready(struct mog_ioq *ioq, struct mog_fd *client_mfd)
+bool mog_ioq_ready(struct mog_ioq *ioq, struct mog_fd *mfd)
 {
         bool good;
 
@@ -73,8 +101,8 @@ bool mog_ioq_ready(struct mog_ioq *ioq, struct mog_fd *client_mfd)
 
                 mog_ioq_current = ioq;
         } else {
-                client_mfd->ioq_blocked = 1;
-                SIMPLEQ_INSERT_TAIL(&ioq->ioq_head, client_mfd, ioqent);
+                mfd->ioq_blocked = 1;
+                SIMPLEQ_INSERT_TAIL(&ioq->ioq_head, mfd, ioqent);
                 ioq_set_contended(ioq);
         }
 
@@ -89,7 +117,7 @@ bool mog_ioq_ready(struct mog_ioq *ioq, struct mog_fd *client_mfd)
  */
 void mog_ioq_next(struct mog_ioq *check_ioq)
 {
-        struct mog_fd *client_mfd = NULL;
+        struct mog_fd *mfd = NULL;
 
         if (mog_ioq_current == NULL)
                 return;
@@ -102,8 +130,8 @@ void mog_ioq_next(struct mog_ioq *check_ioq)
         mog_ioq_current->cur++;
         if (mog_ioq_current->cur <= mog_ioq_current->max) {
                 /* wake up any waiters */
-                client_mfd = SIMPLEQ_FIRST(&mog_ioq_current->ioq_head);
-                if (client_mfd) {
+                mfd = SIMPLEQ_FIRST(&mog_ioq_current->ioq_head);
+                if (mfd) {
                         SIMPLEQ_REMOVE_HEAD(&mog_ioq_current->ioq_head, ioqent);
 
                         /* if there's another head, we're still contended */
@@ -118,8 +146,10 @@ void mog_ioq_next(struct mog_ioq *check_ioq)
         CHECK(int, 0, pthread_mutex_unlock(&mog_ioq_current->mtx));
 
         /* wake up the next sleeper on this queue */
-        if (client_mfd)
-                mog_activeq_push(mog_ioq_current->svc->queue, client_mfd);
+        if (mfd) {
+                TRACE(CMOGSTORED_IOQ_RESCHEDULE(mfd->fd));
+                mog_activeq_push(mog_ioq_current->svc->queue, mfd);
+        }
         /*
          * We may not touch or use client_mfd here anymore.  Another
          * thread may already have it.  In the worst case, it's been
@@ -173,6 +203,12 @@ void mog_ioq_destroy(struct mog_ioq *ioq)
         CHECK(int, 0, pthread_mutex_destroy(&ioq->mtx));
 }
 
+/*
+ * If this returns true, the caller must continue processing a request
+ * without checking other state associated with the mfd.
+ * If this returns false (the common case), the caller continues as
+ * usual.
+ */
 bool mog_ioq_unblock(struct mog_fd *mfd)
 {
         if (mfd->ioq_blocked == 0)
diff --git a/probes.d b/probes.d
index 6a0ced8..20a05a6 100644
--- a/probes.d
+++ b/probes.d
@@ -17,6 +17,10 @@ provider cmogstored {
         /* DWARF: mog_http_get_open */
         probe http_req_end();
 
+        /* DWARF: mog_ioq_ready(.return) */
+        probe ioq_reschedule(int fd);
+        /* DWARF: mog_ioq_unblock(.return) */
+
         probe mgmt_accepted(int fd, const char *host, const char *port,
                                 const char *listen_addr);
         probe mgmt_rderr(struct mog_fd *mfd, size_t buf_len, int err);
diff --git a/tapset/http_request.stp b/tapset/http_request.stp
index 5df785c..ceac883 100644
--- a/tapset/http_request.stp
+++ b/tapset/http_request.stp
@@ -2,6 +2,7 @@
 global cmogstored_http_req_begin;
 global cmogstored_http_pipelined;
 global cmogstored_http_addr;
+global cmogstored_fd_by_tid;
 
 probe process("cmogstored").mark("http_accepted") {
         fd = $arg1;
@@ -13,12 +14,62 @@ probe process("cmogstored").mark("http_accepted") {
         cmogstored_http_addr[pid(),fd] = host_serv;
 }
 
+/*
+ * mog_ioq_ready..ioq_reschedule   = time waiting for others to finish IO
+ * ioq_reschedule..mog_ioq_unblock = time inside epoll ready list
+ */
+
+/* associate the fd with thread for the function return */
+probe process("cmogstored").function("mog_ioq_ready") {
+        cmogstored_fd_by_tid[tid()] = @cast($mfd, "struct mog_fd")->fd;
+}
+
+/*
+ * if mog_ioq_ready returns false, we are blocked waiting on I/O and
+ * will hit ioq_reschedule
+ */
+probe process("cmogstored").function("mog_ioq_ready").return {
+        if (!$return)
+                printf("% 6d % 6d ioq blocked\n",
+                        pid(), cmogstored_fd_by_tid[tid()]);
+}
+
+/*
+ * we hit ioq_reschedule when we are ready to enter the epoll ready-list
+ * we will hit mog_ioq_unblock eventually
+ */
+probe process("cmogstored").mark("ioq_reschedule") {
+        fd = @cast($mfd, "struct mog_fd")->fd;
+        printf("% 6d % 6d ioq reschedule_enter\n", pid(), fd);
+}
+
+/* associate the fd with thread for the function return */
+probe process("cmogstored").function("mog_ioq_unblock") {
+        cmogstored_fd_by_tid[tid()] = @cast($mfd, "struct mog_fd")->fd;
+}
+
+/*
+ * if mog_ioq_unblock returns true, it means the fd was previously
+ * in ioq_reschedule
+ */
+probe process("cmogstored").function("mog_ioq_unblock").return {
+        if ($return)
+                printf("% 6d % 6d ioq reschedule_done\n",
+                        pid(), cmogstored_fd_by_tid[tid()]);
+}
+
+/*
+ * HTTP client disconnected
+ */
 probe process("cmogstored").function("http_close") {
         fd = @cast($mfd, "struct mog_fd")->fd;
         printf("% 6d % 6d closing\n", pid(), fd);
         delete cmogstored_http_addr[pid(),fd];
 }
 
+/*
+ * We start reading/buffering the HTTP request here
+ */
 probe process("cmogstored").mark("http_req_begin") {
         fd = @cast($mfd, "struct mog_fd")->fd;
         is_pipelined = $arg1;
@@ -26,6 +77,9 @@ probe process("cmogstored").mark("http_req_begin") {
         cmogstored_http_pipelined[pid(),fd] = is_pipelined;
 }
 
+/*
+ * we start processing the HTTP request (opening/stat-ing files)
+ */
 probe process("cmogstored").function("http_process_client") {
         fd = @cast($mfd, "struct mog_fd")->fd;
         starttime = cmogstored_http_req_begin[pid(),fd];
@@ -36,6 +90,9 @@ probe process("cmogstored").function("http_process_client") {
                 pid(), fd, diff, is_pipelined ? "true" : "false");
 }
 
+/*
+ * We blocked on writing (HTTP headers) to the client socket
+ */
 probe process("cmogstored").mark("write_buffered") {
         printf("% 6d % 6d blocked with %lu bytes to write\n",
                 pid(), $fd, $len);