lib/fb-direct.c

   1 /*
   2  *      UCW Library -- Fast Buffered I/O on O_DIRECT Files
   3  *
   4  *      (c) 2006 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 /*
  11  *      This is a fastbuf backend for fast streaming I/O using O_DIRECT and
  12  *      the asynchronous I/O module. It's designed for use on large files
  13  *      which don't fit in the disk cache.
  14  *
  15  *      CAVEATS:
  16  *
  17  *        - All operations with a single fbdirect handle must be done
  18  *          within a single thread, unless you provide a custom I/O queue
  19  *          and take care of locking.
  20  *
  21  *      FIXME: what if the OS doesn't support O_DIRECT?
  22  *      FIXME: doc: don't mix threads
  23  *      FIXME: unaligned seeks and partial writes?
  24  *      FIXME: merge with other file-oriented fastbufs
  25  */
  26
  27 #undef LOCAL_DEBUG
  28
  29 #include "lib/lib.h"
  30 #include "lib/fastbuf.h"
  31 #include "lib/lfs.h"
  32 #include "lib/asio.h"
  33 #include "lib/conf.h"
  34
  35 #include <string.h>
  36 #include <fcntl.h>
  37 #include <unistd.h>
  38 #include <pthread.h>
  39
  40 static uns fbdir_cheat;
  41 static uns fbdir_buffer_size = 65536;
  42 static uns fbdir_read_ahead = 1;
  43 static uns fbdir_write_back = 1;
  44
  45 static struct cf_section fbdir_cf = {
  46   CF_ITEMS {
  47     CF_UNS("Cheat", &fbdir_cheat),
  48     CF_UNS("BufferSize", &fbdir_buffer_size),
  49     CF_UNS("ReadAhead", &fbdir_read_ahead),
  50     CF_UNS("WriteBack", &fbdir_write_back),
  51     CF_END
  52   }
  53 };
  54
  55 #define FBDIR_ALIGN 512
  56
  57 static pthread_key_t fbdir_queue_key;
  58
  59 enum fbdir_mode {                               // Current operating mode
  60     M_NULL,
  61     M_READ,
  62     M_WRITE
  63 };
  64
  65 struct fb_direct {
  66   struct fastbuf fb;
  67   int fd;                                       // File descriptor
  68   int is_temp_file;                             // 0=normal file, 1=temporary file, delete on close, -1=shared FD
  69   struct asio_queue *io_queue;                  // I/O queue to use
  70   struct asio_queue *user_queue;                // If io_queue was supplied by the user
  71   struct asio_request *pending_read;
  72   struct asio_request *done_read;
  73   struct asio_request *active_buffer;
  74   enum fbdir_mode mode;
  75   byte name[0];
  76 };
  77 #define FB_DIRECT(f) ((struct fb_direct *)(f)->is_fastbuf)
  78
  79 static void CONSTRUCTOR
  80 fbdir_global_init(void)
  81 {
  82   cf_declare_section("FBDirect", &fbdir_cf, 0);
  83   if (pthread_key_create(&fbdir_queue_key, NULL) < 0)
  84     die("Cannot create fbdir_queue_key: %m");
  85 }
  86
  87 static void
  88 fbdir_read_sync(struct fb_direct *F)
  89 {
  90   while (F->pending_read)
  91     {
  92       struct asio_request *r = asio_wait(F->io_queue);
  93       ASSERT(r);
  94       struct fb_direct *G = r->user_data;
  95       ASSERT(G);
  96       ASSERT(G->pending_read == r && !G->done_read);
  97       G->pending_read = NULL;
  98       G->done_read = r;
  99     }
 100 }
 101
 102 static void
 103 fbdir_change_mode(struct fb_direct *F, enum fbdir_mode mode)
 104 {
 105   if (F->mode == mode)
 106     return;
 107   DBG("FB-DIRECT: Switching mode to %d", mode);
 108   switch (F->mode)
 109     {
 110     case M_NULL:
 111       break;
 112     case M_READ:
 113       fbdir_read_sync(F);                       // Wait for read-ahead requests to finish
 114       if (F->done_read)                         // Return read-ahead requests if any
 115         {
 116           asio_put(F->done_read);
 117           F->done_read = NULL;
 118         }
 119       break;
 120     case M_WRITE:
 121       asio_sync(F->io_queue);                   // Wait for pending writebacks
 122       break;
 123     }
 124   if (F->active_buffer)
 125     {
 126       asio_put(F->active_buffer);
 127       F->active_buffer = NULL;
 128     }
 129   F->mode = mode;
 130 }
 131
 132 static void
 133 fbdir_submit_read(struct fb_direct *F)
 134 {
 135   struct asio_request *r = asio_get(F->io_queue);
 136   r->fd = F->fd;
 137   r->op = ASIO_READ;
 138   r->len = F->io_queue->buffer_size;
 139   r->user_data = F;
 140   asio_submit(r);
 141   F->pending_read = r;
 142 }
 143
 144 static int
 145 fbdir_refill(struct fastbuf *f)
 146 {
 147   struct fb_direct *F = FB_DIRECT(f);
 148
 149   DBG("FB-DIRECT: Refill");
 150
 151   if (!F->done_read)
 152     {
 153       if (!F->pending_read)
 154         {
 155           fbdir_change_mode(F, M_READ);
 156           fbdir_submit_read(F);
 157         }
 158       fbdir_read_sync(F);
 159       ASSERT(F->done_read);
 160     }
 161
 162   struct asio_request *r = F->done_read;
 163   F->done_read = NULL;
 164   if (F->active_buffer)
 165     asio_put(F->active_buffer);
 166   F->active_buffer = r;
 167   if (!r->status)
 168     return 0;
 169   if (r->status < 0)
 170     die("Error reading %s: %s", f->name, strerror(r->returned_errno));
 171   f->bptr = f->buffer = r->buffer;
 172   f->bstop = f->bufend = f->buffer + r->status;
 173   f->pos += r->status;
 174
 175   fbdir_submit_read(F);                         // Read-ahead the next block
 176
 177   return r->status;
 178 }
 179
 180 static void
 181 fbdir_spout(struct fastbuf *f)
 182 {
 183   struct fb_direct *F = FB_DIRECT(f);
 184   struct asio_request *r;
 185
 186   DBG("FB-DIRECT: Spout");
 187
 188   fbdir_change_mode(F, M_WRITE);
 189   r = F->active_buffer;
 190   if (r && f->bptr > f->bstop)
 191     {
 192       r->op = ASIO_WRITE_BACK;
 193       r->fd = F->fd;
 194       r->len = f->bptr - f->bstop;
 195       ASSERT(!(f->pos % FBDIR_ALIGN) || fbdir_cheat);
 196       f->pos += r->len;
 197       if (!fbdir_cheat && r->len % FBDIR_ALIGN)                 // Have to simulate incomplete writes
 198         {
 199           r->len = ALIGN_TO(r->len, FBDIR_ALIGN);
 200           asio_submit(r);
 201           asio_sync(F->io_queue);
 202           DBG("FB-DIRECT: Truncating at %Ld", (long long)f->pos);
 203           if (sh_ftruncate(F->fd, f->pos) < 0)
 204             die("Error truncating %s: %m", f->name);
 205         }
 206       else
 207         asio_submit(r);
 208       r = NULL;
 209     }
 210   if (!r)
 211     r = asio_get(F->io_queue);
 212   f->bstop = f->bptr = f->buffer = r->buffer;
 213   f->bufend = f->buffer + F->io_queue->buffer_size;
 214   F->active_buffer = r;
 215 }
 216
 217 static void
 218 fbdir_seek(struct fastbuf *f, sh_off_t pos, int whence)
 219 {
 220   DBG("FB-DIRECT: Seek %Ld %d", (long long)pos, whence);
 221
 222   if (whence == SEEK_SET && pos == f->pos)
 223     return;
 224
 225   fbdir_change_mode(FB_DIRECT(f), M_NULL);                      // Wait for all async requests to finish
 226   sh_off_t l = sh_seek(FB_DIRECT(f)->fd, pos, whence);
 227   if (l < 0)
 228     die("lseek on %s: %m", f->name);
 229   f->pos = l;
 230 }
 231
 232 static struct asio_queue *
 233 fbdir_get_io_queue(void)
 234 {
 235   struct asio_queue *q = pthread_getspecific(fbdir_queue_key);
 236   if (!q)
 237     {
 238       q = xmalloc_zero(sizeof(struct asio_queue));
 239       q->buffer_size = fbdir_buffer_size;
 240       q->max_writebacks = fbdir_write_back;
 241       asio_init_queue(q);
 242       pthread_setspecific(fbdir_queue_key, q);
 243     }
 244   q->use_count++;
 245   DBG("FB-DIRECT: Got I/O queue, uc=%d", q->use_count);
 246   return q;
 247 }
 248
 249 static void
 250 fbdir_put_io_queue(void)
 251 {
 252   struct asio_queue *q = pthread_getspecific(fbdir_queue_key);
 253   ASSERT(q);
 254   DBG("FB-DIRECT: Put I/O queue, uc=%d", q->use_count);
 255   if (!--q->use_count)
 256     {
 257       asio_cleanup_queue(q);
 258       xfree(q);
 259       pthread_setspecific(fbdir_queue_key, NULL);
 260     }
 261 }
 262
 263 static void
 264 fbdir_close(struct fastbuf *f)
 265 {
 266   struct fb_direct *F = FB_DIRECT(f);
 267
 268   DBG("FB-DIRECT: Close");
 269
 270   fbdir_change_mode(F, M_NULL);
 271   if (!F->user_queue)
 272     fbdir_put_io_queue();
 273
 274   switch (F->is_temp_file)
 275     {
 276     case 1:
 277       if (unlink(f->name) < 0)
 278         log(L_ERROR, "unlink(%s): %m", f->name);
 279     case 0:
 280       close(F->fd);
 281     }
 282
 283   xfree(f);
 284 }
 285
 286 static int
 287 fbdir_config(struct fastbuf *f, uns item, int value)
 288 {
 289   switch (item)
 290     {
 291     case BCONFIG_IS_TEMP_FILE:
 292       FB_DIRECT(f)->is_temp_file = value;
 293       return 0;
 294     default:
 295       return -1;
 296     }
 297 }
 298
 299 static struct fastbuf *
 300 fbdir_open_internal(byte *name, int fd, struct asio_queue *q)
 301 {
 302   int namelen = strlen(name) + 1;
 303   struct fb_direct *F = xmalloc(sizeof(struct fb_direct) + namelen);
 304   struct fastbuf *f = &F->fb;
 305
 306   DBG("FB-DIRECT: Open");
 307   bzero(F, sizeof(*F));
 308   f->name = F->name;
 309   memcpy(f->name, name, namelen);
 310   F->fd = fd;
 311   if (q)
 312     F->io_queue = F->user_queue = q;
 313   else
 314     F->io_queue = fbdir_get_io_queue();
 315   f->refill = fbdir_refill;
 316   f->spout = fbdir_spout;
 317   f->seek = fbdir_seek;
 318   f->close = fbdir_close;
 319   f->config = fbdir_config;
 320   f->can_overwrite_buffer = 2;
 321   return f;
 322 }
 323
 324 struct fastbuf *
 325 fbdir_open_try(byte *name, uns mode, struct asio_queue *q)
 326 {
 327   if (!fbdir_cheat)
 328     mode |= O_DIRECT;
 329   int fd = sh_open(name, mode, 0666);
 330   if (fd < 0)
 331     return NULL;
 332   struct fastbuf *b = fbdir_open_internal(name, fd, q);
 333   if (mode & O_APPEND)
 334     fbdir_seek(b, 0, SEEK_END);
 335   return b;
 336 }
 337
 338 struct fastbuf *
 339 fbdir_open(byte *name, uns mode, struct asio_queue *q)
 340 {
 341   struct fastbuf *b = fbdir_open_try(name, mode, q);
 342   if (!b)
 343     die("Unable to %s file %s: %m",
 344         (mode & O_CREAT) ? "create" : "open", name);
 345   return b;
 346 }
 347
 348 struct fastbuf *
 349 fbdir_open_fd(int fd, struct asio_queue *q)
 350 {
 351   byte x[32];
 352
 353   sprintf(x, "fd%d", fd);
 354   if (!fbdir_cheat && fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_DIRECT) < 0)
 355     log(L_WARN, "Cannot set O_DIRECT on fd %d: %m", fd);
 356   return fbdir_open_internal(x, fd, q);
 357 }
 358
 359 #ifdef TEST
 360
 361 #include "lib/getopt.h"
 362
 363 int main(int argc, char **argv)
 364 {
 365   struct fastbuf *f, *t;
 366
 367   log_init(NULL);
 368   if (cf_getopt(argc, argv, CF_SHORT_OPTS, CF_NO_LONG_OPTS, NULL) >= 0)
 369     die("Hey, whaddya want?");
 370   f = (optind < argc) ? fbdir_open(argv[optind++], O_RDONLY, NULL) : fbdir_open_fd(0, NULL);
 371   t = (optind < argc) ? fbdir_open(argv[optind++], O_RDWR | O_CREAT | O_TRUNC, NULL) : fbdir_open_fd(1, NULL);
 372
 373   bbcopy(f, t, ~0U);
 374   ASSERT(btell(f) == btell(t));
 375
 376 #if 0           // This triggers unaligned write
 377   bflush(t);
 378   bputc(t, '\n');
 379 #endif
 380
 381   brewind(t);
 382   bgetc(t);
 383   ASSERT(btell(t) == 1);
 384
 385   bclose(f);
 386   bclose(t);
 387   return 0;
 388 }
 389
 390 #endif