Initial integration and reimplementation of user-space interface for keyed streams

[palacios.git] / linux_module / iface-stream.c

diff --git a/linux_module/iface-stream.c b/linux_module/iface-stream.c
index c487eb4..36582d9 100644 (file)
--- a/linux_module/iface-stream.c
+++ b/linux_module/iface-stream.c
@@ -1,21 +1,29 @@
-
-/* 
- * VM specific Controls
- * (c) Lei Xia, 2010
+/*
+ * Stream Implementation
+ * (c) Lei Xia  2010
  */
+ 
+ 
 #include <linux/errno.h>
 #include <linux/percpu.h>
 #include <linux/sched.h>
 #include <linux/uaccess.h>
+#include <linux/fs.h>
+#include <linux/poll.h>
+#include <linux/anon_inodes.h>
+#include <linux/file.h>
 
 
 #include <interfaces/vmm_stream.h>
 #include "linux-exts.h"
 #include "util-ringbuffer.h"
 #include "vm.h"
+#include "iface-stream.h"
+
 
+// This is going to need to be a lot bigger...
 #define STREAM_BUF_SIZE 1024
-#define STREAM_NAME_LEN 128
+
 
 
 
@@ -25,6 +33,8 @@ struct stream_buffer {
     char name[STREAM_NAME_LEN];
     struct ringbuf * buf;
 
+    int connected;
+
     wait_queue_head_t intr_queue;
     spinlock_t lock;
 
@@ -38,37 +48,12 @@ struct vm_stream_state {
     struct list_head open_streams;
 };
 
-static int stream_enqueue(struct stream_buffer * stream, char * buf, int len) {
-    int bytes = 0;
-
-    bytes = ringbuf_write(stream->buf, buf, len);
-
-    return bytes;
-}
-
-
-static int stream_dequeue(struct stream_buffer * stream, char * buf, int len) {
-    int bytes = 0;
-
-    bytes = ringbuf_read(stream->buf, buf, len);
-
-    return bytes;
-}
-
-static int stream_datalen(struct stream_buffer * stream){
-    return ringbuf_data_len(stream->buf);
-}
-
-
-
-
 
 static struct stream_buffer * find_stream_by_name(struct v3_guest * guest, const char * name) {
     struct stream_buffer * stream = NULL;
     struct list_head * stream_list = NULL;
     struct vm_stream_state * vm_state = NULL;
 
-
     if (guest == NULL) {
        stream_list = &global_streams;
     } else {
@@ -92,6 +77,25 @@ static struct stream_buffer * find_stream_by_name(struct v3_guest * guest, const
 }
 
 
+
+static ssize_t stream_read(struct file * filp, char __user * buf, size_t size, loff_t * offset) {
+    struct stream_buffer * stream = filp->private_data;
+    
+    wait_event_interruptible(stream->intr_queue, (ringbuf_data_len(stream->buf) != 0));
+
+    return ringbuf_read(stream->buf, buf, size);
+}
+
+
+
+static struct file_operations stream_fops = {
+    .read = stream_read,
+    //    .release = stream_close,
+    //  .poll = stream_poll,
+};
+
+
+
 static void * palacios_stream_open(const char * name, void * private_data) {
     struct v3_guest * guest = (struct v3_guest *)private_data;
     struct stream_buffer * stream = NULL;
@@ -135,7 +139,7 @@ static int palacios_stream_write(void * stream_ptr, char * buf, int len) {
     struct stream_buffer * stream = (struct stream_buffer *)stream_ptr;
     int ret = 0;
 
-    ret = stream_enqueue(stream, buf, len);
+    ret = ringbuf_write(stream->buf, buf, len);
 
     if (ret > 0) {
        wake_up_interruptible(&(stream->intr_queue));
@@ -172,25 +176,61 @@ static int stream_init( void ) {
 static int stream_deinit( void ) {
     if (!list_empty(&(global_streams))) {
        printk("Error removing module with open streams\n");
+       printk("TODO: free old streams... \n");
     }
 
     return 0;
 }
 
+
+
+
+
 static int stream_connect(struct v3_guest * guest, unsigned int cmd, unsigned long arg, void * priv_data) {
     void __user * argp = (void __user *)arg;
-    char path_name[STREAM_NAME_LEN];
+    struct stream_buffer * stream = NULL;
+    int stream_fd = 0;
+    char name[STREAM_NAME_LEN];
+    unsigned long flags = 0;
+    int ret = -1;
+    
     
-    if (copy_from_user(path_name, argp, STREAM_NAME_LEN)) {
+    if (copy_from_user(name, argp, STREAM_NAME_LEN)) {
        printk("%s(%d): copy from user error...\n", __FILE__, __LINE__);
        return -EFAULT;
     }
+
+    stream = find_stream_by_name(guest, name);
+
+    if (stream == NULL) {
+       printk("Could not find stream (%s)\n", name);
+       return -EFAULT;
+    }
+
+    spin_lock_irqsave(&(stream->lock), flags);
+    if (stream->connected == 0) {
+       stream->connected = 1;
+       ret = 1;
+    }
+    spin_unlock_irqrestore(&(stream->lock), flags);
+
+
+    if (ret == -1) {
+       printk("Stream (%s) already connected\n", name);
+       return -EFAULT;
+    }
+
     
+    stream_fd = anon_inode_getfd("v3-stream", &stream_fops, stream, 0);
 
+    if (stream_fd < 0) {
+       printk("Error creating stream inode for (%s)\n", name);
+       return stream_fd;
+    }
 
-    printk("ERROR: Opening Streams is currently not implemented...\n");
+    printk("Stream (%s) connected\n", name);
 
-    return -EFAULT;
+    return stream_fd;
 }