fastwriter/default.c

#define _FASTWRITER_DEFAULT_C

#define _GNU_SOURCE
#define _XOPEN_SOURCE 600
#define _POSIX_C_SOURCE 200112L
#define _LARGEFILE64_SOURCE

#include "config.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <limits.h>
#include <errno.h>

#include <pthread.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>

#include <fcntl.h>


#ifdef HAVE_LINUX_FALLOC_H
# include <linux/falloc.h>
#endif /* HAVE_LINUX_FALLOC_H */

#include "fastwriter.h"
#include "private.h"
#include "sysinfo.h"
#include "default.h"

#define SYNC_MODE
#define HAVE_FALLOCATE
#define EXT4_WRITEBLOCK 4194304
#define EXT4_PREALLOCATE 1073741824


typedef struct {
    int fd;

    int sync_mode;
    
    size_t prior_size;		/**< original size of file */
    size_t preallocated;	/**< preallocated bytes */
    
    size_t wr_block;		/**< minimal block of data to write */
    size_t pa_block;		/**< preallocation setp */
} fastwriter_default_t;


int fastwriter_default_open(fastwriter_t *fw, const char *name, fastwriter_flags_t flags) {
    int err;
    char fs[16];

    int open_flags = (O_CREAT|O_WRONLY|O_NOATIME|O_LARGEFILE);
    int open_mode = (S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);

    
    fastwriter_default_t *ctx;

    err = get_file_fs(name, sizeof(fs) - 1, fs);
    if (err) return err;
    
    ctx = (fastwriter_default_t*)malloc(sizeof(fastwriter_default_t));
    if (!ctx) return ENOMEM;

    memset(ctx, 0, sizeof(fastwriter_default_t));

    fw->ctx = ctx;

#ifdef SYNC_MODE
    open_flags |= O_DIRECT;
    ctx->sync_mode = 1;
#endif /* SYNC_MODE */

    ctx->prior_size = 0;

    if (!strcmp(fs, "raw")) {
	ctx->wr_block = EXT4_WRITEBLOCK;
	ctx->pa_block = 0;
	ctx->prior_size = (size_t)-1;
    } else if (!strcmp(fs, "ext4")) {
	ctx->wr_block = EXT4_WRITEBLOCK;
	ctx->pa_block = EXT4_PREALLOCATE;
    } else if (!strcmp(fs, "btrfs")) {
	ctx->wr_block = EXT4_WRITEBLOCK;
	ctx->pa_block = EXT4_PREALLOCATE;
    } else if (!strcmp(fs, "xfs")) {
	ctx->wr_block = EXT4_WRITEBLOCK;
	ctx->pa_block = EXT4_PREALLOCATE;
    } else {
	ctx->wr_block = EXT4_WRITEBLOCK;
	ctx->pa_block = 0;
    }
    
    if (flags&FASTWRITER_FLAGS_OVERWRITE)
	open_flags |= O_TRUNC;

    ctx->fd = open(name, open_flags, open_mode);
    if (ctx->fd < 0) return errno;

    if (((open_flags&FASTWRITER_FLAGS_OVERWRITE)==0)&&(strcmp(fs, "raw"))) {
	ctx->prior_size = lseek(ctx->fd, 0, SEEK_END);
# ifdef SYNC_MODE	
	if (ctx->prior_size%FASTWRITER_SYNCIO_ALIGN) {
	    close(ctx->fd);
	    
	    ctx->fd = open(name, open_flags&~O_DIRECT, open_mode);
	    if (ctx->fd < 0) return errno;
	    
	    ctx->prior_size = lseek(ctx->fd, 0, SEEK_END);
	    
	    ctx->sync_mode = 0;
	}
# endif /* SYNC_MODE */
    }

    ctx->preallocated = 0;

    return 0;
}


void fastwriter_default_close(fastwriter_t *fw) {
    if (fw->ctx) {
	fastwriter_default_t *ctx = (fastwriter_default_t*)fw->ctx;

	if (ctx->fd >= 0) {
#if defined(SYNC_MODE)||!defined(HAVE_LINUX_FALLOC_H)
	    if (ctx->prior_size != (size_t)-1) {
		ftruncate(ctx->fd, ctx->prior_size + fw->written);
	    }
#endif
	    close(ctx->fd);
	}
	
	free(ctx);
	fw->ctx = NULL;
    }
}


int fastwriter_default_write(fastwriter_t *fw, fastwriter_write_flags_t flags, size_t size, void *data, size_t *written) {
    size_t sum = 0;
    size_t delta = 0;
    ssize_t res;
    fastwriter_default_t *ctx = (fastwriter_default_t*)fw->ctx;

    if ((flags&FASTWRITER_WRITE_FLAG_FORCE)==0) {
	if (size < ctx->wr_block) {
	    *written = 0;
	    return 0;
	}
    
        size -= size % ctx->wr_block;
    }
    

    if ((ctx->pa_block)&&((fw->written + size) > ctx->preallocated)) {
#ifdef HAVE_LINUX_FALLOC_H
    	if (fallocate(ctx->fd, FALLOC_FL_KEEP_SIZE, ctx->preallocated, ctx->pa_block)) {
#else /* HAVE_LINUX_FALLOC_H */
    	if (posix_fallocate(ctx->fd, ctx->preallocated, ctx->pa_block)) {
#endif /* HAVE_LINUX_FALLOC_H */
	    ctx->pa_block = 0;
	} else {
	    ctx->preallocated += ctx->pa_block;
	}
    }

#ifdef SYNC_MODE
	// we expect this to happen only at last iteration (buffer is multiply of the required align)
    if ((ctx->sync_mode)&&(size%FASTWRITER_SYNCIO_ALIGN)) {
	delta = FASTWRITER_SYNCIO_ALIGN - size%FASTWRITER_SYNCIO_ALIGN;
    }
#endif /* SYNC_MODE */
    
    do {
	res = write(ctx->fd, data + sum, size + delta - sum);
//	printf("%i %i %p %zu %i\n", res, ctx->fd, data, size, delta);
	if (res < 0) {
	    *written = sum;
	    return errno;
	}
	
	sum += res;
    } while (sum < size);

#ifdef SYNC_MODE    
    posix_fadvise(ctx->fd, fw->written, size, POSIX_FADV_DONTNEED);
#endif /* SYNC_MODE */
    
    *written = size;
    return 0;
}