linux_module/iface-stream.c

   1 /*
   2  * Stream Implementation
   3  * (c) Lei Xia  2010
   4  */
   5
   6
   7 #include <linux/errno.h>
   8 #include <linux/percpu.h>
   9 #include <linux/sched.h>
  10 #include <linux/uaccess.h>
  11 #include <linux/fs.h>
  12 #include <linux/poll.h>
  13 #include <linux/anon_inodes.h>
  14 #include <linux/file.h>
  15
  16
  17 #include <interfaces/vmm_stream.h>
  18 #include "linux-exts.h"
  19 #include "util-ringbuffer.h"
  20 #include "vm.h"
  21
  22
  23
  24 // This is probably overkill
  25 #define STREAM_RING_LEN 4096
  26 #define STREAM_NAME_LEN 128
  27
  28
  29
  30 static struct list_head global_streams;
  31
  32
  33
  34 struct stream_state {
  35     char name[STREAM_NAME_LEN];
  36
  37     struct ringbuf * out_ring;
  38
  39     int connected;
  40
  41     wait_queue_head_t intr_queue;
  42     spinlock_t lock;
  43
  44     struct v3_guest * guest;
  45     struct list_head stream_node;
  46
  47     struct v3_stream * v3_stream;
  48 };
  49
  50
  51 // Currently just the list of open streams
  52 struct vm_global_streams {
  53     struct list_head open_streams;
  54 };
  55
  56
  57
  58
  59
  60 static struct stream_state * find_stream_by_name(struct v3_guest * guest, const char * name) {
  61     struct stream_state * stream = NULL;
  62     struct list_head * stream_list = NULL;
  63     struct vm_global_streams * vm_state = NULL;
  64
  65     if (guest == NULL) {
  66         stream_list = &global_streams;
  67     } else {
  68         vm_state = get_vm_ext_data(guest, "STREAM_INTERFACE");
  69
  70         if (vm_state == NULL) {
  71             ERROR("ERROR: Could not locate vm stream state for extension STREAM_INTERFACE\n");
  72             return NULL;
  73         }
  74
  75         stream_list = &(vm_state->open_streams);
  76     }
  77
  78     list_for_each_entry(stream,  stream_list, stream_node) {
  79         if (strncmp(stream->name, name, STREAM_NAME_LEN) == 0) {
  80             return stream;
  81         }
  82     }
  83
  84     return NULL;
  85 }
  86
  87
  88
  89 #define TMP_BUF_LEN 128
  90
  91 static ssize_t stream_read(struct file * filp, char __user * buf, size_t size, loff_t * offset) {
  92     struct stream_state * stream = filp->private_data;
  93     ssize_t bytes_read = 0;
  94     ssize_t bytes_left = size;
  95     unsigned long flags;
  96     char tmp_buf[TMP_BUF_LEN];
  97     ssize_t total_bytes_left = 0;
  98
  99     // memset(tmp_buf, 0, TMP_BUF_LEN);
 100
 101     while (bytes_left > 0) {
 102         int tmp_len = (TMP_BUF_LEN > bytes_left) ? bytes_left : TMP_BUF_LEN;
 103         int tmp_read = 0;
 104
 105         palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 106         tmp_read = ringbuf_read(stream->out_ring, tmp_buf, tmp_len);
 107         palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 108
 109         if (tmp_read == 0) {
 110             // If userspace reads more than we have
 111             break;
 112         }
 113
 114         if (copy_to_user(buf + bytes_read, tmp_buf, tmp_read)) {
 115             ERROR("Read Fault\n");
 116             return -EFAULT;
 117         }
 118
 119         bytes_left -= tmp_read;
 120         bytes_read += tmp_read;
 121     }
 122
 123
 124     palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 125     total_bytes_left = ringbuf_data_len(stream->out_ring);
 126     palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 127
 128     if (total_bytes_left > 0) {
 129         wake_up_interruptible(&(stream->intr_queue));
 130     }
 131
 132     return bytes_read;
 133 }
 134
 135 static unsigned int
 136 stream_poll(struct file * filp, struct poll_table_struct * poll_tb) {
 137     struct stream_state * stream = filp->private_data;
 138     unsigned int mask = POLLIN | POLLRDNORM;
 139     unsigned long flags;
 140     int data_avail = 0;
 141
 142     poll_wait(filp, &(stream->intr_queue), poll_tb);
 143
 144     palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 145     data_avail = ringbuf_data_len(stream->out_ring);
 146     palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 147
 148     if (data_avail > 0) {
 149         return mask;
 150     }
 151
 152     return 0;
 153
 154 }
 155
 156 static ssize_t stream_write(struct file * filp, const char __user * buf, size_t size, loff_t * offset) {
 157     struct stream_state * stream = filp->private_data;
 158     char * kern_buf = NULL;
 159     ssize_t bytes_written = 0;
 160
 161     kern_buf = palacios_alloc(size);
 162
 163     if (!kern_buf) {
 164         ERROR("Cannot allocate buffer in stream interface\n");
 165         return -EFAULT;
 166     }
 167
 168     if (copy_from_user(kern_buf, buf, size)) {
 169         ERROR("Stream Write Failed\n");
 170         palacios_free(kern_buf);
 171         return -EFAULT;
 172     };
 173
 174     bytes_written = stream->v3_stream->input(stream->v3_stream, kern_buf, size);
 175
 176     palacios_free(kern_buf);
 177
 178     return bytes_written;
 179 }
 180
 181
 182 static int stream_release(struct inode * i, struct file * filp) {
 183     struct stream_state * stream = filp->private_data;
 184     unsigned long flags;
 185
 186     palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 187     stream->connected = 0;
 188     palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 189
 190
 191     return 0;
 192
 193 }
 194
 195 static struct file_operations stream_fops = {
 196     .read = stream_read,
 197     .write = stream_write,
 198     .release = stream_release,
 199     .poll = stream_poll,
 200 };
 201
 202
 203
 204 static void * palacios_stream_open(struct v3_stream * v3_stream, const char * name, void * private_data) {
 205     struct v3_guest * guest = (struct v3_guest *)private_data;
 206     struct stream_state * stream = NULL;
 207     struct vm_global_streams * vm_state = NULL;
 208
 209     if (guest != NULL) {
 210         vm_state = get_vm_ext_data(guest, "STREAM_INTERFACE");
 211
 212         if (vm_state == NULL) {
 213             ERROR("ERROR: Could not locate vm stream state for extension STREAM_INTERFACE\n");
 214             return NULL;
 215         }
 216     }
 217
 218     if (find_stream_by_name(guest, name) != NULL) {
 219         ERROR("Stream already exists\n");
 220         return NULL;
 221     }
 222
 223     stream = palacios_alloc(sizeof(struct stream_state));
 224     if (!stream) {
 225         ERROR("Unable to allocate stream\n");
 226         return NULL;
 227     }
 228     memset(stream, 0, sizeof(struct stream_state));
 229
 230     stream->out_ring = create_ringbuf(STREAM_RING_LEN);
 231     stream->v3_stream = v3_stream;
 232     stream->guest = guest;
 233     stream->connected = 0;
 234
 235     strncpy(stream->name, name, STREAM_NAME_LEN - 1);
 236
 237     init_waitqueue_head(&(stream->intr_queue));
 238     palacios_spinlock_init(&(stream->lock));
 239
 240     if (guest == NULL) {
 241         list_add(&(stream->stream_node), &(global_streams));
 242     } else {
 243         list_add(&(stream->stream_node), &(vm_state->open_streams));
 244     }
 245
 246     return stream;
 247 }
 248
 249
 250 static uint64_t palacios_stream_output(struct v3_stream * v3_stream, char * buf, int len) {
 251     struct stream_state * stream = (struct stream_state *)v3_stream->host_stream_data;
 252     int bytes_written = 0;
 253     unsigned long flags;
 254
 255
 256     if (stream->connected == 0) {
 257         return 0;
 258     }
 259
 260     while (bytes_written < len) {
 261         palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 262         bytes_written += ringbuf_write(stream->out_ring, buf + bytes_written, len - bytes_written);
 263         palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 264
 265         wake_up_interruptible(&(stream->intr_queue));
 266
 267         if (bytes_written < len) {
 268             // not enough space in ringbuffer, activate user space to drain it
 269             schedule();
 270         }
 271     }
 272
 273
 274     return bytes_written;
 275 }
 276
 277
 278 static void palacios_stream_close(struct v3_stream * v3_stream) {
 279     struct stream_state * stream = (struct stream_state *)v3_stream->host_stream_data;
 280
 281     free_ringbuf(stream->out_ring);
 282     list_del(&(stream->stream_node));
 283     palacios_spinlock_deinit(&(stream->lock));
 284     palacios_free(stream);
 285
 286 }
 287
 288 static struct v3_stream_hooks palacios_stream_hooks = {
 289     .open = palacios_stream_open,
 290     .output = palacios_stream_output,
 291     .close = palacios_stream_close,
 292 };
 293
 294
 295 static int stream_init( void ) {
 296     INIT_LIST_HEAD(&(global_streams));
 297     V3_Init_Stream(&palacios_stream_hooks);
 298
 299     return 0;
 300 }
 301
 302
 303 static int stream_deinit( void ) {
 304     struct stream_state * stream = NULL;
 305     struct stream_state * tmp = NULL;
 306
 307     list_for_each_entry_safe(stream, tmp, &(global_streams), stream_node) {
 308         free_ringbuf(stream->out_ring);
 309         list_del(&(stream->stream_node));
 310         palacios_free(stream);
 311     }
 312
 313     return 0;
 314 }
 315
 316
 317
 318
 319
 320 static int stream_connect(struct v3_guest * guest, unsigned int cmd, unsigned long arg, void * priv_data) {
 321     void __user * argp = (void __user *)arg;
 322     struct stream_state * stream = NULL;
 323     int stream_fd = 0;
 324     char name[STREAM_NAME_LEN];
 325     unsigned long flags = 0;
 326     int ret = -1;
 327
 328
 329     if (copy_from_user(name, argp, STREAM_NAME_LEN)) {
 330         ERROR("%s(%d): copy from user error...\n", __FILE__, __LINE__);
 331         return -EFAULT;
 332     }
 333
 334     stream = find_stream_by_name(guest, name);
 335
 336     if (stream == NULL) {
 337         ERROR("Could not find stream (%s)\n", name);
 338         return -EFAULT;
 339     }
 340
 341     palacios_spinlock_lock_irqsave(&(stream->lock), flags);
 342     if (stream->connected == 0) {
 343         stream->connected = 1;
 344         ret = 1;
 345     }
 346     palacios_spinlock_unlock_irqrestore(&(stream->lock), flags);
 347
 348
 349     if (ret == -1) {
 350         ERROR("Stream (%s) already connected\n", name);
 351         return -EFAULT;
 352     }
 353
 354
 355     stream_fd = anon_inode_getfd("v3-stream", &stream_fops, stream, O_RDWR);
 356
 357     if (stream_fd < 0) {
 358         ERROR("Error creating stream inode for (%s)\n", name);
 359         return stream_fd;
 360     }
 361
 362     INFO("Stream (%s) connected\n", name);
 363
 364     return stream_fd;
 365 }
 366
 367
 368 static int guest_stream_init(struct v3_guest * guest, void ** vm_data) {
 369     struct vm_global_streams * state = palacios_alloc(sizeof(struct vm_global_streams));
 370
 371     if (!state) {
 372         ERROR("Unable to allocate state in stream init\n");
 373         return -1;
 374     }
 375
 376     INIT_LIST_HEAD(&(state->open_streams));
 377     *vm_data = state;
 378
 379     add_guest_ctrl(guest, V3_VM_STREAM_CONNECT, stream_connect, state);
 380
 381     return 0;
 382 }
 383
 384
 385 static int guest_stream_deinit(struct v3_guest * guest, void * vm_data) {
 386     struct vm_global_streams * state = vm_data;
 387
 388     struct stream_state * stream = NULL;
 389     struct stream_state * tmp = NULL;
 390
 391     list_for_each_entry_safe(stream, tmp, &(global_streams), stream_node) {
 392         free_ringbuf(stream->out_ring);
 393         list_del(&(stream->stream_node));
 394         palacios_free(stream);
 395     }
 396
 397     palacios_free(state);
 398
 399     return 0;
 400 }
 401
 402
 403
 404 static struct linux_ext stream_ext = {
 405     .name = "STREAM_INTERFACE",
 406     .init = stream_init,
 407     .deinit = stream_deinit,
 408     .guest_init = guest_stream_init,
 409     .guest_deinit = guest_stream_deinit
 410 };
 411
 412
 413 register_extension(&stream_ext);