int can_overwrite_buffer; /* Can the buffer be altered? (see discussion above) 0=never, 1=temporarily, 2=permanently */
};
- struct fastbuf *bopen_file(const byte *name, int mode, struct fb_params *params); /* Use params==NULL for defaults */
- struct fastbuf *bopen_file_try(const byte *name, int mode, struct fb_params *params);
+/* FastIO on files with run-time parametrization */
+
+enum fb_type { /* Which back-end you want to use */
+ FB_STD, /* Standard buffered I/O */
+ FB_DIRECT, /* Direct I/O bypassing system caches (see fb-direct.c for description) */
+ FB_MMAP /* Memory mapped files */
+};
+
+struct fb_params {
+ enum fb_type type;
+ uns buffer_size;
+ /* FB_STD only */
+ uns keep_back_buf;
+ /* FB_DIRECT only */
+ uns read_ahead;
+ uns write_back;
+ struct asio_queue *asio;
+};
+
+struct cf_section;
+extern struct cf_section fbpar_cf;
+extern struct fb_params fbpar_def;
+
++struct fastbuf *bopen_file(const char *name, int mode, struct fb_params *params); /* Use params==NULL for defaults */
++struct fastbuf *bopen_file_try(const char *name, int mode, struct fb_params *params);
+struct fastbuf *bopen_tmp_file(struct fb_params *params);
+struct fastbuf *bopen_fd(int fd, struct fb_params *params);
+
/* FastIO on standard files (specify buffer size 0 to enable mmaping) */
- struct fastbuf *bfdopen_internal(int fd, const byte *name, uns buflen);
- struct fastbuf *bopen(const byte *name, uns mode, uns buflen);
- struct fastbuf *bopen_try(const byte *name, uns mode, uns buflen);
++struct fastbuf *bfdopen_internal(int fd, const char *name, uns buflen);
+ struct fastbuf *bopen(const char *name, uns mode, uns buflen);
+ struct fastbuf *bopen_try(const char *name, uns mode, uns buflen);
struct fastbuf *bopen_tmp(uns buflen);
struct fastbuf *bfdopen(int fd, uns buflen);
struct fastbuf *bfdopen_shared(int fd, uns buflen);
void bfilesync(struct fastbuf *b);
#define TEMP_FILE_NAME_LEN 256
- void temp_file_name(byte *name);
+ void temp_file_name(char *name);
- struct fastbuf *bfmmopen_internal(int fd, const byte *name, uns mode);
+/* Internal functions of some file back-ends */
+
- struct fastbuf *fbdir_open_fd_internal(int fd, const byte *name, struct asio_queue *io_queue, uns buffer_size, uns read_ahead, uns write_back);
++struct fastbuf *bfmmopen_internal(int fd, const char *name, uns mode);
+
+extern uns fbdir_cheat;
+struct asio_queue;
++struct fastbuf *fbdir_open_fd_internal(int fd, const char *name, struct asio_queue *io_queue, uns buffer_size, uns read_ahead, uns write_back);
+
/* FastIO on in-memory streams */
- struct fastbuf *fbmem_create(unsigned blocksize); /* Create stream and return its writing fastbuf */
+ struct fastbuf *fbmem_create(uns blocksize); /* Create stream and return its writing fastbuf */
struct fastbuf *fbmem_clone_read(struct fastbuf *); /* Create reading fastbuf */
-/* FastIO on memory mapped files */
-
-struct fastbuf *bopen_mm(const char *name, uns mode);
-
/* FastI on file descriptors with limit */
struct fastbuf *bopen_limited_fd(int fd, uns bufsize, uns limit);
--- /dev/null
- fbdir_open_fd_internal(int fd, const byte *name, struct asio_queue *q, uns buffer_size, uns read_ahead UNUSED, uns write_back)
+/*
+ * UCW Library -- Fast Buffered I/O on O_DIRECT Files
+ *
+ * (c) 2006 Martin Mares <mj@ucw.cz>
+ *
+ * This software may be freely distributed and used according to the terms
+ * of the GNU Lesser General Public License.
+ */
+
+/*
+ * This is a fastbuf backend for fast streaming I/O using O_DIRECT and
+ * the asynchronous I/O module. It's designed for use on large files
+ * which don't fit in the disk cache.
+ *
+ * CAVEATS:
+ *
+ * - All operations with a single fbdirect handle must be done
+ * within a single thread, unless you provide a custom I/O queue
+ * and take care of locking.
+ *
+ * FIXME: what if the OS doesn't support O_DIRECT?
+ * FIXME: unaligned seeks and partial writes?
+ * FIXME: merge with other file-oriented fastbufs
+ */
+
+#undef LOCAL_DEBUG
+
+#include "lib/lib.h"
+#include "lib/fastbuf.h"
+#include "lib/lfs.h"
+#include "lib/asio.h"
+#include "lib/conf.h"
+#include "lib/threads.h"
+
+#include <string.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <stdio.h>
+
+uns fbdir_cheat;
+
+static struct cf_section fbdir_cf = {
+ CF_ITEMS {
+ CF_UNS("Cheat", &fbdir_cheat),
+ CF_END
+ }
+};
+
+#define FBDIR_ALIGN 512
+
+enum fbdir_mode { // Current operating mode
+ M_NULL,
+ M_READ,
+ M_WRITE
+};
+
+struct fb_direct {
+ struct fastbuf fb;
+ int fd; // File descriptor
+ int is_temp_file; // 0=normal file, 1=temporary file, delete on close, -1=shared FD
+ struct asio_queue *io_queue; // I/O queue to use
+ struct asio_queue *user_queue; // If io_queue was supplied by the user
+ struct asio_request *pending_read;
+ struct asio_request *done_read;
+ struct asio_request *active_buffer;
+ enum fbdir_mode mode;
+ byte name[0];
+};
+#define FB_DIRECT(f) ((struct fb_direct *)(f)->is_fastbuf)
+
+static void CONSTRUCTOR
+fbdir_global_init(void)
+{
+ cf_declare_section("FBDirect", &fbdir_cf, 0);
+}
+
+static void
+fbdir_read_sync(struct fb_direct *F)
+{
+ while (F->pending_read)
+ {
+ struct asio_request *r = asio_wait(F->io_queue);
+ ASSERT(r);
+ struct fb_direct *G = r->user_data;
+ ASSERT(G);
+ ASSERT(G->pending_read == r && !G->done_read);
+ G->pending_read = NULL;
+ G->done_read = r;
+ }
+}
+
+static void
+fbdir_change_mode(struct fb_direct *F, enum fbdir_mode mode)
+{
+ if (F->mode == mode)
+ return;
+ DBG("FB-DIRECT: Switching mode to %d", mode);
+ switch (F->mode)
+ {
+ case M_NULL:
+ break;
+ case M_READ:
+ fbdir_read_sync(F); // Wait for read-ahead requests to finish
+ if (F->done_read) // Return read-ahead requests if any
+ {
+ asio_put(F->done_read);
+ F->done_read = NULL;
+ }
+ break;
+ case M_WRITE:
+ asio_sync(F->io_queue); // Wait for pending writebacks
+ break;
+ }
+ if (F->active_buffer)
+ {
+ asio_put(F->active_buffer);
+ F->active_buffer = NULL;
+ }
+ F->mode = mode;
+}
+
+static void
+fbdir_submit_read(struct fb_direct *F)
+{
+ struct asio_request *r = asio_get(F->io_queue);
+ r->fd = F->fd;
+ r->op = ASIO_READ;
+ r->len = F->io_queue->buffer_size;
+ r->user_data = F;
+ asio_submit(r);
+ F->pending_read = r;
+}
+
+static int
+fbdir_refill(struct fastbuf *f)
+{
+ struct fb_direct *F = FB_DIRECT(f);
+
+ DBG("FB-DIRECT: Refill");
+
+ if (!F->done_read)
+ {
+ if (!F->pending_read)
+ {
+ fbdir_change_mode(F, M_READ);
+ fbdir_submit_read(F);
+ }
+ fbdir_read_sync(F);
+ ASSERT(F->done_read);
+ }
+
+ struct asio_request *r = F->done_read;
+ F->done_read = NULL;
+ if (F->active_buffer)
+ asio_put(F->active_buffer);
+ F->active_buffer = r;
+ if (!r->status)
+ return 0;
+ if (r->status < 0)
+ die("Error reading %s: %s", f->name, strerror(r->returned_errno));
+ f->bptr = f->buffer = r->buffer;
+ f->bstop = f->bufend = f->buffer + r->status;
+ f->pos += r->status;
+
+ fbdir_submit_read(F); // Read-ahead the next block
+
+ return r->status;
+}
+
+static void
+fbdir_spout(struct fastbuf *f)
+{
+ struct fb_direct *F = FB_DIRECT(f);
+ struct asio_request *r;
+
+ DBG("FB-DIRECT: Spout");
+
+ fbdir_change_mode(F, M_WRITE);
+ r = F->active_buffer;
+ if (r && f->bptr > f->bstop)
+ {
+ r->op = ASIO_WRITE_BACK;
+ r->fd = F->fd;
+ r->len = f->bptr - f->bstop;
+ ASSERT(!(f->pos % FBDIR_ALIGN) || fbdir_cheat);
+ f->pos += r->len;
+ if (!fbdir_cheat && r->len % FBDIR_ALIGN) // Have to simulate incomplete writes
+ {
+ r->len = ALIGN_TO(r->len, FBDIR_ALIGN);
+ asio_submit(r);
+ asio_sync(F->io_queue);
+ DBG("FB-DIRECT: Truncating at %llu", (long long)f->pos);
+ if (sh_ftruncate(F->fd, f->pos) < 0)
+ die("Error truncating %s: %m", f->name);
+ }
+ else
+ asio_submit(r);
+ r = NULL;
+ }
+ if (!r)
+ r = asio_get(F->io_queue);
+ f->bstop = f->bptr = f->buffer = r->buffer;
+ f->bufend = f->buffer + F->io_queue->buffer_size;
+ F->active_buffer = r;
+}
+
+static int
+fbdir_seek(struct fastbuf *f, sh_off_t pos, int whence)
+{
+ DBG("FB-DIRECT: Seek %llu %d", (long long)pos, whence);
+
+ if (whence == SEEK_SET && pos == f->pos)
+ return 1;
+
+ fbdir_change_mode(FB_DIRECT(f), M_NULL); // Wait for all async requests to finish
+ sh_off_t l = sh_seek(FB_DIRECT(f)->fd, pos, whence);
+ if (l < 0)
+ return 0;
+ f->pos = l;
+ return 1;
+}
+
+static struct asio_queue *
+fbdir_get_io_queue(uns buffer_size, uns write_back)
+{
+ struct ucwlib_context *ctx = ucwlib_thread_context();
+ struct asio_queue *q = ctx->io_queue;
+ if (!q)
+ {
+ q = xmalloc_zero(sizeof(struct asio_queue));
+ q->buffer_size = buffer_size;
+ q->max_writebacks = write_back;
+ asio_init_queue(q);
+ ctx->io_queue = q;
+ }
+ q->use_count++;
+ DBG("FB-DIRECT: Got I/O queue, uc=%d", q->use_count);
+ return q;
+}
+
+static void
+fbdir_put_io_queue(void)
+{
+ struct ucwlib_context *ctx = ucwlib_thread_context();
+ struct asio_queue *q = ctx->io_queue;
+ ASSERT(q);
+ DBG("FB-DIRECT: Put I/O queue, uc=%d", q->use_count);
+ if (!--q->use_count)
+ {
+ asio_cleanup_queue(q);
+ xfree(q);
+ ctx->io_queue = NULL;
+ }
+}
+
+static void
+fbdir_close(struct fastbuf *f)
+{
+ struct fb_direct *F = FB_DIRECT(f);
+
+ DBG("FB-DIRECT: Close");
+
+ fbdir_change_mode(F, M_NULL);
+ if (!F->user_queue)
+ fbdir_put_io_queue();
+
+ switch (F->is_temp_file)
+ {
+ case 1:
+ if (unlink(f->name) < 0)
+ msg(L_ERROR, "unlink(%s): %m", f->name);
+ case 0:
+ close(F->fd);
+ }
+
+ xfree(f);
+}
+
+static int
+fbdir_config(struct fastbuf *f, uns item, int value)
+{
+ switch (item)
+ {
+ case BCONFIG_IS_TEMP_FILE:
+ FB_DIRECT(f)->is_temp_file = value;
+ return 0;
+ default:
+ return -1;
+ }
+}
+
+struct fastbuf *
++fbdir_open_fd_internal(int fd, const char *name, struct asio_queue *q, uns buffer_size, uns read_ahead UNUSED, uns write_back)
+{
+ int namelen = strlen(name) + 1;
+ struct fb_direct *F = xmalloc(sizeof(struct fb_direct) + namelen);
+ struct fastbuf *f = &F->fb;
+
+ DBG("FB-DIRECT: Open");
+ bzero(F, sizeof(*F));
+ f->name = F->name;
+ memcpy(f->name, name, namelen);
+ F->fd = fd;
+ if (q)
+ F->io_queue = F->user_queue = q;
+ else
+ F->io_queue = fbdir_get_io_queue(buffer_size, write_back);
+ f->refill = fbdir_refill;
+ f->spout = fbdir_spout;
+ f->seek = fbdir_seek;
+ f->close = fbdir_close;
+ f->config = fbdir_config;
+ f->can_overwrite_buffer = 2;
+ return f;
+}
+
+#ifdef TEST
+
+#include "lib/getopt.h"
+
+int main(int argc, char **argv)
+{
+ struct fastbuf *f, *t;
+
+ log_init(NULL);
+ if (cf_getopt(argc, argv, CF_SHORT_OPTS, CF_NO_LONG_OPTS, NULL) >= 0)
+ die("Hey, whaddya want?");
+ f = (optind < argc) ? fbdir_open(argv[optind++], O_RDONLY, NULL) : fbdir_open_fd(0, NULL);
+ t = (optind < argc) ? fbdir_open(argv[optind++], O_RDWR | O_CREAT | O_TRUNC, NULL) : fbdir_open_fd(1, NULL);
+
+ bbcopy(f, t, ~0U);
+ ASSERT(btell(f) == btell(t));
+
+#if 0 // This triggers unaligned write
+ bflush(t);
+ bputc(t, '\n');
+#endif
+
+ brewind(t);
+ bgetc(t);
+ ASSERT(btell(t) == 1);
+
+ bclose(f);
+ bclose(t);
+ return 0;
+}
+
+#endif
}
}
-static struct fastbuf *
-bfdopen_internal(int fd, uns buflen, const char *name)
+struct fastbuf *
- bfdopen_internal(int fd, const byte *name, uns buflen)
++bfdopen_internal(int fd, const char *name, uns buflen)
{
+ ASSERT(buflen);
int namelen = strlen(name) + 1;
- struct fb_file *F = xmalloc(sizeof(struct fb_file) + buflen + namelen);
+ struct fb_file *F = xmalloc_zero(sizeof(struct fb_file) + buflen + namelen);
struct fastbuf *f = &F->fb;
bzero(F, sizeof(*F));
}
struct fastbuf *
- bopen_try(const byte *name, uns mode, uns buflen)
+ bopen_try(const char *name, uns mode, uns buflen)
{
- int fd = sh_open(name, mode, 0666);
- if (fd < 0)
- return NULL;
- struct fastbuf *b = bfdopen_internal(fd, buflen, name);
- if (mode & O_APPEND)
- bfd_seek(b, 0, SEEK_END);
- return b;
+ return bopen_file_try(name, mode, &(struct fb_params){ .type = FB_STD, .buffer_size = buflen });
}
struct fastbuf *
- bopen(const byte *name, uns mode, uns buflen)
+ bopen(const char *name, uns mode, uns buflen)
{
- if (!buflen)
- return bopen_mm(name, mode);
- struct fastbuf *b = bopen_try(name, mode, buflen);
- if (!b)
- die("Unable to %s file %s: %m",
- (mode & O_CREAT) ? "create" : "open", name);
- return b;
+ return bopen_file(name, mode, &(struct fb_params){ .type = FB_STD, .buffer_size = buflen });
}
struct fastbuf *
}
}
-static struct fastbuf *
+struct fastbuf *
- bfmmopen_internal(int fd, const byte *name, uns mode)
+ bfmmopen_internal(int fd, const char *name, uns mode)
{
int namelen = strlen(name) + 1;
struct fb_mmap *F = xmalloc(sizeof(struct fb_mmap) + namelen);
--- /dev/null
- CF_LOOKUP("Type", (int *)F(type), ((byte *[]){"std", "direct", "mmap", NULL})),
+/*
+ * UCW Library -- FastIO on files with run-time parametrization
+ *
+ * (c) 2007 Pavel Charvat <pchar@ucw.cz>
+ *
+ * This software may be freely distributed and used according to the terms
+ * of the GNU Lesser General Public License.
+ */
+
+#include "lib/lib.h"
+#include "lib/conf.h"
+#include "lib/lfs.h"
+#include "lib/fastbuf.h"
+
+#include <fcntl.h>
+#include <stdio.h>
+
+struct fb_params fbpar_def = {
+ .buffer_size = 65536,
+ .read_ahead = 1,
+ .write_back = 1,
+};
+
+struct cf_section fbpar_cf = {
+# define F(x) PTR_TO(struct fb_params, x)
+ CF_TYPE(struct fb_params),
+ CF_ITEMS {
- bopen_file(const byte *name, int mode, struct fb_params *params)
++ CF_LOOKUP("Type", (int *)F(type), ((char *[]){"std", "direct", "mmap", NULL})),
+ CF_UNS("BufSize", F(buffer_size)),
+ CF_UNS("KeepBackBuf", F(keep_back_buf)),
+ CF_UNS("ReadAhead", F(read_ahead)),
+ CF_UNS("WriteBack", F(write_back)),
+ CF_END
+ }
+# undef F
+};
+
+static struct cf_section fbpar_global_cf = {
+ CF_ITEMS {
+ CF_SECTION("Defaults", &fbpar_def, &fbpar_cf),
+ CF_END
+ }
+};
+
+static void CONSTRUCTOR
+fbpar_global_init(void)
+{
+ cf_declare_section("FBParam", &fbpar_global_cf, 0);
+}
+
+static struct fastbuf *
+bopen_fd_internal(int fd, struct fb_params *params, uns mode, const byte *name)
+{
+ byte buf[32];
+ if (!name)
+ {
+ sprintf(buf, "fd%d", fd);
+ name = buf;
+ }
+ struct fastbuf *fb;
+ switch (params->type)
+ {
+ case FB_STD:
+ fb = bfdopen_internal(fd, name,
+ params->buffer_size ? : fbpar_def.buffer_size);
+ if (params->keep_back_buf)
+ bconfig(fb, BCONFIG_KEEP_BACK_BUF, 1);
+ return fb;
+ case FB_DIRECT:
+ fb = fbdir_open_fd_internal(fd, name, params->asio,
+ params->buffer_size ? : fbpar_def.buffer_size,
+ params->read_ahead ? : fbpar_def.read_ahead,
+ params->write_back ? : fbpar_def.write_back);
+ if (!~mode && !fbdir_cheat && ((int)(mode = fcntl(fd, F_GETFL)) < 0 || fcntl(fd, F_SETFL, mode | O_DIRECT)) < 0)
+ msg(L_WARN, "Cannot set O_DIRECT on fd %d: %m", fd);
+ return fb;
+ case FB_MMAP:
+ if (!~mode && (int)(mode = fcntl(fd, F_GETFL)) < 0)
+ die("Cannot get flags of fd %d: %m", fd);
+ return bfmmopen_internal(fd, name, mode);
+ default:
+ ASSERT(0);
+ }
+}
+
+static struct fastbuf *
+bopen_file_internal(const byte *name, int mode, struct fb_params *params, int try)
+{
+ if (params->type == FB_DIRECT && !fbdir_cheat)
+ mode |= O_DIRECT;
+ if (params->type == FB_MMAP && (mode & O_ACCMODE) == O_WRONLY)
+ mode = (mode & ~O_ACCMODE) | O_RDWR;
+ int fd = sh_open(name, mode, 0666);
+ if (fd < 0)
+ if (try)
+ return NULL;
+ else
+ die("Unable to %s file %s: %m", (mode & O_CREAT) ? "create" : "open", name);
+ struct fastbuf *fb = bopen_fd_internal(fd, params, mode, name);
+ ASSERT(fb);
+ if (mode & O_APPEND)
+ bseek(fb, 0, SEEK_END);
+ return fb;
+}
+
+struct fastbuf *
- bopen_file_try(const byte *name, int mode, struct fb_params *params)
++bopen_file(const char *name, int mode, struct fb_params *params)
+{
+ return bopen_file_internal(name, mode, params ? : &fbpar_def, 0);
+}
+
+struct fastbuf *
++bopen_file_try(const char *name, int mode, struct fb_params *params)
+{
+ return bopen_file_internal(name, mode, params ? : &fbpar_def, 1);
+}
+
+struct fastbuf *
+bopen_fd(int fd, struct fb_params *params)
+{
+ return bopen_fd_internal(fd, params ? : &fbpar_def, ~0U, NULL);
+}
+
+struct fastbuf *
+bopen_tmp_file(struct fb_params *params)
+{
+ byte buf[TEMP_FILE_NAME_LEN];
+ temp_file_name(buf);
+ struct fastbuf *fb = bopen_file_internal(buf, O_RDWR | O_CREAT | O_TRUNC, params, 0);
+ bconfig(fb, BCONFIG_IS_TEMP_FILE, 1);
+ return fb;
+}
projects / libucw.git / commitdiff