X-Git-Url: http://v3vee.org/palacios/gitweb/gitweb.cgi?a=blobdiff_plain;f=linux_module%2Fiface-stream.c;h=207f5f02c8dec12cb984fe52395f79ff7fded7ca;hb=d22c11cec4e8c3390bfe6bf16ed07f5d073f0d4a;hp=62631fe8de8cb9f70cc3bfa88810771dc3a812c1;hpb=07aa8f3c18a33af0961e7546980a63ab5f6fba4f;p=palacios.git

diff --git a/linux_module/iface-stream.c b/linux_module/iface-stream.c
index 62631fe..207f5f0 100644
--- a/linux_module/iface-stream.c
+++ b/linux_module/iface-stream.c
@@ -14,6 +14,7 @@
 #include <linux/file.h>
 
 
+#define sint64_t int64_t
 #include <interfaces/vmm_stream.h>
 #include "linux-exts.h"
 #include "util-ringbuffer.h"
@@ -38,7 +39,8 @@ struct stream_state {
 
     int connected;
 
-    wait_queue_head_t intr_queue;
+    wait_queue_head_t user_poll_queue;
+
     spinlock_t lock;
 
     struct v3_guest * guest;
@@ -85,74 +87,73 @@ static struct stream_state * find_stream_by_name(struct v3_guest * guest, const
 }
 
 
-
-#define TMP_BUF_LEN 128
-
+// host->user nonblocking data flow 
 static ssize_t stream_read(struct file * filp, char __user * buf, size_t size, loff_t * offset) {
-    struct stream_state * stream = filp->private_data;
-    ssize_t bytes_read = 0;
-    ssize_t bytes_left = size;
-    unsigned long flags;
-    char tmp_buf[TMP_BUF_LEN];
-    ssize_t total_bytes_left = 0;
-
-    // memset(tmp_buf, 0, TMP_BUF_LEN);
-
-    while (bytes_left > 0) {
-	int tmp_len = (TMP_BUF_LEN > bytes_left) ? bytes_left : TMP_BUF_LEN;
-	int tmp_read = 0;
-
-	spin_lock_irqsave(&(stream->lock), flags);
-	tmp_read = ringbuf_read(stream->out_ring, tmp_buf, tmp_len);
-	spin_unlock_irqrestore(&(stream->lock), flags);
-
-	if (tmp_read == 0) {
-	    // If userspace reads more than we have
-	    break;
-	}
-
-	if (copy_to_user(buf + bytes_read, tmp_buf, tmp_read)) {
-	    ERROR("Read Fault\n");
-	    return -EFAULT;
-	}
-	
-	bytes_left -= tmp_read;
-	bytes_read += tmp_read;
-    }
+  struct stream_state * stream = filp->private_data;
+  ssize_t bytes_read = 0;
+  unsigned long flags;
+  char *kern_buf;
+  
+  
+  kern_buf = palacios_alloc(size);
+
+  if (!kern_buf) {
+    ERROR("Cannot allocate space for buffer\n");
+    return -1;
+  }
+  
+  palacios_spinlock_lock_irqsave(&(stream->lock), flags);
+  bytes_read = ringbuf_read(stream->out_ring,kern_buf,size);
+  palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
     
-
-    spin_lock_irqsave(&(stream->lock), flags); 
-    total_bytes_left = ringbuf_data_len(stream->out_ring);
-    spin_unlock_irqrestore(&(stream->lock), flags);
-
-    if (total_bytes_left > 0) {
-	wake_up_interruptible(&(stream->intr_queue));
+  if (bytes_read>0) { 
+    if (copy_to_user(buf, kern_buf, bytes_read)) {
+      ERROR("Read Fault\n");
+      palacios_free(kern_buf);
+      return -EFAULT;
+    } else {
+      palacios_free(kern_buf);
+      return bytes_read;
     }
-
-    return bytes_read;
+  } else if (bytes_read==0) { 
+    // out of space
+    palacios_free(kern_buf);
+    return -EWOULDBLOCK;
+  } else { // bytes_read<0
+    ERROR("Read failed\n");
+    palacios_free(kern_buf);
+    return -EFAULT;
+  }
 }
+  
 
 static unsigned int 
 stream_poll(struct file * filp, struct poll_table_struct * poll_tb) {
     struct stream_state * stream = filp->private_data;
-    unsigned int mask = POLLIN | POLLRDNORM;
     unsigned long flags;
     int data_avail = 0;
 
-    poll_wait(filp, &(stream->intr_queue), poll_tb);
+    if (!stream) { 
+      return POLLERR;
+    }
 
-    spin_lock_irqsave(&(stream->lock), flags);
+    poll_wait(filp, &(stream->user_poll_queue), poll_tb);
+
+    palacios_spinlock_lock_irqsave(&(stream->lock), flags);
     data_avail = ringbuf_data_len(stream->out_ring);
-    spin_unlock_irqrestore(&(stream->lock), flags);
+    palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 
     if (data_avail > 0) {
-	return mask;
+	return POLLIN | POLLRDNORM;
     }
 
     return 0;
 
 }
 
+//
+// Non-blocking user->Host->VM data flow
+//
 static ssize_t stream_write(struct file * filp, const char __user * buf, size_t size, loff_t * offset) {
     struct stream_state * stream = filp->private_data;
     char * kern_buf = NULL;
@@ -173,8 +174,21 @@ static ssize_t stream_write(struct file * filp, const char __user * buf, size_t
     
     bytes_written = stream->v3_stream->input(stream->v3_stream, kern_buf, size);
 
-    palacios_free(kern_buf);
+    // could end up with zero here
+    if (bytes_written<0) { 
+      ERROR("Error on writing to stream\n");
+      palacios_free(kern_buf);
+      return -EFAULT;
+    }
+
+    if (bytes_written==0) { 
+      // This is somewhat bogus, since 
+      // the FD is treated as non-blocking regardless
+      palacios_free(kern_buf);
+      return -EWOULDBLOCK;
+    } 
 
+    palacios_free(kern_buf);
     return bytes_written;
 }
 
@@ -183,9 +197,9 @@ static int stream_release(struct inode * i, struct file * filp) {
     struct stream_state * stream = filp->private_data;
     unsigned long flags;
     
-    spin_lock_irqsave(&(stream->lock), flags);
+    palacios_spinlock_lock_irqsave(&(stream->lock), flags);
     stream->connected = 0;
-    spin_unlock_irqrestore(&(stream->lock), flags);
+    palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 
     
     return 0;
@@ -232,10 +246,11 @@ static void * palacios_stream_open(struct v3_stream * v3_stream, const char * na
     stream->guest = guest;
     stream->connected = 0;
 
-    strncpy(stream->name, name, STREAM_NAME_LEN - 1);
+    strncpy(stream->name, name, STREAM_NAME_LEN);
+    stream->name[STREAM_NAME_LEN-1] = 0;
 
-    init_waitqueue_head(&(stream->intr_queue));
-    spin_lock_init(&(stream->lock));
+    init_waitqueue_head(&(stream->user_poll_queue));
+    palacios_spinlock_init(&(stream->lock));
 
     if (guest == NULL) {
 	list_add(&(stream->stream_node), &(global_streams));
@@ -247,31 +262,29 @@ static void * palacios_stream_open(struct v3_stream * v3_stream, const char * na
 }
 
 
-static uint64_t palacios_stream_output(struct v3_stream * v3_stream, char * buf, int len) {
+//
+// Non-blocking VM->host data flow
+//
+static sint64_t palacios_stream_output(struct v3_stream * v3_stream, uint8_t * buf, sint64_t len) {
     struct stream_state * stream = (struct stream_state *)v3_stream->host_stream_data;
-    int bytes_written = 0;
+    sint64_t bytes_written = 0;
     unsigned long flags;
     
 
-    if (stream->connected == 0) {
-	return 0;
-    }
-
-    while (bytes_written < len) {
-	spin_lock_irqsave(&(stream->lock), flags);
-	bytes_written += ringbuf_write(stream->out_ring, buf + bytes_written, len - bytes_written);
-	spin_unlock_irqrestore(&(stream->lock), flags);
 
-	wake_up_interruptible(&(stream->intr_queue));
+    palacios_spinlock_lock_irqsave(&(stream->lock), flags);
+    bytes_written=ringbuf_write(stream->out_ring, buf, len);
+    palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 
-	if (bytes_written < len) {
-	    // not enough space in ringbuffer, activate user space to drain it
-	    schedule();
-	}
-    }
+    if (bytes_written<0) { 
+      // we ended in an error, so push back to VM
+      return bytes_written;
+    } else {
+      // normal situation, tell it how much we handled
+      wake_up_interruptible(&(stream->user_poll_queue));
+      return bytes_written;
+    } 
 
-    
-    return bytes_written;
 }
 
 
@@ -280,6 +293,7 @@ static void palacios_stream_close(struct v3_stream * v3_stream) {
 
     free_ringbuf(stream->out_ring);
     list_del(&(stream->stream_node));
+    palacios_spinlock_deinit(&(stream->lock));
     palacios_free(stream);
 
 }
@@ -300,9 +314,13 @@ static int stream_init( void ) {
 
 
 static int stream_deinit( void ) {
-    if (!list_empty(&(global_streams))) {
-	ERROR("Error removing module with open streams\n");
-	DEBUG("TODO: free old streams... \n");
+    struct stream_state * stream = NULL;
+    struct stream_state * tmp = NULL;
+
+    list_for_each_entry_safe(stream, tmp, &(global_streams), stream_node) {
+        free_ringbuf(stream->out_ring);
+        list_del(&(stream->stream_node));
+        palacios_free(stream);
     }
 
     return 0;
@@ -333,12 +351,12 @@ static int stream_connect(struct v3_guest * guest, unsigned int cmd, unsigned lo
 	return -EFAULT;
     }
 
-    spin_lock_irqsave(&(stream->lock), flags);
+    palacios_spinlock_lock_irqsave(&(stream->lock), flags);
     if (stream->connected == 0) {
 	stream->connected = 1;
 	ret = 1;
     }
-    spin_unlock_irqrestore(&(stream->lock), flags);
+    palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 
 
     if (ret == -1) {
@@ -379,11 +397,22 @@ static int guest_stream_init(struct v3_guest * guest, void ** vm_data) {
 
 static int guest_stream_deinit(struct v3_guest * guest, void * vm_data) {
     struct vm_global_streams * state = vm_data;
-    if (!list_empty(&(state->open_streams))) {
-	ERROR("Error shutting down VM with open streams\n");
-	return -1;
+
+    struct stream_state * stream = NULL;
+    struct stream_state * tmp = NULL;
+
+
+    remove_guest_ctrl(guest, V3_VM_STREAM_CONNECT);
+
+    list_for_each_entry_safe(stream, tmp, &(global_streams), stream_node) {
+        free_ringbuf(stream->out_ring);
+        list_del(&(stream->stream_node));
+        palacios_free(stream);
     }
+    
+    palacios_free(state);
 
+    
     return 0;
 }