plugins/streaming/streaming.c

   1 /* nbdkit
   2  * Copyright (C) 2014 Red Hat Inc.
   3  *
   4  * Redistribution and use in source and binary forms, with or without
   5  * modification, are permitted provided that the following conditions are
   6  * met:
   7  *
   8  * * Redistributions of source code must retain the above copyright
   9  * notice, this list of conditions and the following disclaimer.
  10  *
  11  * * Redistributions in binary form must reproduce the above copyright
  12  * notice, this list of conditions and the following disclaimer in the
  13  * documentation and/or other materials provided with the distribution.
  14  *
  15  * * Neither the name of Red Hat nor the names of its contributors may be
  16  * used to endorse or promote products derived from this software without
  17  * specific prior written permission.
  18  *
  19  * THIS SOFTWARE IS PROVIDED BY RED HAT AND CONTRIBUTORS ''AS IS'' AND
  20  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  21  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  22  * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL RED HAT OR
  23  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  24  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  25  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  26  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  27  * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  29  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  30  * SUCH DAMAGE.
  31  */
  32
  33 #include <config.h>
  34
  35 #include <stdio.h>
  36 #include <stdlib.h>
  37 #include <string.h>
  38 #include <fcntl.h>
  39 #include <unistd.h>
  40 #include <sys/types.h>
  41 #include <sys/stat.h>
  42 #include <errno.h>
  43
  44 #include <nbdkit-plugin.h>
  45
  46 static char *filename = NULL;
  47 static int fd = -1;
  48
  49 /* In theory INT64_MAX, but it breaks qemu's NBD driver. */
  50 static int64_t size = INT64_MAX/2;
  51
  52 /* Flag if we have entered the unrecoverable error state because of
  53  * a seek backwards.
  54  */
  55 static int errorstate = 0;
  56
  57 /* Highest byte (+1) that has been written in the data stream. */
  58 static uint64_t highestwrite = 0;
  59
  60 /* Called for each key=value passed on the command line. */
  61 static int
  62 streaming_config (const char *key, const char *value)
  63 {
  64   if (strcmp (key, "pipe") == 0) {
  65     /* See FILENAMES AND PATHS in nbdkit-plugin(3). */
  66     filename = nbdkit_absolute_path (value);
  67     if (!filename)
  68       return -1;
  69   }
  70   else if (strcmp (key, "size") == 0) {
  71     size = nbdkit_parse_size (value);
  72     if (size == -1)
  73       return -1;
  74   }
  75   else {
  76     nbdkit_error ("unknown parameter '%s'", key);
  77     return -1;
  78   }
  79
  80   return 0;
  81 }
  82
  83 /* Check the user did pass a pipe=<FILENAME> parameter. */
  84 static int
  85 streaming_config_complete (void)
  86 {
  87   if (filename == NULL) {
  88     nbdkit_error ("you must supply the pipe=<FILENAME> parameter "
  89                   "after the plugin name on the command line");
  90     return -1;
  91   }
  92
  93   /* Open the file blindly.  If this fails with ENOENT then we create a
  94    * FIFO and try again.
  95    */
  96  again:
  97   fd = open (filename, O_RDWR|O_CLOEXEC|O_NOCTTY);
  98   if (fd == -1) {
  99     if (errno != ENOENT) {
 100       nbdkit_error ("open: %s: %m", filename);
 101       return -1;
 102     }
 103     if (mknod (filename, S_IFIFO | 0666, 0) == -1) {
 104       nbdkit_error ("mknod: %s: %m", filename);
 105       return -1;
 106     }
 107     goto again;
 108   }
 109
 110   return 0;
 111 }
 112
 113 /* nbdkit is shutting down. */
 114 static void
 115 streaming_unload (void)
 116 {
 117   if (fd >= 0)
 118     close (fd);
 119   free (filename);
 120 }
 121
 122 #define streaming_config_help \
 123   "pipe=<FILENAME>     (required) The filename to serve.\n" \
 124   "size=<SIZE>         (optional) Stream size."
 125
 126 /* Create the per-connection handle. */
 127 static void *
 128 streaming_open (int readonly)
 129 {
 130   void *h;
 131
 132   if (readonly) {
 133     nbdkit_error ("you cannot use the -r option with the streaming plugin");
 134     return NULL;
 135   }
 136
 137   if (errorstate) {
 138     nbdkit_error ("unrecoverable error state, "
 139                   "no new connections can be opened");
 140     return NULL;
 141   }
 142
 143   /* There is no handle, so return an arbitrary non-NULL pointer. */
 144   h = &fd;
 145
 146   return h;
 147 }
 148
 149 /* Free up the per-connection handle. */
 150 static void
 151 streaming_close (void *handle)
 152 {
 153 }
 154
 155 #define THREAD_MODEL NBDKIT_THREAD_MODEL_SERIALIZE_ALL_REQUESTS
 156
 157 /* Return the size of the stream (infinite). */
 158 static int64_t
 159 streaming_get_size (void *handle)
 160 {
 161   return size;
 162 }
 163
 164 /* Write data to the stream. */
 165 static int
 166 streaming_pwrite (void *handle, const void *buf,
 167                   uint32_t count, uint64_t offset)
 168 {
 169   size_t n;
 170   ssize_t r;
 171
 172   if (errorstate) {
 173     nbdkit_error ("unrecoverable error state");
 174     errno = EIO;
 175     return -1;
 176   }
 177
 178   if (offset < highestwrite) {
 179     nbdkit_error ("client tried to seek backwards and write: "
 180                   "the streaming plugin does not currently support this");
 181     errorstate = 1;
 182     errno = EIO;
 183     return -1;
 184   }
 185
 186   /* Need to write some zeroes. */
 187   if (offset > highestwrite) {
 188     int64_t remaining = offset - highestwrite;
 189     static char zerobuf[4096];
 190
 191     while (remaining > 0) {
 192       n = remaining > sizeof zerobuf ? sizeof zerobuf : remaining;
 193       r = write (fd, zerobuf, n);
 194       if (r == -1) {
 195         nbdkit_error ("write: %m");
 196         errorstate = 1;
 197         return -1;
 198       }
 199       highestwrite += r;
 200       remaining -= r;
 201     }
 202   }
 203
 204   /* Write the data. */
 205   while (count > 0) {
 206     r = write (fd, buf, count);
 207     if (r == -1) {
 208       nbdkit_error ("write: %m");
 209       errorstate = 1;
 210       return -1;
 211     }
 212     buf += r;
 213     highestwrite += r;
 214     count -= r;
 215   }
 216
 217   return 0;
 218 }
 219
 220 /* Read data back from the stream. */
 221 static int
 222 streaming_pread (void *handle, void *buf, uint32_t count, uint64_t offset)
 223 {
 224   if (errorstate) {
 225     nbdkit_error ("unrecoverable error state");
 226     errno = EIO;
 227     return -1;
 228   }
 229
 230   /* Allow reads which are entirely >= highestwrite.  These return zeroes. */
 231   if (offset >= highestwrite) {
 232     memset (buf, 0, count);
 233     return 0;
 234   }
 235
 236   nbdkit_error ("client tried to read: "
 237                 "the streaming plugin does not currently support this");
 238   errorstate = 1;
 239   errno = EIO;
 240   return -1;
 241 }
 242
 243 static struct nbdkit_plugin plugin = {
 244   .name              = "streaming",
 245   .longname          = "nbdkit streaming plugin",
 246   .version           = PACKAGE_VERSION,
 247   .unload            = streaming_unload,
 248   .config            = streaming_config,
 249   .config_complete   = streaming_config_complete,
 250   .config_help       = streaming_config_help,
 251   .open              = streaming_open,
 252   .close             = streaming_close,
 253   .get_size          = streaming_get_size,
 254   .pwrite            = streaming_pwrite,
 255   .pread             = streaming_pread,
 256   .errno_is_preserved = 1,
 257 };
 258
 259 NBDKIT_REGISTER_PLUGIN(plugin)