path: root/src/kvstorage_file.c

/* Copyright (c) 2010, Vsevolod Stakhov
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *       * Redistributions of source code must retain the above copyright
 *         notice, this list of conditions and the following disclaimer.
 *       * Redistributions in binary form must reproduce the above copyright
 *         notice, this list of conditions and the following disclaimer in the
 *         documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL AUTHOR BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */


#include "config.h"
#include "kvstorage.h"
#include "kvstorage_file.h"
#include "util.h"
#include "main.h"

struct file_op {
	struct rspamd_kv_element *elt;
	enum {
		FILE_OP_INSERT,
		FILE_OP_DELETE,
		FILE_OP_REPLACE
	} op;
	guint32 ref;
};

/* Main file structure */
struct rspamd_file_backend {
	backend_init init_func;                     /*< this callback is called on kv storage initialization */
	backend_insert insert_func;                 /*< this callback is called when element is inserted */
	backend_replace replace_func;               /*< this callback is called when element is replaced */
	backend_lookup lookup_func;                 /*< this callback is used for lookup of element */
	backend_delete delete_func;                 /*< this callback is called when an element is deleted */
	backend_sync sync_func;                     /*< this callback is called when backend need to be synced */
	backend_incref incref_func;                 /*< this callback is called when element must be ref'd */
	backend_destroy destroy_func;               /*< this callback is used for destroying all elements inside backend */
	gchar *filename;
	gchar *dirname;
	guint dirlen;
	guint sync_ops;
	guint levels;
	GQueue *ops_queue;
	GHashTable *ops_hash;
	gboolean do_fsync;
	gboolean do_ref;
	gboolean initialized;
};

static const gchar hexdigits[] =
{'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e',
	 'f'};

/* Generate file name for operation */
static gboolean
get_file_name (struct rspamd_file_backend *db,
	gchar *key,
	guint keylen,
	gchar *filebuf,
	guint buflen)
{
	gchar *p = filebuf, *end = filebuf + buflen,
	*k = key, t;
	guint i;

	/* First copy backend dirname to file buf */
	if (buflen <= db->dirlen) {
		return FALSE;
	}
	memcpy (p, db->dirname, db->dirlen);
	p += db->dirlen;
	*p++ = G_DIR_SEPARATOR;
	for (i = 0; i < MIN (keylen, db->levels); i++) {
		if (p == end) {
			/* Filebuf is not large enough */
			return FALSE;
		}
		t = *k;
		*p++ = hexdigits[(t & 0xf) ^ ((t & 0xf0) >> 4)];
		*p++ = G_DIR_SEPARATOR;
		k++;
	}
	/* Now we have directory, append base64 encoded filename */
	k = key;
	if (end - p < (gint)keylen * 2 + 1) {
		/* Filebuf is not large enough */
		return FALSE;
	}

	i = 0;
	while (k < key + keylen) {
		t = *k;
		*p++ = hexdigits[(t >> 4) & 0xf];
		*p++ = hexdigits[t & 0xf];
		k++;
	}
	*p = '\0';

	return TRUE;
}

/* Read reference from specified file */
static guint32
file_get_ref (gint fd)
{
	guint32 target;

	if (read (fd, &target, sizeof (guint32)) != sizeof (guint32)) {
		return 0;
	}

	return target;
}

/* Set reference to specified file */
static gboolean
file_set_ref (gint fd, guint32 ref)
{
	if (write (fd, &ref, sizeof (guint32)) != sizeof (guint32)) {
		return FALSE;
	}

	return TRUE;
}

/*
 * Open file, set posix_fadvise and all necessary flags
 */
static gint
file_open_fd (const gchar *path, gsize *len, gint flags)
{
	gint fd;
	struct stat st;

	if ((flags & O_CREAT) != 0) {
		/* Open file */
		if ((fd = open (path, flags, S_IRUSR | S_IWUSR | S_IRGRP)) != -1) {
			rspamd_fallocate (fd, 0, *len);
#ifdef HAVE_FADVISE
			posix_fadvise (fd, 0, *len, POSIX_FADV_SEQUENTIAL);
#endif
		}
	}
	else {
		/* Open file */
		if ((fd = open (path, flags)) == -1) {
			return -1;
		}

		if (fstat (fd, &st) == -1) {
			close (fd);
			return -1;
		}

#ifdef HAVE_FADVISE
		posix_fadvise (fd, 0, st.st_size, POSIX_FADV_SEQUENTIAL);
#endif
		*len = st.st_size;
	}

	return fd;
}

/* Process single file operation */
static gboolean
file_process_single_op (struct rspamd_file_backend *db,
	struct file_op *op,
	gint *pfd)
{
	gchar filebuf[PATH_MAX];
	gint fd;
	gsize len;
	struct iovec iov[2];
	guint32 ref;

	/* Get filename */
	if (!get_file_name (db, ELT_KEY (op->elt), op->elt->keylen, filebuf,
		sizeof (filebuf))) {
		return FALSE;
	}

	if (db->do_ref) {
		len = ELT_SIZE (op->elt) + sizeof (guint32);
	}
	else {
		len = ELT_SIZE (op->elt);
	}

	if (op->op == FILE_OP_DELETE) {
		if (db->do_ref) {
			if ((fd = file_open_fd (filebuf, &len, O_RDWR)) == -1) {
				*pfd = -1;
				return FALSE;
			}
			if ((ref = file_get_ref (fd)) <= 1) {
				/* Refcount is not enough, remove file */
				close (fd);
				*pfd = -1;
				return unlink (filebuf) != -1;
			}
			else {
				/* Decrease ref */
				lseek (fd, 0, SEEK_SET);
				if (!file_set_ref (fd, --ref)) {
					*pfd = fd;
					return FALSE;
				}
			}
		}
		else {
			*pfd = -1;
			return unlink (filebuf) != -1;
		}
	}
	else {
		if ((fd =
			file_open_fd (filebuf, &len, O_CREAT | O_WRONLY | O_TRUNC)) == -1) {
			*pfd = -1;
			return FALSE;
		}
		if (db->do_ref) {
			iov[0].iov_base = &op->ref;
			iov[0].iov_len = sizeof (guint32);
			iov[1].iov_base = op->elt;
			iov[1].iov_len = ELT_SIZE (op->elt);
			if (writev (fd, iov, G_N_ELEMENTS (iov)) == -1) {
				msg_info ("%d: %s", errno, strerror (errno));
				*pfd = fd;
				return FALSE;
			}
		}
		else {
			if (write (fd, op->elt, ELT_SIZE (op->elt)) == -1) {
				msg_info ("%d: %s", errno, strerror (errno));
				*pfd = fd;
				return FALSE;
			}
		}
	}

	*pfd = fd;
	return TRUE;
}

/* Sync vector of descriptors */
static void
file_sync_fds (gint *fds, gint len, gboolean do_fsync)
{
	gint i, fd;

	for (i = 0; i < len; i++) {
		fd = fds[i];
		if (fd != -1) {
			if (do_fsync) {
#ifdef HAVE_FDATASYNC
				fdatasync (fd);
#else
				fsync (fd);
#endif
			}
			close (fd);
		}
	}
}

/* Process operations queue */
static gboolean
file_process_queue (struct rspamd_kv_backend *backend)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	struct file_op *op;
	GList *cur;
	gint *fds, i = 0, len;

	len = g_queue_get_length (db->ops_queue);
	if (len == 0) {
		/* Nothing to process */
		return TRUE;
	}

	fds = g_slice_alloc (len * sizeof (gint));
	cur = db->ops_queue->head;
	while (cur) {
		op = cur->data;
		if (!file_process_single_op (db, op, &fds[i])) {
			file_sync_fds (fds, i, db->do_fsync);
			g_slice_free1 (len * sizeof (gint), fds);
			return FALSE;
		}
		i++;
		cur = g_list_next (cur);
	}

	file_sync_fds (fds, i, db->do_fsync);
	g_slice_free1 (len * sizeof (gint), fds);

	/* Clean the queue */
	g_hash_table_remove_all (db->ops_hash);
	cur = db->ops_queue->head;
	while (cur) {
		op = cur->data;
		if (op->op == FILE_OP_DELETE ||
			((op->elt->flags & KV_ELT_NEED_FREE) != 0 &&
			(op->elt->flags & KV_ELT_NEED_INSERT) == 0)) {
			/* Also clean memory */
			g_slice_free1 (ELT_SIZE (op->elt), op->elt);
		}
		else {
			/* Unset dirty flag */
			op->elt->flags &= ~KV_ELT_DIRTY;
		}
		g_slice_free1 (sizeof (struct file_op), op);
		cur = g_list_next (cur);
	}

	g_queue_clear (db->ops_queue);

	return TRUE;

}


/* Make 16 directories for each level */
static gboolean
rspamd_recursive_mkdir (guint levels)
{
	guint i;
	gchar nbuf[5];

	/* Create directories for backend */
	if (levels > 0) {
		/* Create 16 directories */
		for (i = 0; i < 16; i++) {
			rspamd_snprintf (nbuf, sizeof (nbuf), "./%c", hexdigits[i]);
			if (mkdir (nbuf, 0755) != 0 && errno != EEXIST) {
				msg_info ("cannot create directory %s: %s", nbuf,
					strerror (errno));
				return FALSE;
			}
			else if (levels > 1) {
				if (chdir (nbuf) == -1) {
					msg_err ("chdir to %s failed: %s", nbuf, strerror (errno));
					return FALSE;
				}
				if (!rspamd_recursive_mkdir (levels - 1)) {
					return FALSE;
				}
				if (chdir ("../") == -1) {
					msg_err ("chdir to ../ failed: %s", strerror (errno));
					return FALSE;
				}
			}
		}
	}
	return TRUE;

}

/* Backend callbacks */
static void
rspamd_file_init (struct rspamd_kv_backend *backend)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	gchar pathbuf[PATH_MAX];

	/* Save current directory */
	if (getcwd (pathbuf, sizeof (pathbuf) - 1) == NULL) {
		pathbuf[0] = '\0';
		msg_err ("getcwd failed: %s", strerror (errno));
		goto err;
	}

	/* Chdir to the working dir */
	if (chdir (db->dirname) == -1) {
		msg_err ("chdir failed: %s", strerror (errno));
		goto err;
	}

	/* Create directories for backend */
	if (!rspamd_recursive_mkdir (db->levels)) {
		goto err;
	}

	db->initialized = TRUE;

	if (chdir (pathbuf) == -1) {
		msg_err ("chdir to %s failed: %s", pathbuf, strerror (errno));
	}
	return;
err:
	if (pathbuf[0] != '\0') {
		if (chdir (pathbuf) == -1) {
			msg_err ("chdir to %s failed: %s", pathbuf, strerror (errno));
		}
	}
}

static gboolean
rspamd_file_insert (struct rspamd_kv_backend *backend,
	gpointer key,
	guint keylen,
	struct rspamd_kv_element *elt)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	struct file_op *op;
	struct rspamd_kv_element search_elt;

	search_elt.keylen = keylen;
	search_elt.p = key;

	if (!db->initialized) {
		return FALSE;
	}

	if ((op = g_hash_table_lookup (db->ops_hash, &search_elt)) != NULL) {
		/* We found another op with such key in this queue */
		if (op->op == FILE_OP_DELETE || (op->elt->flags & KV_ELT_NEED_FREE) !=
			0) {
			/* Also clean memory */
			g_hash_table_steal (db->ops_hash, &search_elt);
			g_slice_free1 (ELT_SIZE (op->elt), op->elt);
		}
		op->op = FILE_OP_INSERT;
		op->ref++;
		op->elt = elt;
		elt->flags |= KV_ELT_DIRTY;
		g_hash_table_insert (db->ops_hash, elt, op);
	}
	else {
		op = g_slice_alloc (sizeof (struct file_op));
		op->op = FILE_OP_INSERT;
		op->elt = elt;
		op->ref = 1;
		elt->flags |= KV_ELT_DIRTY;

		g_queue_push_head (db->ops_queue, op);
		g_hash_table_insert (db->ops_hash, elt, op);
	}

	if (db->sync_ops > 0 && g_queue_get_length (db->ops_queue) >=
		db->sync_ops) {
		return file_process_queue (backend);
	}

	return TRUE;
}

static gboolean
rspamd_file_replace (struct rspamd_kv_backend *backend,
	gpointer key,
	guint keylen,
	struct rspamd_kv_element *elt)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	struct file_op *op;
	struct rspamd_kv_element search_elt;

	search_elt.keylen = keylen;
	search_elt.p = key;

	if (!db->initialized) {
		return FALSE;
	}
	if ((op = g_hash_table_lookup (db->ops_hash, &search_elt)) != NULL) {
		/* We found another op with such key in this queue */
		if (op->op == FILE_OP_DELETE || (op->elt->flags & KV_ELT_NEED_FREE) !=
			0) {
			/* Also clean memory */
			g_hash_table_steal (db->ops_hash, &search_elt);
			g_slice_free1 (ELT_SIZE (op->elt), op->elt);
		}
		op->op = FILE_OP_REPLACE;
		op->elt = elt;
		elt->flags |= KV_ELT_DIRTY;
		g_hash_table_insert (db->ops_hash, elt, op);
	}
	else {
		op = g_slice_alloc (sizeof (struct file_op));
		op->op = FILE_OP_REPLACE;
		op->elt = elt;
		op->ref = 1;
		elt->flags |= KV_ELT_DIRTY;

		g_queue_push_head (db->ops_queue, op);
		g_hash_table_insert (db->ops_hash, elt, op);
	}

	if (db->sync_ops > 0 && g_queue_get_length (db->ops_queue) >=
		db->sync_ops) {
		return file_process_queue (backend);
	}

	return TRUE;
}

static struct rspamd_kv_element *
rspamd_file_lookup (struct rspamd_kv_backend *backend,
	gpointer key,
	guint keylen)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	struct file_op *op;
	struct rspamd_kv_element *elt = NULL;
	gchar filebuf[PATH_MAX];
	gint fd;
	struct rspamd_kv_element search_elt;
	gsize len;

	search_elt.keylen = keylen;
	search_elt.p = key;

	if (!db->initialized) {
		return NULL;
	}
	/* First search in ops queue */
	if ((op = g_hash_table_lookup (db->ops_hash, &search_elt)) != NULL) {
		if (op->op == FILE_OP_DELETE) {
			/* To delete, so assume it as not found */
			return NULL;
		}
		return op->elt;
	}

	/* Get filename */
	if (!get_file_name (db, key, keylen, filebuf, sizeof (filebuf))) {
		return NULL;
	}

	if ((fd = file_open_fd (filebuf, &len, O_RDONLY)) == -1) {
		return NULL;
	}

	/* Read element */
	if (db->do_ref) {
		lseek (fd, sizeof (guint32), SEEK_CUR);
		len -= sizeof (guint32);
	}
	elt = g_malloc (len);
	if (read (fd, elt, len) == -1) {
		g_free (elt);
		close (fd);
		return NULL;
	}

	close (fd);

	elt->flags &= ~(KV_ELT_DIRTY | KV_ELT_NEED_FREE);

	return elt;
}

static void
rspamd_file_delete (struct rspamd_kv_backend *backend,
	gpointer key,
	guint keylen)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	gchar filebuf[PATH_MAX];
	struct rspamd_kv_element search_elt;
	struct file_op *op;
	gsize len;
	gint fd;
	guint32 ref;

	if (!db->initialized) {
		return;
	}

	search_elt.keylen = keylen;
	search_elt.p = key;
	/* First search in ops queue */
	if ((op = g_hash_table_lookup (db->ops_hash, &search_elt)) != NULL) {
		op->op = FILE_OP_DELETE;
		if (op->ref > 0) {
			op->ref--;
		}
		return;
	}
	/* Get filename */
	if (!get_file_name (db, key, keylen, filebuf, sizeof (filebuf))) {
		return;
	}

	if (db->do_ref) {
		if ((fd = file_open_fd (filebuf, &len, O_RDWR)) == -1) {
			return;
		}
		if ((ref = file_get_ref (fd)) <= 1) {
			/* Refcount is not enough, remove file */
			close (fd);
			unlink (filebuf);
		}
		else {
			/* Decrease ref */
			lseek (fd, 0, SEEK_SET);
			file_set_ref (fd, --ref);
		}
		return;
	}

	unlink (filebuf);
}

static gboolean
rspamd_file_incref (struct rspamd_kv_backend *backend,
	gpointer key,
	guint keylen)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;
	gchar filebuf[PATH_MAX];
	struct rspamd_kv_element search_elt;
	struct file_op *op;
	gsize len;
	gint fd;
	guint32 ref;

	if (!db->initialized) {
		return FALSE;
	}
	if (!db->do_ref) {
		return TRUE;
	}

	search_elt.keylen = keylen;
	search_elt.p = key;
	/* First search in ops queue */
	if ((op = g_hash_table_lookup (db->ops_hash, &search_elt)) != NULL) {
		op->ref++;
		if (op->op == FILE_OP_DELETE) {
			op->op = FILE_OP_INSERT;
		}
		return TRUE;
	}

	/* Get filename */
	if (!get_file_name (db, key, keylen, filebuf, sizeof (filebuf))) {
		return FALSE;
	}

	if ((fd = file_open_fd (filebuf, &len, O_RDWR)) == -1) {
		return FALSE;
	}

	ref = file_get_ref (fd);

	/* Decrease ref */
	lseek (fd, 0, SEEK_SET);

	if (file_set_ref (fd, ++ref)) {
		close (fd);
		return TRUE;
	}
	else {
		close (fd);
		return FALSE;
	}
}

static void
rspamd_file_destroy (struct rspamd_kv_backend *backend)
{
	struct rspamd_file_backend *db = (struct rspamd_file_backend *)backend;

	if (db->initialized) {
		file_process_queue (backend);
		g_free (db->filename);
		g_free (db->dirname);
		g_queue_free (db->ops_queue);
		g_hash_table_unref (db->ops_hash);
		g_slice_free1 (sizeof (struct rspamd_file_backend), db);

		/* Sync again */
		sync ();
	}
}

/* Create new file backend */
struct rspamd_kv_backend *
rspamd_kv_file_new (const gchar *filename,
	guint sync_ops,
	guint levels,
	gboolean do_fsync,
	gboolean do_ref)
{
	struct rspamd_file_backend *new;
	struct stat st;
	gchar *dirname;

	if (filename == NULL) {
		return NULL;
	}

	dirname = g_path_get_dirname (filename);
	if (dirname == NULL || stat (dirname, &st) == -1 || !S_ISDIR (st.st_mode)) {
		/* Inaccessible path */
		if (dirname != NULL) {
			g_free (dirname);
		}
		msg_err ("invalid file: %s", filename);
		return NULL;
	}

	new = g_slice_alloc0 (sizeof (struct rspamd_file_backend));
	new->dirname = dirname;
	new->dirlen = strlen (dirname);
	new->filename = g_strdup (filename);
	new->sync_ops = sync_ops;
	new->levels = levels;
	new->do_fsync = do_fsync;
	new->do_ref = do_ref;
	new->ops_queue = g_queue_new ();
	new->ops_hash = g_hash_table_new (kv_elt_hash_func, kv_elt_compare_func);

	/* Init callbacks */
	new->init_func = rspamd_file_init;
	new->insert_func = rspamd_file_insert;
	new->lookup_func = rspamd_file_lookup;
	new->delete_func = rspamd_file_delete;
	new->replace_func = rspamd_file_replace;
	new->sync_func = file_process_queue;
	new->incref_func = rspamd_file_incref;
	new->destroy_func = rspamd_file_destroy;

	return (struct rspamd_kv_backend *)new;
}