ec3/src/cluster-cache.c

#include "cluster-cache.h"

#include <blue/io/file.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

struct cluster_cache_entry {
	size_t e_cluster_id;
	size_t e_storage_offset;
	size_t e_length;
	void *e_data;
	b_btree_node e_node;
};

B_BTREE_DEFINE_SIMPLE_GET(
	struct cluster_cache_entry,
	size_t,
	e_node,
	e_cluster_id,
	get_cluster)
B_BTREE_DEFINE_SIMPLE_INSERT(
	struct cluster_cache_entry,
	e_node,
	e_cluster_id,
	put_cluster)

enum ec3_status cluster_cache_init(
	struct cluster_cache *out,
	bool use_disk,
	size_t cluster_size)
{
	memset(out, 0x0, sizeof *out);

	if (use_disk) {
		enum b_status status = b_file_open_temp(
			B_FILE_READ_WRITE | B_FILE_BINARY,
			&out->c_storage);

		if (!B_OK(status)) {
			return ec3_status_from_b_status(
				status,
				EC3_ERR_IO_FAILURE);
		}
	}

	out->c_memcache_count = 4;
	out->c_cluster_size = cluster_size;
	return EC3_SUCCESS;
}

void cluster_cache_finish(struct cluster_cache *cache)
{
	if (cache->c_storage) {
		b_file_release(cache->c_storage);
	}
}

static enum ec3_status evict_one_cache_data(
	struct cluster_cache *cache,
	void **buf)
{
	struct cluster_cache_entry *entry = NULL;
	b_btree_iterator it;

	b_btree_foreach(&it, &cache->c_entries)
	{
		struct cluster_cache_entry *cur
			= b_unbox(struct cluster_cache_entry, it.node, e_node);

		if (cur->e_data) {
			entry = cur;
			break;
		}
	}

	if (!entry) {
		return EC3_ERR_BAD_STATE;
	}

	if (cache->c_storage) {
		size_t nr_written;

		b_status status = b_file_write(
			cache->c_storage,
			entry->e_storage_offset,
			entry->e_length,
			entry->e_data,
			&nr_written);

		if (!B_OK(status)) {
			return ec3_status_from_b_status(
				status,
				EC3_ERR_IO_FAILURE);
		}

		if (nr_written != entry->e_length) {
			return EC3_ERR_IO_FAILURE;
		}
	}

	*buf = entry->e_data;
	entry->e_data = NULL;

	return EC3_SUCCESS;
}

enum ec3_status cluster_cache_get(
	struct cluster_cache *cache,
	size_t cluster_id,
	void *out,
	size_t *cluster_size)
{
	struct cluster_cache_entry *entry
		= get_cluster(&cache->c_entries, cluster_id);

	if (!entry) {
		return EC3_ERR_NO_ENTRY;
	}

	if (entry->e_data) {
		/* this cluster's data is still cached in memory. */
		memcpy(out, entry->e_data, cache->c_cluster_size);
		*cluster_size = entry->e_length;
		return EC3_SUCCESS;
	}

	if (!cache->c_storage) {
		return EC3_ERR_NO_ENTRY;
	}

	/* this cluster's data is stored on-disk. */

	enum ec3_status status = EC3_SUCCESS;
	void *buf;
	if (cache->c_memcache_count >= cache->c_memcache_max) {
		status = evict_one_cache_data(cache, &buf);
	} else {
		buf = malloc(cache->c_cluster_size);
		if (buf) {
			cache->c_memcache_count++;
		}

		status = buf ? EC3_SUCCESS : EC3_ERR_NO_MEMORY;
	}

	if (status != EC3_SUCCESS) {
		return status;
	}

	size_t nr_read = 0;
	enum b_status status2 = b_file_read(
		cache->c_storage,
		entry->e_storage_offset,
		entry->e_length,
		buf,
		&nr_read);

	if (!B_OK(status2)) {
		return ec3_status_from_b_status(status2, EC3_ERR_IO_FAILURE);
	}

	if (nr_read != entry->e_length) {
		return EC3_ERR_IO_FAILURE;
	}

	memcpy(out, buf, nr_read);
	*cluster_size = nr_read;
	return EC3_SUCCESS;
}

enum ec3_status cluster_cache_put(
	struct cluster_cache *cache,
	size_t cluster_id,
	const void *data,
	size_t len)
{
	if (len > cache->c_cluster_size) {
		return EC3_ERR_INVALID_VALUE;
	}

	struct cluster_cache_entry *entry
		= get_cluster(&cache->c_entries, cluster_id);

	if (!entry) {
		entry = malloc(sizeof *entry);

		if (!entry) {
			return EC3_ERR_NO_MEMORY;
		}

		memset(entry, 0x0, sizeof *entry);
		entry->e_cluster_id = cluster_id;
		entry->e_length = len;

		if (cache->c_storage) {
			b_file_size(cache->c_storage, &entry->e_storage_offset);
		}

		put_cluster(&cache->c_entries, entry);
	}

	void *buf = NULL;
	enum ec3_status status = EC3_SUCCESS;

	if (entry && entry->e_data) {
		buf = entry->e_data;
	} else if (cache->c_memcache_count >= cache->c_memcache_max) {
		status = evict_one_cache_data(cache, &buf);
	} else {
		buf = malloc(cache->c_cluster_size);
		if (buf) {
			cache->c_memcache_count++;
		}

		status = buf ? EC3_SUCCESS : EC3_ERR_NO_MEMORY;
	}

	if (status != EC3_SUCCESS) {
		return status;
	}

	memcpy(buf, data, len);

	entry->e_data = buf;

	return EC3_SUCCESS;
}

enum ec3_status cluster_cache_get_highest_cluster_id(
	struct cluster_cache *cache,
	size_t *highest_cluster)
{
	b_btree_node *node = b_btree_last(&cache->c_entries);

	if (!node) {
		return EC3_ERR_NO_ENTRY;
	}

	struct cluster_cache_entry *entry
		= b_unbox(struct cluster_cache_entry, node, e_node);

	*highest_cluster = entry->e_cluster_id;
	return EC3_SUCCESS;
}