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