Move sched_trace ring buffer to kfifo implementation

Use kfifo [kernel/kfifo.c] to implement the ring buffer used
for sched_trace (TRACE() and TRACE_TASK() macros)

This patch also includes some reorganization of sched_trace.c code
and some fixes:

- 1c39c59b3 Fix GFP_KERNEL in rb_alloc_buf with interrupt disabled.
- 193ad2688 Let TRACE() log buffer size and comment converge.
- 6195e2ae8 re-enable capturing of printk() messages in TRACE() logs.

1 files changed, 171 insertions, 272 deletions

diff --git a/litmus/sched_trace.c b/litmus/sched_trace.c
index 5befc1e21b1f..4c1ea7eab635 100644
--- a/litmus/sched_trace.c
+++ b/litmus/sched_trace.c
@@ -1,147 +1,117 @@
 /*
  * sched_trace.c -- record scheduling events to a byte stream.
- *
- * TODO: Move ring buffer to a lockfree implementation.
  */
-
 #include <linux/spinlock.h>
+#include <linux/semaphore.h>
+
 #include <linux/fs.h>
 #include <linux/cdev.h>
-#include <linux/semaphore.h>
 #include <asm/uaccess.h>
 #include <linux/module.h>
 #include <linux/sysrq.h>
 
+#include <linux/kfifo.h>
+
 #include <litmus/sched_trace.h>
 #include <litmus/litmus.h>
 
-typedef struct {
-        /*      guard read and write pointers                   */
-        spinlock_t      lock;
-        /*      guard against concurrent freeing of buffer      */
-        rwlock_t        del_lock;
+/* Allocate a buffer of about 32k per CPU */
+#define LITMUS_TRACE_BUF_PAGES 8
+#define LITMUS_TRACE_BUF_SIZE (PAGE_SIZE * LITMUS_TRACE_BUF_PAGES * NR_CPUS)
 
-        /*      memory allocated for ring buffer                */
-        unsigned long   order;
-        char*           buf;
-        char*           end;
+/* Max length of one read from the buffer */
+#define MAX_READ_LEN (64 * 1024)
 
-        /*      Read/write pointer. May not cross.
-         *      They point to the position of next write and
-         *      last read.
-         */
-        char*           writep;
-        char*           readp;
+/* Max length for one write --- from kernel --- to the buffer */
+#define MSG_SIZE 255
+
+/*
+ * Major number for the tracing char device.
+ * the major numbes are from the unassigned/local use block
+ */
+#define LOG_MAJOR       251
+
+/* Inner ring buffer structure */
+typedef struct {
+        rwlock_t        del_lock;
 
+        /* the buffer */
+        struct kfifo    *kfifo;
 } ring_buffer_t;
 
-#define EMPTY_RING_BUFFER {     \
-        .lock     = SPIN_LOCK_UNLOCKED,         \
-        .del_lock = RW_LOCK_UNLOCKED,           \
-        .buf      = NULL,                       \
-        .end      = NULL,                       \
-        .writep   = NULL,                       \
-        .readp    = NULL                        \
-}
+/* Main buffer structure */
+typedef struct {
+        ring_buffer_t           buf;
+        atomic_t                reader_cnt;
+        struct semaphore        reader_mutex;
+} trace_buffer_t;
+
 
+/*
+ * Inner buffer management functions
+ */
 void rb_init(ring_buffer_t* buf)
 {
-        *buf = (ring_buffer_t) EMPTY_RING_BUFFER;
+        rwlock_init(&buf->del_lock);
+        buf->kfifo = NULL;
 }
 
-int rb_alloc_buf(ring_buffer_t* buf, unsigned long order)
+int rb_alloc_buf(ring_buffer_t* buf, unsigned int size)
 {
         unsigned long flags;
-        int error = 0;
-        char *mem;
 
-        /* do memory allocation while not atomic */
-        mem = (char *) __get_free_pages(GFP_KERNEL, order);
-        if (!mem)
-                return -ENOMEM;
         write_lock_irqsave(&buf->del_lock, flags);
-        BUG_ON(buf->buf);
-        buf->buf = mem;
-        buf->end = buf->buf + PAGE_SIZE * (1 << order) - 1;
-        memset(buf->buf, 0xff, buf->end - buf->buf);
-        buf->order = order;
-        buf->writep = buf->buf + 1;
-        buf->readp  = buf->buf;
+
+        buf->kfifo = kfifo_alloc(size, GFP_ATOMIC, NULL);
+
         write_unlock_irqrestore(&buf->del_lock, flags);
-        return error;
+
+        if(IS_ERR(buf->kfifo)) {
+                printk(KERN_ERR "kfifo_alloc failed\n");
+                return PTR_ERR(buf->kfifo);
+        }
+
+        return 0;
 }
 
 int rb_free_buf(ring_buffer_t* buf)
 {
         unsigned long flags;
-        int error = 0;
+
         write_lock_irqsave(&buf->del_lock, flags);
-        BUG_ON(!buf->buf);
-        free_pages((unsigned long) buf->buf, buf->order);
-        buf->buf    = NULL;
-        buf->end    = NULL;
-        buf->writep = NULL;
-        buf->readp  = NULL;
+
+        BUG_ON(!buf->kfifo);
+        kfifo_free(buf->kfifo);
+
+        buf->kfifo = NULL;
+
         write_unlock_irqrestore(&buf->del_lock, flags);
-        return error;
+
+        return 0;
 }
 
-/* Assumption: concurrent writes are serialized externally
+/*
+ * Assumption: concurrent writes are serialized externally
  *
  * Will only succeed if there is enough space for all len bytes.
  */
 int rb_put(ring_buffer_t* buf, char* mem, size_t len)
 {
         unsigned long flags;
-        char* r , *w;
         int error = 0;
+
         read_lock_irqsave(&buf->del_lock, flags);
-        if (!buf->buf) {
+
+        if (!buf->kfifo) {
                 error = -ENODEV;
                 goto out;
         }
-        spin_lock(&buf->lock);
-        r = buf->readp;
-        w = buf->writep;
-        spin_unlock(&buf->lock);
-        if (r < w && buf->end - w >= len - 1) {
-                /* easy case: there is enough space in the buffer
-                 * to write it in one continous chunk*/
-                memcpy(w, mem, len);
-                w += len;
-                if (w > buf->end)
-                        /* special case: fit exactly into buffer
-                         * w is now buf->end + 1
-                         */
-                        w = buf->buf;
-        } else if (w < r && r - w >= len) { /* >= len because  may not cross */
-                /* we are constrained by the read pointer but we there
-                 * is enough space
-                 */
-                memcpy(w, mem, len);
-                w += len;
-        } else if (r <= w && buf->end - w < len - 1) {
-                /* the wrap around case: there may or may not be space */
-                if ((buf->end - w) + (r - buf->buf) >= len - 1) {
-                        /* copy chunk that fits at the end */
-                        memcpy(w, mem, buf->end - w + 1);
-                        mem += buf->end - w + 1;
-                        len -= (buf->end - w + 1);
-                        w = buf->buf;
-                        /* copy the rest */
-                        memcpy(w, mem, len);
-                        w += len;
-                }
-                else
-                        error = -ENOMEM;
-        } else {
+
+        if((__kfifo_put(buf->kfifo, mem, len)) < len) {
                 error = -ENOMEM;
+                goto out;
         }
-        if (!error) {
-                spin_lock(&buf->lock);
-                buf->writep = w;
-                spin_unlock(&buf->lock);
-        }
+
  out:
         read_unlock_irqrestore(&buf->del_lock, flags);
         return error;
@@ -151,143 +121,107 @@ int rb_put(ring_buffer_t* buf, char* mem, size_t len)
 int rb_get(ring_buffer_t* buf, char* mem, size_t len)
 {
         unsigned long flags;
-        char* r , *w;
         int error = 0;
+
         read_lock_irqsave(&buf->del_lock, flags);
-        if (!buf->buf) {
+        if (!buf->kfifo) {
                 error = -ENODEV;
                 goto out;
         }
-        spin_lock(&buf->lock);
-        r = buf->readp;
-        w = buf->writep;
-        spin_unlock(&buf->lock);
-
-        if (w <= r && buf->end - r >= len) {
-                /* easy case: there is enough data in the buffer
-                 * to get it in one  chunk*/
-                memcpy(mem, r + 1, len);
-                r += len;
-                error = len;
-
-        } else if (r + 1 < w && w - r - 1 >= len) {
-                /* we are constrained by the write pointer but
-                 * there is enough data
-                 */
-                memcpy(mem, r + 1, len);
-                r += len;
-                error = len;
-
-        } else if (r + 1 < w && w - r - 1 < len) {
-                /* we are constrained by the write pointer and there
-                 * there is not enough data
-                 */
-                memcpy(mem, r + 1, w - r - 1);
-                error = w - r - 1;
-                r    += w - r - 1;
-
-        } else if (w <= r && buf->end - r < len) {
-                /* the wrap around case: there may or may not be enough data
-                 * first let's get what is available
-                 */
-                memcpy(mem, r + 1, buf->end - r);
-                error += (buf->end - r);
-                mem   += (buf->end - r);
-                len   -= (buf->end - r);
-                r     += (buf->end - r);
-
-                if (w > buf->buf) {
-                        /* there is more to get */
-                        r = buf->buf - 1;
-                        if (w - r >= len) {
-                                /* plenty */
-                                memcpy(mem, r + 1, len);
-                                error += len;
-                                r     += len;
-                        } else {
-                                memcpy(mem, r + 1, w - r - 1);
-                                error += w - r - 1;
-                                r     += w - r - 1;
-                        }
-                }
-        } /* nothing available */
 
-        if (error > 0) {
-                spin_lock(&buf->lock);
-                buf->readp = r;
-                spin_unlock(&buf->lock);
-        }
+        error = __kfifo_get(buf->kfifo, (unsigned char*)mem, len);
+
  out:
         read_unlock_irqrestore(&buf->del_lock, flags);
         return error;
 }
 
+/*
+ * Device Driver management
+ */
+static spinlock_t log_buffer_lock = SPIN_LOCK_UNLOCKED;
+static trace_buffer_t log_buffer;
 
+static void init_log_buffer(void)
+{
+        rb_init(&log_buffer.buf);
+        atomic_set(&log_buffer.reader_cnt,0);
+        init_MUTEX(&log_buffer.reader_mutex);
+}
 
-/******************************************************************************/
-/*                        DEVICE FILE DRIVER                                  */
-/******************************************************************************/
-
-
+static DEFINE_PER_CPU(char[MSG_SIZE], fmt_buffer);
 
-/* Allocate a buffer of about 1 MB per CPU.
+/*
+ * sched_trace_log_message - Write to the trace buffer (log_buffer)
  *
+ * This is the only function accessing the log_buffer from inside the
+ * kernel for writing.
+ * Concurrent access to sched_trace_log_message must be serialized using
+ * log_buffer_lock
+ * The maximum length of a formatted message is 255
  */
-#define BUFFER_ORDER 8
-
-typedef struct {
-        ring_buffer_t           buf;
-        atomic_t                reader_cnt;
-        struct semaphore        reader_mutex;
-} trace_buffer_t;
-
-
-/* This does not initialize the semaphore!! */
+void sched_trace_log_message(const char* fmt, ...)
+{
+        unsigned long   flags;
+        va_list         args;
+        size_t          len;
+        char*           buf;
 
-#define EMPTY_TRACE_BUFFER \
-        { .buf = EMPTY_RING_BUFFER, .reader_cnt = ATOMIC_INIT(0)}
+        va_start(args, fmt);
+        local_irq_save(flags);
 
-static spinlock_t               log_buffer_lock = SPIN_LOCK_UNLOCKED;
-static trace_buffer_t           log_buffer = EMPTY_TRACE_BUFFER;
+        /* format message */
+        buf = __get_cpu_var(fmt_buffer);
+        len = vscnprintf(buf, MSG_SIZE, fmt, args);
 
-static void init_log_buffer(void)
-{
-        /* only initialize the mutex, the rest was initialized as part
-         * of the static initialization macro
+        spin_lock(&log_buffer_lock);
+        /* Don't copy the trailing null byte, we don't want null bytes
+         * in a text file.
          */
-        init_MUTEX(&log_buffer.reader_mutex);
+        rb_put(&log_buffer.buf, buf, len);
+        spin_unlock(&log_buffer_lock);
+
+        local_irq_restore(flags);
+        va_end(args);
 }
 
+/*
+ * log_read - Read the trace buffer
+ *
+ * This function is called as a file operation from userspace.
+ * Readers can sleep. Access is serialized through reader_mutex
+ */
 static ssize_t log_read(struct file *filp, char __user *to, size_t len,
                       loff_t *f_pos)
 {
-        /*      we ignore f_pos, this is strictly sequential */
+        /* we ignore f_pos, this is strictly sequential */
 
         ssize_t error = -EINVAL;
         char*   mem;
-        trace_buffer_t *buf = filp->private_data;
+        trace_buffer_t *tbuf = filp->private_data;
 
-        if (down_interruptible(&buf->reader_mutex)) {
+        if (down_interruptible(&tbuf->reader_mutex)) {
                 error = -ERESTARTSYS;
                 goto out;
         }
 
-        if (len > 64 * 1024)
-                len = 64 * 1024;
+        if (len > MAX_READ_LEN)
+                len = MAX_READ_LEN;
+
         mem = kmalloc(len, GFP_KERNEL);
         if (!mem) {
                 error = -ENOMEM;
                 goto out_unlock;
         }
 
-        error = rb_get(&buf->buf, mem, len);
+        error = rb_get(&tbuf->buf, mem, len);
         while (!error) {
                 set_current_state(TASK_INTERRUPTIBLE);
                 schedule_timeout(110);
                 if (signal_pending(current))
                         error = -ERESTARTSYS;
                 else
-                        error = rb_get(&buf->buf, mem, len);
+                        error = rb_get(&tbuf->buf, mem, len);
         }
 
         if (error > 0 && copy_to_user(to, mem, error))
@@ -295,89 +229,87 @@ static ssize_t log_read(struct file *filp, char __user *to, size_t len,
 
         kfree(mem);
  out_unlock:
-        up(&buf->reader_mutex);
+        up(&tbuf->reader_mutex);
  out:
         return error;
 }
 
-/* defined in kernel/printk.c */
+/*
+ * Enable redirection of printk() messages to the trace buffer.
+ * Defined in kernel/printk.c
+ */
 extern int trace_override;
 extern int trace_recurse;
 
-/* log_open - open the global log message ring buffer.
+/*
+ * log_open - open the global log message ring buffer.
  */
 static int log_open(struct inode *in, struct file *filp)
 {
-        int error               = -EINVAL;
-        trace_buffer_t* buf;
+        int error = -EINVAL;
+        trace_buffer_t* tbuf;
 
-        buf = &log_buffer;
+        tbuf = &log_buffer;
 
-        if (down_interruptible(&buf->reader_mutex)) {
+        if (down_interruptible(&tbuf->reader_mutex)) {
                 error = -ERESTARTSYS;
                 goto out;
         }
 
-        /*      first open must allocate buffers        */
-        if (atomic_inc_return(&buf->reader_cnt) == 1) {
-                if ((error = rb_alloc_buf(&buf->buf, BUFFER_ORDER)))
+        /* first open must allocate buffers */
+        if (atomic_inc_return(&tbuf->reader_cnt) == 1) {
+                if ((error = rb_alloc_buf(&tbuf->buf, LITMUS_TRACE_BUF_SIZE)))
                 {
-                        atomic_dec(&buf->reader_cnt);
+                        atomic_dec(&tbuf->reader_cnt);
                         goto out_unlock;
                 }
         }
 
         error = 0;
-        filp->private_data = buf;
-        printk(KERN_DEBUG "sched_trace buf: from 0x%p to 0x%p  length: %x\n",
-               buf->buf.buf, buf->buf.end,
-               (unsigned int) (buf->buf.end - buf->buf.buf));
+        filp->private_data = tbuf;
+
+        printk(KERN_DEBUG
+               "sched_trace kfifo at 0x%p with buffer starting at: 0x%p\n",
+               tbuf->buf.kfifo, &((tbuf->buf.kfifo)->buffer));
 
         /* override printk() */
         trace_override++;
 
  out_unlock:
-        up(&buf->reader_mutex);
+        up(&tbuf->reader_mutex);
  out:
         return error;
 }
 
 static int log_release(struct inode *in, struct file *filp)
 {
-        int error               = -EINVAL;
-        trace_buffer_t* buf     = filp->private_data;
+        int error = -EINVAL;
+        trace_buffer_t* tbuf = filp->private_data;
 
         BUG_ON(!filp->private_data);
 
-        if (down_interruptible(&buf->reader_mutex)) {
+        if (down_interruptible(&tbuf->reader_mutex)) {
                 error = -ERESTARTSYS;
                 goto out;
         }
 
-        /*      last release must deallocate buffers    */
-        if (atomic_dec_return(&buf->reader_cnt) == 0) {
-                error = rb_free_buf(&buf->buf);
+        /* last release must deallocate buffers */
+        if (atomic_dec_return(&tbuf->reader_cnt) == 0) {
+                error = rb_free_buf(&tbuf->buf);
         }
 
         /* release printk() overriding */
         trace_override--;
 
-        up(&buf->reader_mutex);
+        printk(KERN_DEBUG "sched_trace kfifo released\n");
+
+        up(&tbuf->reader_mutex);
  out:
         return error;
 }
 
-/******************************************************************************/
-/*                          Device Registration                               */
-/******************************************************************************/
-
-/* the major numbes are from the unassigned/local use block
- *
- * This should be converted to dynamic allocation at some point...
- */
-#define LOG_MAJOR       251
-
-/* log_fops  - The file operations for accessing the global LITMUS log message
+/*
+ * log_fops  - The file operations for accessing the global LITMUS log message
  *             buffer.
  *
  * Except for opening the device file it uses the same operations as trace_fops.
@@ -389,6 +321,9 @@ struct file_operations log_fops = {
         .read    = log_read,
 };
 
+/*
+ * Device registration
+ */
 static int __init register_buffer_dev(const char* name,
                                       struct file_operations* fops,
                                       int major, int count)
@@ -398,7 +333,7 @@ static int __init register_buffer_dev(const char* name,
         int error = 0;
 
         trace_dev = MKDEV(major, 0);
-        error     = register_chrdev_region(trace_dev, count, name);
+        error = register_chrdev_region(trace_dev, count, name);
         if (error)
         {
                 printk(KERN_WARNING "sched trace: "
@@ -424,6 +359,21 @@ static int __init register_buffer_dev(const char* name,
 }
 
 #ifdef CONFIG_MAGIC_SYSRQ
+void dump_trace_buffer(int max)
+{
+        char line[80];
+        int len;
+        int count = 0;
+
+        /* potentially, but very unlikely race... */
+        trace_recurse = 1;
+        while ((max == 0 || count++ < max) &&
+               (len = rb_get(&log_buffer.buf, line, sizeof(line) - 1)) > 0) {
+                line[len] = '\0';
+                printk("%s", line);
+        }
+        trace_recurse = 0;
+}
 
 static void sysrq_dump_trace_buffer(int key, struct tty_struct *tty)
 {
@@ -435,7 +385,6 @@ static struct sysrq_key_op sysrq_dump_trace_buffer_op = {
         .help_msg       = "dump-trace-buffer(Y)",
         .action_msg     = "writing content of TRACE() buffer",
 };
-
 #endif
 
 static int __init init_sched_trace(void)
@@ -458,53 +407,3 @@ static int __init init_sched_trace(void)
 
 module_init(init_sched_trace);
 
-#define MSG_SIZE 255
-static DEFINE_PER_CPU(char[MSG_SIZE], fmt_buffer);
-
-/* sched_trace_log_message - This is the only function that accesses the the
- *                           log buffer inside the kernel for writing.
- *                           Concurrent access to it is serialized via the
- *                           log_buffer_lock.
- *
- *                           The maximum length of a formatted message is 255.
- */
-void sched_trace_log_message(const char* fmt, ...)
-{
-        unsigned long   flags;
-        va_list         args;
-        size_t          len;
-        char*           buf;
-
-        va_start(args, fmt);
-        local_irq_save(flags);
-
-        /* format message */
-        buf = __get_cpu_var(fmt_buffer);
-        len = vscnprintf(buf, MSG_SIZE, fmt, args);
-
-        spin_lock(&log_buffer_lock);
-        /* Don't copy the trailing null byte, we don't want null bytes
-         * in a text file.
-         */
-        rb_put(&log_buffer.buf, buf, len);
-        spin_unlock(&log_buffer_lock);
-
-        local_irq_restore(flags);
-        va_end(args);
-}
-
-void dump_trace_buffer(int max)
-{
-        char line[80];
-        int len;
-        int count = 0;
-
-        /* potentially, but very unlikely race... */
-        trace_recurse = 1;
-        while ((max == 0 || count++ < max) &&
-               (len = rb_get(&log_buffer.buf, line, sizeof(line) - 1)) > 0) {
-                line[len] = '\0';
-                printk("%s", line);
-        }
-        trace_recurse = 0;
-}