A major overhaul of the whole thing.

It has come to my attention that previously, the project was built on top
of some *very* incorrect assumptions. This patch mostly addresses that.
And while it does NOT leave the project in otherwise working state,
it does weed out most, if not all, of effects of previously incorrect
assumptions, so it can be built the right way from here on forth.

Signed-off-by: Gediminas Jakutis <gediminas@varciai.lt>

10 files changed, 344 insertions, 143 deletions

diff --git a/src/cache.c b/src/cache.c
index de182ec..4fe0c46 100644
--- a/src/cache.c
+++ b/src/cache.c
@@ -4,32 +4,44 @@
 
 #include <errno.h>
 #include <stdlib.h>
+#include <string.h>
+#include <rin/definitions.h>
+#include <rin/diagnostic.h>
 #include "defs.h"
 
-static off_t cache_walker(struct stream * const restrict in, ssize_t idx);
-
-int cache_create(struct stream * const restrict in, const struct settings * const restrict s)
+int cache_create(struct stream * const in, const struct settings * const restrict s)
 {
-	int ret;
+	int ret = 0;
 	void *cache;
 
 	try(!in->cached, err, EINVAL, "cannot create cache: stream is uncached");
 	try(!(cache = calloc(in->n, s->stride)), err, ENOMEM, "out of memory");
 
+	/* yeah... */
 	in->cache = cache;
+	in->cache_a = cache;
+	in->cache_l = cache;
 err:
 	return ret;
 }
 
-int cache_populate(struct stream * const restrict in)
+int cache_populate(struct stream * const in)
 {
 	int ret = 0;
 	size_t i;
+	struct entry_l *tmp;
 
 	try(!in->cached, err, EINVAL, "cannot populate cache: stream is uncached");
 
 	for (i = 0; i < in->n && !ret; ++i) {
-		ret = in->get_element(in, i, in->cache + i);
+		errno = 0;
+		tmp = in->get_next_element_direct(in);
+		if (tmp) { /* non-cache reads CAN fail */
+			put(in, tmp);
+		} else {
+			ret = errno;
+			break;
+		}
 	}
 
 err:
@@ -38,14 +50,14 @@ err:
 
 int cache_flush(struct stream * const in)
 {
-	int ret;
-	size_t i;
+	int ret = 0;
+	struct entry_l *tmp;
 
 	try(!in->cached, err, EINVAL, "no cache to flush: stream is uncached");
 	try(in->out != 1, err, EINVAL, "cannot flush a non-output cache");
 
-	for (i = 0; i < in->n && !ret; ++i) {
-		ret = in->put_element(in, i, in->cache + i);
+	while((tmp = get(in))) {
+		in->place_next_element_direct(in, tmp);
 	}
 
 err:
@@ -54,89 +66,141 @@ err:
 
 int cache_destroy(struct stream * const in)
 {
-	int ret;
+	int ret = 0;
 
 	try(!in->cached, err, EINVAL, "no cache to destroy: stream uncached");
 	free(in->cache);
+	in->cache_l = NULL;
+	in->cache_a = NULL;
 	in->cache = NULL;
 err:
 	return ret;
 }
 
-int cache_transfer(struct stream * const from, struct stream * const to)
+int cache_transfer(struct stream * const src, struct stream * const dest)
 {
-	int ret;
+	int ret = 0;
 
-	try(!from->cached || !to->cached, err, EINVAL, "cannot transfer caches of uncached streams");
-	try(!from->cache, err, EINVAL, "no cache to transfer");
-	try(!to->cache, err, EINVAL, "cannot transfer cache: recipient cache already exists");
+	try(!src->cached || !dest->cached, err, EINVAL, "cannot transfer caches of uncached streams");
+	try(!src->cache, err, EINVAL, "no cache to transfer");
+	try(dest->cache, err, EINVAL, "cannot transfer cache: recipient cache already exists");
 
-	to->cache = from->cache;
-	from->cache = NULL;
+	dest->cache = src->cache;
+	dest->cache_a = (struct entry *) src->cache;
+	dest->cache_l = (struct entry_l *) src->cache;
+	src->cache = NULL;
+	src->cache_a = NULL;
+	src->cache_l = NULL;
 err:
 	return ret;
 }
 
-int cached_get_array(struct stream * const in, ssize_t idx, struct entry_l * const data)
+int cache_block_copy(struct stream const * const src, struct stream * const dest, const struct settings * const s)
 {
 	int ret = 0;
-	const struct entry * const cache = in->cache;
 
-	data->val = cache[idx].val;
+	try(!src->cached || !dest->cached, err, EINVAL, "cannot cache-copy between uncached streams");
+	try(!src->cache, err, EINVAL, "no cache to transfer");
+	try(!(src->n < s->to), err, EINVAL, "invalid copy size");
+	try(!dest->cache, err, EINVAL, "no cache to transfer to");
 
+	memcpy(dest->cache, src->cache_a + s->ss, (s->to - s->ss) * s->stride);
+
+err:
 	return ret;
 }
 
-int cached_get_list(struct stream * const restrict in, ssize_t idx, struct entry_l * const data)
+int cache_list_copy(struct stream const * const src, struct stream * const dest, const struct settings * const s)
 {
-	int ret = 0;
-	off_t offset;
+	int ret = ENOSYS;
+
+	(void) src;
+	(void) dest;
+	(void) s;
 
-	offset = cache_walker(in, idx);
-	*data = in->cache[offset];
+	rin_warn("stub!");
 
 	return ret;
 }
 
-int cached_put_array(struct stream * const in, ssize_t idx, const struct entry_l * const data)
+int cache_block_split(struct stream * const src, struct stream * const A, struct stream * const B)
 {
 	int ret = 0;
-	struct entry * const cache = in->cache;
 
-	cache[idx].val = data->val;
+	try(src->n < 2, err, EINVAL, "cannot split single element stream.");
+
+	/* copy everything, then change whatever's neccesary */
+	*A = *B = *src;
 
+	A->n = A->n / 2 + A->n % 2;
+	B->n = B->n / 2;
+	A->index = B->index = 0;
+	B->cache_a = B->cache_a + A->n;
+	B->cache_l = (struct entry_l *) B->cache_a;
+	B->cache = (char *) B->cache_a;
+
+err:
 	return ret;
 }
 
-int cached_put_list(struct stream * const restrict in, ssize_t idx, const struct entry_l * const data)
+int cache_list_split(struct stream * const src, struct stream * const A, struct stream * const B)
 {
-	int ret = 0;
-	off_t offset;
+	int ret = ENOSYS;
+
+	(void) src;
+	(void) A;
+	(void) B;
 
-	offset = cache_walker(in, idx);
-	in->cache[offset] = *data;
+	rin_warn("stub!");
 
 	return ret;
 }
 
-static off_t cache_walker(struct stream * const restrict in, ssize_t idx)
+struct entry_l *cached_get_array(struct stream * const in)
 {
-	size_t i;
-	off_t ret = in->prev_off;
+	struct entry_l *ret = NULL;
 
-	if (in->prev_idx < idx) { /* previous index smaller, walk forward */
-		for (i = labs(idx - in->prev_idx); i; --i) {
-			ret = in->cache[ret].next;
-		}
+	if (in->index < in->n) {
+		ret = (struct entry_l *)(in->cache_a + in->index);
+		++in->index;
+	}
 
-	} else if (in->prev_idx > idx) { /* previous index larger, walk backwards */
-		for (i = in->prev_idx - idx; i; --i) {
-			ret = in->cache[ret].prev;
-		}
-	} /* ...or we're standing on it already, if neither */
+	return ret;
+}
+
+struct entry_l *cached_get_list(struct stream * const in)
+{
+	struct entry_l *ret = NULL;
+
+	if (in->index < in->n) {
+		ret = in->cnode;
+		in->cnode = (struct entry_l *) ret->next;
+		++in->index;
+	}
+
+	return ret;
+}
+
+int cached_put_array(struct stream * const in, const struct entry_l * const data)
+{
+	int ret = EINVAL;
+
+	if (in->index < in->n) {
+		in->cache_a[in->index].val = data->val;
+		++in->index;
+	}
+
+	return ret;
+}
+
+int cached_put_list(struct stream * const restrict in, const struct entry_l * const node)
+{
+	int ret = ENOSYS;
+
+	(void) in;
+	(void) node;
 
-	in->prev_idx = idx;
-	in->prev_off = ret;
+	rin_warn("stub!");
 
 	return ret;
 }
diff --git a/src/cache.h b/src/cache.h
index 45215d4..9f333d3 100644
--- a/src/cache.h
+++ b/src/cache.h
@@ -8,16 +8,25 @@
 #include <stddef.h>
 #include "defs.h"
 
-int cache_create(struct stream * const restrict in, const struct settings * const restrict s);
-int cache_populate(struct stream * const restrict in);
+/* INIT|DESTROY */
+int cache_create(struct stream * const in, const struct settings * const restrict s);
+int cache_populate(struct stream * const in);
 int cache_flush(struct stream * const in);
 int cache_destroy(struct stream * const in);
-int cache_transfer(struct stream * const from, struct stream * const to);
 
-int cached_get_array(struct stream * const in, ssize_t idx, struct entry_l * const data);
-int cached_get_list(struct stream * const restrict in, ssize_t idx, struct entry_l * const data);
+/* BLOCKMANIP */
+int cache_transfer(struct stream * const src, struct stream * const dest);
+int cache_block_copy(struct stream const * const src, struct stream * const dest, const struct settings * const s);
+int cache_list_copy(struct stream const * const src, struct stream * const dest, const struct settings * const s);
+int cache_block_split(struct stream * const src, struct stream * const A, struct stream * const B);
+int cache_list_split(struct stream * const src, struct stream * const A, struct stream * const B);
 
-int cached_put_array(struct stream * const in, ssize_t idx, const struct entry_l * const data);
-int cached_put_list(struct stream * const restrict in, ssize_t idx, const struct entry_l * const data);
+/* GET */
+struct entry_l *cached_get_array(struct stream * const in);
+struct entry_l *cached_get_list(struct stream * const in);
+
+/* PUT */
+int cached_put_array(struct stream * const in, const struct entry_l * const data);
+int cached_put_list(struct stream * const restrict in, const struct entry_l * const node);
 
 #endif /* ALGOS_CACHE_H_INCLUDED */
diff --git a/src/datagen.c b/src/datagen.c
index 25002e2..7eab23f 100644
--- a/src/datagen.c
+++ b/src/datagen.c
@@ -6,37 +6,36 @@
 #include <stddef.h>
 #include <errno.h>
 #include <stdlib.h>
+#include <string.h>
 #include <sys/types.h>
 #include "datagen.h"
 #include "defs.h"
 
-int gen_get_array(struct stream * const in, ssize_t idx, struct entry_l * const data)
-{
-	int ret = 0;
-	(void) idx;
+static struct entry_l buf;
 
-	in->prev_idx = -1;
-	ret = read(in->fd, &data->val, sizeof(data->val));
+struct entry_l *gen_get_array(struct stream * const in)
+{
+	struct entry_l *ret;
+	int errn;
 
-	if (ret != sizeof(data->val)) {
-		ret = ret > 0 ? EAGAIN : errno;
+	if (read(in->fd, &buf.val, sizeof(buf.val)) == sizeof(buf.val)) {
+		ret = &buf;
+	} else {
+		errn = errno;
+		ret = NULL;
+		rin_err("error reading /dev/urandom: %s", strerror(errn));
 	}
 
 	return ret;
 }
 
-int gen_get_list(struct stream * const restrict in, ssize_t idx, struct entry_l * const data)
+struct entry_l *gen_get_list(struct stream * const in)
 {
-	int ret = 0;
+	struct entry_l *ret;
 
-	data->prev = idx > 0 ? idx - 1 : labs(idx) - 1;
-	data->next = idx + 1;
-
-	if (!idx) {
-		data->val = 0; /* header lmao */
-	} else {
-		ret = gen_get_array(in, idx, data);
-	}
+	try_s((ret = gen_get_array(in)), err);
+	ret->next = 0;
 
+err:
 	return ret;
 }
diff --git a/src/datagen.h b/src/datagen.h
index 0b8aa35..92a7878 100644
--- a/src/datagen.h
+++ b/src/datagen.h
@@ -7,7 +7,7 @@
 
 #include "defs.h"
 
-int gen_get_array(struct stream * const in, ssize_t idx, struct entry_l * const data);
-int gen_get_list(struct stream * const restrict in, ssize_t idx, struct entry_l * const data);
+struct entry_l *gen_get_array(struct stream * const in);
+struct entry_l *gen_get_list(struct stream * const in);
 
 #endif /* ALGOS_DATAGEN_H_INCLUDED */
diff --git a/src/defs.h b/src/defs.h
index f3eb48f..cb03c81 100644
--- a/src/defs.h
+++ b/src/defs.h
@@ -23,8 +23,12 @@
 	}} while (0);
 
 
-#define get(in, idx, data) (in->cached ? in->get_element_cache(in, idx, data) | in->get_element(in, idx, data))
-#define put(in, idx, data) (in->cached ? in->put_element_cache(in, idx, data) | in->put_element(in, idx, data))
+#define get(in) (in->cached ? in->get_next_element_cache(in) : in->get_next_element_direct(in))
+#define put(in, data) (in->cached ? in->place_next_element_cache(in, data) : in->place_next_element_direct(in, data))
+
+#define stream_blank .fd = -1, .settings = &settings, .get_next_element_direct = stub_getnext, \
+		.get_next_element_cache = stub_getnext, .place_next_element_direct = stub_put, \
+		.place_next_element_cache = stub_put, .split = stub_split, .flush = stub_flush
 
 /* for array implementation */
 struct entry {
@@ -34,8 +38,7 @@ struct entry {
 /* for linked list implementation */
 struct entry_l {
 	struct entry;
-	int32_t next; /* """pointer""" to the next element. */
-	int32_t prev; /* """pointer""" to the previous element. */
+	int64_t next; /* """pointer""" to the next element. */
 };
 
 enum opmode {
@@ -55,18 +58,25 @@ enum dataformat {
 };
 
 struct stream {
+	struct stream *parent;
 	size_t n;
-	ssize_t prev_idx;
-	off_t prev_off;
 	int fd;
 	int out;
 	int cached;
 	char *name;
-	struct entry_l *cache;
-	int (*get_element)(struct stream * const, ssize_t, struct entry_l * const);
-	int (*put_element)(struct stream * const restrict, ssize_t, struct entry_l const * const);
-	int (*get_element_cache)(struct stream * const, ssize_t, struct entry_l * const);
-	int (*put_element_cache)(struct stream * const restrict, ssize_t, struct entry_l const * const);
+	size_t stride;
+	size_t index;
+	struct entry_l *cnode;
+	char *cache;
+	struct settings *settings;
+	struct entry *cache_a;
+	struct entry_l *cache_l;
+	struct entry_l *(*get_next_element_direct)(struct stream * const);
+	struct entry_l *(*get_next_element_cache)(struct stream * const);
+	int (*place_next_element_direct)(struct stream * const, struct entry_l const * const);
+	int (*place_next_element_cache)(struct stream * const, struct entry_l const * const);
+	int (*split)(struct stream * const, struct stream * const, struct stream * const);
+	int (*flush)(struct stream * const);
 };
 
 struct settings {
diff --git a/src/io.c b/src/io.c
index 688ec35..d323573 100644
--- a/src/io.c
+++ b/src/io.c
@@ -17,8 +17,9 @@
 #include "datagen.h"
 #include "cache.h"
 
-static int stream_open_in(struct stream * const in, const struct settings * const s);
 static int stream_open_out(struct stream * const in, const struct settings * const s);
+static int stream_open_out_lite(struct stream * const in, const struct settings * const s);
+static int stream_open_in(struct stream * const in, const struct settings * const s);
 static int stream_open_special(struct stream * const in);
 
 int stream_open(struct stream * const in, const struct settings * const s)
@@ -27,9 +28,13 @@ int stream_open(struct stream * const in, const struct settings * const s)
 
 	try(!in || in->fd > 0 || !in->name || !s, err, EINVAL, "invalid argunent");
 
-	if (in->out == 1) {
+	if (in->out == 2) {
+		ret = stream_open_out_lite(in, s);
+	} else if (in->out == 1) {
+		try(!in->name, err, EINVAL, "no filename given");
 		ret = stream_open_out(in, s);
 	} else if (!in->out) {
+		try(!in->name, err, EINVAL, "no filename given");
 		ret = stream_open_in(in, s);
 	} else {
 		ret = stream_open_special(in);
@@ -101,7 +106,31 @@ static int stream_open_out(struct stream * const in, const struct settings * con
 		mode = st.st_mode;
 	}
 
-	in->fd = open(dname, O_TMPFILE | O_WRONLY, mode);
+	in->fd = open(dname, O_TMPFILE | O_RDWR, mode);
+	try(in->fd < 0, err, errno, "failed creating temporary output file: %s", strerror(ret));
+	try(ftruncate(in->fd, s->stride * in->n), err, errno, "failed setting output file size: %s", strerror(ret));
+
+err:
+	free(dname);
+	return ret;
+}
+
+static int stream_open_out_lite(struct stream * const in, const struct settings * const s)
+{
+	struct stat st;
+	char *dname = NULL;
+	char *tmp[2];
+	int ret = 0;
+
+	tmp[0] = strdup(in->name);
+	tmp[1] = dirname(tmp[0]);
+	dname = strdup(tmp[1]);
+	free(tmp[0]);
+
+	try(stat(dname, &st), err, errno, "stat failed: %s", strerror(ret));
+	try(!(st.st_mode & S_IFDIR), err, EINVAL, "invalid output path");
+
+	in->fd = open(dname, O_TMPFILE | O_RDWR, S_IRUSR | S_IWUSR);
 	try(in->fd < 0, err, errno, "failed creating temporary output file: %s", strerror(ret));
 	try(ftruncate(in->fd, s->stride * in->n), err, errno, "failed setting output file size: %s", strerror(ret));
 
@@ -120,10 +149,6 @@ static int stream_open_in(struct stream * const in, const struct settings * cons
 	in->n = st.st_size / s->stride;
 	in->fd = open(in->name, O_RDONLY | O_NOATIME);
 	try(in->fd < 0, err, errno, "failed opening input file: %s", strerror(ret));
-	if (in->cached) {
-		cache_create(in, s);
-		cache_populate(in);
-	}
 
 err:
 	return ret;
@@ -141,9 +166,3 @@ static int stream_open_special(struct stream * const in)
 err:
 	return ret;
 }
-
-int direct_get_array(struct stream * const restrict in, ssize_t idx, struct entry_l * const data);
-int direct_get_list(struct stream * const restrict in, ssize_t idx, struct entry_l * const data);
-
-int direct_put_array(struct stream * const restrict in, ssize_t idx, const struct entry_l * const data);
-int direct_put_list(struct stream * const restrict in, ssize_t idx, const struct entry_l * const data);
diff --git a/src/main.c b/src/main.c
index 1050741..352c531 100644
--- a/src/main.c
+++ b/src/main.c
@@ -17,14 +17,18 @@ static char randfile[] = "/dev/urandom";
 
 static struct settings settings = {0};
 
-static struct stream file_in = {.prev_idx = -1, .fd = -1};
-static struct stream file_out = {.prev_idx = -1, .fd = -1, .out = 1};
-
 static int parseargs(int argc, char **argv, struct settings * settings);
-int prime_io_functions(struct settings const * const s, struct stream * const in);
+int load_io_functions(struct settings const * const s, struct stream * const in);
 void printhelp(const char * const name);
-static int stub_get(struct stream * const restrict in, ssize_t idx, struct entry_l * const data);
-static int stub_put(struct stream * const restrict in, ssize_t idx, struct entry_l const * const data);
+struct entry_l *stub_getnext(struct stream * const restrict in);
+static int stub_put(struct stream * const restrict in, struct entry_l const * const data);
+static int stub_split(struct stream * const in, struct stream * const a, struct stream * const b);
+static int stub_flush(struct stream * const in);
+
+/* these need to go AFTER the stub declarations */
+static struct stream file_in = {stream_blank};
+static struct stream file_out = {stream_blank, .out = 1};
+static struct stream file_tmp = {stream_blank, .out = 2};
 
 int main(int argc, char **argv)
 {
@@ -37,21 +41,58 @@ int main(int argc, char **argv)
 		file_in.name = randfile;
 		file_in.out = -1;
 		file_in.n = settings.to;
+		file_in.cached = 1;
+		file_out.cached = 1;
 	} else {
 		file_in.name = settings.filein;
 	}
 
-	ret = prime_io_functions(&settings, &file_in);
-	ret = prime_io_functions(&settings, &file_out);
+	load_io_functions(&settings, &file_in);
 
 	file_out.name = settings.fileout ? settings.fileout : settings.filein;
 	try_s((ret = stream_open(&file_in, &settings)), out);
-	file_out.n = file_in.n;
+	file_out.n = file_in.n - settings.ss;
 	try_s((ret = stream_open(&file_out, &settings)), out);
 
-out:
+	load_io_functions(&settings, &file_out);
+
+	if (settings.access == cached) {
+		try_s((ret = cache_create(&file_in, &settings)), out);
+		try_s((ret = cache_populate(&file_in)), out);
+
+		switch (settings.opmode) {
+		case mode_fetch:
+			if (settings.format == array) {
+				try_s((ret = cache_block_copy(&file_in, &file_out, &settings)), out);
+			} else { /* settings.format == list */
+				try_s((ret = cache_list_copy(&file_in, &file_out, &settings)), out);
+			}
+			break;
+		case mode_generate:
+				try_s((ret = cache_transfer(&file_in, &file_out)), out);
+				break;
+		case mode_normal:
+				/* TODO */
+				;
+		}
+
+		try_s((ret = cache_flush(&file_out)), out);
+
+	} else { /* uncached */
+		/* TODO */
+	}
+
 	stream_close(&file_in);
 	stream_close(&file_out);
+
+	while (0) {
+out:
+		stream_close(&file_in);
+		file_out.out = 2; /* in case of error-exit, just close, don't link */
+		stream_close(&file_out);
+		stream_close(&file_tmp);
+	}
+
 early_out:
 	return ret;
 }
@@ -89,7 +130,7 @@ static int parseargs(int argc, char **argv, struct settings * settings)
 			}
 		} else if (!(strncmp(argv[i], "--num=", 6))) {
 			if (strlen(argv[i]) > 6) {
-				s.to = strtoul(argv[i] + 11, NULL, 10);
+				s.to = strtoul(argv[i] + 6, NULL, 10);
 			} else {
 				goto err;
 			}
@@ -139,45 +180,29 @@ err:
 	return ret;
 }
 
-int prime_io_functions(struct settings const * const s, struct stream * const in)
+int load_io_functions(struct settings const * const s, struct stream * const in)
 {
 	int ret = 0;
 
 	if (in->out == 1) {
 		if (s->format == array) {
-			in->get_element = stub_get;
-			in->put_element = stub_put;
-			in->get_element_cache = cached_get_array;
-			in->put_element_cache = cached_put_array;
+			/* TODO */
 		} else {
-			in->get_element = stub_get;
-			in->put_element = stub_put;
-			in->get_element_cache = cached_get_list;
-			in->put_element_cache = cached_put_list;
+			/* TODO */
 		}
 	} else if (!in->out) { /* reading streams do not support dumping */
 		if (s->format == array) {
-			in->get_element = stub_get;
-			in->put_element = stub_put;
-			in->get_element_cache = stub_get;
-			in->put_element_cache = stub_put;
+			/* TODO */
 		} else {
-			in->get_element = stub_get;
-			in->put_element = stub_put;
-			in->get_element_cache = stub_get;
-			in->put_element_cache = stub_put;
+			/* TODO */
 		}
 	} else { /* data generation streams do not support dumping nor any cache I/O beyond initial data generation */
 		if (s->format == array) {
-			in->get_element = gen_get_array;
-			in->put_element = stub_put;
-			in->get_element_cache = stub_get;
-			in->put_element_cache = stub_put;
+			in->get_next_element_direct = gen_get_array;
+			in->place_next_element_cache = cached_put_array;
 		} else {
-			in->get_element = gen_get_list;
-			in->put_element = stub_put;
-			in->get_element_cache = stub_get;
-			in->put_element_cache = stub_put;
+			in->get_next_element_direct = gen_get_list;
+			in->place_next_element_cache = cached_put_list;
 		}
 	}
 
@@ -188,7 +213,7 @@ void printhelp(const char * const name)
 {
 	printf(	"This is a mergesort program and such\n"
 		"\n"
-		"usage:\t%s [options]\n FILE"
+		"usage:\t%s [OPTION]... FILE"
 		"\n"
 		"Options:\n"
 		" --sort              sort mode (default)\n"
@@ -207,22 +232,49 @@ void printhelp(const char * const name)
 	return;
 }
 
-static int stub_get(struct stream * const restrict in, ssize_t idx, struct entry_l * const data)
+struct entry_l *stub_getnext(struct stream * const restrict in)
 {
+	struct entry_l *ret = NULL;
+
 	(void) in;
-	(void) idx;
-	(void) data;
 
 	rin_warn("stub!");
-	return ENOSYS;
+
+	return ret;
 }
 
-static int stub_put(struct stream * const restrict in, ssize_t idx, struct entry_l const * const data)
+static int stub_put(struct stream * const restrict in, struct entry_l const * const data)
 {
+	int ret = ENOSYS;
+
 	(void) in;
-	(void) idx;
 	(void) data;
 
 	rin_warn("stub!");
-	return ENOSYS;
+
+	return ret;
+}
+
+static int stub_split(struct stream * const in, struct stream * const a, struct stream * const b)
+{
+	int ret = ENOSYS;
+
+	(void) in;
+	(void) a;
+	(void) b;
+
+	rin_warn("stub!");
+
+	return ret;
+}
+
+static int stub_flush(struct stream * const in)
+{
+	int ret = ENOSYS;
+
+	(void) in;
+
+	rin_warn("stub!");
+
+	return ret;
 }
diff --git a/src/mergesort.c b/src/mergesort.c
new file mode 100644
index 0000000..2562d9d
--- /dev/null
+++ b/src/mergesort.c
@@ -0,0 +1,34 @@
+/* SPDX-License-Identifier: LGPL-2.1-only */
+
+/* Copyright (C) 2020 Gediminas Jakutis */
+
+#include <errno.h>
+#include <stdlib.h>
+#include "defs.h"
+#include "mergesort.h"
+
+int merge(struct stream * const dest, struct stream * const A, struct stream * const B)
+{
+	int ret;
+	struct entry_l *a;
+	struct entry_l *b;
+
+	try(A->parent != B->parent, err, EINVAL, "cannot merge blocks: uncommon parent!");
+
+	a = get(A);
+	b = get(B);
+
+	while (a || b) {
+		if (a && (!b || a->val <= b->val)) {
+			put(dest, a);
+			a = get(A);
+		} else {
+			put(dest, b);
+			b = get(B);
+		}
+	}
+
+err:
+	return ret;
+}
+
diff --git a/src/mergesort.h b/src/mergesort.h
new file mode 100644
index 0000000..8ac3ab5
--- /dev/null
+++ b/src/mergesort.h
@@ -0,0 +1,12 @@
+/* SPDX-License-Identifier: LGPL-2.1-only */
+
+/* Copyright (C) 2020 Gediminas Jakutis */
+
+#ifndef ALGOS_MERGESORT_H_INCLUDED
+#define ALGOS_MERGESORT_H_INCLUDED
+
+#include "defs.h"
+
+int merge(struct stream * const dest, struct stream * const A, struct stream * const B);
+
+#endif /* ALGOS_MERGESORT_H_INCLUDED */
diff --git a/src/meson.build b/src/meson.build
index f94cce4..b714823 100644
--- a/src/meson.build
+++ b/src/meson.build
@@ -3,6 +3,7 @@ source_files = [
 	'datagen.c',
 	'io.c',
 	'main.c',
+	'mergesort.c',
 ]
 
 header_files = [
@@ -10,6 +11,7 @@ header_files = [
 	'defs.h',
 	'datagen.h',
 	'io.h',
+	'mergesort.h',
 ]
 
 sources = files(source_files)